libsubprocess: create write_buffer on demand

Problem: Profiling shows that creating the write buffer for
remote subprocesses eats up a healthy amount of cycles.  However,
the buffer is not needed for many circumstances.  It is only needed
when there is an attemp to write data to the subprocess before
the subprocess is running.

Create the write buffer only when it is needed.

src/common/libsubprocess/remote.c

@@ -226,25 +226,12 @@ static int remote_channel_setup (flux_subprocess_t *p,
     struct subprocess_channel *c = NULL;
     char *e = NULL;
     int save_errno;
-    int buffer_size;
 
     if (!(c = channel_create (p, output_cb, name, channel_flags))) {
         llog_debug (p, "channel_create: %s", strerror (errno));
         goto error;
     }
 
-    if ((buffer_size = cmd_option_bufsize (p, name)) < 0) {
-        llog_debug (p, "cmd_option_bufsize: %s", strerror (errno));
-        goto error;
-    }
-
-    if (channel_flags & CHANNEL_WRITE) {
-        if (!(c->write_buffer = fbuf_create (buffer_size))) {
-            llog_debug (p, "fbuf_create: %s", strerror (errno));
-            goto error;
-        }
-    }
-
     if (channel_flags & CHANNEL_READ) {
         int wflag;
 
@@ -257,6 +244,11 @@ static int remote_channel_setup (flux_subprocess_t *p,
             c->line_buffered = true;
 
         if (!(p->flags & FLUX_SUBPROCESS_FLAGS_LOCAL_UNBUF)) {
+            int buffer_size;
+            if ((buffer_size = cmd_option_bufsize (p, name)) < 0) {
+                llog_debug (p, "cmd_option_bufsize: %s", strerror (errno));
+                goto error;
+            }
             if (!(c->read_buffer = fbuf_create (buffer_size))) {
                 llog_debug (p, "fbuf_create: %s", strerror (errno));
                 goto error;
@@ -525,6 +517,9 @@ static int send_channel_data (flux_subprocess_t *p)
                 set_failed (p, "internal close error");
                 return -1;
             }
+            /* Don't need this anymore, reclaim the space */
+            fbuf_destroy (c->write_buffer);
+            c->write_buffer = NULL;
         }
         if (c->closed) {
             if (subprocess_write (p->h,

src/common/libsubprocess/subprocess.c

@@ -708,6 +708,17 @@ int flux_subprocess_write (flux_subprocess_t *p,
             return -1;
         }
         if (p->state == FLUX_SUBPROCESS_INIT) {
+            if (!c->write_buffer) {
+                int buffer_size;
+                if ((buffer_size = cmd_option_bufsize (p, stream)) < 0) {
+                    log_err ("cmd_option_bufsize: %s", strerror (errno));
+                    return -1;
+                }
+                if (!(c->write_buffer = fbuf_create (buffer_size))) {
+                    log_err ("fbuf_create");
+                    return -1;
+                }
+            }
             if (fbuf_space (c->write_buffer) < len) {
                 errno = ENOSPC;
                 return -1;