ENT-14195 2. cf-reactor data structure and polling

cf-reactor/Makefile.am

@@ -39,7 +39,9 @@ libcf_reactor_la_LIBADD = ../libpromises/libpromises.la
 
 libcf_reactor_la_SOURCES = \
 	cf-reactor.c \
-	reactor_context.c reactor_context.h
+	reactor_context.c reactor_context.h \
+	watcher.c watcher.h \
+	wakeup_channel.c wakeup_channel.h
 
 if !BUILTIN_EXTENSIONS
  bin_PROGRAMS = cf-reactor

cf-reactor/README.md

@@ -26,23 +26,6 @@ On dispatch, the bundle corresponding to an event is run, however it doesn't do
 
 ## Implementation details
 
-### Moving stuff around
-
-Rather than exposing the raw file-descriptor bookkeeping required for `select(2)`, we introduce a unified interface that serves both the reactor-plugin and event-driven code paths. This is achieved by encapsulating all relevant state in a context struct, `ReactorContext`:
-
-```C
-typedef struct ReactorContext
-{
-    Seq *fds // array of ReactorFd, which holds the fd and some metadata
-    fd_set readfds;
-} ReactorContext;
-```
-
-- `ReactorContextInitialize()`: initializes the reactor-plugin and event-driven code. Wraps `ReactorNovaInitialize()`
-- `ReactorContextSetupFileDescriptors()`: populates readfds with the file descriptors to monitor, prior to the select() call.
-- `ReactorContextHandleEvents()`: iterates over the file descriptors and dispatches the appropriate action based on which ones were signaled as ready. Wraps `ReactorNovaHandleTimeout` and `ReactorNovaHandleEvents()`.
-- `ReactorContextFinalize()`: releases the daemon's associated resources. Wraps `ReactorNovaFinalize()`.
-
 ### Tracking spec & Events
 
 In order to track all the events promises, we use two datastructures: a global list of `"Watcher"`, which is a struct associated with an event type and the promise name (also called `key`) and a global hashmap mapping this `key` to a `bundle` which is parsed from the policy.

cf-reactor/reactor_context.c

@@ -25,6 +25,7 @@
 #include <reactor_context.h>
 #include <prototypes3.h>        /* ReactorNova*() */
 #include <signals.h>            /* GetSignalPipe() */
+#include <watcher.h>
 #include <alloc.h>
 
 #define INIT_FD_COUNT 8
@@ -37,106 +38,129 @@ static ReactorFd *ReactorFdNew(int fd, ReactorFdType type)
     return rfd;
 }
 
-bool ReactorContextInitialize(ReactorContext *ctx)
+bool ReactorContextInitialize(ReactorContext *reactor_context)
 {
-    assert(ctx != NULL);
+    assert(reactor_context != NULL);
 
-    ctx->fds = SeqNew(INIT_FD_COUNT, free);
+    reactor_context->fds = SeqNew(INIT_FD_COUNT, free);
+
+    WatcherRegistryInitialize();
 
     // Initialize Nova fds
     {
-        ctx->max_nova_fds = ReactorNovaMaxFds();
-        int *nova_fds = (int *) xmalloc(ctx->max_nova_fds * sizeof(int));
+        reactor_context->max_nova_fds = ReactorNovaMaxFds();
+        int *nova_fds = (int *) xmalloc(reactor_context->max_nova_fds * sizeof(int));
         size_t num_nova_fds = 0;
 
-        if (!ReactorNovaInitialize(nova_fds, ctx->max_nova_fds, &num_nova_fds))
+        if (!ReactorNovaInitialize(nova_fds, reactor_context->max_nova_fds, &num_nova_fds))
         {
             free(nova_fds);
-            SeqDestroy(ctx->fds);
-            ctx->fds = NULL;
+            SeqDestroy(reactor_context->fds);
+            reactor_context->fds = NULL;
             return false;
         }
 
+        // num_nova_fds can never end up less than max_nova_fds here: Nova
+        // asserts internally that it always reports back the same fd count
+        // it advertises as its max (see the assert on poll_fd_idx in
+        // SetupEventProcessing(), nova/reactor-plugin/cf-reactor.c),
+        // so this loop always appends exactly max_nova_fds entries.
         for (size_t i = 0; i < num_nova_fds; i++)
         {
-            SeqAppend(ctx->fds, ReactorFdNew(nova_fds[i], REACTOR_FD_NOVA));
+            SeqAppend(reactor_context->fds, ReactorFdNew(nova_fds[i], REACTOR_FD_NOVA));
         }
         free(nova_fds);
     }
 
-    // TODO: initialize other event sources here.
-
+    // Initialize watcher fd
+    {
+        int watcher_fd;
+        if (!EventWatcherInitialize(&watcher_fd))
+        {
+            ReactorNovaFinalize();
+            WatcherRegistryFinalize();
+            SeqDestroy(reactor_context->fds);
+            reactor_context->fds = NULL;
+            return false;
+        }
+        SeqAppend(reactor_context->fds, ReactorFdNew(watcher_fd, REACTOR_FD_WATCHER));
+    }
+
     return true;
 }
 
-int ReactorContextSetupFileDescriptors(ReactorContext *ctx)
+int ReactorContextSetupFileDescriptors(ReactorContext *reactor_context)
 {
-    assert(ctx != NULL);
+    assert(reactor_context != NULL);
 
-    FD_ZERO(&ctx->readfds);
+    FD_ZERO(&reactor_context->readfds);
     int signal_pipe = GetSignalPipe();
-    FD_SET(signal_pipe, &ctx->readfds);
+    FD_SET(signal_pipe, &reactor_context->readfds);
 
     int max_fd = signal_pipe;
-    for (size_t i = 0; i < SeqLength(ctx->fds); i++)
+    for (size_t i = 0; i < SeqLength(reactor_context->fds); i++)
     {
-        const ReactorFd *rfd = SeqAt(ctx->fds, i);
-        FD_SET(rfd->fd, &ctx->readfds);
+        const ReactorFd *rfd = SeqAt(reactor_context->fds, i);
+        FD_SET(rfd->fd, &reactor_context->readfds);
         max_fd = MAX(rfd->fd, max_fd);
     }
     return max_fd + 1;
 }
 
-static bool NovaHasTimedOut(const ReactorContext *ctx)
+static bool NovaHasTimedOut(const ReactorContext *reactor_context)
 {
-    assert(ctx != NULL);
-    for (size_t i = 0; i < SeqLength(ctx->fds); i++)
+    assert(reactor_context != NULL);
+    for (size_t i = 0; i < SeqLength(reactor_context->fds); i++)
     {
-        const ReactorFd *rfd = SeqAt(ctx->fds, i);
+        const ReactorFd *rfd = SeqAt(reactor_context->fds, i);
 
         if (rfd->type != REACTOR_FD_NOVA)
         {
             continue;
         }
 
-        if (FD_ISSET(rfd->fd, &ctx->readfds))
+        if (FD_ISSET(rfd->fd, &reactor_context->readfds))
         {
             return false;
         }
     }
     return true;
 }
 
-static int *GetNovaFds(const ReactorContext *ctx)
+static int *GetNovaFds(const ReactorContext *reactor_context)
 {
-    assert(ctx != NULL);
-    int *nova_fds = (int *) xmalloc(ctx->max_nova_fds * sizeof(int));
+    assert(reactor_context != NULL);
+    int *nova_fds = (int *) xmalloc(reactor_context->max_nova_fds * sizeof(int));
     size_t num_nova_fds = 0;
 
-    for (size_t i = 0; i < SeqLength(ctx->fds); i++)
+    for (size_t i = 0; i < SeqLength(reactor_context->fds); i++)
     {
-        const ReactorFd *rfd = SeqAt(ctx->fds, i);
+        const ReactorFd *rfd = SeqAt(reactor_context->fds, i);
 
         if (rfd->type != REACTOR_FD_NOVA)
         {
             continue;
         }
-        // Since we came so far, this should be always true
-        assert(num_nova_fds < ctx->max_nova_fds);
+        // This can never go out of bounds: reactor_context->fds always
+        // holds exactly max_nova_fds REACTOR_FD_NOVA entries (see the
+        // comment in ReactorContextInitialize()), so num_nova_fds cannot
+        // exceed max_nova_fds and nova_fds is always fully populated by
+        // the time this function returns.
+        assert(num_nova_fds < reactor_context->max_nova_fds);
         nova_fds[num_nova_fds++] = rfd->fd;
     }
 
     return nova_fds;
 }
 
-static void SetNovaFds(ReactorContext *ctx, const int *nova_fds)
+static void SetNovaFds(ReactorContext *reactor_context, const int *nova_fds)
 {
-    assert(ctx != NULL);
+    assert(reactor_context != NULL);
     size_t num_nova_fds = 0;
 
-    for (size_t i = 0; i < SeqLength(ctx->fds); i++)
+    for (size_t i = 0; i < SeqLength(reactor_context->fds); i++)
     {
-        ReactorFd *rfd = SeqAt(ctx->fds, i);
+        ReactorFd *rfd = SeqAt(reactor_context->fds, i);
 
         if (rfd->type != REACTOR_FD_NOVA)
         {
@@ -145,21 +169,39 @@ static void SetNovaFds(ReactorContext *ctx, const int *nova_fds)
         rfd->fd = nova_fds[num_nova_fds++];
     }
 }
+static int GetWatcherFd(const ReactorContext *reactor_context)
+{
+    assert(reactor_context != NULL);
+    for (size_t i = 0; i < SeqLength(reactor_context->fds); i++)
+    {
+        const ReactorFd *rfd = SeqAt(reactor_context->fds, i);
+
+        if (rfd->type == REACTOR_FD_WATCHER)
+        {
+            return rfd->fd;
+        }
+    }
+    /**
+     * We should never reach this point: ReactorContextInitialize would fail if no watcher fd is set up.
+     */
+    ProgrammingError("Couldn't find the watcher fd");
+    return 0;
+}
 
-void ReactorContextHandleEvents(ReactorContext *ctx, time_t *next_tick)
+void ReactorContextHandleEvents(ReactorContext *reactor_context, time_t *next_tick)
 {
-    assert(ctx != NULL);
+    assert(reactor_context != NULL);
 
-    if (NovaHasTimedOut(ctx))
+    if (NovaHasTimedOut(reactor_context))
     {
         ReactorNovaHandleTimeout(next_tick);
     }
     else
     {
-        int *nova_fds = GetNovaFds(ctx);
-        ReactorNovaHandleEvents(&ctx->readfds, nova_fds, next_tick);
+        int *nova_fds = GetNovaFds(reactor_context);
+        ReactorNovaHandleEvents(&reactor_context->readfds, nova_fds, next_tick);
         // ReactorNova replaces the fd of broken connection
-        SetNovaFds(ctx, nova_fds);
+        SetNovaFds(reactor_context, nova_fds);
 
         free(nova_fds);
     }
@@ -168,22 +210,23 @@ void ReactorContextHandleEvents(ReactorContext *ctx, time_t *next_tick)
     * 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(), &ctx->readfds))
+    if (FD_ISSET(GetSignalPipe(), &reactor_context->readfds))
     {
         unsigned char buf;
         while (recv(GetSignalPipe(), &buf, 1, 0) > 0) { /* drain */ }
     }
 
-    // TODO: handle events for other event sources here.
+    EventWatcherHandleEvents(GetWatcherFd(reactor_context), &reactor_context->readfds);
 }
 
-void ReactorContextFinalize(ReactorContext *ctx)
+void ReactorContextFinalize(ReactorContext *reactor_context)
 {
-    assert(ctx != NULL);
+    assert(reactor_context != NULL);
 
+    EventWatcherFinalize();
     ReactorNovaFinalize();
-    ctx->max_nova_fds = 0;
+    reactor_context->max_nova_fds = 0;
 
-    SeqDestroy(ctx->fds);
-    ctx->fds = NULL;
+    SeqDestroy(reactor_context->fds);
+    reactor_context->fds = NULL;
 }

cf-reactor/reactor_context.h

@@ -30,7 +30,8 @@
 
 typedef enum
 {
-  REACTOR_FD_NOVA
+  REACTOR_FD_NOVA,
+  REACTOR_FD_WATCHER
 } ReactorFdType;
 
 /**