libsubprocess: avoid using child watchers

Problem: child watchers are only used by libsubprocess, and probably
do not need to be in the public API since libsubprocess is provided
for managing child processes.  In addition, we are considering porting
Flux to libuv and libuv does not offer similar functionality.

Register a signal watcher for SIGCHLD within libsubprocess that
persists as long as there are subprocesses to monitor, and calls
waitpid(2) to consume all subprocess state changes.  Add a hash by pid
and allow subprocess objects to register a callback to receive these
changes for a given pid.

Have all subprocess users create reactors without FLUX_REACTOR_SIGCHLD,
as the default ev_loop registers a SIGCHLD watcher that conflicts with
this one.

Add EVFLAG_SIGNALFD to the non-default loop, as that flag was used with
reactors created with FLUX_REACTOR_SIGCHLD, and appears to be be required
to avoid sharness tests hanging randomly.

Author: garlick

src/broker/broker.c

@@ -248,16 +248,15 @@ int main (int argc, char *argv[])
      * ctx.h is used conventionally within the broker for RPCs, message
      * handlers, etc.  ctx.h_internal belongs to the broker's routing logic
      * and is accessed using flux_send() and flux_recv() only.  Both handles
-     * share a reactor, which is created with FLUX_REACTOR_SIGCHLD in order
-     * to support libsubprocess.
+     * share a reactor.
      *
      * N.B. since both handles are in the same thread, synchronous RPCs on
      * ctx.h will deadlock.  The main broker reactor must run in order
      * to move messages from the interthread queue to the routing logic.
      * Careful with flux_attr_get(), which hides a synchronous RPC if the
      * requested value is not cached.
      */
-    if (!(ctx.reactor = flux_reactor_create (FLUX_REACTOR_SIGCHLD))
+    if (!(ctx.reactor = flux_reactor_create (0))
         || !(ctx.h = flux_open ("interthread://broker", 0))
         || flux_set_reactor (ctx.h, ctx.reactor) < 0
         || !(ctx.h_internal = flux_open ("interthread://broker", 0))

src/broker/test/runat.c

@@ -350,7 +350,6 @@ void badinput (flux_t *h)
 int main (int argc, char *argv[])
 {
     flux_t *h;
-    flux_reactor_t *r;
 
     plan (NO_PLAN);
 
@@ -360,12 +359,8 @@ int main (int argc, char *argv[])
 
     if (!(logs = zlist_new ()))
         BAIL_OUT ("zlist_new failed");
-    if (!(r = flux_reactor_create (FLUX_REACTOR_SIGCHLD)))
-        BAIL_OUT ("flux_reactor_create failed");
     if (!(h = flux_open ("loop://", 0)))
         BAIL_OUT ("could not create loop handle");
-    if (flux_set_reactor (h, r) < 0)
-        BAIL_OUT ("flux_set_reactor failed");
     if (flux_attr_set_cacheonly (h, "rank", "0") < 0)
         BAIL_OUT ("flux_attr_set_cacheonly rank failed");
     flux_log_set_redirect (h, diag_logger, NULL);
@@ -374,7 +369,6 @@ int main (int argc, char *argv[])
     basic (h);
     badinput (h);
 
-    flux_reactor_destroy (r);
     flux_close (h);
 
     clear_list (logs);

src/cmd/builtin/proxy.c

@@ -385,10 +385,8 @@ static int cmd_proxy (optparse_t *p, int ac, char *av[])
     free (uri);
     flux_log_set_appname (ctx.h, "proxy");
     ctx.proxy_user = getuid ();
-    if (!(r = flux_reactor_create (FLUX_REACTOR_SIGCHLD)))
-        log_err_exit ("flux_reactor_create");
-    if (flux_set_reactor (ctx.h, r) < 0)
-        log_err_exit ("flux_set_reactor");
+    if (!(r = flux_get_reactor (ctx.h)))
+        log_err_exit ("flux_get_reactor");
 
     /* Register handler for loss of broker connection if --reconnect
      */

src/cmd/flux-start.c

@@ -1074,7 +1074,7 @@ int start_session (const char *cmd_argz,
         if (signal (SIGTTOU, SIG_IGN) == SIG_ERR)
             log_err_exit ("signal");
     }
-    if (!(ctx.reactor = flux_reactor_create (FLUX_REACTOR_SIGCHLD)))
+    if (!(ctx.reactor = flux_reactor_create (0)))
         log_err_exit ("flux_reactor_create");
     if (!(ctx.timer = flux_timer_watcher_create (ctx.reactor,
                                                  ctx.exit_timeout,

src/cmd/flux-terminus.c

@@ -293,23 +293,16 @@ static int run_service (const char *service, int fd)
     struct flux_terminus_server *ts;
     flux_future_t *f;
     int rc = -1;
-    flux_reactor_t *r;
 
     if (fdwalk (terminus_server_closefd, &fd) < 0) {
         log_err ("fdwalk");
         goto err;
     }
 
-    r = flux_reactor_create (FLUX_REACTOR_SIGCHLD);
-    if (!r) {
-        log_err ("flux_reactor_create");
-        goto err;
-    }
     if (!(h = flux_open (NULL, 0))) {
         log_err ("flux_open");
         goto err;
     }
-    flux_set_reactor (h, r);
     if (!(f = flux_service_register (h, service))
         || flux_future_get (f, NULL) < 0) {
         log_err ("flux_service_register (%s)", service);
@@ -330,11 +323,10 @@ static int run_service (const char *service, int fd)
 
     /*  Set up to exit when the last session exits
      */
-    flux_terminus_server_notify_empty (ts, empty_cb, r);
+    flux_terminus_server_notify_empty (ts, empty_cb, flux_get_reactor (h));
 
     rc = flux_reactor_run (flux_get_reactor (h), 0);
     flux_terminus_server_destroy (ts);
-    flux_reactor_destroy (r);
     flux_close (h);
     return rc;
 err:

src/common/libflux/reactor.c

@@ -72,7 +72,7 @@ flux_reactor_t *flux_reactor_create (int flags)
     if ((flags & FLUX_REACTOR_SIGCHLD))
         r->loop = ev_default_loop (EVFLAG_SIGNALFD);
     else
-        r->loop = ev_loop_new (EVFLAG_NOSIGMASK);
+        r->loop = ev_loop_new (EVFLAG_NOSIGMASK | EVFLAG_SIGNALFD);
     if (!r->loop) {
         errno = ENOMEM;
         flux_reactor_destroy (r);

src/common/libsubprocess/Makefile.am

@@ -20,6 +20,8 @@ noinst_LTLIBRARIES = \
 libsubprocess_la_SOURCES = \
 	command.c \
 	command_private.h \
+	sigchld.c \
+	sigchld.h \
 	local.c \
 	local.h \
 	fork.c \
@@ -48,6 +50,7 @@ fluxcoreinclude_HEADERS = \
 	subprocess.h
 
 TESTS = \
+	test_sigchld.t \
 	test_command.t \
 	test_subprocess.t \
 	test_stdio.t \
@@ -102,6 +105,11 @@ bulk_exec_LDADD = \
 
 bulk_exec_LDFLAGS = $(test_ldflags)
 
+test_sigchld_t_SOURCES = test/sigchld.c
+test_sigchld_t_CPPFLAGS = $(test_cppflags)
+test_sigchld_t_LDADD = $(test_ldadd)
+test_sigchld_t_LDFLAGS = $(test_ldflags)
+
 test_command_t_SOURCES = test/command.c
 test_command_t_CPPFLAGS = $(test_cppflags)
 test_command_t_LDADD = $(test_ldadd)

src/common/libsubprocess/local.c

@@ -33,6 +33,7 @@
 #include "fork.h"
 #include "posix_spawn.h"
 #include "util.h"
+#include "sigchld.h"
 
 static void local_channel_flush (struct subprocess_channel *c)
 {
@@ -449,20 +450,9 @@ static void close_child_fds (flux_subprocess_t *p)
     }
 }
 
-static void child_watch_cb (flux_reactor_t *r,
-                            flux_watcher_t *w,
-                            int revents,
-                            void *arg)
+static void sigchld_cb (pid_t pid, int status, void *arg)
 {
     flux_subprocess_t *p = arg;
-    int status;
-
-    if ((status = flux_child_watcher_get_rstatus (w)) < 0) {
-        llog_error (p,
-                    "flux_child_watcher_get_rstatus: %s",
-                    strerror (errno));
-        return;
-    }
 
     p->status = status;
 
@@ -480,9 +470,8 @@ static void child_watch_cb (flux_reactor_t *r,
             state_change_start (p);
         }
 
-        /*  Child watcher no longer needed, pid now invalid */
-        if (p->child_w)
-            flux_watcher_stop (p->child_w);
+        /*  callback no longer needed, pid now invalid */
+        sigchld_unregister (pid);
     }
 
     if (p->state == FLUX_SUBPROCESS_EXITED)
@@ -502,16 +491,10 @@ static int start_local_watchers (flux_subprocess_t *p)
 {
     struct subprocess_channel *c;
 
-    /* no-op if reactor is !FLUX_REACTOR_SIGCHLD */
-    if (!(p->child_w = flux_child_watcher_create (p->reactor,
-                                                  p->pid,
-                                                  true,
-                                                  child_watch_cb,
-                                                  p))) {
-        llog_debug (p, "flux_child_watcher_create: %s", strerror (errno));
+    if (sigchld_register (p->reactor, p->pid, sigchld_cb, p) < 0) {
+        llog_debug (p, "sigchld_register: %s", strerror (errno));
         return -1;
     }
-    flux_watcher_start (p->child_w);
 
     c = zhash_first (p->channels);
     while (c) {

src/common/libsubprocess/server.c

@@ -30,6 +30,7 @@
 #include "server.h"
 #include "client.h"
 #include "util.h"
+#include "sigchld.h"
 
 extern char **environ;
 
@@ -55,6 +56,7 @@ struct subprocess_server {
     // The shutdown future is created when user calls shutdown,
     //  and fulfilled once subprocesses list becomes empty.
     flux_future_t *shutdown;
+    bool has_sigchld_ctx;
 };
 
 static void server_kill (flux_subprocess_t *p, int signum);
@@ -709,6 +711,8 @@ void subprocess_server_destroy (subprocess_server_t *s)
         flux_future_destroy (s->shutdown);
         free (s->service_name);
         free (s->local_uri);
+        if (s->has_sigchld_ctx)
+            sigchld_finalize ();
         free (s);
         errno = saved_errno;
     }
@@ -750,6 +754,11 @@ subprocess_server_t *subprocess_server_create (flux_t *h,
                                     &s->handlers) < 0)
         goto error;
 
+    /* Avoid unnecessary on-demand create/destroy of SIGCHLD handler + hash.
+     */
+    if (sigchld_initialize (flux_get_reactor (h)) < 0)
+        goto error;
+    s->has_sigchld_ctx = true;
     return s;
 
 error:

src/common/libsubprocess/server.h

@@ -19,11 +19,10 @@ typedef int (*subprocess_server_auth_f) (const flux_msg_t *msg,
                                          void *arg,
                                          flux_error_t *error);
 
-/* Create a subprocess server.  The handle 'h' must contain a reactor
- * created with the FLUX_REACTOR_SIGCHLD flag.  Note that there can be
- * only one reactor per process with this flag set.  Also, it may be wise
- * to block SIGPIPE to avoid termination when writing to stdin of a subprocess
- * that has terminated.
+/* Create a subprocess server.
+ * This sets up a signal watcher for SIGCHLD.  Make sure SIGCHLD cannot be
+ * delivered to other threads. Also, it may be wise to block SIGPIPE to
+ * avoid termination when writing to stdin of a subprocess that has terminated.
  */
 subprocess_server_t *subprocess_server_create (flux_t *h,
                                                const char *service_name,