From bea617d8ec344a0299edb5cee99cef130b2a4518 Mon Sep 17 00:00:00 2001 From: Victor Moene Date: Tue, 1 Sep 2026 11:34:19 +0200 Subject: [PATCH 1/2] Moved reactor-plugin code into the cf-reactor daemon The code for the reactor-plugin was split up into different functions to be integrated inside the new cf-reactor daemon in core. The implementation was tweaked to use select(2) instead of poll(2), to support cf-reactor on different platforms. However, the reactor-plugin remains linux only. Ticket: ENT-14434 Signed-off-by: Victor Moene --- cf-reactor/cf-reactor.c | 180 ++++++++++++++++++++++++++++++++- libpromises/enterprise_stubs.c | 18 +++- libpromises/prototypes3.h | 5 +- 3 files changed, 196 insertions(+), 7 deletions(-) diff --git a/cf-reactor/cf-reactor.c b/cf-reactor/cf-reactor.c index 9e9e87cb12f..c720b1a0ff5 100644 --- a/cf-reactor/cf-reactor.c +++ b/cf-reactor/cf-reactor.c @@ -33,6 +33,9 @@ #include #include #include +#include /* signal, kill */ +#include /* GetSignalPipe, MakeSignalPipe */ +#include /*****************************************************************************/ /* Globals */ @@ -40,6 +43,12 @@ int NO_FORK = false; +#define DEFAULT_POLL_INTERVAL_SECS 30 +// this is just an arbitrary number that has to be higher or equal to the number of fds used by reactor-plugin +#define N_ALL_FDS 8 + +static volatile sig_atomic_t terminate = 0; + /*******************************************************************/ /* Command line options */ /*******************************************************************/ @@ -179,16 +188,183 @@ static GenericAgentConfig *CheckOpts(int argc, char **argv) /*****************************************************************************/ + +static void HandleReactorSignals(int signum) +{ + HandleSignalsForDaemon(signum); + + if (IsPendingTermination()) + { + terminate = 1; + } + + /* HandleSignalsForDaemon() re-arms itself, which would take this wrapper + * out of the chain. Re-arm ours afterwards, so that a second delivery + * still reaches the code above. */ + signal(signum, HandleReactorSignals); +} + +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); EvalContext *ctx = EvalContextNew(); GenericAgentConfigApply(ctx, config); - int ret = ReactorEnterpriseMain(NO_FORK); +#ifdef __MINGW32__ + + if (!NO_FORK) + { + Log(LOG_LEVEL_VERBOSE, "Windows does not support starting processes in the background - starting in foreground"); + } + +#else /* !__MINGW32__ */ + pid_t existing_pid = ReadPID("cf-reactor.pid"); + if ((existing_pid != -1) && (kill(existing_pid, 0) == 0)) + { + Log(LOG_LEVEL_ERR, "Another instance of cf-reactor is already running (pid %jd), terminating", + (intmax_t) existing_pid); + return 1; + } + + if ((!NO_FORK) && (fork() != 0)) + { + Log(LOG_LEVEL_INFO, "cf-reactor: starting"); + _exit(EXIT_SUCCESS); + } + + if (!NO_FORK) + { + ActAsDaemon(); + } + +#endif /* !__MINGW32__ */ + + umask(077); + WritePID("cf-reactor.pid"); + MakeSignalPipe(); + + signal(SIGINT, HandleReactorSignals); + signal(SIGTERM, HandleReactorSignals); + signal(SIGBUS, HandleReactorSignals); + signal(SIGHUP, HandleReactorSignals); + signal(SIGUSR1, HandleReactorSignals); + signal(SIGUSR2, HandleReactorSignals); + + int all_fds[N_ALL_FDS]; + // the first num_nova_fds fds are populated with nova fds + size_t num_nova_fds; + if (!ReactorNovaInitialize(all_fds, N_ALL_FDS, &num_nova_fds, &terminate)) + { + 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 + + /* 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(), + * so that the spawner and the processes it execs keep the default handling. */ + signal(SIGPIPE, SIG_IGN); + + /* We need an initial value here for the first iteration of the cycle + * below. */ + time_t next_tick = time(NULL) + DEFAULT_POLL_INTERVAL_SECS; + while (!terminate) + { + fd_set readfds; + int max_fd = SetupFileDescriptors(&readfds, all_fds, num_fds); + + /* 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); + + next_tick = last_tick + DEFAULT_POLL_INTERVAL_SECS; + + if (ret < 0) + { + if (errno == EINTR) + { + /* Not an error, just a signal delivered while blocked in + * select(). Loop around: the top of the loop re-checks + * `terminate` and rebuilds the fd set from scratch. */ + continue; + } + + /*** error ***/ + Log(LOG_LEVEL_ERR, "Failed to poll events: %s", GetErrorStr()); + break; + } + else if (ret == 0) + { + /*** timeout ***/ + Log(LOG_LEVEL_DEBUG, "Timed-out waiting for next notification"); + + next_tick += DEFAULT_POLL_INTERVAL_SECS; + 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 < N_ALL_FDS */ + if (ReactorNovaHasTimedOut(&readfds, all_fds, num_nova_fds)) + { + ReactorNovaHandleTimeout(&next_tick); + continue; + } + + ReactorNovaHandleEvents(&readfds, all_fds, &next_tick, &terminate); + } + ReactorNovaFinalize(); GenericAgentFinalize(ctx, config); CallCleanupFunctions(); - return ret; + return 0; } diff --git a/libpromises/enterprise_stubs.c b/libpromises/enterprise_stubs.c index 98ebe1696df..054ee25204e 100644 --- a/libpromises/enterprise_stubs.c +++ b/libpromises/enterprise_stubs.c @@ -232,9 +232,19 @@ ENTERPRISE_VOID_FUNC_2ARG_DEFINE_STUB(void, Nova_ClassHistoryEnable, { } -ENTERPRISE_FUNC_1ARG_DEFINE_STUB(int, ReactorEnterpriseMain, ARG_UNUSED bool, no_fork) +ENTERPRISE_FUNC_4ARG_DEFINE_STUB(bool, ReactorNovaInitialize, ARG_UNUSED int*, fds, ARG_UNUSED size_t, max_size, ARG_UNUSED size_t *, num_fds, volatile sig_atomic_t *, terminate) +{ + return false; +} + +ENTERPRISE_VOID_FUNC_1ARG_DEFINE_STUB(void, ReactorNovaHandleTimeout, ARG_UNUSED time_t *, next_tick) +{ +} + +ENTERPRISE_VOID_FUNC_4ARG_DEFINE_STUB(void, ReactorNovaHandleEvents, ARG_UNUSED fd_set *, readfds, ARG_UNUSED int *, fds, ARG_UNUSED time_t *, next_tick, volatile sig_atomic_t *, terminate) +{ +} + +ENTERPRISE_VOID_FUNC_0ARG_DEFINE_STUB(void, ReactorNovaFinalize) { - Log(LOG_LEVEL_VERBOSE, "Nova extension library is not available."); - Log(LOG_LEVEL_VERBOSE, "Running cf-reactor community edition."); - return 0; } diff --git a/libpromises/prototypes3.h b/libpromises/prototypes3.h index 7ba1d9f051c..4ae40b680f7 100644 --- a/libpromises/prototypes3.h +++ b/libpromises/prototypes3.h @@ -85,7 +85,10 @@ ENTERPRISE_VOID_FUNC_0ARG_DECLARE(void, ReloadHAConfig); ENTERPRISE_VOID_FUNC_2ARG_DECLARE(void, Nova_ClassHistoryAddContextName, const StringSet *, list, const char *, context_name); ENTERPRISE_VOID_FUNC_2ARG_DECLARE(void, Nova_ClassHistoryEnable, StringSet **, list, bool, enable); -ENTERPRISE_FUNC_1ARG_DECLARE(int, ReactorEnterpriseMain, bool, no_fork); +ENTERPRISE_FUNC_4ARG_DECLARE(bool, ReactorNovaInitialize, int*, fds, size_t, max_size, size_t *, num_fds, volatile sig_atomic_t *, terminate); +ENTERPRISE_VOID_FUNC_1ARG_DECLARE(void, ReactorNovaHandleTimeout, time_t *, next_tick); +ENTERPRISE_VOID_FUNC_4ARG_DECLARE(void, ReactorNovaHandleEvents, fd_set *, readfds, int *, fds, time_t *, next_tick, volatile sig_atomic_t *, terminate); +ENTERPRISE_VOID_FUNC_0ARG_DECLARE(void, ReactorNovaFinalize); /* manual.c */ From c891e5922b389d51d97f30737a0c2c26ba7b5d7e Mon Sep 17 00:00:00 2001 From: Victor Moene Date: Wed, 2 Sep 2026 16:21:57 +0200 Subject: [PATCH 2/2] Uncommented valgrind tests for cf-reactor Now that cf-reactor is its own daemon, we can uncomment the valgrind tests for it Signed-off-by: Victor Moene --- cf-reactor/cf-reactor.c | 73 ++++++++++++++++---------------- libpromises/enterprise_stubs.c | 14 ++++-- libpromises/prototypes3.h | 5 ++- tests/valgrind-check/valgrind.sh | 6 +-- 4 files changed, 54 insertions(+), 44 deletions(-) diff --git a/cf-reactor/cf-reactor.c b/cf-reactor/cf-reactor.c index c720b1a0ff5..05783808f05 100644 --- a/cf-reactor/cf-reactor.c +++ b/cf-reactor/cf-reactor.c @@ -34,8 +34,9 @@ #include #include #include /* signal, kill */ -#include /* GetSignalPipe, MakeSignalPipe */ +#include /* GetSignalPipe, MakeSignalPipe, IsPendingTermination, HandleSignalsForDaemon */ #include +#include /* xmalloc */ /*****************************************************************************/ /* Globals */ @@ -43,11 +44,11 @@ int NO_FORK = false; -#define DEFAULT_POLL_INTERVAL_SECS 30 -// this is just an arbitrary number that has to be higher or equal to the number of fds used by reactor-plugin -#define N_ALL_FDS 8 +/*****************************************************************************/ +/* Constants */ +/*****************************************************************************/ -static volatile sig_atomic_t terminate = 0; +#define DEFAULT_POLL_INTERVAL_SECS 30 /*******************************************************************/ /* Command line options */ @@ -189,21 +190,6 @@ static GenericAgentConfig *CheckOpts(int argc, char **argv) /*****************************************************************************/ -static void HandleReactorSignals(int signum) -{ - HandleSignalsForDaemon(signum); - - if (IsPendingTermination()) - { - terminate = 1; - } - - /* HandleSignalsForDaemon() re-arms itself, which would take this wrapper - * out of the chain. Re-arm ours afterwards, so that a second delivery - * still reaches the code above. */ - signal(signum, HandleReactorSignals); -} - static int SetupFileDescriptors(fd_set *readfds, int *fds, size_t num_fds) { assert(readfds != NULL); @@ -275,24 +261,31 @@ int main(int argc, char *argv[]) WritePID("cf-reactor.pid"); MakeSignalPipe(); - signal(SIGINT, HandleReactorSignals); - signal(SIGTERM, HandleReactorSignals); - signal(SIGBUS, HandleReactorSignals); - signal(SIGHUP, HandleReactorSignals); - signal(SIGUSR1, HandleReactorSignals); - signal(SIGUSR2, HandleReactorSignals); - - int all_fds[N_ALL_FDS]; + signal(SIGINT, HandleSignalsForDaemon); + signal(SIGTERM, HandleSignalsForDaemon); + signal(SIGBUS, HandleSignalsForDaemon); + signal(SIGHUP, HandleSignalsForDaemon); + 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, N_ALL_FDS, &num_nova_fds, &terminate)) + if (!ReactorNovaInitialize(all_fds, max_nova_fds, &num_nova_fds)) { + 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 + // 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 @@ -303,7 +296,7 @@ int main(int argc, char *argv[]) /* We need an initial value here for the first iteration of the cycle * below. */ time_t next_tick = time(NULL) + DEFAULT_POLL_INTERVAL_SECS; - while (!terminate) + while (!IsPendingTermination()) { fd_set readfds; int max_fd = SetupFileDescriptors(&readfds, all_fds, num_fds); @@ -315,7 +308,13 @@ int main(int argc, char *argv[]) struct timeval timeout = { .tv_sec = remaining }; int ret = select(max_fd, &readfds, NULL, NULL, &timeout); - next_tick = last_tick + DEFAULT_POLL_INTERVAL_SECS; + /* 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. */ + next_tick = time(NULL) + DEFAULT_POLL_INTERVAL_SECS; if (ret < 0) { @@ -323,7 +322,8 @@ int main(int argc, char *argv[]) { /* Not an error, just a signal delivered while blocked in * select(). Loop around: the top of the loop re-checks - * `terminate` and rebuilds the fd set from scratch. */ + * whether termination is pending and rebuilds the fd set + * from scratch. */ continue; } @@ -336,7 +336,6 @@ int main(int argc, char *argv[]) /*** timeout ***/ Log(LOG_LEVEL_DEBUG, "Timed-out waiting for next notification"); - next_tick += DEFAULT_POLL_INTERVAL_SECS; ReactorNovaHandleTimeout(&next_tick); continue; } @@ -352,16 +351,18 @@ int main(int argc, char *argv[]) while (recv(GetSignalPipe(), &buf, 1, 0) > 0) { /* drain */ } } - /* This is needed since num_nova_fds < N_ALL_FDS */ + /* 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, &terminate); + ReactorNovaHandleEvents(&readfds, all_fds, &next_tick); } ReactorNovaFinalize(); + free(all_fds); GenericAgentFinalize(ctx, config); CallCleanupFunctions(); diff --git a/libpromises/enterprise_stubs.c b/libpromises/enterprise_stubs.c index 054ee25204e..dccf8bc2f67 100644 --- a/libpromises/enterprise_stubs.c +++ b/libpromises/enterprise_stubs.c @@ -232,16 +232,24 @@ ENTERPRISE_VOID_FUNC_2ARG_DEFINE_STUB(void, Nova_ClassHistoryEnable, { } -ENTERPRISE_FUNC_4ARG_DEFINE_STUB(bool, ReactorNovaInitialize, ARG_UNUSED int*, fds, ARG_UNUSED size_t, max_size, ARG_UNUSED size_t *, num_fds, volatile sig_atomic_t *, terminate) +ENTERPRISE_FUNC_0ARG_DEFINE_STUB(size_t, ReactorNovaMaxFds) { - return false; + return 0; +} + +ENTERPRISE_FUNC_3ARG_DEFINE_STUB(bool, ReactorNovaInitialize, ARG_UNUSED int*, fds, ARG_UNUSED size_t, max_size, size_t *, num_fds) +{ + Log(LOG_LEVEL_VERBOSE, "Nova extension library is not available."); + Log(LOG_LEVEL_VERBOSE, "Running cf-reactor community edition."); + *num_fds = 0; + return true; } ENTERPRISE_VOID_FUNC_1ARG_DEFINE_STUB(void, ReactorNovaHandleTimeout, ARG_UNUSED time_t *, next_tick) { } -ENTERPRISE_VOID_FUNC_4ARG_DEFINE_STUB(void, ReactorNovaHandleEvents, ARG_UNUSED fd_set *, readfds, ARG_UNUSED int *, fds, ARG_UNUSED time_t *, next_tick, volatile sig_atomic_t *, terminate) +ENTERPRISE_VOID_FUNC_3ARG_DEFINE_STUB(void, ReactorNovaHandleEvents, ARG_UNUSED fd_set *, readfds, ARG_UNUSED int *, fds, ARG_UNUSED time_t *, next_tick) { } diff --git a/libpromises/prototypes3.h b/libpromises/prototypes3.h index 4ae40b680f7..2b963ed2c0a 100644 --- a/libpromises/prototypes3.h +++ b/libpromises/prototypes3.h @@ -85,9 +85,10 @@ ENTERPRISE_VOID_FUNC_0ARG_DECLARE(void, ReloadHAConfig); ENTERPRISE_VOID_FUNC_2ARG_DECLARE(void, Nova_ClassHistoryAddContextName, const StringSet *, list, const char *, context_name); ENTERPRISE_VOID_FUNC_2ARG_DECLARE(void, Nova_ClassHistoryEnable, StringSet **, list, bool, enable); -ENTERPRISE_FUNC_4ARG_DECLARE(bool, ReactorNovaInitialize, int*, fds, size_t, max_size, size_t *, num_fds, volatile sig_atomic_t *, terminate); +ENTERPRISE_FUNC_0ARG_DECLARE(size_t, ReactorNovaMaxFds); +ENTERPRISE_FUNC_3ARG_DECLARE(bool, ReactorNovaInitialize, int*, fds, size_t, max_size, size_t *, num_fds); ENTERPRISE_VOID_FUNC_1ARG_DECLARE(void, ReactorNovaHandleTimeout, time_t *, next_tick); -ENTERPRISE_VOID_FUNC_4ARG_DECLARE(void, ReactorNovaHandleEvents, fd_set *, readfds, int *, fds, time_t *, next_tick, volatile sig_atomic_t *, terminate); +ENTERPRISE_VOID_FUNC_3ARG_DECLARE(void, ReactorNovaHandleEvents, fd_set *, readfds, int *, fds, time_t *, next_tick); ENTERPRISE_VOID_FUNC_0ARG_DECLARE(void, ReactorNovaFinalize); /* manual.c */ diff --git a/tests/valgrind-check/valgrind.sh b/tests/valgrind-check/valgrind.sh index 0eed132b757..4c125e37f7e 100644 --- a/tests/valgrind-check/valgrind.sh +++ b/tests/valgrind-check/valgrind.sh @@ -225,14 +225,14 @@ tail reactor.txt echo "Checking that serverd, execd and reactor PIDs are still correct/alive:" ps -p $exec_pid ps -p $server_pid -# ps -p $reactor_pid +ps -p $reactor_pid echo "Killing valgrind cf-execd" kill $exec_pid echo "Killing valgrind cf-serverd" kill $server_pid -# echo "Killing valgrind cf-reactor" -# kill $reactor_pid +echo "Killing valgrind cf-reactor" +kill $reactor_pid wait $exec_pid wait $server_pid