diff --git a/cf-reactor/Makefile.am b/cf-reactor/Makefile.am index c6cd6f2fa8b..ddd49dc0f0f 100644 --- a/cf-reactor/Makefile.am +++ b/cf-reactor/Makefile.am @@ -38,7 +38,10 @@ AM_CFLAGS = $(CF3_CFLAGS) \ libcf_reactor_la_LIBADD = ../libpromises/libpromises.la libcf_reactor_la_SOURCES = \ - cf-reactor.c + cf-reactor.c cf-reactor.h \ + reactor_context.c reactor_context.h \ + watcher.c watcher.h \ + wakeup_channel.c wakeup_channel.h if !BUILTIN_EXTENSIONS bin_PROGRAMS = cf-reactor diff --git a/cf-reactor/cf-reactor.c b/cf-reactor/cf-reactor.c index 05783808f05..e6825c0d2e7 100644 --- a/cf-reactor/cf-reactor.c +++ b/cf-reactor/cf-reactor.c @@ -36,7 +36,7 @@ #include /* signal, kill */ #include /* GetSignalPipe, MakeSignalPipe, IsPendingTermination, HandleSignalsForDaemon */ #include -#include /* xmalloc */ +#include /*****************************************************************************/ /* Globals */ @@ -190,38 +190,6 @@ 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); @@ -268,28 +236,16 @@ int main(int argc, char *argv[]) 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)) + ReactorContext reactor_ctx; + if (!ReactorContextInitialize(&reactor_ctx)) { - 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(), + * rather than terminate the whole daemon. Set after ReactorContextInitialize(), * so that the spawner and the processes it execs keep the default handling. */ signal(SIGPIPE, SIG_IGN); @@ -298,22 +254,20 @@ int main(int argc, char *argv[]) time_t next_tick = time(NULL) + DEFAULT_POLL_INTERVAL_SECS; while (!IsPendingTermination()) { - fd_set readfds; - int max_fd = SetupFileDescriptors(&readfds, all_fds, num_fds); + int max_fd = ReactorContextSetupFileDescriptors(&reactor_ctx); /* 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); + int ret = select(max_fd, &reactor_ctx.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. */ + * sites of ReactorNovaHandleTimeout() 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) @@ -334,35 +288,14 @@ int main(int argc, char *argv[]) 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); + ReactorContextHandleEvents(&reactor_ctx, &next_tick); } - ReactorNovaFinalize(); - free(all_fds); + ReactorContextFinalize(&reactor_ctx); GenericAgentFinalize(ctx, config); CallCleanupFunctions(); diff --git a/cf-reactor/cf-reactor.h b/cf-reactor/cf-reactor.h new file mode 100644 index 00000000000..73aa41eb238 --- /dev/null +++ b/cf-reactor/cf-reactor.h @@ -0,0 +1,60 @@ +/* + 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_REACTOR_H +#define CFENGINE_REACTOR_H + +#include + +/** + * @brief Shared state for the cf-reactor daemon's single select(2) loop. + * + * `all_fds` is a single flat array shared by every event source the daemon + * watches. Nova's fds always occupy the first `num_nova_fds` slots (Nova + * owns that sub-range and is the only thing allowed to populate it); any + * other event source (currently just the watcher subsystem, see watcher.h) + * appends its own fd(s) after that, and `num_fds` tracks the total number + * of slots in use. Adding a new event source means: + * + * 1. Have it report how many fds it needs, and add that to the capacity + * computed in ReactorContextInitialize() (reactor_context.c). + * 2. Give it an Initialize(fds, max_size, num_fds)/HandleEvents(readfds)/ + * Finalize(void) triplet shaped like ReactorNova*() or EventWatcher*(), + * and wire the three calls into reactor_context.c next to the existing + * ones. + * + * No other file needs to know how many event sources exist or in what + * order their fds appear. + */ +typedef struct ReactorContext +{ + int *all_fds; + size_t all_fds_capacity; + size_t num_nova_fds; + size_t num_fds; + + fd_set readfds; +} ReactorContext; + +#endif diff --git a/cf-reactor/reactor_context.c b/cf-reactor/reactor_context.c new file mode 100644 index 00000000000..052fa768365 --- /dev/null +++ b/cf-reactor/reactor_context.c @@ -0,0 +1,130 @@ +/* + 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 /* ReactorNova*() */ +#include /* GetSignalPipe() */ +#include +#include +#include + +bool ReactorContextInitialize(ReactorContext *ctx) +{ + assert(ctx != NULL); + + WatcherRegistryInitialize(); + // TODO: Register watchers with `WatcherRegister()` + + size_t max_nova_fds = ReactorNovaMaxFds(); + ctx->all_fds_capacity = max_nova_fds + 1; + ctx->all_fds = xmalloc(ctx->all_fds_capacity * sizeof(int)); + + size_t num_nova_fds = 0; + if (!ReactorNovaInitialize(ctx->all_fds, max_nova_fds, &num_nova_fds)) + { + WatcherRegistryFinalize(); + free(ctx->all_fds); + ctx->all_fds = NULL; + return false; + } + ctx->num_nova_fds = num_nova_fds; + ctx->num_fds = num_nova_fds; + + if (!EventWatcherInitialize(ctx->all_fds, ctx->all_fds_capacity, &ctx->num_fds)) + { + ReactorNovaFinalize(); + WatcherRegistryFinalize(); + free(ctx->all_fds); + ctx->all_fds = NULL; + return false; + } + + return true; +} + +int ReactorContextSetupFileDescriptors(ReactorContext *ctx) +{ + assert(ctx != NULL); + + FD_ZERO(&ctx->readfds); + int signal_pipe = GetSignalPipe(); + FD_SET(signal_pipe, &ctx->readfds); + + int max_fd = signal_pipe; + for (size_t i = 0; i < ctx->num_fds; i++) + { + FD_SET(ctx->all_fds[i], &ctx->readfds); + max_fd = MAX(ctx->all_fds[i], max_fd); + } + return max_fd + 1; +} + +static bool NovaHasTimedOut(ReactorContext *ctx) +{ + assert(ctx != NULL); + for (size_t i = 0; i < ctx->num_nova_fds; i++) + { + if (FD_ISSET(ctx->all_fds[i], &ctx->readfds)) + { + return false; + } + } + return true; +} + +void ReactorContextHandleEvents(ReactorContext *ctx, time_t *next_tick) +{ + assert(ctx != NULL); + + if (NovaHasTimedOut(ctx)) + { + ReactorNovaHandleTimeout(next_tick); + } + else + { + ReactorNovaHandleEvents(&ctx->readfds, ctx->all_fds, next_tick); + } + + /* 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(), &ctx->readfds)) + { + unsigned char buf; + while (recv(GetSignalPipe(), &buf, 1, 0) > 0) { /* drain */ } + } + + EventWatcherHandleEvents(&ctx->readfds); +} + +void ReactorContextFinalize(ReactorContext *ctx) +{ + assert(ctx != NULL); + + EventWatcherFinalize(); + ReactorNovaFinalize(); + free(ctx->all_fds); + ctx->all_fds = NULL; +} diff --git a/cf-reactor/reactor_context.h b/cf-reactor/reactor_context.h new file mode 100644 index 00000000000..dab47e8e6e6 --- /dev/null +++ b/cf-reactor/reactor_context.h @@ -0,0 +1,36 @@ +/* + 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_REACTOR_CONTEXT_H +#define CFENGINE_REACTOR_CONTEXT_H + +#include + + +bool ReactorContextInitialize(ReactorContext *ctx); +int ReactorContextSetupFileDescriptors(ReactorContext *ctx); +void ReactorContextHandleEvents(ReactorContext *ctx, time_t *next_tick); +void ReactorContextFinalize(ReactorContext *ctx); + +#endif diff --git a/cf-reactor/wakeup_channel.c b/cf-reactor/wakeup_channel.c new file mode 100644 index 00000000000..99a8de9cc89 --- /dev/null +++ b/cf-reactor/wakeup_channel.c @@ -0,0 +1,106 @@ +/* + 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() */ + +bool WakeupChannelOpen(WakeupChannel *channel) +{ + assert(channel != NULL); + + channel->fds[0] = -1; + channel->fds[1] = -1; + + /* Windows' select() only works on sockets, and a plain pipe(2) isn't + * one, so this always goes through socketpair() -- which works just as + * well on POSIX -- rather than keeping two code paths in sync. This + * mirrors MakeSignalPipe() in libpromises/signals.c. */ + if (socketpair(AF_UNIX, SOCK_STREAM, 0, channel->fds) != 0) + { + Log(LOG_LEVEL_ERR, "Could not create wakeup channel (socketpair: '%s')", GetErrorStr()); + return false; + } + + for (int i = 0; i < 2; i++) + { +#ifdef __MINGW32__ + u_long enable = 1; + int ret = ioctlsocket(channel->fds[i], FIONBIO, &enable); +#define CNTLNAME "ioctlsocket" +#else + int ret = fcntl(channel->fds[i], F_SETFL, O_NONBLOCK); +#define CNTLNAME "fcntl" +#endif + if (ret != 0) + { + Log(LOG_LEVEL_ERR, "Could not set wakeup channel to non-blocking (" CNTLNAME ": '%s')", + GetErrorStr()); + WakeupChannelClose(channel); + return false; + } +#undef CNTLNAME + } + + return true; +} + +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; + send(channel->fds[1], (const char *) &byte, sizeof(byte), 0); +} + +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..ff3c28a1439 --- /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 diff --git a/cf-reactor/watcher.c b/cf-reactor/watcher.c new file mode 100644 index 00000000000..25cc61008b6 --- /dev/null +++ b/cf-reactor/watcher.c @@ -0,0 +1,257 @@ +/* + 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 /* MaskTerminationSignalsInThread() */ +#include /* IsPendingTermination() */ +#include +#include +#include /* StringHash_untyped(), StringEqual_untyped() */ +#include +#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 + +/* Not currently caller-configurable -- WatcherRegister() doesn't take an + * interval parameter -- so every watcher polls at this rate for now. */ +#define DEFAULT_WATCHER_POLL_INTERVAL_SECS 10 + +typedef struct +{ + char *key; + WatcherCheckFn check_fn; /* resolved from `type` at WatcherRegister() time */ + WatcherPayloadDestroyFn destroy_payload; + void *payload; + time_t poll_interval_secs; + time_t next_due; +} Watcher; + +static void WatcherDestroy(void *item); /* defined below, next to WatcherRegister() */ + +static Seq *watchers = NULL; +static Map *event_to_bundle = NULL; + +static WakeupChannel wakeup_channel; +static ThreadedQueue *event_queue = NULL; +static pthread_t watcher_thread; + +/*****************************************************************************/ + +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; +} + +/*****************************************************************************/ +/* WatcherRegister() / WatcherDestroy() -- create and destroy one Watcher. */ +/*****************************************************************************/ + +void WatcherRegister(const char *key, EventType type, void *payload, Bundle *bundle, time_t interval) +{ + assert(key != NULL); + assert(bundle != NULL); + assert(watchers != NULL && event_to_bundle != NULL); + + WatcherCheckFn check_fn; + WatcherPayloadDestroyFn destroy_payload; + switch (type) + { + + // TODO: add more cases + + 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_payload != NULL) + { + destroy_payload(payload); + } + return; + } + + Watcher *w = xmalloc(sizeof(Watcher)); + w->key = xstrdup(key); + w->payload = payload; + w->poll_interval_secs = interval; + w->next_due = 0; /* due immediately on the watcher thread's first pass */ + w->check_fn = check_fn; + w->destroy_payload = destroy_payload; + + SeqAppend(watchers, w); + MapInsert(event_to_bundle, w->key, bundle); +} + +static void WatcherDestroy(void *item) +{ + Watcher *w = item; + if (w->destroy_payload != NULL) + { + w->destroy_payload(w->payload); + } + free(w->key); + free(w); +} + +/*****************************************************************************/ +/* Watcher thread */ +/*****************************************************************************/ + +static void *WatcherThreadMain(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()) + { + 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); + + if (now >= w->next_due) + { + bool fired = w->check_fn(w->payload); + w->next_due = now + w->poll_interval_secs; + if (fired) + { + 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); + } + + if (sleep_for > 0) + { + sleep((unsigned int) sleep_for); + } + } + + return NULL; +} + +/*****************************************************************************/ +/* Subsystem lifecycle */ +/*****************************************************************************/ + +bool EventWatcherInitialize(int *fds, size_t max_size, size_t *num_fds) +{ + assert(fds != NULL); + assert(num_fds != NULL); + assert(watchers != NULL && event_to_bundle != NULL); /* WatcherRegistryInitialize() first */ + + if (*num_fds >= max_size) + { + ProgrammingError("No fd slots left for the reactor watcher subsystem (allocated %zu)", max_size); + } + + if (!WakeupChannelOpen(&wakeup_channel)) + { + return false; + } + + event_queue = ThreadedQueueNew(16, NULL); + + int ret = pthread_create(&watcher_thread, NULL, WatcherThreadMain, NULL); + if (ret != 0) + { + Log(LOG_LEVEL_ERR, "Unable to start cf-reactor watcher thread: %s", GetErrorStrFromCode(ret)); + ThreadedQueueDestroy(event_queue); + event_queue = NULL; + WakeupChannelClose(&wakeup_channel); + return false; + } + + fds[*num_fds] = WakeupChannelReadFd(&wakeup_channel); + (*num_fds)++; + + Log(LOG_LEVEL_VERBOSE, "Started reactor watcher subsystem with %zu watcher(s)", SeqLength(watchers)); + return true; +} + +void EventWatcherHandleEvents(fd_set *readfds) +{ + assert(readfds != NULL); + if (!FD_ISSET(WakeupChannelReadFd(&wakeup_channel), 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); /* never NULL, see the invariant above */ + Log(LOG_LEVEL_NOTICE, "Reactor watcher '%s' fired", key); + // TODO: run bundle + } +} + +void EventWatcherFinalize(void) +{ + pthread_join(watcher_thread, NULL); + 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..f806082e476 --- /dev/null +++ b/cf-reactor/watcher.h @@ -0,0 +1,56 @@ +/* + 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 +#include /* Bundle */ + +typedef enum +{ + EVENT_FILE_DELETED, +} EventType; + +typedef bool (*WatcherCheckFn)(void *payload); +typedef void (*WatcherPayloadDestroyFn)(void *payload); + +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 payload 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 *payload, Bundle *bundle, time_t interval); +bool EventWatcherInitialize(int *fds, size_t max_size, size_t *num_fds); +void EventWatcherHandleEvents(fd_set *readfds); +void EventWatcherFinalize(void); + +#endif