From 92c395740f44641b93e26c7e048baecfbdc6f344 Mon Sep 17 00:00:00 2001 From: Victor Moene Date: Mon, 7 Sep 2026 13:44:51 +0200 Subject: [PATCH] Refactored cf-reactor event handling with one unified interface Signed-off-by: Victor Moene --- cf-reactor/Makefile.am | 3 +- cf-reactor/README.md | 7 +- cf-reactor/cf-reactor.c | 87 ++---------------- cf-reactor/reactor_context.c | 171 +++++++++++++++++++++++++++++++++++ cf-reactor/reactor_context.h | 59 ++++++++++++ 5 files changed, 243 insertions(+), 84 deletions(-) create mode 100644 cf-reactor/reactor_context.c create mode 100644 cf-reactor/reactor_context.h diff --git a/cf-reactor/Makefile.am b/cf-reactor/Makefile.am index c6cd6f2fa8b..443e919f112 100644 --- a/cf-reactor/Makefile.am +++ b/cf-reactor/Makefile.am @@ -38,7 +38,8 @@ AM_CFLAGS = $(CF3_CFLAGS) \ libcf_reactor_la_LIBADD = ../libpromises/libpromises.la libcf_reactor_la_SOURCES = \ - cf-reactor.c + cf-reactor.c \ + reactor_context.c reactor_context.h if !BUILTIN_EXTENSIONS bin_PROGRAMS = cf-reactor diff --git a/cf-reactor/README.md b/cf-reactor/README.md index 5577d56a393..261123b4340 100644 --- a/cf-reactor/README.md +++ b/cf-reactor/README.md @@ -33,12 +33,7 @@ Rather than exposing the raw file-descriptor bookkeeping required for `select(2) ```C typedef struct ReactorContext { - int *all_fds; // heap allocated array of fds - size_t all_fds_capacity; // total number of fds. number of nova fds + number of event fds - size_t num_nova_fds; // this is returned by the reactor-plugin - size_t num_fds; // this is 1 - // the first (num_nova_fds - 1) slots in the array are reserved for the reactor-plugin, the last one is reserved for the event driven code. - + Seq *fds // array of ReactorFd, which holds the fd and some metadata fd_set readfds; } ReactorContext; ``` diff --git a/cf-reactor/cf-reactor.c b/cf-reactor/cf-reactor.c index 913b05b7c6c..80169153a91 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 */ @@ -191,38 +191,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); @@ -269,28 +237,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); @@ -299,22 +255,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) @@ -335,35 +289,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/reactor_context.c b/cf-reactor/reactor_context.c new file mode 100644 index 00000000000..4225656e809 --- /dev/null +++ b/cf-reactor/reactor_context.c @@ -0,0 +1,171 @@ +/* + 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 + +#define INIT_FD_COUNT 8 + +static size_t max_nova_fds = 0; + +static ReactorFd *ReactorFdNew(int fd, ReactorFdType type) +{ + ReactorFd *rfd = xmalloc(sizeof(ReactorFd)); + rfd->fd = fd; + rfd->type = type; + return rfd; +} + +bool ReactorContextInitialize(ReactorContext *ctx) +{ + assert(ctx != NULL); + + ctx->fds = SeqNew(INIT_FD_COUNT, free); + + // Initialize Nova fds + { + max_nova_fds = ReactorNovaMaxFds(); + int *nova_fds = (int *) xmalloc(max_nova_fds * sizeof(int)); + size_t num_nova_fds = 0; + + if (!ReactorNovaInitialize(nova_fds, max_nova_fds, &num_nova_fds)) + { + free(nova_fds); + SeqDestroy(ctx->fds); + ctx->fds = NULL; + return false; + } + + for (size_t i = 0; i < num_nova_fds; i++) + { + SeqAppend(ctx->fds, ReactorFdNew(nova_fds[i], REACTOR_FD_NOVA)); + } + free(nova_fds); + } + + // TODO: initialize other event sources here. + + 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 < SeqLength(ctx->fds); i++) + { + const ReactorFd *rfd = SeqAt(ctx->fds, i); + FD_SET(rfd->fd, &ctx->readfds); + max_fd = MAX(rfd->fd, max_fd); + } + return max_fd + 1; +} + +static bool NovaHasTimedOut(const ReactorContext *ctx) +{ + assert(ctx != NULL); + for (size_t i = 0; i < SeqLength(ctx->fds); i++) + { + const ReactorFd *rfd = SeqAt(ctx->fds, i); + + if (rfd->type != REACTOR_FD_NOVA) + { + continue; + } + + if (FD_ISSET(rfd->fd, &ctx->readfds)) + { + return false; + } + } + return true; +} + +static int *GetNovaFds(Seq *fds) +{ + int *nova_fds = (int *) xmalloc(max_nova_fds * sizeof(int)); + size_t num_nova_fds = 0; + + for (size_t i = 0; i < SeqLength(fds); i++) + { + const ReactorFd *rfd = SeqAt(fds, i); + + if (rfd->type != REACTOR_FD_NOVA) + { + continue; + } + // Since we came so far, this should be always true + assert(num_nova_fds < max_nova_fds); + nova_fds[num_nova_fds++] = rfd->fd; + } + + return nova_fds; +} + +void ReactorContextHandleEvents(ReactorContext *ctx, time_t *next_tick) +{ + assert(ctx != NULL); + + if (NovaHasTimedOut(ctx)) + { + ReactorNovaHandleTimeout(next_tick); + } + else + { + int *nova_fds = GetNovaFds(ctx->fds); + ReactorNovaHandleEvents(&ctx->readfds, nova_fds, next_tick); + + free(nova_fds); + } + + /* 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 */ } + } + + // TODO: handle events for other event sources here. +} + +void ReactorContextFinalize(ReactorContext *ctx) +{ + assert(ctx != NULL); + + ReactorNovaFinalize(); + max_nova_fds = 0; + + SeqDestroy(ctx->fds); + ctx->fds = NULL; +} diff --git a/cf-reactor/reactor_context.h b/cf-reactor/reactor_context.h new file mode 100644 index 00000000000..11d9ec821c2 --- /dev/null +++ b/cf-reactor/reactor_context.h @@ -0,0 +1,59 @@ +/* + 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 +#include + +typedef enum +{ + REACTOR_FD_NOVA +} ReactorFdType; + +/** + * @brief Single file descriptor watched by daemon's select(2) loop as well as metadata of its origin + */ +typedef struct +{ + ReactorFdType type; + int fd; +} ReactorFd; + +/** + * @brief Shared state for the cf-reactor daemon's single select(2) loop. fds is an array of ReactorFd + */ +typedef struct +{ + Seq *fds; + fd_set readfds; +} ReactorContext; + +bool ReactorContextInitialize(ReactorContext *ctx); +int ReactorContextSetupFileDescriptors(ReactorContext *ctx); +void ReactorContextHandleEvents(ReactorContext *ctx, time_t *next_tick); +void ReactorContextFinalize(ReactorContext *ctx); + +#endif