ENT-14434: Move reactor-plugin logic into cf-reactor daemon loop

cf-reactor/cf-reactor.c

@@ -33,13 +33,23 @@
 #include <man.h>
 #include <cleanup.h>
 #include <prototypes3.h>
+#include <signal.h>             /* signal, kill */
+#include <signals.h>            /* GetSignalPipe, MakeSignalPipe, IsPendingTermination, HandleSignalsForDaemon */
+#include <exec_tools.h>
+#include <alloc.h>              /* xmalloc */
 
 /*****************************************************************************/
 /* Globals                                                                   */
 /*****************************************************************************/
 
 int NO_FORK = false;
 
+/*****************************************************************************/
+/* Constants                                                                 */
+/*****************************************************************************/
+
+#define DEFAULT_POLL_INTERVAL_SECS 30
+
 /*******************************************************************/
 /* Command line options                                            */
 /*******************************************************************/
@@ -179,16 +189,183 @@ static GenericAgentConfig *CheckOpts(int argc, char **argv)
 
 /*****************************************************************************/
 
+
+static int SetupFileDescriptors(fd_set *readfds, int *fds, size_t num_fds)
+{
+    assert(readfds != NULL);
+
+    FD_ZERO(readfds);
+    int signal_pipe = GetSignalPipe();
+    FD_SET(signal_pipe, readfds);
+
+    int max_fd = signal_pipe;
+
+    for (size_t i = 0; i < num_fds; i++)
+    {
+        FD_SET(fds[i], readfds);
+        max_fd = MAX(fds[i], max_fd);
+    }
+    return max_fd + 1;
+}
+
+static bool ReactorNovaHasTimedOut(fd_set *readfds, int *fds, size_t num_fds)
+{
+    assert(readfds != NULL);
+
+    for (size_t i = 0; i < num_fds; i++)
+    {
+        if (FD_ISSET(fds[i], readfds))
+        {
+            return false;
+        }
+    }
+    return true;
+}
+
 int main(int argc, char *argv[])
 {
     GenericAgentConfig *config = CheckOpts(argc, argv);
     EvalContext *ctx = EvalContextNew();
     GenericAgentConfigApply(ctx, config);
 
-    int ret = ReactorEnterpriseMain(NO_FORK);
+#ifdef __MINGW32__
+
+    if (!NO_FORK)
+    {
+        Log(LOG_LEVEL_VERBOSE, "Windows does not support starting processes in the background - starting in foreground");
+    }
+
+#else /* !__MINGW32__ */
+    pid_t existing_pid = ReadPID("cf-reactor.pid");
+    if ((existing_pid != -1) && (kill(existing_pid, 0) == 0))
+    {
+        Log(LOG_LEVEL_ERR, "Another instance of cf-reactor is already running (pid %jd), terminating",
+            (intmax_t) existing_pid);
+        return 1;
+    }
+
+    if ((!NO_FORK) && (fork() != 0))
+    {
+        Log(LOG_LEVEL_INFO, "cf-reactor: starting");
+        _exit(EXIT_SUCCESS);
+    }
+
+    if (!NO_FORK)
+    {
+        ActAsDaemon();
+    }
+
+#endif /* !__MINGW32__ */
+
+    umask(077);
+    WritePID("cf-reactor.pid");
+    MakeSignalPipe();
+
+    signal(SIGINT, HandleSignalsForDaemon);
+    signal(SIGTERM, HandleSignalsForDaemon);
+    signal(SIGBUS, HandleSignalsForDaemon);
+    signal(SIGHUP, HandleSignalsForDaemon);
+    signal(SIGUSR1, HandleSignalsForDaemon);
+    signal(SIGUSR2, HandleSignalsForDaemon);
+
+    /* Ask Nova how many fds it needs, rather than guessing a number here that
+     * really belongs to reactor-plugin (and would silently go stale if the
+     * two drift apart across releases). */
+    size_t max_nova_fds = ReactorNovaMaxFds();
+    int *all_fds = xmalloc(max_nova_fds * sizeof(int));
+    // the first num_nova_fds fds are populated with nova fds
+    size_t num_nova_fds;
+    if (!ReactorNovaInitialize(all_fds, max_nova_fds, &num_nova_fds))
+    {
+        free(all_fds);
+        GenericAgentFinalize(ctx, config);
+        DoCleanupAndExit(EXIT_FAILURE);
+    }
+    // returns the number of fds used by nova reactor
+    size_t num_fds = num_nova_fds;
+    // TODO: populate all_fds with other fd used for event driven code (the
+    // allocation above will need to grow accordingly, e.g. by adding a fixed
+    // count on top of max_nova_fds before calling xmalloc())
+
+    /* Writing to a pipe whose spawned process already exited (e.g. cfbs
+     * rejecting its arguments before reading its stdin) must fail with EPIPE
+     * rather than terminate the whole daemon. Set after ReactorNovaInitialize(),
+     * so that the spawner and the processes it execs keep the default handling. */
+    signal(SIGPIPE, SIG_IGN);
+
+    /* We need an initial value here for the first iteration of the cycle
+     * below. */
+    time_t next_tick = time(NULL) + DEFAULT_POLL_INTERVAL_SECS;
+    while (!IsPendingTermination())
+    {
+        fd_set readfds;
+        int max_fd = SetupFileDescriptors(&readfds, all_fds, num_fds);
+
+        /* Determine how much time is remaining until the next tick. */
+        time_t last_tick = time(NULL);
+        time_t remaining = next_tick > last_tick ? next_tick - last_tick : 0;
+
+        struct timeval timeout = { .tv_sec = remaining };
+        int ret = select(max_fd, &readfds, NULL, NULL, &timeout);
+
+        /* Reschedule the backstop tick against the current time (not
+         * `last_tick`, which was captured before select() potentially
+         * blocked for the whole `remaining` duration), so that both call
+         * sites of ReactorNovaHandleTimeout() below agree on what "the next
+         * tick" means, instead of one of them silently doubling the
+         * interval. */
+        next_tick = time(NULL) + DEFAULT_POLL_INTERVAL_SECS;
+
+        if (ret < 0)
+        {
+            if (errno == EINTR)
+            {
+                /* Not an error, just a signal delivered while blocked in
+                 * select(). Loop around: the top of the loop re-checks
+                 * whether termination is pending and rebuilds the fd set
+                 * from scratch. */
+                continue;
+            }
+
+            /*** error ***/
+            Log(LOG_LEVEL_ERR, "Failed to poll events: %s", GetErrorStr());
+            break;
+        }
+        else if (ret == 0)
+        {
+            /*** timeout ***/
+            Log(LOG_LEVEL_DEBUG, "Timed-out waiting for next notification");
+
+            ReactorNovaHandleTimeout(&next_tick);
+            continue;
+        }
+        /* else */
+
+        /* The signal pipe is always in the watched set so we wake up
+         * promptly on a pending signal, but (per its own contract in
+         * signals.c) it must be drained or it stays "ready" forever, which
+         * would stop select() from ever blocking again. */
+        if (FD_ISSET(GetSignalPipe(), &readfds))
+        {
+            unsigned char buf;
+            while (recv(GetSignalPipe(), &buf, 1, 0) > 0) { /* drain */ }
+        }
+
+        /* This is needed since num_nova_fds may end up smaller than num_fds
+         * once other event-driven fds are added (see the TODO above). */
+        if (ReactorNovaHasTimedOut(&readfds, all_fds, num_nova_fds))
+        {
+            ReactorNovaHandleTimeout(&next_tick);
+            continue;
+        }
+
+        ReactorNovaHandleEvents(&readfds, all_fds, &next_tick);
+    }
+    ReactorNovaFinalize();
+    free(all_fds);
 
     GenericAgentFinalize(ctx, config);
     CallCleanupFunctions();
 
-    return ret;
+    return 0;
 }

libpromises/enterprise_stubs.c

@@ -232,9 +232,27 @@ ENTERPRISE_VOID_FUNC_2ARG_DEFINE_STUB(void, Nova_ClassHistoryEnable,
 {
 }
 
-ENTERPRISE_FUNC_1ARG_DEFINE_STUB(int, ReactorEnterpriseMain, ARG_UNUSED bool, no_fork)
+ENTERPRISE_FUNC_0ARG_DEFINE_STUB(size_t, ReactorNovaMaxFds)
+{
+    return 0;
+}
+
+ENTERPRISE_FUNC_3ARG_DEFINE_STUB(bool, ReactorNovaInitialize, ARG_UNUSED int*, fds, ARG_UNUSED size_t, max_size, size_t *, num_fds)
 {
     Log(LOG_LEVEL_VERBOSE, "Nova extension library is not available.");
     Log(LOG_LEVEL_VERBOSE, "Running cf-reactor community edition.");
-    return 0;
+    *num_fds = 0;
+    return true;
+}
+
+ENTERPRISE_VOID_FUNC_1ARG_DEFINE_STUB(void, ReactorNovaHandleTimeout, ARG_UNUSED time_t *, next_tick)
+{
+}
+
+ENTERPRISE_VOID_FUNC_3ARG_DEFINE_STUB(void, ReactorNovaHandleEvents, ARG_UNUSED fd_set *, readfds, ARG_UNUSED int *, fds, ARG_UNUSED time_t *, next_tick)
+{
+}
+
+ENTERPRISE_VOID_FUNC_0ARG_DEFINE_STUB(void, ReactorNovaFinalize)
+{
 }

libpromises/prototypes3.h

@@ -85,7 +85,11 @@ ENTERPRISE_VOID_FUNC_0ARG_DECLARE(void, ReloadHAConfig);
 ENTERPRISE_VOID_FUNC_2ARG_DECLARE(void, Nova_ClassHistoryAddContextName, const StringSet *, list, const char *, context_name);
 ENTERPRISE_VOID_FUNC_2ARG_DECLARE(void, Nova_ClassHistoryEnable, StringSet **, list, bool, enable);
 
-ENTERPRISE_FUNC_1ARG_DECLARE(int, ReactorEnterpriseMain, bool, no_fork);
+ENTERPRISE_FUNC_0ARG_DECLARE(size_t, ReactorNovaMaxFds);
+ENTERPRISE_FUNC_3ARG_DECLARE(bool, ReactorNovaInitialize, int*, fds, size_t, max_size, size_t *, num_fds);
+ENTERPRISE_VOID_FUNC_1ARG_DECLARE(void, ReactorNovaHandleTimeout, time_t *, next_tick);
+ENTERPRISE_VOID_FUNC_3ARG_DECLARE(void, ReactorNovaHandleEvents, fd_set *, readfds, int *, fds, time_t *, next_tick);
+ENTERPRISE_VOID_FUNC_0ARG_DECLARE(void, ReactorNovaFinalize);
 
 /* manual.c */
 

tests/valgrind-check/valgrind.sh

@@ -225,14 +225,14 @@ tail reactor.txt
 echo "Checking that serverd, execd and reactor PIDs are still correct/alive:"
 ps -p $exec_pid
 ps -p $server_pid
-# ps -p $reactor_pid
+ps -p $reactor_pid
 
 echo "Killing valgrind cf-execd"
 kill $exec_pid
 echo "Killing valgrind cf-serverd"
 kill $server_pid
-# echo "Killing valgrind cf-reactor"
-# kill $reactor_pid
+echo "Killing valgrind cf-reactor"
+kill $reactor_pid
 
 wait $exec_pid
 wait $server_pid