diff --git a/cf-reactor/Makefile.am b/cf-reactor/Makefile.am index 443e919f112..ef959371b00 100644 --- a/cf-reactor/Makefile.am +++ b/cf-reactor/Makefile.am @@ -39,7 +39,9 @@ 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 \ + 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..53680cb108a 100644 --- a/cf-reactor/README.md +++ b/cf-reactor/README.md @@ -26,23 +26,6 @@ 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. diff --git a/cf-reactor/reactor_context.c b/cf-reactor/reactor_context.c index 71f7b7bc3a1..cbd48947032 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,88 @@ 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; 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 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 +127,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 +169,39 @@ 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; + } + } + /** + * We should never reach this point: ReactorContextInitialize would fail if no watcher fd is set up. + */ + ProgrammingError("Couldn't find the 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 +210,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..a5f7f5f4967 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; /** 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..f679a20b4da --- /dev/null +++ b/cf-reactor/watcher.c @@ -0,0 +1,251 @@ +/* + 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 = NULL; + WatcherPayloadDestroyFn destroy_payload = NULL; + 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 *fd) +{ + assert(fd != NULL); + assert(watchers != NULL && event_to_bundle != NULL); /* WatcherRegistryInitialize() first */ + + 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; + } + + *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) +{ + 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..2d6a2c616a0 --- /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 *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 *fd); +void EventWatcherHandleEvents(int fd, fd_set *readfds); +void EventWatcherFinalize(void); + +#endif