Merge pull request #4332 from chu11/job_list_cleanup

job-list: misc cleanup

Author: mergify[bot]

src/modules/job-archive/job-archive.c

@@ -412,7 +412,7 @@ int job_info_lookup (struct job_archive_ctx *ctx, json_t *job)
     double t_run = 0.0;
 
     if (json_unpack (job, "{s:I s?:f}", "id", &id, "t_run", &t_run) < 0) {
-        flux_log (ctx->h, LOG_ERR, "%s: parse t_inactive", __FUNCTION__);
+        flux_log (ctx->h, LOG_ERR, "%s: parse t_run", __FUNCTION__);
         goto error;
     }
 

src/modules/job-list/idsync.c

@@ -17,11 +17,9 @@
 #include <flux/core.h>
 
 #include "src/common/libczmqcontainers/czmq_containers.h"
-#include "src/common/libutil/errno_safe.h"
 #include "src/common/libjob/job_hash.h"
 
 #include "idsync.h"
-#include "job_state.h"
 
 void idsync_data_destroy (void *data)
 {

src/modules/job-list/job_state.c

@@ -118,14 +118,14 @@ static void job_destroy (void *data)
 {
     struct job *job = data;
     if (job) {
-        json_decref (job->exception_context);
+        free (job->ranks);
+        free (job->nodelist);
         json_decref (job->annotations);
+        grudgeset_destroy (job->dependencies);
         json_decref (job->jobspec_job);
         json_decref (job->jobspec_cmd);
         json_decref (job->R);
-        grudgeset_destroy (job->dependencies);
-        free (job->ranks);
-        free (job->nodelist);
+        json_decref (job->exception_context);
         zlist_destroy (&job->next_states);
         free (job);
     }
@@ -145,25 +145,27 @@ static struct job *job_create (struct list_ctx *ctx, flux_jobid_t id)
         return NULL;
     job->ctx = ctx;
     job->id = id;
-    job->state = FLUX_JOB_STATE_NEW;
     job->userid = FLUX_USERID_UNKNOWN;
-    job->ntasks = -1;
-    job->nnodes = -1;
     job->urgency = -1;
-    job->expiration = -1.0;
-    job->wait_status = -1;
     /* pending jobs that are not yet assigned a priority shall be
      * listed after those who do, so we set the job priority to MIN */
     job->priority = FLUX_JOB_PRIORITY_MIN;
+    job->state = FLUX_JOB_STATE_NEW;
+    job->ntasks = -1;
+    job->nnodes = -1;
+    job->expiration = -1.0;
+    job->wait_status = -1;
     job->result = FLUX_JOB_RESULT_FAILED;
-    job->eventlog_seq = -1;
 
     if (!(job->next_states = zlist_new ())) {
         errno = ENOMEM;
         job_destroy (job);
         return NULL;
     }
 
+    job->states_mask = FLUX_JOB_STATE_NEW;
+    job->states_events_mask = FLUX_JOB_STATE_NEW;
+    job->eventlog_seq = -1;
     return job;
 }
 
@@ -824,14 +826,16 @@ static void process_next_state (struct list_ctx *ctx, struct job *job)
             if (st->state == FLUX_JOB_STATE_DEPEND) {
                 /* get initial jobspec */
                 if (!(f = state_depend_lookup (jsctx, job))) {
-                    flux_log_error (jsctx->h, "%s: state_depend_lookup", __FUNCTION__);
+                    flux_log_error (jsctx->h, "%s: state_depend_lookup",
+                                    __FUNCTION__);
                     return;
                 }
             }
             else { /* st->state == FLUX_JOB_STATE_RUN */
                 /* get R to get node count, etc. */
                 if (!(f = state_run_lookup (jsctx, job))) {
-                    flux_log_error (jsctx->h, "%s: state_run_lookup", __FUNCTION__);
+                    flux_log_error (jsctx->h, "%s: state_run_lookup",
+                                    __FUNCTION__);
                     return;
                 }
             }
@@ -1231,14 +1235,12 @@ static int submit_context_parse (flux_t *h,
 {
     int urgency;
     int userid;
-    int flags;
 
     if (!context
         || json_unpack (context,
-                        "{ s:i s:i s:i }",
+                        "{ s:i s:i }",
                         "urgency", &urgency,
-                        "userid", &userid,
-                        "flags", &flags) < 0) {
+                        "userid", &userid) < 0) {
         flux_log (h, LOG_ERR, "%s: submit context invalid: %ju",
                   __FUNCTION__, (uintmax_t)job->id);
         errno = EPROTO;

src/modules/job-list/job_state.h

@@ -59,6 +59,17 @@ struct job_state_ctx {
     flux_future_t *events;
 };
 
+/* timestamp of when we enter the state
+ *
+ * associated eventlog entries when restarting
+ *
+ * t_depend - "submit"
+ * t_priority - "priority" (not saved, can be entered multiple times)
+ * t_sched - "depend" (not saved, can be entered multiple times)
+ * t_run - "alloc"
+ * t_cleanup - "finish" or "exception" w/ severity == 0
+ * t_inactive - "clean"
+ */
 struct job {
     struct list_ctx *ctx;
 
@@ -67,7 +78,11 @@ struct job {
     int urgency;
     int64_t priority;
     double t_submit;
-    int flags;
+    // t_depend is identical to t_submit
+    // double t_depend;
+    double t_run;
+    double t_cleanup;
+    double t_inactive;
     flux_job_state_t state;
     const char *name;
     int ntasks;
@@ -78,19 +93,18 @@ struct job {
     int wait_status;
     bool success;
     bool exception_occurred;
-    json_t *exception_context;
     int exception_severity;
     const char *exception_type;
     const char *exception_note;
     flux_job_result_t result;
     json_t *annotations;
     struct grudgeset *dependencies;
-    int eventlog_seq;           /* last event seq read */
 
     /* cache of job information */
     json_t *jobspec_job;
     json_t *jobspec_cmd;
     json_t *R;
+    json_t *exception_context;
 
     /* Track which states we have seen and have completed transition
      * to.  We do not immediately update to the new state and place
@@ -109,22 +123,7 @@ struct job {
     unsigned int states_events_mask;
     void *list_handle;
 
-    /* timestamp of when we enter the state
-     *
-     * associated eventlog entries when restarting
-     *
-     * depend - "submit"
-     * priority - "priority" (not saved, can be entered multiple times)
-     * sched - "depend" (not saved, can be entered multiple times)
-     * run - "alloc"
-     * cleanup - "finish" or "exception" w/ severity == 0
-     * inactive - "clean"
-     */
-    // t_depend is identical to t_submit above, use that
-    // double t_depend;
-    double t_run;
-    double t_cleanup;
-    double t_inactive;
+    int eventlog_seq;           /* last event seq read */
 };
 
 struct job_state_ctx *job_state_create (struct list_ctx *ctx);

src/modules/job-list/list.c

@@ -20,6 +20,7 @@
 #include "src/common/libutil/errno_safe.h"
 #include "src/common/libczmqcontainers/czmq_containers.h"
 
+#include "job-list.h"
 #include "idsync.h"
 #include "list.h"
 #include "job_util.h"

src/modules/job-list/list.h

@@ -13,8 +13,6 @@
 
 #include <flux/core.h>
 
-#include "job-list.h"
-
 void list_cb (flux_t *h, flux_msg_handler_t *mh,
               const flux_msg_t *msg, void *arg);
 

t/Makefile.am

@@ -323,6 +323,7 @@ dist_check_SCRIPTS = \
 	job-exec/imp-fail.sh \
 	job-list/list-id.py \
 	job-list/list-rpc.py \
+	job-list/job-list-helper.sh \
 	job-list/jobspec-permissive.jsonschema \
 	ingest/bad-validate.py
 

t/job-list/job-list-helper.sh

@@ -0,0 +1,51 @@
+#!/bin/sh
+#
+
+# job-list test helper functions
+
+JOB_LIST_WAIT_ITERS=50
+
+# Return the expected jobids list in a given state:
+#   "all", "pending", "running", "inactive", "active",
+#   "completed", "canceled", "failed"
+#
+job_list_state_ids() {
+    for f in "$@"; do
+        cat ${f}.ids
+    done
+}
+
+# Return the expected count of jobs in a given state (See above for list)
+#
+job_list_state_count() {
+    job_list_state_ids "$@" | wc -l
+}
+
+# the job-list module has eventual consistency with the jobs stored in
+# the job-manager's queue.  To ensure no raciness in tests, we spin
+# until all of the pending jobs have reached SCHED state, running jobs
+# have reached RUN state, and inactive jobs have reached INACTIVE
+# state.
+#
+# job ids for jobs in these states are expected to be in pending.ids,
+# running.ids, and inactive.ids respectively.
+
+job_list_wait_states() {
+        pending=$(job_list_state_count pending)
+        running=$(job_list_state_count running)
+        inactive=$(job_list_state_count inactive)
+        local i=0
+        while ( [ "$(flux job list --states=sched | wc -l)" != "$pending" ] \
+                || [ "$(flux job list --states=run | wc -l)" != "$running" ] \
+                || [ "$(flux job list --states=inactive | wc -l)" != "$inactive" ]) \
+               && [ $i -lt ${JOB_LIST_WAIT_ITERS} ]
+        do
+                sleep 0.1
+                i=$((i + 1))
+        done
+        if [ "$i" -eq "${JOB_LIST_WAIT_ITERS}" ]
+        then
+            return 1
+        fi
+        return 0
+}