Merge pull request #5267 from garlick/job-exec-barrier

mergify[bot] · web-flow · commit 7a64c5bee980 · 2023-06-15T23:19:55.000Z
job-exec: always use stdio for exec barrier
diff --git a/src/modules/job-exec/exec.c b/src/modules/job-exec/exec.c
@@ -56,7 +56,11 @@ struct exec_ctx {
 
 static void exec_ctx_destroy (struct exec_ctx *tc)
 {
-    free (tc);
+    if (tc) {
+        int saved_errno = errno;
+        free (tc);
+        errno = saved_errno;
+    }
 }
 
 static struct exec_ctx *exec_ctx_create (json_t *jobspec)
@@ -84,27 +88,6 @@ static void start_cb (struct bulk_exec *exec, void *arg)
 {
     struct jobinfo *job = arg;
     jobinfo_started (job);
-    /*  This is going to be really slow. However, it should at least
-     *   work for now. We wait for all imp's to start, then send input
-     */
-    if (job->multiuser && !config_use_imp_helper ()) {
-        char *input = NULL;
-        json_t *o = json_pack ("{s:s}", "J", job->J);
-        if (!o || !(input = json_dumps (o, JSON_COMPACT))) {
-            jobinfo_fatal_error (job, errno, "Failed to get input to IMP");
-            goto out;
-        }
-        if (bulk_exec_write (exec, "stdin", input, strlen (input)) < 0)
-            jobinfo_fatal_error (job,
-                                 errno,
-                                 "Failed to write %ld bytes input to IMP",
-                                 strlen (input));
-        (void) bulk_exec_close (exec, "stdin");
-out:
-        json_decref (o);
-        free (input);
-    }
-
 }
 
 static void complete_cb (struct bulk_exec *exec, void *arg)
@@ -118,13 +101,11 @@ static void complete_cb (struct bulk_exec *exec, void *arg)
 static int exec_barrier_enter (struct bulk_exec *exec)
 {
     struct exec_ctx *ctx = bulk_exec_aux_get (exec, "ctx");
-    const char *stream = config_use_imp_helper () ?
-                         "stdin" :
-                         "FLUX_EXEC_PROTOCOL_FD";
+
     if (!ctx)
         return -1;
     if (++ctx->barrier_enter_count == bulk_exec_total (exec)) {
-        if (bulk_exec_write (exec, stream, "exit=0\n", 7) < 0)
+        if (bulk_exec_write (exec, "stdin", "exit=0\n", 7) < 0)
             return -1;
         ctx->barrier_enter_count = 0;
         ctx->barrier_completion_count++;
@@ -136,7 +117,7 @@ static int exec_barrier_enter (struct bulk_exec *exec)
          *   case where a shell exits while a barrier is already in progress
          *   is handled in exit_cb().
          */
-        if (bulk_exec_write (exec, stream, "exit=1\n", 7) < 0)
+        if (bulk_exec_write (exec, "stdin", "exit=1\n", 7) < 0)
             return -1;
     }
     return 0;
@@ -151,8 +132,7 @@ static void output_cb (struct bulk_exec *exec, flux_subprocess_t *p,
     struct jobinfo *job = arg;
     const char *cmd = flux_cmd_arg (flux_subprocess_get_cmd (p), 0);
 
-    if (streq (stream, "FLUX_EXEC_PROTOCOL_FD")
-        || (config_use_imp_helper () && streq (stream, "stdout"))) {
+    if (streq (stream, "stdout")) {
         if (streq (data, "enter\n")
             && exec_barrier_enter (exec) < 0) {
             jobinfo_fatal_error (job,
@@ -292,10 +272,7 @@ static void exit_cb (struct bulk_exec *exec,
      */
     if (ctx->barrier_completion_count == 0
         || ctx->barrier_enter_count > 0) {
-        const char *stream = config_use_imp_helper () ?
-                             "stdin" :
-                             "FLUX_EXEC_PROTOCOL_FD";
-        if (bulk_exec_write (exec, stream, "exit=1\n", 7) < 0)
+        if (bulk_exec_write (exec, "stdin", "exit=1\n", 7) < 0)
             jobinfo_fatal_error (job, 0,
                                  "failed to terminate barrier: %s",
                                  strerror (errno));
@@ -338,6 +315,7 @@ static int exec_init (struct jobinfo *job)
     }
     if (bulk_exec_aux_set (exec, "ctx", ctx,
                           (flux_free_f) exec_ctx_destroy) < 0) {
+        exec_ctx_destroy (ctx);
         flux_log_error (job->h, "exec_init: bulk_exec_aux_set");
         goto err;
     }
@@ -350,12 +328,11 @@ static int exec_init (struct jobinfo *job)
         goto err;
     }
     if (job->multiuser) {
-            if (config_use_imp_helper ()
-                && flux_cmd_setenvf (cmd,
-                                     1,
-                                     "FLUX_IMP_EXEC_HELPER",
-                                     "flux imp-exec-helper %ju",
-                                     (uintmax_t) job->id) < 0) {
+        if (flux_cmd_setenvf (cmd,
+                              1,
+                              "FLUX_IMP_EXEC_HELPER",
+                              "flux imp-exec-helper %ju",
+                              (uintmax_t) job->id) < 0) {
             flux_log_error (job->h, "exec_init: flux_cmd_setenvf");
             goto err;
         }
@@ -370,19 +347,6 @@ static int exec_init (struct jobinfo *job)
         flux_log_error (job->h, "exec_init: flux_cmd_argv_append");
         goto err;
     }
-
-    /*  If more than one shell is involved in this job, set up a channel
-     *   for exec system based barrier:
-     */
-    if (idset_count (ranks) > 1 && !config_use_imp_helper ()) {
-        if (flux_cmd_add_channel (cmd, "FLUX_EXEC_PROTOCOL_FD") < 0
-            || flux_cmd_setopt (cmd,
-                                "FLUX_EXEC_PROTOCOL_FD_LINE_BUFFER",
-                                "true") < 0) {
-            flux_log_error (job->h, "exec_init: flux_cmd_add_channel");
-            goto err;
-        }
-    }
     if (bulk_exec_push_cmd (exec, ranks, cmd, 0) < 0) {
         flux_log_error (job->h, "exec_init: bulk_exec_push_cmd");
         goto err;
diff --git a/src/modules/job-exec/exec_config.c b/src/modules/job-exec/exec_config.c
@@ -27,7 +27,6 @@
 static const char *default_cwd = "/tmp";
 static const char *default_job_shell = NULL;
 static const char *flux_imp_path = NULL;
-static bool use_imp_helper = false;
 
 static const char *jobspec_get_job_shell (json_t *jobspec)
 {
@@ -74,11 +73,6 @@ const char *config_get_imp_path (void)
     return flux_imp_path;
 }
 
-bool config_use_imp_helper (void)
-{
-    return use_imp_helper;
-}
-
 /*  Initialize common configurations for use by job-exec exec modules.
  */
 int config_init (flux_t *h, int argc, char **argv)
@@ -114,33 +108,22 @@ int config_init (flux_t *h, int argc, char **argv)
         return -1;
     }
 
-#if HAVE_FLUX_SECURITY
-    /* Use IMP helper by default for flux-security >= 0.9.0
-     */
-    if (FLUX_SECURITY_VERSION_MAJOR >= 0
-        && FLUX_SECURITY_VERSION_MINOR >= 9)
-        use_imp_helper = true;
-#endif /* HAVE_FLUX_SECURITY */
-
     if (argv && argc) {
         /* Finally, override values on cmdline */
         for (int i = 0; i < argc; i++) {
             if (strstarts (argv[i], "job-shell="))
                 default_job_shell = argv[i]+10;
             else if (strstarts (argv[i], "imp="))
                 flux_imp_path = argv[i]+4;
-            else if (streq (argv[i], "no-imp-helper"))
-                use_imp_helper = false;
         }
     }
 
     flux_log (h, LOG_DEBUG, "using default shell path %s", default_job_shell);
     if (flux_imp_path) {
         flux_log (h,
                   LOG_DEBUG,
-                  "using imp path %s (%s helper)",
-                  flux_imp_path,
-                  use_imp_helper ? "with" : "without");
+                  "using imp path %s (with helper)",
+                  flux_imp_path);
     }
     return 0;
 }
diff --git a/src/modules/job-exec/exec_config.h b/src/modules/job-exec/exec_config.h
@@ -21,8 +21,6 @@ const char *config_get_cwd (struct jobinfo *job);
 
 const char *config_get_imp_path (void);
 
-bool config_use_imp_helper (void);
-
 int config_init (flux_t *h, int argc, char **argv);
 
 #endif /* !HAVE_JOB_EXEC_CONFIG_EXEC_H */
diff --git a/src/modules/job-exec/job-exec.c b/src/modules/job-exec/job-exec.c
@@ -147,7 +147,6 @@ void jobinfo_decref (struct jobinfo *job)
         job->ctx = NULL;
         flux_msg_decref (job->req);
         job->req = NULL;
-        free (job->J);
         resource_set_destroy (job->R);
         json_decref (job->jobspec);
         free (job->rootref);
@@ -892,13 +891,6 @@ static void jobinfo_start_continue (flux_future_t *f, void *arg)
         jobinfo_fatal_error (job, errno, "initializing critical ranks");
         goto done;
     }
-    if (job->multiuser && !config_use_imp_helper ()) {
-        const char *J = jobinfo_kvs_lookup_get (f, "J");
-        if (!J || !(job->J = strdup (J))) {
-            jobinfo_fatal_error (job, errno, "reading J: %s", error.text);
-            goto done;
-        }
-    }
     if (jobinfo_load_implementation (job) < 0) {
         jobinfo_fatal_error (job, errno, "failed to initialize implementation");
         goto done;
@@ -1057,12 +1049,6 @@ static flux_future_t *jobinfo_start_init (struct jobinfo *job)
     if (!(f_kvs = flux_jobid_kvs_lookup (h, job->id, 0, "R"))
         || flux_future_push (f, "R", f_kvs) < 0)
         goto err;
-    if (job->multiuser
-        && !config_use_imp_helper ()
-        && (!(f_kvs = flux_jobid_kvs_lookup (h, job->id, 0, "J"))
-        || flux_future_push (f, "J", f_kvs) < 0)) {
-        goto err;
-    }
     if (job->reattach)
         f_kvs = ns_get_rootref (h, job, 0);
     else
diff --git a/src/modules/job-exec/job-exec.h b/src/modules/job-exec/job-exec.h
@@ -68,7 +68,6 @@ struct jobinfo {
 
     struct resource_set * R;         /* Fetched and parsed resource set R */
     json_t *              jobspec;   /* Fetched jobspec */
-    char *                J;         /* Signed jobspec */
 
     struct idset *        critical_ranks;  /* critical shell ranks */
 
diff --git a/src/modules/job-exec/sdexec.c b/src/modules/job-exec/sdexec.c
@@ -210,6 +210,14 @@ static struct sdexec *sdexec_create (flux_t *h,
             flux_log_error (job->h, "flux_cmd_argv_append");
             goto cleanup;
         }
+        if (flux_cmd_setenvf (se->cmd,
+                              1,
+                              "FLUX_IMP_EXEC_HELPER",
+                              "flux imp-exec-helper %ju",
+                              (uintmax_t) job->id) < 0) {
+            flux_log_error (job->h, "flux_cmd_setenvf");
+            goto cleanup;
+        }
     }
 
     if (flux_cmd_argv_append (se->cmd, job_shell) < 0
@@ -357,26 +365,7 @@ static void state_cb (sdprocess_t *sdp, sdprocess_state_t state, void *arg)
     if (!se->job->reattach && !se->job->running)
         jobinfo_started (se->job);
 
-    if (state == SDPROCESS_ACTIVE) {
-        /*  Don't try to write J to stdin_fd of -1
-         *  This probably indicates we've reattached to this job
-         */
-        if (se->job->multiuser && se->stdin_fds[0] >= 0) {
-            char *input = NULL;
-            json_t *o = json_pack ("{s:s}", "J", se->job->J);
-            if (!o || !(input = json_dumps (o, JSON_COMPACT))) {
-                jobinfo_fatal_error (se->job, errno, "Failed to get input to IMP");
-                return;
-            }
-            if (write (se->stdin_fds[0], input, strlen (input)) < 0) {
-                jobinfo_fatal_error (se->job, errno, "write");
-                return;
-            }
-            close (se->stdin_fds[0]);
-            se->stdin_fds[0] = -1;
-        }
-    }
-    else if (state == SDPROCESS_EXITED
+    if (state == SDPROCESS_EXITED
              && !se->jobinfo_tasks_complete_called) {
 
         /* Since we are calling jobinfo_tasks_complete(), the
