job-exec: always use stdio for exec barrier

Problem: job-exec uses either stdio or a subprocess channel
to implement the exec barrier, but now that we require a
minimum flux-security version of 0.9.0, stdio may be used in
all cases.

The critical change in flux-security is that the IMP can now
use a helper program to obtain J in lieu of reading it from stdin.

Simplify the code in job-exec so that stdio is always used
for the exec barrier and the IMP always uses the helper.

This also makes it easier to integrate pending work to execute jobs
as systemd transient units, since subprocess channels will not work
in that environment.

Author: garlick

src/modules/job-exec/exec.c

@@ -88,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)
@@ -122,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++;
@@ -140,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;
@@ -155,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,
@@ -296,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));
@@ -355,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;
         }
@@ -375,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;

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;
 }

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 */

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

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 */
 

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