@cryptotaxi247 / netdata-1 / commits / 3153adc70

Adjust functions event loop to propagate exit code (#21455)

* Add status handling to functions_evloop and update plugins for proper exit status tracking * Fix compilation of ebpf, freeipmi, windows events plugins (adjust functions_evloop_init call parameters) * Use atomic load for plugin exit checks across all collectors to prevent race conditions during shutdown. * Use atomic store for `ebpf_plugin_exit` to ensure thread-safe shutdown handling (just for consistency) * Use atomic load for `plugin_should_exit` in functions_evloop for thread-safe shutdown handling and clean up formatting inconsistencies. * Use atomic store for `plugin_should_exit` in functions_evloop and freeipmi for thread-safe shutdown handling. * Cannot be infinite loop now. Use atomic load for `function_plugin_should_exit` in freeipmi plugin to ensure thread-safe shutdown handling.

Stelios Fragkakis committed Jan 5, 2026 at 10:22 UTC 3153adc70df6c342ca25242f87d81f18a02398b2

11 files changed +32 -23

src/collectors/apps.plugin/apps_plugin.c

+4 -2

@@ -784,12 +784,13 @@ int main(int argc, char **argv) {

784

785 apps_pids_init();

786 OS_FUNCTION(apps_os_init)();

787 + int exit_status = 0;

788

789 // ------------------------------------------------------------------------

790 // the event loop for functions

791

792 struct functions_evloop_globals *wg =

792 - functions_evloop_init(1, "APPS", &apps_and_stdout_mutex, &apps_plugin_exit);

793 + functions_evloop_init(1, "APPS", &apps_and_stdout_mutex, &apps_plugin_exit, &exit_status);

794

795 functions_evloop_add_function(wg, "processes", function_processes, PLUGINS_FUNCTIONS_TIMEOUT_DEFAULT, NULL);

796

@@ -801,7 +802,7 @@ int main(int argc, char **argv) {

802 global_iterations_counter = 1;

803 heartbeat_t hb;

804 heartbeat_init(&hb, update_every * USEC_PER_SEC);

804 - for(; !apps_plugin_exit ; global_iterations_counter++) {

805 + for (; !__atomic_load_n(&apps_plugin_exit, __ATOMIC_ACQUIRE); global_iterations_counter++) {

806 netdata_mutex_unlock(&apps_and_stdout_mutex);

807

808 usec_t dt;

@@ -881,4 +882,5 @@ int main(int argc, char **argv) {

882 debug_log("done Loop No %zu", global_iterations_counter);

883 }

884 netdata_mutex_unlock(&apps_and_stdout_mutex);

885 + exit(exit_status);

886 }

src/collectors/ebpf.plugin/ebpf.c

+1 -1

@@ -1126,7 +1126,7 @@ void ebpf_stop_threads(int sig)

1126 nd_thread_join(ebpf_threads[i].thread);

1127 }

1128

1129 - ebpf_plugin_exit = true;

1129 + __atomic_store_n(&ebpf_plugin_exit, true, __ATOMIC_RELEASE);

1130

1131 netdata_mutex_lock(&mutex_cgroup_shm);

1132 nd_thread_signal_cancel(cgroup_integration_thread.thread);

src/collectors/ebpf.plugin/ebpf.h

+1 -1

@@ -366,7 +366,7 @@ extern uint64_t collect_pids;

366

367 static inline bool ebpf_plugin_stop(void)

368 {

369 - return ebpf_plugin_exit || nd_thread_signaled_to_cancel();

369 + return __atomic_load_n(&ebpf_plugin_exit, __ATOMIC_ACQUIRE) || nd_thread_signaled_to_cancel();

370 }

371

372 void ebpf_stop_threads(int sig);

src/collectors/ebpf.plugin/ebpf_functions.c

+1 -1

@@ -805,7 +805,7 @@ void ebpf_function_thread(void *ptr)

805 {

806 (void)ptr;

807

808 - struct functions_evloop_globals *wg = functions_evloop_init(1, "EBPF", &lock, &ebpf_plugin_exit);

808 + struct functions_evloop_globals *wg = functions_evloop_init(1, "EBPF", &lock, &ebpf_plugin_exit, NULL);

809

810 functions_evloop_add_function(

811 wg, EBPF_FUNCTION_SOCKET, ebpf_function_socket_manipulation, PLUGINS_FUNCTIONS_TIMEOUT_DEFAULT, NULL);

src/collectors/freeipmi.plugin/freeipmi_plugin.c

+3 -3

@@ -1645,7 +1645,7 @@ close_and_send:

1645

1646 static void plugin_exit(int code) {

1647 fflush(stdout);

1648 - function_plugin_should_exit = true;

1648 + __atomic_store_n(&function_plugin_should_exit, true, __ATOMIC_RELEASE);

1649 exit(code);

1650 }

1651

@@ -2006,7 +2006,7 @@ int main (int argc, char **argv) {

2006

2007 heartbeat_t hb;

2008 heartbeat_init(&hb, update_every * USEC_PER_SEC);

2009 - for(iteration = 0; 1 ; iteration++) {

2009 + for(iteration = 0; !__atomic_load_n(&function_plugin_should_exit, __ATOMIC_ACQUIRE) ; iteration++) {

2010 usec_t dt = heartbeat_next(&hb);

2011

2012 if (!tty) {

@@ -2072,7 +2072,7 @@ int main (int argc, char **argv) {

2072 if (add_func_sensors) {

2073 add_func_sensors = false;

2074 struct functions_evloop_globals *wg =

2075 - functions_evloop_init(1, "FREEIPMI", &stdout_mutex, &function_plugin_should_exit);

2075 + functions_evloop_init(1, "FREEIPMI", &stdout_mutex, &function_plugin_should_exit, NULL);

2076 functions_evloop_add_function(

2077 wg, "ipmi-sensors", freeimi_function_sensors, PLUGINS_FUNCTIONS_TIMEOUT_DEFAULT, NULL);

2078 FREEIPMI_GLOBAL_FUNCTION_SENSORS();

src/collectors/network-viewer.plugin/network-viewer.c

+2 -2

@@ -1016,7 +1016,7 @@ int main(int argc __maybe_unused, char **argv __maybe_unused) {

1016 // ----------------------------------------------------------------------------------------------------------------

1017

1018 struct functions_evloop_globals *wg =

1019 - functions_evloop_init(5, "Network-Viewer", &stdout_mutex, &plugin_should_exit);

1019 + functions_evloop_init(5, "Network-Viewer", &stdout_mutex, &plugin_should_exit, NULL);

1020

1021 functions_evloop_add_function(wg, NETWORK_CONNECTIONS_VIEWER_FUNCTION,

1022 network_viewer_function,

@@ -1030,8 +1030,8 @@ int main(int argc __maybe_unused, char **argv __maybe_unused) {

1030

1031 heartbeat_t hb;

1032 heartbeat_init(&hb, USEC_PER_SEC);

1033 - while(!plugin_should_exit) {

1033

1034 + while(!__atomic_load_n(&plugin_should_exit, __ATOMIC_ACQUIRE)) {

1035 usec_t dt_ut = heartbeat_next(&hb);

1036 send_newline_ut += dt_ut;

1037

src/collectors/systemd-journal.plugin/systemd-main.c

+2 -2

@@ -73,7 +73,7 @@ int main(int argc __maybe_unused, char **argv __maybe_unused)

73 // the event loop for functions

75 struct functions_evloop_globals *wg =

76 - functions_evloop_init(ND_SD_JOURNAL_WORKER_THREADS, "SDJ", &stdout_mutex, &plugin_should_exit);

76 + functions_evloop_init(ND_SD_JOURNAL_WORKER_THREADS, "SDJ", &stdout_mutex, &plugin_should_exit, NULL);

78 functions_evloop_add_function(

79 wg, ND_SD_JOURNAL_FUNCTION_NAME, function_systemd_journal, ND_SD_JOURNAL_DEFAULT_TIMEOUT, NULL);

@@ -106,7 +106,7 @@ int main(int argc __maybe_unused, char **argv __maybe_unused)

106

107 heartbeat_t hb;

108 heartbeat_init(&hb, USEC_PER_SEC);

109 - while (!plugin_should_exit) {

109 + while (!__atomic_load_n(&plugin_should_exit, __ATOMIC_ACQUIRE)) {

110 if (since_last_scan_ut > ND_SD_JOURNAL_ALL_FILES_SCAN_EVERY_USEC) {

111 nd_journal_files_registry_update();

112 since_last_scan_ut = 0;

src/collectors/systemd-units.plugin/plugin_systemd_units.c

+2 -2

@@ -2224,7 +2224,7 @@ int main(int argc __maybe_unused, char **argv __maybe_unused)

2224 // the event loop for functions

2225

2226 struct functions_evloop_globals *wg =

2227 - functions_evloop_init(ND_SD_JOURNAL_WORKER_THREADS, "SDU", &stdout_mutex, &plugin_should_exit);

2227 + functions_evloop_init(ND_SD_JOURNAL_WORKER_THREADS, "SDU", &stdout_mutex, &plugin_should_exit, NULL);

2228

2229 functions_evloop_add_function(

2230 wg, ND_SD_UNITS_FUNCTION_NAME, function_systemd_units, ND_SD_UNITS_DEFAULT_TIMEOUT, NULL);

@@ -2253,7 +2253,7 @@ int main(int argc __maybe_unused, char **argv __maybe_unused)

2253

2254 heartbeat_t hb;

2255 heartbeat_init(&hb, USEC_PER_SEC);

2256 - while (!plugin_should_exit) {

2256 + while (!__atomic_load_n(&plugin_should_exit, __ATOMIC_ACQUIRE)) {

2257 usec_t dt_ut = heartbeat_next(&hb);

2258 send_newline_ut += dt_ut;

2259

src/collectors/windows-events.plugin/windows-events.c

+2 -2

@@ -1355,7 +1355,7 @@ int main(int argc __maybe_unused, char **argv __maybe_unused) {

1355 // the event loop for functions

1356

1357 struct functions_evloop_globals *wg =

1358 - functions_evloop_init(WINDOWS_EVENTS_WORKER_THREADS, "WEVT", &stdout_mutex, &plugin_should_exit);

1358 + functions_evloop_init(WINDOWS_EVENTS_WORKER_THREADS, "WEVT", &stdout_mutex, &plugin_should_exit, NULL);

1359

1360 functions_evloop_add_function(wg,

1361 WEVT_FUNCTION_NAME,

@@ -1385,7 +1385,7 @@ int main(int argc __maybe_unused, char **argv __maybe_unused) {

1385

1386 heartbeat_t hb;

1387 heartbeat_init(&hb, USEC_PER_SEC);

1388 - while(!plugin_should_exit) {

1388 + while(!__atomic_load_n(&plugin_should_exit, __ATOMIC_ACQUIRE)) {

1389

1390 if(since_last_scan_ut > WINDOWS_EVENTS_SCAN_EVERY_USEC) {

1391 wevt_sources_scan();

src/libnetdata/functions_evloop/functions_evloop.c

+13 -6

@@ -49,6 +49,7 @@ struct functions_evloop_globals {

50 netdata_mutex_t *stdout_mutex;

51 bool *plugin_should_exit;

52 + int *status;

53 bool workers_exit; // all workers are waiting on the same condition - this makes them all exit, when any is cancelled

55 ND_THREAD *reader_thread;

@@ -289,7 +290,7 @@ static bool rrd_function_worker_global_process_input(struct functions_evloop_glo

290 nd_log(NDLS_COLLECTORS, NDLP_NOTICE, "Received PROGRESS for transaction '%s', but it not available here", transaction);

291 }

292 else if(keyword && strcmp(keyword, PLUGINSD_CALL_QUIT) == 0) {

292 - *wg->plugin_should_exit = true;

293 + __atomic_store_n(wg->plugin_should_exit, true, __ATOMIC_RELEASE);

294 return true;

295 }

296 else

@@ -306,7 +307,7 @@ static void rrd_functions_worker_globals_reader_main(void *arg) {

307 buffered_reader_init(&wg->reader);

308 wg->buffer = buffer_create(sizeof(wg->reader.read_buffer) + 2, NULL);

309

309 - while(!(*wg->plugin_should_exit)) {

310 + while(!__atomic_load_n(wg->plugin_should_exit, __ATOMIC_ACQUIRE)) {

311 if(unlikely(!buffered_reader_next_line(&wg->reader, wg->buffer))) {

312 buffered_reader_ret_t ret = buffered_reader_read_timeout(

313 &wg->reader,

@@ -326,14 +327,16 @@ static void rrd_functions_worker_globals_reader_main(void *arg) {

327 }

328

329 int status = 0;

329 - if(!(*wg->plugin_should_exit)) {

330 + if(!__atomic_load_n(wg->plugin_should_exit, __ATOMIC_ACQUIRE)) {

331 nd_log(NDLS_COLLECTORS, NDLP_ERR, "Read error on stdin");

332 status = 1;

333 }

334

334 - *wg->plugin_should_exit = true;

335 buffer_free(wg->buffer);

336 - exit(status);

336 +

337 + if (wg->status)

338 + __atomic_store_n(wg->status, status, __ATOMIC_RELEASE);

339 + __atomic_store_n(wg->plugin_should_exit, true, __ATOMIC_RELEASE);

340 }

341

342 void worker_queue_delete_cb(const DICTIONARY_ITEM *item __maybe_unused, void *value, void *data __maybe_unused) {

@@ -341,7 +344,10 @@ void worker_queue_delete_cb(const DICTIONARY_ITEM *item __maybe_unused, void *va

344 worker_job_cleanup(j);

345 }

346

344 -struct functions_evloop_globals *functions_evloop_init(size_t worker_threads, const char *tag, netdata_mutex_t *stdout_mutex, bool *plugin_should_exit) {

347 +struct functions_evloop_globals *functions_evloop_init(size_t worker_threads, const char *tag,

348 + netdata_mutex_t *stdout_mutex, bool *plugin_should_exit,

349 + int *status)

350 +{

351 struct functions_evloop_globals *wg = callocz(1, sizeof(struct functions_evloop_globals));

352

353 wg->worker_queue = dictionary_create(DICT_OPTION_DONT_OVERWRITE_VALUE);

@@ -357,6 +363,7 @@ struct functions_evloop_globals *functions_evloop_init(size_t worker_threads, co

363 wg->workers = worker_threads;

364 wg->worker_threads = callocz(wg->workers, sizeof(ND_THREAD *));

365 wg->tag = tag;

366 + wg->status = status;

367

368 char tag_buffer[NETDATA_THREAD_TAG_MAX + 1];

369 snprintfz(tag_buffer, NETDATA_THREAD_TAG_MAX, "%s_READER", wg->tag);

src/libnetdata/functions_evloop/functions_evloop.h

+1 -1

@@ -90,7 +90,7 @@ typedef void (*functions_evloop_worker_execute_t)(const char *transaction, char

90 const char *source, void *data);

92 struct functions_evloop_worker_job;

93 -struct functions_evloop_globals *functions_evloop_init(size_t worker_threads, const char *tag, netdata_mutex_t *stdout_mutex, bool *plugin_should_exit);

93 +struct functions_evloop_globals *functions_evloop_init(size_t worker_threads, const char *tag, netdata_mutex_t *stdout_mutex, bool *plugin_should_exit, int *status);

94 void functions_evloop_add_function(struct functions_evloop_globals *wg, const char *function, functions_evloop_worker_execute_t cb, time_t default_timeout, void *data);

95 void functions_evloop_cancel_threads(struct functions_evloop_globals *wg);