Merge pull request #6197 from garlick/free_final_flag

mergify[bot] · web-flow · commit 59934202adcb · 2024-08-07T03:59:13.000Z
job-manager: add final flag to sched.free
diff --git a/src/modules/job-manager/alloc.c b/src/modules/job-manager/alloc.c
@@ -98,16 +98,20 @@ static void interface_teardown (struct alloc *alloc, char *s, int errnum)
 
 /* Send sched.free request.
  */
-int free_request (struct alloc *alloc, flux_jobid_t id, json_t *R)
+int free_request (struct alloc *alloc,
+                  flux_jobid_t id,
+                  json_t *R,
+                  bool final)
 {
     flux_msg_t *msg;
 
     if (!(msg = flux_request_encode ("sched.free", NULL)))
         return -1;
     if (flux_msg_pack (msg,
-                       "{s:I s:O}",
+                       "{s:I s:O s:b}",
                        "id", id,
-                       "R", R) < 0)
+                       "R", R,
+                       "final", final) < 0)
         goto error;
     if (flux_send (alloc->ctx->h, msg, 0) < 0)
         goto error;
@@ -180,7 +184,7 @@ static void alloc_response_cb (flux_t *h,
         (void)json_object_del (R, "scheduling");
 
         if (!job) {
-            (void)free_request (alloc, id, R);
+            (void)free_request (alloc, id, R, true);
             break;
         }
         if (job_priority_queue_delete (alloc->sent, job) < 0)
@@ -486,10 +490,13 @@ static void check_cb (flux_reactor_t *r,
                                    NULL);
 }
 
-int alloc_send_free_request (struct alloc *alloc, json_t *R, flux_jobid_t id)
+int alloc_send_free_request (struct alloc *alloc,
+                             json_t *R,
+                             flux_jobid_t id,
+                             bool final)
 {
     if (alloc->scheduler_is_online) {
-        if (free_request (alloc, id, R) < 0)
+        if (free_request (alloc, id, R, final) < 0)
             return -1;
     }
     return 0;
diff --git a/src/modules/job-manager/alloc.h b/src/modules/job-manager/alloc.h
@@ -50,7 +50,10 @@ int alloc_pending_count (struct alloc *alloc);
 
 /* Release resources back to the scheduler.
  */
-int alloc_send_free_request (struct alloc *alloc, json_t *R, flux_jobid_t id);
+int alloc_send_free_request (struct alloc *alloc,
+                             json_t *R,
+                             flux_jobid_t id,
+                             bool final);
 
 /* List pending jobs
  */
diff --git a/src/modules/job-manager/housekeeping.c b/src/modules/job-manager/housekeeping.c
@@ -227,16 +227,19 @@ static void allocation_release (struct allocation *a)
     struct idset *ranks = NULL;
     struct rlist *rl = NULL;
     json_t *R = NULL;
+    bool final = false;
 
     if ((ranks = get_housekept_ranks (a)) && idset_count (ranks) == 0) {
         idset_destroy (ranks);
         return; // nothing to do
     }
+    if (idset_empty (a->pending))
+        final = true;
 
     if (!ranks
         || !(rl = rlist_copy_ranks (a->rl, ranks))
         || !(R = rlist_to_R (rl))
-        || alloc_send_free_request (ctx->alloc, R, a->id) < 0
+        || alloc_send_free_request (ctx->alloc, R, a->id, final) < 0
         || rlist_remove_ranks (a->rl, ranks) < 0) {
         char *s = idset_encode (ranks, IDSET_FLAG_RANGE);
         flux_log (ctx->h,
@@ -457,7 +460,7 @@ int housekeeping_start (struct housekeeping *hk,
     }
     return 0;
 skip:
-    return alloc_send_free_request (hk->ctx->alloc, R, id);
+    return alloc_send_free_request (hk->ctx->alloc, R, id, true);
 }
 
 static int housekeeping_hello_respond_one (struct housekeeping *hk,
diff --git a/src/modules/sched-simple/sched.c b/src/modules/sched-simple/sched.c
@@ -113,7 +113,8 @@ jobreq_create (const flux_msg_t *msg)
     if (job == NULL)
         return NULL;
 
-    if (flux_msg_unpack (msg, "{s:I s:i s:i s:f s:o}",
+    if (flux_msg_unpack (msg,
+                         "{s:I s:i s:i s:f s:o}",
                          "id", &job->id,
                          "priority", &job->priority,
                          "userid", &job->uid,
@@ -133,8 +134,8 @@ jobreq_create (const flux_msg_t *msg)
     if (json_unpack (jobspec,
                      "{s:{s?{s?O}}}",
                      "attributes",
-                     "system",
-                     "constraints", &job->constraints) < 0) {
+                       "system",
+                         "constraints", &job->constraints) < 0) {
         job->errnum = errno;
         goto err;
     }
@@ -253,15 +254,14 @@ static int try_alloc (flux_t *h, struct simple_sched *ss)
                 flux_log_error (h, "try_alloc: rlist_free");
             rlist_destroy (alloc);
             alloc = NULL;
-        } else if (errno == ENOSPC)
+        }
+        else if (errno == ENOSPC)
             return rc;
         else if (errno == EOVERFLOW)
             note = "unsatisfiable request";
         else if (fail_alloc)
             note = "DEBUG_FAIL_ALLOC";
-        if (schedutil_alloc_respond_deny (ss->util_ctx,
-                                          job->msg,
-                                          note) < 0)
+        if (schedutil_alloc_respond_deny (ss->util_ctx, job->msg, note) < 0)
             flux_log_error (h, "schedutil_alloc_respond_deny");
         goto out;
     }
@@ -272,9 +272,9 @@ static int try_alloc (flux_t *h, struct simple_sched *ss)
                                               R,
                                               "{ s:{s:s s:n s:n} }",
                                               "sched",
-                                              "resource_summary", s,
-                                              "reason_pending",
-                                              "jobs_ahead") < 0)
+                                                "resource_summary", s,
+                                                "reason_pending",
+                                                "jobs_ahead") < 0)
         flux_log_error (h, "schedutil_alloc_respond_success_pack");
 
     flux_log (h, LOG_DEBUG, "alloc: %s: %s", idf58 (job->id), s);
@@ -310,8 +310,10 @@ static void annotate_reason_pending (struct simple_sched *ss)
     }
 }
 
-static void prep_cb (flux_reactor_t *r, flux_watcher_t *w,
-                     int revents, void *arg)
+static void prep_cb (flux_reactor_t *r,
+                     flux_watcher_t *w,
+                     int revents,
+                     void *arg)
 {
     struct simple_sched *ss = arg;
     /* if there is at least one job to schedule, start check and idle */
@@ -322,8 +324,10 @@ static void prep_cb (flux_reactor_t *r, flux_watcher_t *w,
     }
 }
 
-static void check_cb (flux_reactor_t *r, flux_watcher_t *w,
-                      int revents, void *arg)
+static void check_cb (flux_reactor_t *r,
+                      flux_watcher_t *w,
+                      int revents,
+                      void *arg)
 {
     struct simple_sched *ss = arg;
     flux_watcher_stop (ss->idle);
@@ -339,7 +343,11 @@ static void check_cb (flux_reactor_t *r, flux_watcher_t *w,
     }
 }
 
-static int try_free (flux_t *h, struct simple_sched *ss, json_t *R)
+static int try_free (flux_t *h,
+                     struct simple_sched *ss,
+                     flux_jobid_t id,
+                     json_t *R,
+                     bool final)
 {
     int rc = -1;
     char *r = NULL;
@@ -354,8 +362,14 @@ static int try_free (flux_t *h, struct simple_sched *ss, json_t *R)
     r = rlist_dumps (alloc);
     if ((rc = rlist_free (ss->rlist, alloc)) < 0)
         flux_log_error (h, "free: %s", r);
-    else
-        flux_log (h, LOG_DEBUG, "free: %s", r);
+    else {
+        flux_log (h,
+                  LOG_DEBUG,
+                  "free: %s %s%s",
+                  r,
+                  idf58 (id),
+                  final ? " (final)" : "");
+    }
     free (r);
     rlist_destroy (alloc);
     return rc;
@@ -365,13 +379,20 @@ void free_cb (flux_t *h, const flux_msg_t *msg, const char *R_str, void *arg)
 {
     struct simple_sched *ss = arg;
     json_t *R;
+    flux_jobid_t id;
+    int final = 0;
 
-    if (flux_request_unpack (msg, NULL, "{s:o}", "R", &R) < 0) {
+    if (flux_request_unpack (msg,
+                             NULL,
+                             "{s:I s:o s?b}",
+                             "id", &id,
+                             "R", &R,
+                             "final", &final) < 0) {
         flux_log (h, LOG_ERR, "free: error unpacking sched.free request");
         return;
     }
 
-    if (try_free (h, ss, R) < 0) {
+    if (try_free (h, ss, id, R, final) < 0) {
         flux_log_error (h, "free: could not free R");
         return;
     }
@@ -393,7 +414,8 @@ static void alloc_cb (flux_t *h, const flux_msg_t *msg, void *arg)
 
     if (ss->alloc_limit
         && zlistx_size (ss->queue) >= ss->alloc_limit) {
-        flux_log (h, LOG_ERR,
+        flux_log (h,
+                  LOG_ERR,
                   "alloc received above max concurrency: %d",
                   ss->alloc_limit);
         errno = EINVAL;
@@ -411,14 +433,18 @@ static void alloc_cb (flux_t *h, const flux_msg_t *msg, void *arg)
         jobreq_destroy (job);
         return;
     }
-    flux_log (h, LOG_DEBUG, "req: %s: spec={%d,%d,%d} duration=%.1f",
-                            idf58 (job->id), job->jj.nnodes,
-                            job->jj.nslots, job->jj.slot_size,
-                            job->jj.duration);
+
+    flux_log (h,
+              LOG_DEBUG,
+              "req: %s: spec={%d,%d,%d} duration=%.1f",
+              idf58 (job->id),
+              job->jj.nnodes,
+              job->jj.nslots,
+              job->jj.slot_size,
+              job->jj.duration);
+
     search_dir = job->priority > FLUX_JOB_URGENCY_DEFAULT;
-    job->handle = zlistx_insert (ss->queue,
-                                 job,
-                                 search_dir);
+    job->handle = zlistx_insert (ss->queue, job, search_dir);
     flux_watcher_start (ss->prep);
     return;
 err:
@@ -430,9 +456,7 @@ static void alloc_cb (flux_t *h, const flux_msg_t *msg, void *arg)
  * If a matching job found in queue, respond to the alloc request
  * and "dequeue" it.
  */
-static void cancel_cb (flux_t *h,
-                       const flux_msg_t *msg,
-                       void *arg)
+static void cancel_cb (flux_t *h, const flux_msg_t *msg, void *arg)
 {
     struct simple_sched *ss = arg;
     flux_jobid_t id;
@@ -457,9 +481,7 @@ static void cancel_cb (flux_t *h,
  * matching job found in queue, update the priority and reorder queue
  * as necessary.
  */
-static void prioritize_cb (flux_t *h,
-                           const flux_msg_t *msg,
-                           void *arg)
+static void prioritize_cb (flux_t *h, const flux_msg_t *msg, void *arg)
 {
     static int min_sort_size = 4;
     struct simple_sched *ss = arg;
@@ -520,7 +542,8 @@ static int hello_cb (flux_t *h,
     uint32_t userid;
     double t_submit;
 
-    if (flux_msg_unpack (msg, "{s:I s:i s:i s:f}",
+    if (flux_msg_unpack (msg,
+                         "{s:I s:i s:i s:f}",
                          "id", &id,
                          "priority", &priority,
                          "userid", &userid,
@@ -529,7 +552,8 @@ static int hello_cb (flux_t *h,
         return -1;
     }
 
-    flux_log (h, LOG_DEBUG,
+    flux_log (h,
+              LOG_DEBUG,
               "hello: id=%s priority=%u userid=%u t_submit=%0.1f",
               idf58 (id),
               priority,
@@ -594,7 +618,9 @@ static void status_cb (flux_t *h, flux_msg_handler_t *mh,
     }
     rlist_destroy (rl);
 
-    if (flux_respond_pack (h, msg, "{s:o s:o s:o}",
+    if (flux_respond_pack (h,
+                           msg,
+                           "{s:o s:o s:o}",
                            "all", all,
                            "allocated", alloc,
                            "down", down) < 0)
@@ -622,14 +648,16 @@ static void feasibility_cb (flux_t *h,
     const char *errmsg = NULL;
     flux_error_t error;
 
-    if (flux_request_unpack (msg, NULL, "{s:o}",
+    if (flux_request_unpack (msg,
+                             NULL,
+                             "{s:o}",
                             "jobspec", &jobspec) < 0)
         goto err;
     if (json_unpack (jobspec,
                      "{s:{s?{s?o}}}",
                      "attributes",
-                     "system",
-                     "constraints", &constraints) < 0)
+                       "system",
+                         "constraints", &constraints) < 0)
         goto err;
 
     if (jj_get_counts_json (jobspec, &jj) < 0) {
@@ -701,9 +729,7 @@ static void expiration_cb (flux_t *h,
         errno = EINVAL;
         goto err;
     }
-    if (flux_module_debug_test (ss->h,
-                                DEBUG_EXPIRATION_UPDATE_DENY,
-                                false)) {
+    if (flux_module_debug_test (ss->h, DEBUG_EXPIRATION_UPDATE_DENY, false)) {
         errmsg = "Rejecting expiration update for testing";
         goto err;
     }
@@ -722,7 +748,8 @@ static int ss_resource_update (struct simple_sched *ss, flux_future_t *f)
     double expiration = -1.;
     const char *s;
 
-    int rc = flux_rpc_get_unpack (f, "{s?s s?s s?F}",
+    int rc = flux_rpc_get_unpack (f,
+                                  "{s?s s?s s?F}",
                                   "up", &up,
                                   "down", &down,
                                   "expiration", &expiration);
@@ -760,7 +787,8 @@ static void acquire_continuation (flux_future_t *f, void *arg)
 {
     struct simple_sched *ss = arg;
     if (flux_future_get (f, NULL) < 0) {
-        flux_log (ss->h, LOG_ERR,
+        flux_log (ss->h,
+                  LOG_ERR,
                   "exiting due to resource update failure: %s",
                   future_strerror (f, errno));
         flux_reactor_stop (flux_get_reactor (ss->h));
@@ -780,7 +808,8 @@ static int ss_acquire_resources (flux_t *h, struct simple_sched *ss)
     json_t *R;
     json_error_t e;
 
-    if (!(f = flux_rpc (h, "resource.acquire",
+    if (!(f = flux_rpc (h,
+                        "resource.acquire",
                         NULL,
                         FLUX_NODEID_ANY,
                         FLUX_RPC_STREAMING))) {
@@ -789,7 +818,9 @@ static int ss_acquire_resources (flux_t *h, struct simple_sched *ss)
     }
     ss->acquire_f = f;
     if (flux_rpc_get_unpack (f, "{s:o}", "resources", &R) < 0) {
-        flux_log (h, LOG_ERR, "resource.acquire failed: %s",
+        flux_log (h,
+                  LOG_ERR,
+                  "resource.acquire failed: %s",
                   future_strerror (f, errno));
         goto out;
     }
@@ -846,8 +877,12 @@ static int simple_sched_init (flux_t *h, struct simple_sched *ss)
         goto out;
     }
     s = rlist_dumps (ss->rlist);
-    flux_log (h, LOG_DEBUG, "ready: %d of %d cores: %s",
-                            ss->rlist->avail, ss->rlist->total, s);
+    flux_log (h,
+              LOG_DEBUG,
+              "ready: %d of %d cores: %s",
+              ss->rlist->avail,
+              ss->rlist->total,
+              s);
     free (s);
     rc = 0;
 out:
diff --git a/t/t2226-housekeeping.t b/t/t2226-housekeeping.t
