flux-exec: ensure stdin is restored to blocking mode on exit

doc/man1/flux-exec.adoc

@@ -25,8 +25,8 @@ stdout and stderr.
 On receipt of SIGINT and SIGTERM signals, flux-exec(1) shall forward
 the received signal to all currently running remote processes.
 
-flux-exec(1) is meant as an administrative and test utility, and should not
-be used for executing lightweight jobs (LWJs) or user commands.
+flux-exec(1) is meant as an administrative and test utility, and cannot
+be used to launch Flux jobs.
 
 EXIT STATUS
 -----------
@@ -46,7 +46,7 @@ OPTIONS
 Label lines of output with the source RANK.
 
 *-n, --noinput*::
-Redirect `stdin` from `/dev/null`.
+Do not attempt to forward stdin.  Send EOF to remote process stdin.
 
 *-d, --dir*'=DIR'::
 Set the working directory of remote 'COMMANDS' to 'DIR'. The default is to

src/cmd/builtin/proxy.c

@@ -38,7 +38,6 @@
 #include <errno.h>
 #include <libgen.h>
 #include <stdbool.h>
-#include <fcntl.h>
 #include <argz.h>
 #include <glob.h>
 #include <czmq.h>
@@ -47,6 +46,7 @@
 #include "src/common/libutil/xzmalloc.h"
 #include "src/common/libutil/oom.h"
 #include "src/common/libutil/cleanup.h"
+#include "src/common/libutil/fdutils.h"
 #include "src/common/libsubprocess/subprocess.h"
 
 #define LISTEN_BACKLOG      5
@@ -126,20 +126,6 @@ static proxy_ctx_t *ctx_create (flux_t *h)
     return ctx;
 }
 
-static int set_nonblock (int fd, bool nonblock)
-{
-    int flags = fcntl (fd, F_GETFL);
-    if (flags < 0)
-        return -1;
-    if (nonblock)
-        flags |= O_NONBLOCK;
-    else
-        flags &= ~O_NONBLOCK;
-    if (fcntl (fd, F_SETFL, flags) < 0)
-        return -1;
-    return 0;
-}
-
 static client_t * client_create (proxy_ctx_t *ctx, int rfd, int wfd)
 {
     client_t *c;
@@ -168,12 +154,12 @@ static client_t * client_create (proxy_ctx_t *ctx, int rfd, int wfd)
     flux_watcher_start (c->inw);
     flux_msg_iobuf_init (&c->inbuf);
     flux_msg_iobuf_init (&c->outbuf);
-    if (set_nonblock (c->rfd, true) < 0) {
-        flux_log_error (h, "set_nonblock");
+    if (fd_set_nonblocking (c->rfd) < 0) {
+        flux_log_error (h, "fd_set_nonblocking");
         goto error;
     }
-    if (c->wfd != c->rfd && set_nonblock (c->wfd, true) < 0) {
-        flux_log_error (h, "set_nonblock");
+    if (c->wfd != c->rfd && fd_set_nonblocking (c->wfd) < 0) {
+        flux_log_error (h, "fd_set_nonblocking");
         goto error;
     }
 

src/cmd/flux-exec.c

@@ -68,8 +68,8 @@ zlist_t *subprocesses;
 
 optparse_t *opts = NULL;
 
+int stdin_flags;
 flux_watcher_t *stdin_w;
-int stdin_fd = STDIN_FILENO;
 
 void completion_cb (flux_subprocess_t *p)
 {
@@ -108,10 +108,12 @@ void state_cb (flux_subprocess_t *p, flux_subprocess_state_t state)
         exited++;
     }
 
-    if (started == rank_count)
-        flux_watcher_start (stdin_w);
-    if (exited == rank_count)
-        flux_watcher_stop (stdin_w);
+    if (stdin_w) {
+        if (started == rank_count)
+            flux_watcher_start (stdin_w);
+        if (exited == rank_count)
+            flux_watcher_stop (stdin_w);
+    }
 
     if (state == FLUX_SUBPROCESS_EXEC_FAILED
         || state == FLUX_SUBPROCESS_FAILED) {
@@ -221,6 +223,15 @@ void subprocess_destroy (void *arg)
     flux_subprocess_destroy (p);
 }
 
+/* atexit handler
+ * This is a good faith attempt to restore stdin flags to what they were
+ * before we set O_NONBLOCK per bug #1803.
+ */
+void restore_stdin_flags (void)
+{
+    (void)fcntl (STDIN_FILENO, F_SETFL, stdin_flags);
+}
+
 int main (int argc, char *argv[])
 {
     const char *optargp;
@@ -315,16 +326,31 @@ int main (int argc, char *argv[])
     if (optparse_getopt (opts, "verbose", NULL) > 0)
         fprintf (stderr, "%03fms: Sent all requests\n", monotime_since (t0));
 
+    /* -n,--noinput: close subprocess STDIN
+     */
     if (optparse_getopt (opts, "noinput", NULL) > 0) {
-        if ((stdin_fd = open ("/dev/null", O_RDONLY)) < 0)
-            log_err_exit ("open");
+        flux_subprocess_t *p;
+        p = zlist_first (subprocesses);
+        while (p) {
+            if (flux_subprocess_close (p, "STDIN") < 0)
+                log_err_exit ("flux_subprocess_close");
+            p = zlist_next (subprocesses);
+        }
+    }
+    /* configure stdin watcher
+     */
+    else {
+        if ((stdin_flags = fcntl (STDIN_FILENO, F_GETFL)) < 0)
+            log_err_exit ("fcntl F_GETFL stdin");
+        if (atexit (restore_stdin_flags) != 0)
+            log_err_exit ("atexit");
+        if (fcntl (STDIN_FILENO, F_SETFL, stdin_flags | O_NONBLOCK) < 0)
+            log_err_exit ("fcntl F_SETFL stdin");
+        if (!(stdin_w = flux_buffer_read_watcher_create (r, STDIN_FILENO,
+                                                         1 << 20, stdin_cb,
+                                                         0, NULL)))
+            log_err_exit ("flux_buffer_read_watcher_create");
     }
-
-    if (!(stdin_w = flux_buffer_read_watcher_create (r, stdin_fd,
-                                                     1 << 20, stdin_cb,
-                                                     0, NULL)))
-        log_err_exit ("flux_buffer_read_watcher_create");
-
     if (signal (SIGINT, signal_cb) == SIG_ERR)
         log_err_exit ("signal");
 

src/common/libflux/reactor.c

@@ -42,6 +42,7 @@
 #include "src/common/libev/ev.h"
 #include "src/common/libutil/ev_zmq.h"
 #include "src/common/libutil/log.h"
+#include "src/common/libutil/fdutils.h"
 
 struct flux_reactor {
     struct ev_loop *loop;
@@ -379,30 +380,6 @@ int flux_fd_watcher_get_fd (flux_watcher_t *w)
 /* buffer
  */
 
-static int fd_set_nonblocking (int fd, int *flags_orig)
-{
-    int fval;
-
-    if ((fval = fcntl (fd, F_GETFL, 0)) < 0)
-        return (-1);
-
-    if (fcntl (fd, F_SETFL, fval | O_NONBLOCK) < 0)
-        return (-1);
-
-    if (flags_orig)
-        (*flags_orig) = fval;
-
-    return (0);
-}
-
-static int fd_set_flags (int fd, int flags)
-{
-    if (fcntl (fd, F_SETFL, flags) < 0)
-        return (-1);
-
-    return (0);
-}
-
 static void buffer_read_start (flux_watcher_t *w)
 {
     struct ev_buffer_read *ebr = (struct ev_buffer_read *)w->data;
@@ -442,15 +419,18 @@ flux_watcher_t *flux_buffer_read_watcher_create (flux_reactor_t *r, int fd,
 {
     struct ev_buffer_read *ebr;
     flux_watcher_t *w = NULL;
-    int flags_orig;
+    int fd_flags;
 
     if (fd < 0) {
         errno = EINVAL;
         return NULL;
     }
-
-    if (fd_set_nonblocking (fd, &flags_orig) < 0)
+    if ((fd_flags = fd_get_flags (fd)) < 0)
         return NULL;
+    if (!(fd_flags & O_NONBLOCK)) {
+        errno = EINVAL;
+        return NULL;
+    }
 
     if (!(w = flux_watcher_create (r,
                                    sizeof (*ebr),
@@ -477,7 +457,6 @@ flux_watcher_t *flux_buffer_read_watcher_create (flux_reactor_t *r, int fd,
 
 cleanup:
     flux_watcher_destroy (w);
-    fd_set_flags (fd, flags_orig);
     return NULL;
 }
 
@@ -527,15 +506,19 @@ flux_watcher_t *flux_buffer_write_watcher_create (flux_reactor_t *r, int fd,
 {
     struct ev_buffer_write *ebw;
     flux_watcher_t *w = NULL;
-    int flags_orig;
+    int fd_flags;
 
     if (fd < 0) {
         errno = EINVAL;
         return NULL;
     }
 
-    if (fd_set_nonblocking (fd, &flags_orig) < 0)
+    if ((fd_flags = fd_get_flags (fd)) < 0)
         return NULL;
+    if (!(fd_flags & O_NONBLOCK)) {
+        errno = EINVAL;
+        return NULL;
+    }
 
     if (!(w = flux_watcher_create (r,
                                    sizeof (*ebw),
@@ -559,7 +542,6 @@ flux_watcher_t *flux_buffer_write_watcher_create (flux_reactor_t *r, int fd,
 
 cleanup:
     flux_watcher_destroy (w);
-    fd_set_flags (fd, flags_orig);
     return NULL;
 }
 

src/common/libflux/test/reactor.c

@@ -6,6 +6,7 @@
 
 #include "src/common/libflux/reactor.h"
 #include "src/common/libutil/xzmalloc.h"
+#include "src/common/libutil/fdutils.h"
 #include "src/common/libtap/tap.h"
 
 static const size_t zmqwriter_msgcount = 1024;
@@ -170,23 +171,12 @@ static void fdreader (flux_reactor_t *r, flux_watcher_t *w,
     flux_reactor_stop_error (r);
 }
 
-static int set_nonblock (int fd)
-{
-    int flags = fcntl (fd, F_GETFL, NULL);
-    if (flags < 0 || fcntl (fd, F_SETFL, flags | O_NONBLOCK) < 0) {
-        fprintf (stderr, "fcntl: %s\n", strerror (errno));
-        return -1;
-    }
-    return 0;
-}
-
 static void test_fd (flux_reactor_t *reactor)
 {
     int fd[2];
     flux_watcher_t *r, *w;
 
-    ok (socketpair (PF_LOCAL, SOCK_STREAM, 0, fd) == 0
-        && set_nonblock (fd[0]) == 0 && set_nonblock (fd[1]) == 0,
+    ok (socketpair (PF_LOCAL, SOCK_STREAM|SOCK_NONBLOCK, 0, fd) == 0,
         "fd: successfully created non-blocking socketpair");
     r = flux_fd_watcher_create (reactor, fd[0], FLUX_POLLIN, fdreader, NULL);
     w = flux_fd_watcher_create (reactor, fd[1], FLUX_POLLOUT, fdwriter, NULL);
@@ -374,7 +364,7 @@ static void test_buffer (flux_reactor_t *reactor)
     int count;
     char buf[1024];
 
-    ok (socketpair (PF_LOCAL, SOCK_STREAM, 0, fd) == 0,
+    ok (socketpair (PF_LOCAL, SOCK_STREAM|SOCK_NONBLOCK, 0, fd) == 0,
         "buffer: successfully created socketpair");
 
     /* read buffer test */
@@ -587,6 +577,19 @@ static void test_buffer (flux_reactor_t *reactor)
     count = 0;
     ok (pipe (pfds) == 0,
         "buffer: hey I can has a pipe!");
+
+    w = flux_buffer_write_watcher_create (reactor,
+                                          pfds[1],
+                                          1024,
+                                          buffer_write,
+                                          0,
+                                          &count);
+    ok (w == NULL && errno == EINVAL,
+        "buffer: write_watcher_create fails with EINVAL if fd !nonblocking");
+
+    ok (fd_set_nonblocking (pfds[1]) >= 0,
+        "buffer: fd_set_nonblocking");
+
     w = flux_buffer_write_watcher_create (reactor,
                                           pfds[1],
                                           1024,
@@ -811,7 +814,7 @@ static void test_buffer_corner_case (flux_reactor_t *reactor)
 
     /* read buffer corner case test - other end closes stream */
 
-    ok (socketpair (PF_LOCAL, SOCK_STREAM, 0, fd) == 0,
+    ok (socketpair (PF_LOCAL, SOCK_STREAM|SOCK_NONBLOCK, 0, fd) == 0,
         "buffer corner case: successfully created socketpair");
 
     bfc.count = 0;
@@ -848,7 +851,7 @@ static void test_buffer_corner_case (flux_reactor_t *reactor)
 
     /* read line buffer corner case test - other end closes stream */
 
-    ok (socketpair (PF_LOCAL, SOCK_STREAM, 0, fd) == 0,
+    ok (socketpair (PF_LOCAL, SOCK_STREAM|SOCK_NONBLOCK, 0, fd) == 0,
         "buffer corner case: successfully created socketpair");
 
     bfc.count = 0;
@@ -885,7 +888,7 @@ static void test_buffer_corner_case (flux_reactor_t *reactor)
 
     /* read line buffer corner case test - left over data not a line */
 
-    ok (socketpair (PF_LOCAL, SOCK_STREAM, 0, fd) == 0,
+    ok (socketpair (PF_LOCAL, SOCK_STREAM|SOCK_NONBLOCK, 0, fd) == 0,
         "buffer corner case: successfully created socketpair");
 
     bfc.count = 0;

src/common/libsubprocess/local.c

@@ -37,6 +37,7 @@
 
 #include "src/common/libutil/log.h"
 #include "src/common/libutil/fdwalk.h"
+#include "src/common/libutil/fdutils.h"
 
 #include "subprocess.h"
 #include "subprocess_private.h"
@@ -192,6 +193,7 @@ static int channel_local_setup (flux_subprocess_t *p,
     int fds[2] = { -1, -1 };
     char *e = NULL;
     int save_errno;
+    int fd_flags;
 
     if (!(c = channel_create (p, output_f, name, channel_flags))) {
         flux_log_error (p->h, "calloc");
@@ -212,6 +214,11 @@ static int channel_local_setup (flux_subprocess_t *p,
     fds[0] = -1;
     fds[1] = -1;
 
+    if ((fd_flags = fd_set_nonblocking (c->parent_fd)) < 0) {
+        flux_log_error (p->h, "fd_set_nonblocking");
+        goto error;
+    }
+
     if ((channel_flags & CHANNEL_WRITE) && in_cb) {
         c->buffer_write_w = flux_buffer_write_watcher_create (p->reactor,
                                                               c->parent_fd,
@@ -426,9 +433,7 @@ static void closefd_child (void *arg, int fd)
     c = zhash_first (p->channels);
     while (c) {
         if (c->child_fd == fd) {
-            int flags = fcntl (fd, F_GETFD, 0);
-            if (flags >= 0)
-                (void) fcntl (fd, F_SETFD, flags & ~FD_CLOEXEC);
+            (void) fd_unset_cloexec (fd);
             return;
         }
         c = zhash_next (p->channels);