diff --git a/package/confd/confd.mk b/package/confd/confd.mk index 4350985a..eeab1fe2 100644 --- a/package/confd/confd.mk +++ b/package/confd/confd.mk @@ -10,7 +10,7 @@ CONFD_SITE = $(BR2_EXTERNAL_INFIX_PATH)/src/confd CONFD_LICENSE = BSD-3-Clause CONFD_LICENSE_FILES = LICENSE CONFD_REDISTRIBUTE = NO -CONFD_DEPENDENCIES = host-sysrepo sysrepo rousette netopeer2 jansson libite sysrepo libsrx libglib2 +CONFD_DEPENDENCIES = host-sysrepo sysrepo rousette netopeer2 jansson libite sysrepo libsrx libglib2 libev CONFD_AUTORECONF = YES CONFD_CONF_OPTS += --disable-silent-rules --with-crypt=$(BR2_PACKAGE_CONFD_DEFAULT_CRYPT) CONFD_SYSREPO_SHM_PREFIX = sr_buildroot$(subst /,_,$(CONFIG_DIR))_confd diff --git a/src/confd/configure.ac b/src/confd/configure.ac index f52e683f..71e4717a 100644 --- a/src/confd/configure.ac +++ b/src/confd/configure.ac @@ -89,6 +89,15 @@ PKG_CHECK_MODULES([sysrepo], [sysrepo >= 4.2.10]) PKG_CHECK_MODULES([libyang], [libyang >= 4.2.2]) PKG_CHECK_MODULES([libsrx], [libsrx >= 1.0.0]) +AC_CHECK_HEADER([ev.h], + [saved_LIBS="$LIBS" + AC_CHECK_LIB([ev], [ev_loop_new], + [EV_LIBS="-lev"], + [AC_MSG_ERROR("libev not found")]) + LIBS="$saved_LIBS"], + [AC_MSG_ERROR("ev.h not found")]) +AC_SUBST([EV_LIBS]) + # Control build with automake flags AM_CONDITIONAL(CONTAINERS, [test "x$enable_containers" != "xno"]) diff --git a/src/confd/src/Makefile.am b/src/confd/src/Makefile.am index 67b2538f..4ab20ba0 100644 --- a/src/confd/src/Makefile.am +++ b/src/confd/src/Makefile.am @@ -7,7 +7,7 @@ plugin_LTLIBRARIES = confd-plugin.la sbin_PROGRAMS = confd confd_CFLAGS = $(sysrepo_CFLAGS) $(libyang_CFLAGS) $(jansson_CFLAGS) $(libite_CFLAGS) -confd_LDADD = $(sysrepo_LIBS) $(libyang_LIBS) $(jansson_LIBS) $(libite_LIBS) -ldl +confd_LDADD = $(sysrepo_LIBS) $(libyang_LIBS) $(jansson_LIBS) $(libite_LIBS) $(EV_LIBS) -ldl confd_SOURCES = main.c confd_plugin_la_LDFLAGS = -module -avoid-version -shared diff --git a/src/confd/src/core.c b/src/confd/src/core.c index ef958985..dd035a3a 100644 --- a/src/confd/src/core.c +++ b/src/confd/src/core.c @@ -461,10 +461,10 @@ static inline int subscribe_model(char *model, struct confd *confd, int flags) { return sr_module_change_subscribe(confd->session, model, "//.", change_cb, confd, CB_PRIO_PRIMARY, SR_SUBSCR_CHANGE_ALL_MODULES | - SR_SUBSCR_DEFAULT | flags, &confd->sub) || + SR_SUBSCR_NO_THREAD | flags, &confd->sub) || sr_module_change_subscribe(confd->startup, model, "//.", startup_save, NULL, CB_PRIO_PASSIVE, SR_SUBSCR_CHANGE_ALL_MODULES | - SR_SUBSCR_PASSIVE, &confd->sub); + SR_SUBSCR_PASSIVE | SR_SUBSCR_NO_THREAD, &confd->sub); } int sr_plugin_init_cb(sr_session_ctx_t *session, void **priv) @@ -645,6 +645,15 @@ err: return rc; } +void confd_get_subscriptions(void *priv, sr_subscription_ctx_t **out_sub, + sr_subscription_ctx_t **out_fsub) +{ + struct confd *c = (struct confd *)priv; + + *out_sub = c->sub; + *out_fsub = c->fsub; +} + void sr_plugin_cleanup_cb(sr_session_ctx_t *session, void *priv) { struct confd *ptr = (struct confd *)priv; diff --git a/src/confd/src/core.h b/src/confd/src/core.h index 928584fe..9262fde6 100644 --- a/src/confd/src/core.h +++ b/src/confd/src/core.h @@ -145,7 +145,7 @@ static inline int register_change(sr_session_ctx_t *session, const char *module, int flags, sr_module_change_cb cb, void *arg, sr_subscription_ctx_t **sub) { int rc = sr_module_change_subscribe(session, module, xpath, cb, arg, - CB_PRIO_PRIMARY, flags | SR_SUBSCR_DEFAULT, sub); + CB_PRIO_PRIMARY, flags | SR_SUBSCR_NO_THREAD, sub); if (rc) { ERROR("failed subscribing to changes of %s: %s", xpath, sr_strerror(rc)); return rc; @@ -159,7 +159,7 @@ static inline int register_monitor(sr_session_ctx_t *session, const char *module int flags, sr_module_change_cb cb, void *arg, sr_subscription_ctx_t **sub) { int rc = sr_module_change_subscribe(session, module, xpath, cb, arg, - 0, flags | SR_SUBSCR_PASSIVE, sub); + 0, flags | SR_SUBSCR_PASSIVE | SR_SUBSCR_NO_THREAD, sub); if (rc) { ERROR("failed subscribing to monitor %s: %s", xpath, sr_strerror(rc)); return rc; @@ -172,7 +172,7 @@ static inline int register_oper(sr_session_ctx_t *session, const char *module, c sr_oper_get_items_cb cb, void *arg, int flags, sr_subscription_ctx_t **sub) { int rc = sr_oper_get_subscribe(session, module, xpath, cb, arg, - flags | SR_SUBSCR_DEFAULT, sub); + flags | SR_SUBSCR_NO_THREAD, sub); if (rc) ERROR("failed subscribing to %s oper: %s", xpath, sr_strerror(rc)); return rc; @@ -181,7 +181,7 @@ static inline int register_oper(sr_session_ctx_t *session, const char *module, c static inline int register_rpc(sr_session_ctx_t *session, const char *xpath, sr_rpc_cb cb, void *arg, sr_subscription_ctx_t **sub) { - int rc = sr_rpc_subscribe(session, xpath, cb, arg, 0, SR_SUBSCR_DEFAULT, sub); + int rc = sr_rpc_subscribe(session, xpath, cb, arg, 0, SR_SUBSCR_NO_THREAD, sub); if (rc) ERROR("failed subscribing to %s rpc: %s", xpath, sr_strerror(rc)); return rc; diff --git a/src/confd/src/main.c b/src/confd/src/main.c index 38966a47..7e454e71 100644 --- a/src/confd/src/main.c +++ b/src/confd/src/main.c @@ -13,9 +13,11 @@ #include #include #include +#include #include #include #include +#include #include #include #include @@ -51,14 +53,15 @@ struct plugin { char *name; int (*init_cb)(sr_session_ctx_t *session, void **private_data); void (*cleanup_cb)(sr_session_ctx_t *session, void *private_data); + void (*get_subs)(void *priv, sr_subscription_ctx_t **sub, sr_subscription_ctx_t **fsub); void *private_data; + sr_subscription_ctx_t *sub; + sr_subscription_ctx_t *fsub; int initialized; }; -/* Protected flag for terminating */ -static int loop_finish; -static pthread_mutex_t lock = PTHREAD_MUTEX_INITIALIZER; -static pthread_cond_t cond = PTHREAD_COND_INITIALIZER; +/* Maximum number of sysrepo event pipe file descriptors across all plugins */ +#define MAX_EVENT_FDS 64 static void error_print(int sr_error, const char *format, ...) { @@ -129,45 +132,11 @@ static void help_print(void) "\n"); } -static void signal_handler(int sig) -{ - switch (sig) { - case SIGINT: - case SIGQUIT: - case SIGABRT: - case SIGTERM: - case SIGHUP: - pthread_mutex_lock(&lock); - if (!loop_finish) { - loop_finish = 1; - pthread_cond_signal(&cond); - } else { - error_print(0, "Exiting without a proper cleanup"); - exit(EXIT_FAILURE); - } - pthread_mutex_unlock(&lock); - break; - default: - error_print(0, "Exiting on receiving an unhandled signal"); - exit(EXIT_FAILURE); - } -} - -static void handle_signals(void) +static void ignore_signals(void) { struct sigaction action; - sigset_t block_mask; - - sigfillset(&block_mask); - action.sa_handler = signal_handler; - action.sa_mask = block_mask; - action.sa_flags = 0; - sigaction(SIGINT, &action, NULL); - sigaction(SIGQUIT, &action, NULL); - sigaction(SIGABRT, &action, NULL); - sigaction(SIGTERM, &action, NULL); - sigaction(SIGHUP, &action, NULL); + memset(&action, 0, sizeof(action)); action.sa_handler = SIG_IGN; sigaction(SIGPIPE, &action, NULL); sigaction(SIGTSTP, &action, NULL); @@ -175,6 +144,68 @@ static void handle_signals(void) sigaction(SIGTTOU, &action, NULL); } +/* libev callbacks for steady-state operation */ +static void signal_cb(struct ev_loop *loop, struct ev_signal *w, int revents) +{ + (void)revents; + (void)w; + ev_break(loop, EVBREAK_ALL); +} + +static void sr_event_cb(struct ev_loop *loop, struct ev_io *w, int revents) +{ + (void)loop; + (void)revents; + sr_subscription_process_events(w->data, NULL, NULL); +} + +/* + * Temporary event pump thread for bootstrap. + * + * With SR_SUBSCR_NO_THREAD, sysrepo writes events to a pipe and waits + * for the application to call sr_subscription_process_events(). During + * bootstrap, sr_replace_config() blocks waiting for callbacks — this + * thread ensures those callbacks get dispatched. + */ +struct event_pump { + struct plugin *plugins; + int plugin_count; + int running; +}; + +static void *event_pump_thread(void *arg) +{ + struct event_pump *pump = arg; + struct pollfd fds[MAX_EVENT_FDS]; + sr_subscription_ctx_t *subs[MAX_EVENT_FDS]; + int nfds = 0; + + for (int i = 0; i < pump->plugin_count; i++) { + struct plugin *p = &pump->plugins[i]; + + if (p->sub && sr_get_event_pipe(p->sub, &fds[nfds].fd) == SR_ERR_OK) { + fds[nfds].events = POLLIN; + subs[nfds] = p->sub; + nfds++; + } + if (p->fsub && sr_get_event_pipe(p->fsub, &fds[nfds].fd) == SR_ERR_OK) { + fds[nfds].events = POLLIN; + subs[nfds] = p->fsub; + nfds++; + } + } + + while (pump->running) { + if (poll(fds, nfds, 100) > 0) { + for (int i = 0; i < nfds; i++) + if (fds[i].revents & POLLIN) + sr_subscription_process_events(subs[i], NULL, NULL); + } + } + + return NULL; +} + static void quiet_now(void) { int fd = -1; @@ -196,8 +227,9 @@ static void daemon_init(int debug, sr_log_level_t log_level) nice(-20); + ignore_signals(); + if (debug) { - handle_signals(); if (debug < 0) goto done; sr_log_stderr(log_level); @@ -212,8 +244,6 @@ static void daemon_init(int debug, sr_log_level_t log_level) if (pid > 0) exit(EXIT_SUCCESS); - handle_signals(); - sid = setsid(); if (sid < 0) { error_print(0, "setsid() failed (%s).", strerror(errno)); @@ -357,6 +387,9 @@ static int load_plugins(struct plugin **plugins, int *plugin_count) break; } + /* Optional: allows main to collect subscription contexts */ + *(void **)&plugin->get_subs = dlsym(handle, "confd_get_subscriptions"); + plugin->handle = handle; name_len = path_len_no_ext(ent->d_name); @@ -861,18 +894,47 @@ int main(int argc, char **argv) } conout(0, "\n"); + /* Phase 8: Collect subscription contexts from plugins */ + for (i = 0; i < plugin_count; i++) { + if (plugins[i].initialized && plugins[i].get_subs) + plugins[i].get_subs(plugins[i].private_data, + &plugins[i].sub, &plugins[i].fsub); + } + + /* Phase 9: Start event pump thread for bootstrap. + * With SR_SUBSCR_NO_THREAD, sr_replace_config() blocks waiting + * for callbacks. The pump thread processes those events. */ + struct event_pump pump = { + .plugins = plugins, + .plugin_count = plugin_count, + .running = 1, + }; + pthread_t pump_tid; + + if (pthread_create(&pump_tid, NULL, event_pump_thread, &pump)) { + SRPLG_LOG_ERR("confd", "Failed to create event pump thread: %s", strerror(errno)); + goto cleanup; + } + /* - * Phase 8: Load startup config -- plugins are now subscribed, so + * Phase 10: Load startup config -- plugins are now subscribed, so * sr_replace_config() will trigger their change callbacks. + * The event pump thread processes those callbacks. */ conout(3, "Loading startup-config"); if (bootstrap_config(conn, sess, factory_path, startup_path, failure_path, test_path, timeout_ms)) { + pump.running = 0; + pthread_join(pump_tid, NULL); conout(1, "\n"); goto cleanup; } conout(0, "\n"); + /* Phase 11: Stop event pump — bootstrap is done */ + pump.running = 0; + pthread_join(pump_tid, NULL); + /* No more progress to show, go to quiet daemon mode */ quiet_now(); @@ -883,11 +945,42 @@ int main(int argc, char **argv) if (pidfile && write_pidfile(pidfd) < 0) goto cleanup; - /* Wait for a terminating signal */ - pthread_mutex_lock(&lock); - while (!loop_finish) - pthread_cond_wait(&cond, &lock); - pthread_mutex_unlock(&lock); + /* Phase 12: Steady-state — libev event loop replaces pthread_cond_wait */ + { + struct ev_loop *loop = EV_DEFAULT; + struct ev_signal sigterm_w, sigint_w, sighup_w, sigquit_w; + struct ev_io io_watchers[MAX_EVENT_FDS]; + int nio = 0; + + ev_signal_init(&sigterm_w, signal_cb, SIGTERM); + ev_signal_init(&sigint_w, signal_cb, SIGINT); + ev_signal_init(&sighup_w, signal_cb, SIGHUP); + ev_signal_init(&sigquit_w, signal_cb, SIGQUIT); + ev_signal_start(loop, &sigterm_w); + ev_signal_start(loop, &sigint_w); + ev_signal_start(loop, &sighup_w); + ev_signal_start(loop, &sigquit_w); + + for (i = 0; i < plugin_count; i++) { + int fd; + + if (plugins[i].sub && sr_get_event_pipe(plugins[i].sub, &fd) == SR_ERR_OK) { + ev_io_init(&io_watchers[nio], sr_event_cb, fd, EV_READ); + io_watchers[nio].data = plugins[i].sub; + ev_io_start(loop, &io_watchers[nio]); + nio++; + } + if (plugins[i].fsub && sr_get_event_pipe(plugins[i].fsub, &fd) == SR_ERR_OK) { + ev_io_init(&io_watchers[nio], sr_event_cb, fd, EV_READ); + io_watchers[nio].data = plugins[i].fsub; + ev_io_start(loop, &io_watchers[nio]); + nio++; + } + } + + ev_run(loop, 0); + ev_loop_destroy(loop); + } rc = EXIT_SUCCESS;