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

cf-reactor/Makefile.am

@@ -39,7 +39,10 @@ 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 \
+	stoppable_thread.c stoppable_thread.h \
+	wakeup_channel.c wakeup_channel.h
 
 if !BUILTIN_EXTENSIONS
  bin_PROGRAMS = cf-reactor

cf-reactor/README.md

@@ -6,7 +6,7 @@
 
 cf-reactor periodically reads the policy to set up the in-memory datastructures needed to react to events. It evaluates reactor bundles and parses events promises to create the corresponding `watchers`. These watchers stay active until the next policy read, when they are rebuilt.
 
-Events promises are parsed into watcher instances, each holding: the event `key` from the events promise, an event `type` (e.g. file deletion), a `payload` with whatever state the check needs (e.g. the name of the file being watched), and a `check function` that detects the event by comparing the current state to the state recorded at the last check (e.g. whether the file still exists). It returns `true` only on the specific transition being watched for, so it fires once per change and does not stay `true` afterwards.
+Events promises are parsed into watcher instances, each holding: the event `key` from the events promise, an event `type` (e.g. file deletion), a `state` with whatever state the check needs (e.g. the name of the file being watched), and a `check function` that detects the event by comparing the current state to the state recorded at the last check (e.g. whether the file still exists). It returns `true` only on the specific transition being watched for, so it fires once per change and does not stay `true` afterwards.
 
 ### Events
 
@@ -26,28 +26,11 @@ 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 reads the policy periodically, and when it does, rebuilds the list of watchers and the hashmap using the single function `WatcherRegister(key, event_type, payload, bundle, interval)`. Each events promise is associated with an event type, which is defined in `when` bodies:
+cf-reactor reads the policy periodically, and when it does, rebuilds the list of watchers and the hashmap using the single function `WatcherRegister(key, event_type, state, bundle, interval)`. Each events promise is associated with an event type, which is defined in `when` bodies:
 
 ```cf3
 body when file_deleted(filename)
@@ -57,11 +40,11 @@ body when file_deleted(filename)
 ```
 
 Every event type must have defined:
-- A check function (called `check_fn` in `Watcher`): This is a function defined specifically for the event that checks if the conditions holds. For example, in case of file deletion, we check if the file doesn't exist anymore compare to the last time we checked. If yes, then it returns `true`.
-- A `payload`: This is a struct whose interpretation depends on the event type (thus being declared as `void *`). We typically need some state that we compare between each event-check. In the case of file deletion, we need to know the name of the file we are watching, and whether the file existed last time we checked.
-- A payload destroying function (called `destroy_payload`): This is simply a function to free the payload associated with the event type.
+- A check function (called `check_callback` in `Watcher`): This is a function defined specifically for the event that checks if the conditions holds. For example, in case of file deletion, we check if the file doesn't exist anymore compare to the last time we checked. If yes, then it returns `true`.
+- A `state`: This is a struct whose interpretation depends on the event type (thus being declared as `void *`). We typically need some state that we compare between each event-check. In the case of file deletion, we need to know the name of the file we are watching, and whether the file existed last time we checked.
+- A state destroying function (called `destroy_state`): This is simply a function to free the state associated with the event type.
 
-Also, we need a function that will create the payload. That's what `FileWatcherPayloadNew()` does. 
+Also, we need a function that will create the state. That's what `FileWatcherPayloadNew()` does.
 
 So each event type we add in the future just need to have these four things defined, and we need to create the `Watcher` object with the right functions inside `WatcherRegister()` and also call the right `"...PayloadNew()"` function.
 
@@ -71,7 +54,7 @@ So each event type we add in the future just need to have these four things defi
 `ReactorContextInitialize()` sets up all the necessary data structures for polling, and then starts `WatcherThreadMain`, which polls for events as follows:
 
 - It iterates through each watcher in the global list of watchers.
-- If the elapsed time exceeds the watcher's `interval`, it runs `check_fn` to determine whether an event has been triggered.
+- If the elapsed time exceeds the watcher's `interval`, it runs `check_callback` to determine whether an event has been triggered.
 - If an event was triggered, it pushes the watcher's `key` (the promise name) onto a thread-safe queue, then signals the file descriptor via `WakeupChannelNotify`, which `select(2)` will pick up on its next iteration. It then goes back to sleep.
 
 In parallel, `EventWatcherHandleEvents`, called from within `ReactorContextHandleEvents`, reads from the file descriptor with `WakeupChannelReadFd()` once notified that an event has occurred, and pops the thread-safe queue until it's empty. Each key popped from the queue is looked up in the global hashmap to retrieve the corresponding bundle, which `cf-reactor` then runs (in another thread or subprocess)

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,130 @@ 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;
+            WatcherRegistryFinalize();
             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 event 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 +170,36 @@ 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;
+        }
+    }
+    ProgrammingError("Reactor was not initialized with event 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 +208,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;
 
 /**
@@ -52,9 +53,9 @@ typedef struct
   size_t max_nova_fds;
 } ReactorContext;
 
-bool ReactorContextInitialize(ReactorContext *ctx);
-int ReactorContextSetupFileDescriptors(ReactorContext *ctx);
-void ReactorContextHandleEvents(ReactorContext *ctx, time_t *next_tick);
-void ReactorContextFinalize(ReactorContext *ctx);
+bool ReactorContextInitialize(ReactorContext *reactor_context);
+int ReactorContextSetupFileDescriptors(ReactorContext *reactor_context);
+void ReactorContextHandleEvents(ReactorContext *reactor_context, time_t *next_tick);
+void ReactorContextFinalize(ReactorContext *reactor_context);
 
 #endif