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.

Also update a unit test that uses the raw protocol.

src/common/libsubprocess/client.c

@@ -226,7 +226,7 @@ bool subprocess_rexec_is_output (flux_future_t *f,
 int subprocess_write (flux_t *h,
                       const char *service_name,
                       uint32_t rank,
-                      pid_t pid,
+                      uint32_t exec_matchtag,
                       const char *stream,
                       const char *data,
                       int len,
@@ -237,7 +237,7 @@ int subprocess_write (flux_t *h,
     char *topic;
     int rc = -1;
 
-    if (!h || pid < 0 || !stream || !service_name) {
+    if (!h || exec_matchtag == FLUX_MATCHTAG_NONE || !stream || !service_name) {
         errno = EINVAL;
         return -1;
     }
@@ -249,7 +249,7 @@ int subprocess_write (flux_t *h,
                                 rank,
                                 FLUX_RPC_NORESPONSE,
                                 "{s:i s:O}",
-                                "pid", pid,
+                                "matchtag", exec_matchtag,
                                 "io", io)))
         goto out;
     rc = 0;

src/common/libsubprocess/client.h

@@ -43,7 +43,7 @@ bool subprocess_rexec_is_output (flux_future_t *f,
 int subprocess_write (flux_t *h,
                       const char *service_name,
                       uint32_t rank,
-                      pid_t pid,
+                      uint32_t exec_matchtag,
                       const char *stream,
                       const char *data,
                       int len,

src/common/libsubprocess/remote.c

@@ -211,7 +211,7 @@ static int remote_write (struct subprocess_channel *c)
     if (subprocess_write (c->p->h,
                           c->p->service_name,
                           c->p->rank,
-                          c->p->pid,
+                          flux_rpc_get_matchtag (c->p->f),
                           c->name,
                           ptr,
                           lenp,
@@ -235,7 +235,7 @@ static int remote_close (struct subprocess_channel *c)
     if (subprocess_write (c->p->h,
                           c->p->service_name,
                           c->p->rank,
-                          c->p->pid,
+                          flux_rpc_get_matchtag (c->p->f),
                           c->name,
                           NULL,
                           0,

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 {