Merge pull request #5282 from grondo/pty-output-ref

shell: ensure captured pty data does not come after stdout EOF in output eventlog

Author: mergify[bot]

src/cmd/flux-job.c

@@ -1933,6 +1933,9 @@ void attach_stdin_cb (flux_reactor_t *r, flux_watcher_t *w,
  */
 void attach_output_start (struct attach_ctx *ctx)
 {
+    if (ctx->output_f)
+        return;
+
     if (!(ctx->output_f = flux_job_event_watch (ctx->h,
                                                 ctx->id,
                                                 "guest.output",

src/common/libflux/ev_buffer_read.c

@@ -87,7 +87,7 @@ static void buffer_read_cb (struct ev_loop *loop, ev_io *iow, int revents)
             return;
 
         if (!ret) {
-            ebr->eof_read = true;
+            ev_buffer_read_decref (ebr);
             (void)flux_buffer_readonly (ebr->fb);
             ev_io_stop (ebr->loop, iow);
         }
@@ -138,6 +138,7 @@ int ev_buffer_read_init (struct ev_buffer_read *ebr,
     ebr->start = false;
     ebr->eof_read = false;
     ebr->eof_sent = false;
+    ebr->refcnt = 1;
 
     if (!(ebr->fb = flux_buffer_create (size)))
         goto cleanup;
@@ -195,6 +196,17 @@ void ev_buffer_read_stop (struct ev_loop *loop, struct ev_buffer_read *ebr)
     }
 }
 
+void ev_buffer_read_incref (struct ev_buffer_read *ebr)
+{
+    ebr->refcnt++;
+}
+
+void ev_buffer_read_decref (struct ev_buffer_read *ebr)
+{
+    if (--ebr->refcnt == 0)
+        ebr->eof_read = true;
+}
+
 /*
  * vi:tabstop=4 shiftwidth=4 expandtab
  */

src/common/libflux/ev_buffer_read.h

@@ -21,6 +21,7 @@ typedef void (*ev_buffer_read_f)(struct ev_loop *loop,
                                  int revents);
 
 struct ev_buffer_read {
+    int              refcnt;
     ev_io            io_w;
     ev_prepare       prepare_w;
     ev_idle          idle_w;
@@ -44,5 +45,7 @@ int ev_buffer_read_init (struct ev_buffer_read *ebr,
 void ev_buffer_read_cleanup (struct ev_buffer_read *ebr);
 void ev_buffer_read_start (struct ev_loop *loop, struct ev_buffer_read *ebr);
 void ev_buffer_read_stop (struct ev_loop *loop, struct ev_buffer_read *ebr);
+void ev_buffer_read_incref (struct ev_buffer_read *ebr);
+void ev_buffer_read_decref (struct ev_buffer_read *ebr);
 
 #endif /* !_EV_BUFFER_READ_H */

src/common/libflux/reactor.c

@@ -463,6 +463,18 @@ const char *flux_buffer_read_watcher_get_data (flux_watcher_t *w, int *lenp)
     return NULL;
 }
 
+void flux_buffer_read_watcher_incref (flux_watcher_t *w)
+{
+    if (w)
+        ev_buffer_read_incref ((struct ev_buffer_read *)w->data);
+}
+
+void flux_buffer_read_watcher_decref (flux_watcher_t *w)
+{
+    if (w)
+        ev_buffer_read_decref ((struct ev_buffer_read *)w->data);
+}
+
 static void buffer_write_start (flux_watcher_t *w)
 {
     struct ev_buffer_write *ebw = (struct ev_buffer_write *)w->data;

src/common/libflux/reactor.h

@@ -115,6 +115,12 @@ flux_buffer_t *flux_buffer_read_watcher_get_buffer (flux_watcher_t *w);
 const char *flux_buffer_read_watcher_get_data (flux_watcher_t *w,
                                                int *lenp);
 
+/* Take a reference on read watcher to prevent read of EOF
+ * EOF will be delayed until decref drops refcount to 0.
+ */
+void flux_buffer_read_watcher_incref (flux_watcher_t *w);
+void flux_buffer_read_watcher_decref (flux_watcher_t *w);
+
 /* 'cb' only called after fd closed (FLUX_POLLOUT) or error (FLUX_POLLERR) */
 flux_watcher_t *flux_buffer_write_watcher_create (flux_reactor_t *r, int fd,
                                                   int size, flux_watcher_f cb,

src/common/libflux/test/reactor.c

@@ -708,10 +708,72 @@ static void test_buffer (flux_reactor_t *reactor)
 
 struct buffer_fd_close
 {
+    flux_watcher_t *w;
     int count;
     int fd;
 };
 
+static void buffer_decref (flux_reactor_t *r,
+                           flux_watcher_t *w,
+                           int revents,
+                           void *arg)
+{
+    struct buffer_fd_close *bfc = arg;
+    bfc->count++;
+    flux_buffer_read_watcher_decref (bfc->w);
+    ok (true, "flux_buffer_read_watcher_decref");
+    flux_watcher_destroy (w);
+}
+
+static void buffer_read_fd_decref (flux_reactor_t *r,
+                                   flux_watcher_t *w,
+                                   int revents,
+                                   void *arg)
+{
+    struct buffer_fd_close *bfc = arg;
+    flux_buffer_t *fb;
+    const void *ptr;
+    int len;
+
+    if (revents & FLUX_POLLERR) {
+        fail ("buffer decref: got FLUX_POLLERR");
+        return;
+    }
+    if (!(revents & FLUX_POLLIN)) {
+        fail ("buffer decref: got FLUX_POLLERR");
+        return;
+    }
+
+    fb = flux_buffer_read_watcher_get_buffer (w);
+    ok ((ptr = flux_buffer_read (fb, -1, &len)) != NULL,
+        "buffer decref: read from buffer success");
+    if (!bfc->count) {
+        flux_watcher_t *w;
+        ok (len == 6,
+            "buffer decref: read returned correct length");
+        ok (!memcmp (ptr, "foobar", 6),
+            "buffer decref: read returned correct data");
+        diag ("closing write side of read buffer");
+        close (bfc->fd);
+
+        /* Schedule decref of read buffer
+         */
+        w = flux_timer_watcher_create (r, 0.01, 0., buffer_decref, bfc);
+        flux_watcher_start (w);
+    }
+    else {
+        ok (bfc->count == 2,
+            "buffer decref: EOF called only after manual decref");
+        ok ((ptr = flux_buffer_read (fb, -1, &len)) != NULL,
+            "buffer decref: read from buffer success");
+
+        ok (len == 0,
+            "buffer decref: read returned 0, socketpair is closed");
+        flux_watcher_stop (w);
+    }
+    bfc->count++;
+}
+
 static void buffer_read_fd_close (flux_reactor_t *r, flux_watcher_t *w,
                                   int revents, void *arg)
 {
@@ -865,6 +927,49 @@ static void buffer_read_line_fd_close_and_left_over_data (flux_reactor_t *r,
     return;
 }
 
+static void test_buffer_refcnt (flux_reactor_t *reactor)
+{
+    int fd[2];
+    flux_watcher_t *w;
+    struct buffer_fd_close bfc;
+
+    /* read buffer decref test - other end closes stream */
+
+    ok (socketpair (PF_LOCAL, SOCK_STREAM|SOCK_NONBLOCK, 0, fd) == 0,
+        "buffer decref: successfully created socketpair");
+
+    bfc.count = 0;
+    bfc.fd = fd[1];
+    w = flux_buffer_read_watcher_create (reactor,
+                                         fd[0],
+                                         1024,
+                                         buffer_read_fd_decref,
+                                         0,
+                                         &bfc);
+    ok (w != NULL,
+        "buffer decref: read created");
+    bfc.w = w;
+
+    ok (write (fd[1], "foobar", 6) == 6,
+        "buffer decref: write to socketpair success");
+
+    flux_watcher_start (w);
+
+    diag ("calling flux_buffer_read_watcher_incref");
+    flux_buffer_read_watcher_incref (w);
+
+    ok (flux_reactor_run (reactor, 0) == 0,
+        "buffer decref: reactor ran to completion");
+
+    ok (bfc.count == 3,
+        "buffer decref: read callback successfully called thrice");
+
+    flux_watcher_stop (w);
+    flux_watcher_destroy (w);
+
+    close (fd[0]);
+}
+
 static void test_buffer_corner_case (flux_reactor_t *reactor)
 {
     int fd[2];
@@ -1521,6 +1626,7 @@ int main (int argc, char *argv[])
     test_periodic (reactor);
     test_fd (reactor);
     test_buffer (reactor);
+    test_buffer_refcnt (reactor);
     test_buffer_corner_case (reactor);
     test_idle (reactor);
     test_prepcheck (reactor);

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
 }

src/common/libsubprocess/test/iochan.c

@@ -36,6 +36,7 @@ struct iochan_ctx {
     int recvcount;
     int sendcount;
     int count;
+    int refcount;
     const char *name;
 };
 
@@ -108,6 +109,11 @@ static void iochan_state_cb (flux_subprocess_t *p,
                 diag ("%s: %s", ctx->name, strsignal (WTERMSIG (status)));
             // completion callback will exit the reactor, but just in case
             iochan_start_doomsday (ctx, 2.);
+            // if testing refcnt, release stdout reference now
+            if (streq (ctx->name, "refcnt")) {
+                ctx->refcount--;
+                flux_subprocess_channel_decref (ctx->p, "stdout");
+            }
             break;
         }
         case FLUX_SUBPROCESS_FAILED:
@@ -200,6 +206,10 @@ bool iochan_run_check (flux_t *h, const char *name, int count)
                                  tap_logger,
                                  NULL)))
         BAIL_OUT ("flux_rexec_ex failed");
+    if (streq (ctx.name, "refcnt")) {
+        ctx.refcount++;
+        flux_subprocess_channel_incref (ctx.p, "stdout");
+    }
     if (flux_subprocess_aux_set (ctx.p, "ctx", &ctx, NULL) < 0)
         BAIL_OUT ("flux_subprocess_aux_set failed");
 
@@ -224,6 +234,12 @@ bool iochan_run_check (flux_t *h, const char *name, int count)
     if (ctx.recvcount < ctx.sendcount)
         ret = false;
 
+    if (streq (ctx.name, "refcnt")) {
+        diag ("%s: final refcount: %d", ctx.name, ctx.refcount);
+        if (ctx.refcount != 0)
+            ret = false;
+    }
+
     flux_watcher_destroy (ctx.source);
     flux_watcher_destroy (ctx.timer);
     diag ("%s: destroying subprocess", name);
@@ -247,6 +263,8 @@ int main (int argc, char *argv[])
         "medium check worked");
     ok (iochan_run_check (h, "simple", linesize * 10000),
         "large check worked");
+    ok (iochan_run_check (h, "refcnt", linesize * 10),
+        "refcount check worked");
     test_server_stop (h);
     flux_close (h);
 

src/common/libterminus/pty.c

@@ -425,6 +425,12 @@ void pty_client_send_data (struct flux_pty *pty, void *data, int len)
     }
 }
 
+void pty_client_monitor_send_eof (struct flux_pty *pty)
+{
+    if (pty->monitor)
+        (*pty->monitor) (pty, NULL, 0);
+}
+
 static void pty_read (flux_reactor_t *r,
                       flux_watcher_t *w,
                       int revents,
@@ -452,6 +458,7 @@ static void pty_read (flux_reactor_t *r,
             flux_watcher_stop (pty->fdw);
             pty->wait_on_close = false;
             check_pty_complete (pty);
+            pty_client_monitor_send_eof (pty);
             return;
         }
         llog_error (pty, "read: %s", strerror (errno));