libsubprocess: use matchtag in client write req

Problem: RFC 42 now requires write requests to reference the
subprocess via the exec matchtag rather than the pid.

Send the matchtag in the write request instead of the pid and
alter the internal function prototype for subprocess_write() to
accept the exec future in lieu of rank, service, and matchtag.
Update users of that function.

Also update a unit test that uses the raw protocol.

Author: garlick

src/common/libsubprocess/client.c

@@ -44,6 +44,9 @@ struct rexec_ctx {
     json_t *cmd;
     int flags;
     struct rexec_response response;
+    uint32_t matchtag;
+    uint32_t rank;
+    char *service_name;
 };
 
 static void rexec_response_clear (struct rexec_response *resp)
@@ -62,12 +65,16 @@ static void rexec_ctx_destroy (struct rexec_ctx *ctx)
         int saved_errno = errno;
         rexec_response_clear (&ctx->response);
         json_decref (ctx->cmd);
+        free (ctx->service_name);
         free (ctx);
         errno = saved_errno;
     }
 }
 
-static struct rexec_ctx *rexec_ctx_create (flux_cmd_t *cmd, int flags)
+static struct rexec_ctx *rexec_ctx_create (flux_cmd_t *cmd,
+                                           const char *service_name,
+                                           uint32_t rank,
+                                           int flags)
 {
     struct rexec_ctx *ctx;
     int valid_flags = SUBPROCESS_REXEC_STDOUT
@@ -80,10 +87,12 @@ static struct rexec_ctx *rexec_ctx_create (flux_cmd_t *cmd, int flags)
     }
     if (!(ctx = calloc (1, sizeof (*ctx))))
         return NULL;
-    if (!(ctx->cmd = cmd_tojson (cmd)))
+    if (!(ctx->cmd = cmd_tojson (cmd))
+        || !(ctx->service_name = strdup (service_name)))
         goto error;
     ctx->flags = flags;
     ctx->response.pid = -1;
+    ctx->rank = rank;
     return ctx;
 error:
     rexec_ctx_destroy (ctx);
@@ -106,7 +115,7 @@ flux_future_t *subprocess_rexec (flux_t *h,
     }
     if (asprintf (&topic, "%s.exec", service_name) < 0)
         return NULL;
-    if (!(ctx = rexec_ctx_create (cmd, flags)))
+    if (!(ctx = rexec_ctx_create (cmd, service_name, rank, flags)))
         goto error;
     if (!(f = flux_rpc_pack (h,
                              topic,
@@ -122,6 +131,7 @@ flux_future_t *subprocess_rexec (flux_t *h,
         rexec_ctx_destroy (ctx);
         goto error;
     }
+    ctx->matchtag = flux_rpc_get_matchtag (f);
     free (topic);
     return f;
 error:
@@ -223,33 +233,32 @@ bool subprocess_rexec_is_output (flux_future_t *f,
     return false;
 }
 
-int subprocess_write (flux_t *h,
-                      const char *service_name,
-                      uint32_t rank,
-                      pid_t pid,
+int subprocess_write (flux_future_t *f_exec,
                       const char *stream,
                       const char *data,
                       int len,
                       bool eof)
 {
+    struct rexec_ctx *ctx = flux_future_aux_get (f_exec, "flux::rexec");
+    flux_t *h = flux_future_get_flux (f_exec);
     flux_future_t *f = NULL;
     json_t *io;
     char *topic;
     int rc = -1;
 
-    if (!h || pid < 0 || !stream || !service_name) {
+    if (!stream || !ctx) {
         errno = EINVAL;
         return -1;
     }
-    if (asprintf (&topic, "%s.write", service_name) < 0)
+    if (asprintf (&topic, "%s.write", ctx->service_name) < 0)
         return -1;
     if (!(io = ioencode (stream, "0", data, len, eof))
         || !(f = flux_rpc_pack (h,
                                 topic,
-                                rank,
+                                ctx->rank,
                                 FLUX_RPC_NORESPONSE,
                                 "{s:i s:O}",
-                                "pid", pid,
+                                "matchtag", ctx->matchtag,
                                 "io", io)))
         goto out;
     rc = 0;

src/common/libsubprocess/client.h

@@ -40,10 +40,7 @@ bool subprocess_rexec_is_output (flux_future_t *f,
                                  int *len,
                                  bool *eof);
 
-int subprocess_write (flux_t *h,
-                      const char *service_name,
-                      uint32_t rank,
-                      pid_t pid,
+int subprocess_write (flux_future_t *f,
                       const char *stream,
                       const char *data,
                       int len,

src/common/libsubprocess/remote.c

@@ -487,14 +487,7 @@ static int send_channel_data (flux_subprocess_t *p)
                     return -1;
                 }
             }
-            if (subprocess_write (p->h,
-                                  p->service_name,
-                                  p->rank,
-                                  p->pid,
-                                  c->name,
-                                  ptr,
-                                  len,
-                                  c->closed) < 0) {
+            if (subprocess_write (p->f, c->name, ptr, len, c->closed) < 0) {
                 llog_debug (p,
                             "error sending rexec.write request: %s",
                             strerror (errno));

src/common/libsubprocess/subprocess.c

@@ -729,14 +729,7 @@ int flux_subprocess_write (flux_subprocess_t *p,
             }
         }
         else { /* p->state == FLUX_SUBPROCESS_RUNNING */
-            if (subprocess_write (p->h,
-                                  p->service_name,
-                                  p->rank,
-                                  p->pid,
-                                  c->name,
-                                  buf,
-                                  len,
-                                  false) < 0) {
+            if (subprocess_write (p->f, c->name, buf, len, false) < 0) {
                 log_err ("error sending rexec.write request: %s",
                          strerror (errno));
                 return -1;
@@ -787,14 +780,7 @@ int flux_subprocess_close (flux_subprocess_t *p, const char *stream)
          */
         c->closed = true;
         if (p->state == FLUX_SUBPROCESS_RUNNING) {
-            if (subprocess_write (p->h,
-                                  p->service_name,
-                                  p->rank,
-                                  p->pid,
-                                  c->name,
-                                  NULL,
-                                  0,
-                                  true) < 0) {
+            if (subprocess_write (p->f, c->name, NULL, 0, true) < 0) {
                 log_err ("error sending rexec.write request: %s",
                          strerror (errno));
                 return -1;

src/common/libsubprocess/test/iostress.c

@@ -120,7 +120,7 @@ static void iostress_state_cb (flux_subprocess_t *p,
     }
 }
 
-static int rexec_write (flux_t *h, pid_t pid, const char *buf, int len)
+static int rexec_write (flux_t *h, uint32_t matchtag, const char *buf, int len)
 {
     flux_future_t *f;
     json_t *io;
@@ -133,7 +133,7 @@ static int rexec_write (flux_t *h, pid_t pid, const char *buf, int len)
                              0,
                              FLUX_RPC_NORESPONSE,
                              "{s:i s:O}",
-                             "pid", pid,
+                             "matchtag", matchtag,
                              "io", io))) {
         json_decref (io);
         return -1;
@@ -150,15 +150,18 @@ static void iostress_source_cb (flux_reactor_t *r,
 {
     struct iostress_ctx *ctx = arg;
     char *buf;
+    uint32_t matchtag;
 
     if (!(buf = malloc (ctx->linesize)))
         BAIL_OUT ("out of memory");
     memset (buf, 'F', ctx->linesize - 1);
     buf[ctx->linesize - 1] = '\n';
 
+    matchtag = flux_rpc_get_matchtag (ctx->p->f);
+
     for (int i = 0; i < ctx->batchlines; i++) {
         if (ctx->direct) {
-            if (rexec_write (ctx->h, ctx->pid, buf, ctx->linesize) < 0)
+            if (rexec_write (ctx->h, matchtag, buf, ctx->linesize) < 0)
                 BAIL_OUT ("rexec_write failed");
         }
         else {