libsubprocess: add flux_subprocess_channel_incref/decref

Problem: It would be useful to take a reference on an output stream
of a subprocess so that stream doesn't generate EOF until all references
have been dropped. This can be used when multiple sources are being
spliced onto a subprocess output to ensure that the subprocess doesn't
register completion until the other outputs have also been fully read.

Add flux_subprocess_channel_incref/decref which wrap the underlying
flux_buffer_read_watcher_incref/decref for the channel in question.

Author: grondo

src/common/libsubprocess/subprocess.c

@@ -1232,6 +1232,27 @@ int flux_set_default_subprocess_log (flux_t *h,
     return 0;
 }
 
+void flux_subprocess_channel_incref (flux_subprocess_t *p, const char *name)
+{
+    struct subprocess_channel *c;
+    if (!p || !p->local)
+        return;
+    if (!(c = zhash_lookup (p->channels, name)))
+        return;
+    flux_buffer_read_watcher_incref (c->buffer_read_w);
+}
+
+void flux_subprocess_channel_decref (flux_subprocess_t *p, const char *name)
+{
+    struct subprocess_channel *c;
+    if (!p || !p->local)
+        return;
+    if (!(c = zhash_lookup (p->channels, name)))
+        return;
+    flux_buffer_read_watcher_decref (c->buffer_read_w);
+}
+
+
 /*
  * vi: ts=4 sw=4 expandtab
  */

src/common/libsubprocess/subprocess.h

@@ -493,6 +493,13 @@ int flux_subprocess_aux_set (flux_subprocess_t *p,
  */
 void *flux_subprocess_aux_get (flux_subprocess_t *p, const char *name);
 
+/*
+ *  Take/drop a reference on a subprocess output channel `name` (e.g. "stdout"
+ *   or "stderr"). EOF will not be produced from this channel the reference
+ *   count drops to zero.
+ */
+void flux_subprocess_channel_incref (flux_subprocess_t *p, const char *name);
+void flux_subprocess_channel_decref (flux_subprocess_t *p, const char *name);
 
 #ifdef __cplusplus
 }