diff --git a/cf-reactor/Makefile.am b/cf-reactor/Makefile.am index 443e919f112..3dc15308837 100644 --- a/cf-reactor/Makefile.am +++ b/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 diff --git a/cf-reactor/README.md b/cf-reactor/README.md index 261123b4340..0e9c41ad481 100644 --- a/cf-reactor/README.md +++ b/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) diff --git a/cf-reactor/reactor_context.c b/cf-reactor/reactor_context.c index 71f7b7bc3a1..907fe6d5974 100644 --- a/cf-reactor/reactor_context.c +++ b/cf-reactor/reactor_context.c @@ -25,6 +25,7 @@ #include #include /* ReactorNova*() */ #include /* GetSignalPipe() */ +#include #include #define INIT_FD_COUNT 8 @@ -37,69 +38,89 @@ 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; } @@ -107,36 +128,40 @@ static bool NovaHasTimedOut(const ReactorContext *ctx) 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; } diff --git a/cf-reactor/reactor_context.h b/cf-reactor/reactor_context.h index 632664a5f74..3950b94696a 100644 --- a/cf-reactor/reactor_context.h +++ b/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 diff --git a/cf-reactor/stoppable_thread.c b/cf-reactor/stoppable_thread.c new file mode 100644 index 00000000000..5920bfacb89 --- /dev/null +++ b/cf-reactor/stoppable_thread.c @@ -0,0 +1,147 @@ +/* + Copyright 2026 Northern.tech AS + + This file is part of CFEngine 3 - written and maintained by Northern.tech AS. + + This program is free software; you can redistribute it and/or modify it + under the terms of the GNU General Public License as published by the + Free Software Foundation; version 3. + + This program is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + GNU General Public License for more details. + + You should have received a copy of the GNU General Public License + along with this program; if not, write to the Free Software + Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA + + To the extent this program is licensed as part of the Enterprise + versions of CFEngine, the applicable Commercial Open Source License + (COSL) may apply to this file if you as a licensee so wish it. See + included file COSL.txt. +*/ + +#include +#include +#include +#include /* ThreadLock(), ThreadUnlock(), ThreadWait() */ +#include + +struct StoppableThread_ +{ + pthread_t thread; + StoppableThreadFn fn; + void *arg; + + /* Protects stop_requested and exited. cond is signalled when either of + * them changes. */ + pthread_mutex_t lock; + pthread_cond_t cond; + bool stop_requested; /* set by StoppableThreadStop() */ + bool exited; /* set when fn returns */ +}; + +static void StoppableThreadDestroy(StoppableThread *thread) +{ + assert(thread != NULL); + pthread_cond_destroy(&thread->cond); + pthread_mutex_destroy(&thread->lock); + free(thread); +} + +static void *StoppableThreadMain(void *data) +{ + StoppableThread *thread = data; + + thread->fn(thread, thread->arg); + + ThreadLock(&thread->lock); + thread->exited = true; + pthread_cond_broadcast(&thread->cond); + ThreadUnlock(&thread->lock); + + return NULL; +} + +StoppableThread *StoppableThreadStart(StoppableThreadFn fn, void *arg) +{ + assert(fn != NULL); + + StoppableThread *thread = xcalloc(1, sizeof(StoppableThread)); + thread->fn = fn; + thread->arg = arg; + pthread_mutex_init(&thread->lock, NULL); + pthread_cond_init(&thread->cond, NULL); + + int ret = pthread_create(&thread->thread, NULL, StoppableThreadMain, thread); + if (ret != 0) + { + Log(LOG_LEVEL_ERR, "Unable to start thread: %s", GetErrorStrFromCode(ret)); + StoppableThreadDestroy(thread); + return NULL; + } + + return thread; +} + +bool StoppableThreadShouldStop(StoppableThread *thread) +{ + assert(thread != NULL); + + ThreadLock(&thread->lock); + bool ret = thread->stop_requested; + ThreadUnlock(&thread->lock); + return ret; +} + +void StoppableThreadSleep(StoppableThread *thread, time_t secs) +{ + assert(thread != NULL); + + ThreadLock(&thread->lock); + if (!thread->stop_requested && secs > 0) + { + ThreadWait(&thread->cond, &thread->lock, (int) secs); + } + ThreadUnlock(&thread->lock); +} + +bool StoppableThreadStop(StoppableThread *thread, int timeout_secs) +{ + assert(thread != NULL); + + ThreadLock(&thread->lock); + thread->stop_requested = true; + pthread_cond_broadcast(&thread->cond); + + const time_t deadline = time(NULL) + timeout_secs; + time_t now; + while (!thread->exited && (now = time(NULL)) < deadline) + { + ThreadWait(&thread->cond, &thread->lock, (int) (deadline - now)); + } + const bool exited = thread->exited; + ThreadUnlock(&thread->lock); + + if (exited) + { + pthread_join(thread->thread, NULL); + StoppableThreadDestroy(thread); + return true; + } + + Log(LOG_LEVEL_ERR, "Thread did not exit within %d seconds, cancelling it", timeout_secs); +#ifdef HAVE_PTHREAD_CANCEL + int ret = pthread_cancel(thread->thread); + if (ret != 0) + { + Log(LOG_LEVEL_ERR, "Failed to cancel thread: %s", GetErrorStrFromCode(ret)); + } +#endif + /* Cancellation is only acted upon at a cancellation point, which the + * thread may never reach, so don't risk blocking in pthread_join(). The + * handle is leaked on purpose: the thread may still be using it. */ + pthread_detach(thread->thread); + return false; +} diff --git a/cf-reactor/stoppable_thread.h b/cf-reactor/stoppable_thread.h new file mode 100644 index 00000000000..59b9d5bf80f --- /dev/null +++ b/cf-reactor/stoppable_thread.h @@ -0,0 +1,76 @@ +/* + Copyright 2026 Northern.tech AS + + This file is part of CFEngine 3 - written and maintained by Northern.tech AS. + + This program is free software; you can redistribute it and/or modify it + under the terms of the GNU General Public License as published by the + Free Software Foundation; version 3. + + This program is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + GNU General Public License for more details. + + You should have received a copy of the GNU General Public License + along with this program; if not, write to the Free Software + Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA + + To the extent this program is licensed as part of the Enterprise + versions of CFEngine, the applicable Commercial Open Source License + (COSL) may apply to this file if you as a licensee so wish it. See + included file COSL.txt. +*/ + +#ifndef CFENGINE_STOPPABLE_THREAD_H +#define CFENGINE_STOPPABLE_THREAD_H + +#include + +/** + * @brief A background thread that the thread that started it can ask to stop, + * and give up on if it doesn't stop in time. + * + * The thread routine is expected to loop until StoppableThreadShouldStop() + * returns true, and to sleep with StoppableThreadSleep() so that + * StoppableThreadStop() can wake it up immediately. + */ +typedef struct StoppableThread_ StoppableThread; + +typedef void (*StoppableThreadFn)(StoppableThread *thread, void *arg); + +/** + * @brief Start a thread running fn(thread, arg). + * @return the thread handle, or NULL on failure (the cause is logged). + */ +StoppableThread *StoppableThreadStart(StoppableThreadFn fn, void *arg); + +/** + * @brief To be called from the thread routine. + * @return true once StoppableThreadStop() has been called. + */ +bool StoppableThreadShouldStop(StoppableThread *thread); + +/** + * @brief To be called from the thread routine. Sleep for up to secs seconds, + * returning early if StoppableThreadStop() is called. Returns immediately if + * secs <= 0 or a stop has already been requested. + */ +void StoppableThreadSleep(StoppableThread *thread, time_t secs); + +/** + * @brief Ask the thread to stop and wait up to timeout_secs for it to return. + * + * If the thread doesn't return in time, it's cancelled (where supported) and + * detached, since a thread stuck outside a cancellation point would otherwise + * block pthread_join() forever. + * + * In both cases the handle must not be used afterwards. + * + * @return true if the thread returned and was joined, false if it was + * abandoned. In the latter case it may still be running, so anything + * it uses must not be freed. + */ +bool StoppableThreadStop(StoppableThread *thread, int timeout_secs); + +#endif /* CFENGINE_STOPPABLE_THREAD_H */ diff --git a/cf-reactor/wakeup_channel.c b/cf-reactor/wakeup_channel.c new file mode 100644 index 00000000000..e2aa7e740dd --- /dev/null +++ b/cf-reactor/wakeup_channel.c @@ -0,0 +1,92 @@ +/* + Copyright 2026 Northern.tech AS + + This file is part of CFEngine 3 - written and maintained by Northern.tech AS. + + This program is free software; you can redistribute it and/or modify it + under the terms of the GNU General Public License as published by the + Free Software Foundation; version 3. + + This program is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + GNU General Public License for more details. + + You should have received a copy of the GNU General Public License + along with this program; if not, write to the Free Software + Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA + + To the extent this program is licensed as part of the Enterprise + versions of CFEngine, the applicable Commercial Open Source License + (COSL) may apply to this file if you as a licensee so wish it. See + included file COSL.txt. +*/ + +#include +#include +#include /* cf_closesocket() */ +#include /* MakeNonBlockingSocketPair() */ + +bool WakeupChannelOpen(WakeupChannel *channel) +{ + assert(channel != NULL); + + return MakeNonBlockingSocketPair(channel->fds); +} + +int WakeupChannelReadFd(const WakeupChannel *channel) +{ + assert(channel != NULL); + return channel->fds[0]; +} + +void WakeupChannelNotify(const WakeupChannel *channel) +{ + assert(channel != NULL); + + /* One byte is enough to wake the reader up; if the channel happens to + * already be full, the reader is already guaranteed to wake up because + * of what's queued, so a transient EAGAIN/EWOULDBLOCK here is fine. */ + unsigned char byte = 1; + if (send(channel->fds[1], (const char *) &byte, sizeof(byte), 0) < 0) + { + // These signal contention. Everything else is an error. + if (errno != EAGAIN +#ifndef __MINGW32__ + && errno != EWOULDBLOCK +#endif + ) + { + /* Not fatal: unlike SignalNotify(), this doesn't run in a + * signal handler, so there's no async-signal-safety reason to + * _exit() here. But it's still worth logging loudly: on a real + * failure the reactor's select() loop has no other way to + * notice a fired watcher event, since a select() timeout only + * re-arms ReactorNova, it doesn't drain event_queue (see + * EventWatcherHandleEvents()). */ + Log(LOG_LEVEL_ERR, "Could not notify wakeup channel (send: '%s')", GetErrorStr()); + } + } +} + +void WakeupChannelDrain(const WakeupChannel *channel) +{ + assert(channel != NULL); + + unsigned char buf; + while (recv(channel->fds[0], (char *) &buf, sizeof(buf), 0) > 0) { /* drain */ } +} + +void WakeupChannelClose(WakeupChannel *channel) +{ + assert(channel != NULL); + + for (int i = 0; i < 2; i++) + { + if (channel->fds[i] != -1) + { + cf_closesocket(channel->fds[i]); + channel->fds[i] = -1; + } + } +} diff --git a/cf-reactor/wakeup_channel.h b/cf-reactor/wakeup_channel.h new file mode 100644 index 00000000000..8fd2ab197fa --- /dev/null +++ b/cf-reactor/wakeup_channel.h @@ -0,0 +1,49 @@ +/* + Copyright 2026 Northern.tech AS + + This file is part of CFEngine 3 - written and maintained by Northern.tech AS. + + This program is free software; you can redistribute it and/or modify it + under the terms of the GNU General Public License as published by the + Free Software Foundation; version 3. + + This program is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + GNU General Public License for more details. + + You should have received a copy of the GNU General Public License + along with this program; if not, write to the Free Software + Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA + + To the extent this program is licensed as part of the Enterprise + versions of CFEngine, the applicable Commercial Open Source License + (COSL) may apply to this file if you as a licensee so wish it. See + included file COSL.txt. +*/ + +#ifndef CFENGINE_WAKEUP_CHANNEL_H +#define CFENGINE_WAKEUP_CHANNEL_H + +#include + +/** + * @brief A cross-platform self-pipe: lets a background thread wake up the + * daemon's select(2) loop on demand. + * + * Any current or future background event source (the watcher subsystem + * today, potentially others later) that needs to interrupt select() should + * own one of these rather than inventing its own pipe/socketpair handling. + */ +typedef struct +{ + int fds[2]; /* [0] = read end, add to the select() fd_set; [1] = write end */ +} WakeupChannel; + +bool WakeupChannelOpen(WakeupChannel *channel); +int WakeupChannelReadFd(const WakeupChannel *channel); +void WakeupChannelNotify(const WakeupChannel *channel); +void WakeupChannelDrain(const WakeupChannel *channel); +void WakeupChannelClose(WakeupChannel *channel); + +#endif /* CFENGINE_WAKEUP_CHANNEL_H */ diff --git a/cf-reactor/watcher.c b/cf-reactor/watcher.c new file mode 100644 index 00000000000..28fda475cdd --- /dev/null +++ b/cf-reactor/watcher.c @@ -0,0 +1,254 @@ +/* + Copyright 2026 Northern.tech AS + + This file is part of CFEngine 3 - written and maintained by Northern.tech AS. + + This program is free software; you can redistribute it and/or modify it + under the terms of the GNU General Public License as published by the + Free Software Foundation; version 3. + + This program is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + GNU General Public License for more details. + + You should have received a copy of the GNU General Public License + along with this program; if not, write to the Free Software + Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA + + To the extent this program is licensed as part of the Enterprise + versions of CFEngine, the applicable Commercial Open Source License + (COSL) may apply to this file if you as a licensee so wish it. See + included file COSL.txt. +*/ + +#include +#include +#include +#include // MaskTerminationSignalsInThread() +#include // IsPendingTermination() +#include +#include +#include // StringHash_untyped(), StringEqual_untyped() +#include +#include +#include + +/* Upper bound on how long the watcher thread ever sleeps in one go, so that + * IsPendingTermination() is re-checked at least this often during shutdown, + * regardless of what poll intervals watchers asked for. */ +#define MAX_WATCHER_THREAD_SLEEP_SECS 1 + +#define WATCHER_THREAD_EXIT_TIMEOUT_SECS 5 + +typedef struct +{ + char *key; + WatcherCheckFn check_callback; // resolved from `type` at WatcherRegister() time + WatcherStateDestroyFn destroy_state; + void *state; + time_t poll_interval_secs; + time_t next_due; +} Watcher; + +static void WatcherDestroy(void *item); /* defined below, next to WatcherRegister() */ + +// TODO: potential race condition. If the policy is reparsed and watchers are re-registered +// while the watcher thread is iterating over this Seq, +// it may read a Watcher that is being freed or reallocated concurrently +static Seq *watchers = NULL; +static Map *event_to_bundle = NULL; + +static WakeupChannel wakeup_channel = { .fds = { -1, -1 } }; +static ThreadedQueue *event_queue = NULL; +static StoppableThread *watcher_thread = NULL; + +/*****************************************************************************/ + +void WatcherRegistryInitialize(void) +{ + assert(watchers == NULL); + assert(event_to_bundle == NULL); + + watchers = SeqNew(4, WatcherDestroy); + event_to_bundle = MapNew(StringHash_untyped, StringEqual_untyped, NULL, NULL); +} + +void WatcherRegistryFinalize(void) +{ + SeqDestroy(watchers); + watchers = NULL; + MapDestroy(event_to_bundle); + event_to_bundle = NULL; +} + +// Expects interval to be strictly greater than 0, otherwise the watcher thread will busy spin +void WatcherRegister(const char *key, EventType type, void *state, Bundle *bundle, time_t interval) +{ + assert(key != NULL); + assert(bundle != NULL); + assert(watchers != NULL && event_to_bundle != NULL); + assert(interval > 0); + + WatcherCheckFn check_callback = NULL; + WatcherStateDestroyFn destroy_state = NULL; + switch (type) + { + + case EVENT_FILE_DELETED: + // TODO: initialize check_callback and destroy_state + break; + + default: + ProgrammingError("Unknown reactor event type %d for watcher '%s'", (int) type, key); + } + + if (MapHasKey(event_to_bundle, key)) + { + Log(LOG_LEVEL_ERR, "Reactor watcher key '%s' is already registered, ignoring the duplicate", key); + if (destroy_state != NULL) + { + destroy_state(state); + } + return; + } + + Watcher *w = xmalloc(sizeof(Watcher)); + w->key = xstrdup(key); + w->state = state; + w->poll_interval_secs = interval; + w->next_due = 0; /* due immediately on the watcher thread's first pass */ + w->check_callback = check_callback; + w->destroy_state = destroy_state; + + SeqAppend(watchers, w); + MapInsert(event_to_bundle, w->key, bundle); +} + +static void WatcherDestroy(void *item) +{ + Watcher *w = item; + if (w->destroy_state != NULL) + { + w->destroy_state(w->state); + } + free(w->key); + free(w); +} + +static void WatcherThreadMain(StoppableThread *thread, ARG_UNUSED void *unused) +{ + /* Keep termination signals landing on the main thread (which owns the + * daemon's HandleSignalsForDaemon()-based shutdown), never on this one. + * No-op on Windows -- see signal_lib.h. */ +#ifndef __MINGW32__ + MaskTerminationSignalsInThread(); +#endif + + while (!IsPendingTermination() && !StoppableThreadShouldStop(thread)) + { + time_t now = time(NULL); + time_t sleep_for = MAX_WATCHER_THREAD_SLEEP_SECS; + bool any_event = false; + + for (size_t i = 0; i < SeqLength(watchers); i++) + { + Watcher *w = SeqAt(watchers, i); + + // check_callback cannot be null if it is correctly assigned during policy evaluation + assert(w->check_callback != NULL); + + if (now >= w->next_due) + { + bool fired = w->check_callback(w->state); + w->next_due = now + w->poll_interval_secs; + if (fired) + { + /** TODO: potential use after free. If a policy reparse destroys this Watcher (and frees w->key) + * before the queued key is consumed, the consumer will read freed memory */ + ThreadedQueuePush(event_queue, w->key); + any_event = true; + } + } + + time_t until_due = (w->next_due > now) ? (w->next_due - now) : 0; + sleep_for = MIN(sleep_for, until_due); + } + + if (any_event) + { + WakeupChannelNotify(&wakeup_channel); + } + + StoppableThreadSleep(thread, sleep_for); + } +} + +bool EventWatcherInitialize(int *fd) +{ + assert(fd != NULL); + assert(watchers != NULL && event_to_bundle != NULL); /* WatcherRegistryInitialize() first */ + + if (!WakeupChannelOpen(&wakeup_channel)) + { + Log(LOG_LEVEL_ERR, "Failed to open wakeup channel for cf-reactor event watcher"); + return false; + } + + event_queue = ThreadedQueueNew(16, NULL); + + watcher_thread = StoppableThreadStart(WatcherThreadMain, NULL); + if (watcher_thread == NULL) + { + Log(LOG_LEVEL_ERR, "Unable to start cf-reactor watcher thread"); + ThreadedQueueDestroy(event_queue); + event_queue = NULL; + WakeupChannelClose(&wakeup_channel); + return false; + } + + *fd = WakeupChannelReadFd(&wakeup_channel); + + Log(LOG_LEVEL_VERBOSE, "Started reactor watcher subsystem with %zu watcher(s)", SeqLength(watchers)); + return true; +} + +void EventWatcherHandleEvents(int fd, fd_set *readfds) +{ + assert(readfds != NULL); + + if (!FD_ISSET(fd, readfds)) + { + return; + } + + WakeupChannelDrain(&wakeup_channel); + + void *item; + while (ThreadedQueuePop(event_queue, &item, 0)) + { + const char *key = item; + ARG_UNUSED const Bundle *bundle = MapGet(event_to_bundle, key); + Log(LOG_LEVEL_NOTICE, "Reactor watcher '%s' fired", key); + // TODO: run bundle + } +} + +void EventWatcherFinalize(void) +{ + bool joined = StoppableThreadStop(watcher_thread, WATCHER_THREAD_EXIT_TIMEOUT_SECS); + watcher_thread = NULL; + if (!joined) + { + /* The thread may still be running and using the event queue, the + * wakeup channel and the watcher registry, so leak them rather than + * risk a use-after-free. We're shutting down anyway. */ + return; + } + + ThreadedQueueDestroy(event_queue); + event_queue = NULL; + WakeupChannelClose(&wakeup_channel); + + WatcherRegistryFinalize(); +} diff --git a/cf-reactor/watcher.h b/cf-reactor/watcher.h new file mode 100644 index 00000000000..4f458386e84 --- /dev/null +++ b/cf-reactor/watcher.h @@ -0,0 +1,55 @@ +/* + Copyright 2026 Northern.tech AS + + This file is part of CFEngine 3 - written and maintained by Northern.tech AS. + + This program is free software; you can redistribute it and/or modify it + under the terms of the GNU General Public License as published by the + Free Software Foundation; version 3. + + This program is distributed in the hope that it will be useful, + but WITHOUT ANY WARRANTY; without even the implied warranty of + MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + GNU General Public License for more details. + + You should have received a copy of the GNU General Public License + along with this program; if not, write to the Free Software + Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA + + To the extent this program is licensed as part of the Enterprise + versions of CFEngine, the applicable Commercial Open Source License + (COSL) may apply to this file if you as a licensee so wish it. See + included file COSL.txt. +*/ + +#ifndef CFENGINE_WATCHER_H +#define CFENGINE_WATCHER_H + +#include /* Bundle */ + +typedef enum +{ + EVENT_FILE_DELETED, +} EventType; + +typedef bool (*WatcherCheckFn)(void *state); +typedef void (*WatcherStateDestroyFn)(void *state); + +void WatcherRegistryInitialize(void); +void WatcherRegistryFinalize(void); + +/** + * @brief Register a specific watcher instance. + * + * @param key the events promise identifier + * @param type the type of watcher, defined in when bodies + * @param state the data used for by the watcher, depending on the type + * @param bundle the bundle to run on event + * @param interval interval between runs + */ +void WatcherRegister(const char *key, EventType type, void *state, Bundle *bundle, time_t interval); +bool EventWatcherInitialize(int *fd); +void EventWatcherHandleEvents(int fd, fd_set *readfds); +void EventWatcherFinalize(void); + +#endif diff --git a/libpromises/signals.c b/libpromises/signals.c index 21d150cd65f..75370656ffe 100644 --- a/libpromises/signals.c +++ b/libpromises/signals.c @@ -27,6 +27,7 @@ #include /* GetStateDir() */ #include /* FILE_SEPARATOR */ #include /* TerminateCustomPromises */ +#include /* cf_closesocket() */ static bool PENDING_TERMINATION = false; /* GLOBAL_X */ @@ -72,45 +73,63 @@ static void CloseSignalPipe(void) } /** - * Make a pipe that can be used to flag that a signal has arrived. - * Using a pipe avoids race conditions, since it saves its values until emptied. - * Use GetSignalPipe() to get the pipe. + * Creates an AF_UNIX SOCK_STREAM socket pair and sets both ends + * non-blocking. * Note that we use a real socket as the pipe, because Windows only supports * using select() with real sockets. This means also using send() and recv() * instead of write() and read(). */ -void MakeSignalPipe(void) +bool MakeNonBlockingSocketPair(int fds[2]) { - if (socketpair(AF_UNIX, SOCK_STREAM, 0, SIGNAL_PIPE) != 0) + fds[0] = -1; + fds[1] = -1; + + if (socketpair(AF_UNIX, SOCK_STREAM, 0, fds) != 0) { - Log(LOG_LEVEL_CRIT, "Could not create internal communication pipe. Cannot continue. (socketpair: '%s')", - GetErrorStr()); - DoCleanupAndExit(EXIT_FAILURE); + Log(LOG_LEVEL_ERR, "Could not create socket pair (socketpair: '%s')", GetErrorStr()); + return false; } - RegisterCleanupFunction(&CloseSignalPipe); - for (int c = 0; c < 2; c++) { #ifdef __MINGW32__ u_long enable = 1; - int ret = ioctlsocket(SIGNAL_PIPE[c], FIONBIO, &enable); + int ret = ioctlsocket(fds[c], FIONBIO, &enable); #define CNTLNAME "ioctlsocket" #else /* Unix: */ - int ret = fcntl(SIGNAL_PIPE[c], F_SETFL, O_NONBLOCK); + int ret = fcntl(fds[c], F_SETFL, O_NONBLOCK); #define CNTLNAME "fcntl" #endif /* __MINGW32__ */ if (ret != 0) { - Log(LOG_LEVEL_CRIT, - "Could not unblock internal communication pipe. " - "Cannot continue. (" CNTLNAME ": '%s')", - GetErrorStr()); - DoCleanupAndExit(EXIT_FAILURE); + Log(LOG_LEVEL_ERR, "Could not set socket pair to non-blocking (" CNTLNAME ": '%s')", GetErrorStr()); + cf_closesocket(fds[0]); + cf_closesocket(fds[1]); + fds[0] = -1; + fds[1] = -1; + return false; } #undef CNTLNAME } + + return true; +} + +/** + * Make a pipe that can be used to flag that a signal has arrived. + * Using a pipe avoids race conditions, since it saves its values until emptied. + * Use GetSignalPipe() to get the pipe. + */ +void MakeSignalPipe(void) +{ + if (!MakeNonBlockingSocketPair(SIGNAL_PIPE)) + { + Log(LOG_LEVEL_CRIT, "Could not create internal communication pipe. Cannot continue."); + DoCleanupAndExit(EXIT_FAILURE); + } + + RegisterCleanupFunction(&CloseSignalPipe); } /** diff --git a/libpromises/signals.h b/libpromises/signals.h index 4ddbaf60266..03b73f74b78 100644 --- a/libpromises/signals.h +++ b/libpromises/signals.h @@ -36,6 +36,14 @@ void RequestReloadConfig(void); void MakeSignalPipe(void); int GetSignalPipe(void); + +/** + * Creates an AF_UNIX SOCK_STREAM socket pair with both ends set + * non-blocking, the way MakeSignalPipe() and cf-reactor's WakeupChannel + * both need. On failure, fds[0] and fds[1] are left at -1 and the cause is + * logged at LOG_LEVEL_ERR. + */ +bool MakeNonBlockingSocketPair(int fds[2]); void HandleSignalsForDaemon(int signum); void HandleSignalsForAgent(int signum); diff --git a/tests/unit/Makefile.am b/tests/unit/Makefile.am index 696c34e29c2..2a6499c0f00 100644 --- a/tests/unit/Makefile.am +++ b/tests/unit/Makefile.am @@ -37,6 +37,7 @@ AM_CPPFLAGS = $(CORE_CPPFLAGS) \ -I$(srcdir)/../../cf-execd \ -I$(srcdir)/../../cf-key \ -I$(srcdir)/../../cf-check \ + -I$(srcdir)/../../cf-reactor \ -DTESTDATADIR='"$(srcdir)/data"' LDADD = ../../libpromises/libpromises.la libtest.la @@ -154,7 +155,10 @@ check_PROGRAMS = \ split_process_line_test \ new_packages_promise_test \ iteration_test \ - protocol_recv_overflow_test + protocol_recv_overflow_test \ + watcher_test \ + stoppable_thread_test \ + wakeup_channel_test if HAVE_AVAHI_CLIENT if HAVE_AVAHI_COMMON @@ -225,6 +229,17 @@ set_domainname_test_LDADD = libtest.la ../../libpromises/libpromises.la regex_test_SOURCES = regex_test.c ../../libpromises/match_scope.c +watcher_test_SOURCES = watcher_test.c \ + ../../cf-reactor/watcher.c \ + ../../cf-reactor/stoppable_thread.c \ + ../../cf-reactor/wakeup_channel.c + +stoppable_thread_test_SOURCES = stoppable_thread_test.c \ + ../../cf-reactor/stoppable_thread.c + +wakeup_channel_test_SOURCES = wakeup_channel_test.c \ + ../../cf-reactor/wakeup_channel.c + protocol_test_SOURCES = protocol_test.c \ ../../cf-serverd/server_common.c \ ../../cf-serverd/server_tls.c \ diff --git a/tests/unit/stoppable_thread_test.c b/tests/unit/stoppable_thread_test.c new file mode 100644 index 00000000000..c935003b52c --- /dev/null +++ b/tests/unit/stoppable_thread_test.c @@ -0,0 +1,237 @@ +#include + +#include +#include +#include +#include /* ThreadLock(), ThreadUnlock() */ + +/* How long the thread routines below wait for the test to do something + * before failing, so that a broken StoppableThread can't hang the test. */ +#define TEST_TIMEOUT_SECS 10 + +typedef struct +{ + ThreadedQueue *started; /* the routine pushes to it once running */ + bool should_stop_at_start; /* StoppableThreadShouldStop() on entry */ + bool saw_stop; /* the routine returned because of a stop */ + int iterations; +} RoutineData; + +static RoutineData *RoutineDataNew(void) +{ + RoutineData *data = xcalloc(1, sizeof(RoutineData)); + data->started = ThreadedQueueNew(1, NULL); + return data; +} + +static void RoutineDataDestroy(RoutineData *data) +{ + ThreadedQueueDestroy(data->started); + free(data); +} + +/* Block until the routine has pushed to data->started. */ +static void WaitForStart(RoutineData *data) +{ + void *item; + assert_true(ThreadedQueuePop(data->started, &item, TEST_TIMEOUT_SECS)); +} + +/* Loops until stopped, sleeping for a long time on each iteration, so it only + * returns promptly if StoppableThreadStop() wakes it up. */ +static void LoopUntilStopped(StoppableThread *thread, void *arg) +{ + RoutineData *data = arg; + data->should_stop_at_start = StoppableThreadShouldStop(thread); + ThreadedQueuePush(data->started, data); + + while (!StoppableThreadShouldStop(thread)) + { + data->iterations++; + StoppableThreadSleep(thread, 60); + } + data->saw_stop = true; +} + +/* Returns straight away without waiting for a stop. */ +static void ReturnImmediately(ARG_UNUSED StoppableThread *thread, void *arg) +{ + RoutineData *data = arg; + data->iterations++; +} + +/* Sleeping for 0 or a negative number of seconds must return straight away, + * which is checked by the test timing the whole routine. */ +static void SleepNonPositive(StoppableThread *thread, void *arg) +{ + RoutineData *data = arg; + StoppableThreadSleep(thread, 0); + StoppableThreadSleep(thread, -1); + data->iterations++; +} + +/* Sleeping after a stop was requested must return straight away. */ +static void SleepAfterStop(StoppableThread *thread, void *arg) +{ + RoutineData *data = arg; + ThreadedQueuePush(data->started, data); + + while (!StoppableThreadShouldStop(thread)) + { + StoppableThreadSleep(thread, 60); + } + StoppableThreadSleep(thread, 60); + data->saw_stop = true; +} + +static void test_stop_running_thread(void) +{ + RoutineData *data = RoutineDataNew(); + + StoppableThread *thread = StoppableThreadStart(LoopUntilStopped, data); + assert_true(thread != NULL); + WaitForStart(data); + + assert_true(StoppableThreadStop(thread, TEST_TIMEOUT_SECS)); + + /* The routine got the right arg, saw no stop before one was requested, + * and returned because of it. Reading data is safe here: the thread has + * been joined. */ + assert_false(data->should_stop_at_start); + assert_true(data->saw_stop); + assert_true(data->iterations >= 1); + + RoutineDataDestroy(data); +} + +static void test_stop_wakes_sleep(void) +{ + RoutineData *data = RoutineDataNew(); + + StoppableThread *thread = StoppableThreadStart(LoopUntilStopped, data); + assert_true(thread != NULL); + WaitForStart(data); + + /* The routine sleeps 60s per iteration, so this only returns quickly if + * the stop interrupts the sleep. */ + const time_t start = time(NULL); + assert_true(StoppableThreadStop(thread, TEST_TIMEOUT_SECS)); + assert_true(time(NULL) - start < 2); + + RoutineDataDestroy(data); +} + +static void test_stop_already_returned(void) +{ + RoutineData *data = RoutineDataNew(); + + StoppableThread *thread = StoppableThreadStart(ReturnImmediately, data); + assert_true(thread != NULL); + + /* Whether or not the routine has already returned, it gets joined. */ + assert_true(StoppableThreadStop(thread, TEST_TIMEOUT_SECS)); + assert_int_equal(data->iterations, 1); + + RoutineDataDestroy(data); +} + +static void test_sleep_non_positive(void) +{ + RoutineData *data = RoutineDataNew(); + + const time_t start = time(NULL); + StoppableThread *thread = StoppableThreadStart(SleepNonPositive, data); + assert_true(thread != NULL); + assert_true(StoppableThreadStop(thread, TEST_TIMEOUT_SECS)); + assert_true(time(NULL) - start < 2); + assert_int_equal(data->iterations, 1); + + RoutineDataDestroy(data); +} + +static void test_sleep_after_stop(void) +{ + RoutineData *data = RoutineDataNew(); + + StoppableThread *thread = StoppableThreadStart(SleepAfterStop, data); + assert_true(thread != NULL); + WaitForStart(data); + + const time_t start = time(NULL); + assert_true(StoppableThreadStop(thread, TEST_TIMEOUT_SECS)); + assert_true(time(NULL) - start < 2); + assert_true(data->saw_stop); + + RoutineDataDestroy(data); +} + +/* State for the timeout test. It's static rather than passed through arg, + * because the thread is abandoned and may outlive the test function. */ +static pthread_mutex_t ignore_stop_lock = PTHREAD_MUTEX_INITIALIZER; +static bool ignore_stop_release = false; +static ThreadedQueue *ignore_stop_started = NULL; +/* Keeps the abandoned handle reachable, so that it doesn't show up as a + * leak under valgrind. */ +static StoppableThread *abandoned_thread = NULL; + +static bool IgnoreStopReleased(void) +{ + ThreadLock(&ignore_stop_lock); + bool ret = ignore_stop_release; + ThreadUnlock(&ignore_stop_lock); + return ret; +} + +/* Ignores stop requests, like a routine stuck in a blocking call would, until + * the test releases it (or it's cancelled while in usleep()). */ +static void IgnoreStop(ARG_UNUSED StoppableThread *thread, ARG_UNUSED void *arg) +{ + ThreadedQueuePush(ignore_stop_started, NULL); + + const time_t give_up = time(NULL) + TEST_TIMEOUT_SECS; + while (!IgnoreStopReleased() && time(NULL) < give_up) + { + usleep(10000); + } +} + +static void test_stop_timeout(void) +{ + ignore_stop_started = ThreadedQueueNew(1, NULL); + + abandoned_thread = StoppableThreadStart(IgnoreStop, NULL); + assert_true(abandoned_thread != NULL); + void *item; + assert_true(ThreadedQueuePop(ignore_stop_started, &item, TEST_TIMEOUT_SECS)); + + /* The routine never checks for a stop, so this must give up after the + * timeout, not wait for the routine to return. */ + const time_t start = time(NULL); + assert_false(StoppableThreadStop(abandoned_thread, 1)); + const time_t elapsed = time(NULL) - start; + assert_true(elapsed >= 1); + assert_true(elapsed < TEST_TIMEOUT_SECS); + + /* Let the routine return in case it wasn't cancelled (no pthread_cancel() + * on this platform). ignore_stop_started is leaked on purpose, since the + * thread may not have finished with it. */ + ThreadLock(&ignore_stop_lock); + ignore_stop_release = true; + ThreadUnlock(&ignore_stop_lock); +} + +int main() +{ + PRINT_TEST_BANNER(); + const UnitTest tests[] = + { + unit_test(test_stop_running_thread), + unit_test(test_stop_wakes_sleep), + unit_test(test_stop_already_returned), + unit_test(test_sleep_non_positive), + unit_test(test_sleep_after_stop), + unit_test(test_stop_timeout), + }; + + return run_tests(tests); +} diff --git a/tests/unit/wakeup_channel_test.c b/tests/unit/wakeup_channel_test.c new file mode 100644 index 00000000000..aa9e2c6d47c --- /dev/null +++ b/tests/unit/wakeup_channel_test.c @@ -0,0 +1,164 @@ +#include + +#include +#include +#include /* LoggingPrivContext, LoggingPrivSetContext() */ + +static int captured_err_count = 0; + +static char *CaptureErrorLogHook(ARG_UNUSED LoggingPrivContext *pctx, LogLevel level, const char *message) +{ + if (level == LOG_LEVEL_ERR) + { + captured_err_count++; + } + return (char *) message; +} + +/* Non-blocking check of whether fd is readable, the way the reactor's + * select() loop sees it. */ +static bool IsReadable(int fd) +{ + fd_set readfds; + FD_ZERO(&readfds); + FD_SET(fd, &readfds); + struct timeval timeout = { 0 }; + int ret = select(fd + 1, &readfds, NULL, NULL, &timeout); + assert_true(ret >= 0); + return ret > 0 && FD_ISSET(fd, &readfds); +} + +static void test_open_close(void) +{ + WakeupChannel channel; + assert_true(WakeupChannelOpen(&channel)); + + assert_true(channel.fds[0] >= 0); + assert_true(channel.fds[1] >= 0); + assert_int_equal(WakeupChannelReadFd(&channel), channel.fds[0]); + + WakeupChannelClose(&channel); + assert_int_equal(channel.fds[0], -1); + assert_int_equal(channel.fds[1], -1); + + /* Closing again must be a no-op, not close whatever now reuses the fds. */ + WakeupChannelClose(&channel); + assert_int_equal(channel.fds[0], -1); + assert_int_equal(channel.fds[1], -1); +} + +static void test_close_unopened(void) +{ + /* A channel that was never opened must not close fd 0 (stdin). */ + WakeupChannel channel = { .fds = { -1, -1 } }; + WakeupChannelClose(&channel); + assert_int_equal(channel.fds[0], -1); + assert_int_equal(channel.fds[1], -1); +} + +static void test_not_readable_initially(void) +{ + WakeupChannel channel; + assert_true(WakeupChannelOpen(&channel)); + + assert_false(IsReadable(WakeupChannelReadFd(&channel))); + + WakeupChannelClose(&channel); +} + +static void test_notify_then_drain(void) +{ + WakeupChannel channel; + assert_true(WakeupChannelOpen(&channel)); + const int fd = WakeupChannelReadFd(&channel); + + WakeupChannelNotify(&channel); + assert_true(IsReadable(fd)); + + WakeupChannelDrain(&channel); + assert_false(IsReadable(fd)); + + WakeupChannelClose(&channel); +} + +static void test_drain_clears_multiple_notifies(void) +{ + WakeupChannel channel; + assert_true(WakeupChannelOpen(&channel)); + const int fd = WakeupChannelReadFd(&channel); + + for (int i = 0; i < 10; i++) + { + WakeupChannelNotify(&channel); + } + assert_true(IsReadable(fd)); + + /* One drain must consume every pending notification, otherwise select() + * would keep returning immediately. */ + WakeupChannelDrain(&channel); + assert_false(IsReadable(fd)); + + WakeupChannelClose(&channel); +} + +static void test_drain_empty_does_not_block(void) +{ + WakeupChannel channel; + assert_true(WakeupChannelOpen(&channel)); + + /* The channel is non-blocking, so this must return straight away. */ + WakeupChannelDrain(&channel); + assert_false(IsReadable(WakeupChannelReadFd(&channel))); + + WakeupChannelClose(&channel); +} + +static void test_notify_when_full(void) +{ + WakeupChannel channel; + assert_true(WakeupChannelOpen(&channel)); + const int fd = WakeupChannelReadFd(&channel); + + captured_err_count = 0; + LoggingPrivContext log_ctx = { .log_hook = CaptureErrorLogHook }; + LoggingPrivSetContext(&log_ctx); + + /* Far more notifications than the socket buffer holds: once it's full, + * Notify() must neither block nor log an error, since the reader is + * already guaranteed to wake up. */ + for (int i = 0; i < 1000000; i++) + { + WakeupChannelNotify(&channel); + } + + LoggingPrivSetContext(NULL); + + assert_int_equal(captured_err_count, 0); + assert_true(IsReadable(fd)); + + WakeupChannelDrain(&channel); + assert_false(IsReadable(fd)); + + /* Still usable after having been full. */ + WakeupChannelNotify(&channel); + assert_true(IsReadable(fd)); + + WakeupChannelClose(&channel); +} + +int main() +{ + PRINT_TEST_BANNER(); + const UnitTest tests[] = + { + unit_test(test_open_close), + unit_test(test_close_unopened), + unit_test(test_not_readable_initially), + unit_test(test_notify_then_drain), + unit_test(test_drain_clears_multiple_notifies), + unit_test(test_drain_empty_does_not_block), + unit_test(test_notify_when_full), + }; + + return run_tests(tests); +} diff --git a/tests/unit/watcher_test.c b/tests/unit/watcher_test.c new file mode 100644 index 00000000000..8991d7d9f53 --- /dev/null +++ b/tests/unit/watcher_test.c @@ -0,0 +1,106 @@ +#include + +#include +#include /* Bundle */ +#include /* LoggingPrivContext, LoggingPrivSetContext() */ + +#include /* strstr() */ + +static int captured_err_count = 0; +static char captured_err_message[256]; + +static char *CaptureErrorLogHook(ARG_UNUSED LoggingPrivContext *pctx, LogLevel level, const char *message) +{ + if (level == LOG_LEVEL_ERR) + { + captured_err_count++; + strlcpy(captured_err_message, message, sizeof(captured_err_message)); + } + return (char *) message; +} + +static void test_registry_initialize_finalize(void) +{ + WatcherRegistryInitialize(); + WatcherRegistryFinalize(); +} + +static void test_watcher_register_single(void) +{ + WatcherRegistryInitialize(); + + /* WatcherRegister()/WatcherDestroy() never dereference the bundle + * pointer, they only store it in a map, so a fake non-NULL pointer is + * fine here. */ + Bundle *fake_bundle = (Bundle *) 0x1; + + WatcherRegister("test-event", EVENT_FILE_DELETED, NULL, fake_bundle, 5); + + WatcherRegistryFinalize(); +} + +static void test_watcher_register_duplicate_key_ignored(void) +{ + WatcherRegistryInitialize(); + + Bundle *fake_bundle_a = (Bundle *) 0x1; + Bundle *fake_bundle_b = (Bundle *) 0x2; + + WatcherRegister("dup-event", EVENT_FILE_DELETED, NULL, fake_bundle_a, 5); + + captured_err_count = 0; + captured_err_message[0] = '\0'; + LoggingPrivContext log_ctx = { .log_hook = CaptureErrorLogHook }; + LoggingPrivSetContext(&log_ctx); + + /* Registering the same key again must be rejected: an error is logged + * (not silently swallowed) and the first registration is kept, not + * replaced. */ + WatcherRegister("dup-event", EVENT_FILE_DELETED, NULL, fake_bundle_b, 5); + + LoggingPrivSetContext(NULL); + + assert_int_equal(captured_err_count, 1); + assert_true(strstr(captured_err_message, "dup-event") != NULL); + assert_true(strstr(captured_err_message, "already registered") != NULL); + + WatcherRegistryFinalize(); +} + +static void test_event_watcher_lifecycle(void) +{ + WatcherRegistryInitialize(); + + int fd = -1; + assert_true(EventWatcherInitialize(&fd)); + assert_true(fd >= 0); + + fd_set readfds; + FD_ZERO(&readfds); + /* fd not set: must return early without touching the wakeup channel or + * the event queue. */ + EventWatcherHandleEvents(fd, &readfds); + + FD_SET(fd, &readfds); + /* fd set but nothing queued: must drain the channel (no-op) and find + * nothing to pop from the event queue. */ + EventWatcherHandleEvents(fd, &readfds); + + /* Must wake up and stop the watcher thread by itself (IsPendingTermination() + * is false here), join it, and finalize the watcher registry. */ + EventWatcherFinalize(); +} + +int main() +{ + PRINT_TEST_BANNER(); + const UnitTest tests[] = + { + unit_test(test_registry_initialize_finalize), + unit_test(test_watcher_register_single), + unit_test(test_watcher_register_duplicate_key_ignored), + unit_test(test_event_watcher_lifecycle), + }; + + return run_tests(tests); +}