confd: replace sysrepo threads with libev event loop

Use SR_SUBSCR_NO_THREAD for all subscriptions and integrate sysrepo
event pipes into a libev event loop.  This eliminates approximately 30
per-subscription threads, reducing overhead on embedded ARM hardware.

A temporary poll-based "event pump" thread handles callback dispatch
during bootstrap (where sr_replace_config blocks waiting for callbacks),
then exits.  After bootstrap, the single-threaded libev loop takes over
for steady-state event processing.

Note, the confd-test-mode plugin still requires use of threads so we do
not create deadlocks when calling sr_replace_config().

Signed-off-by: Joachim Wiberg <troglobit@gmail.com>

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

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"])
 

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

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;

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;

src/confd/src/main.c

@@ -13,9 +13,11 @@
 #include <dirent.h>
 #include <dlfcn.h>
 #include <errno.h>
+#include <ev.h>
 #include <fcntl.h>
 #include <getopt.h>
 #include <glob.h>
+#include <poll.h>
 #include <pthread.h>
 #include <stdint.h>
 #include <signal.h>
@@ -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;