@cryptotaxi247 / netdata-1 / commits / cb0cc8361

Make all threads joinable and join on agent shutdown (#20228)

Make all threads joinable / join on exit

Stelios Fragkakis committed May 6, 2025 at 16:57 UTC cb0cc8361bd66592e8e0370fdd082546ea158195

22 files changed +71 -75

src/collectors/ebpf.plugin/ebpf.c

+2 -2

@@ -4244,7 +4244,7 @@ static void ebpf_initialize_data_sharing()

4244 switch (integration_with_collectors) {

4245 case NETDATA_EBPF_INTEGRATION_SOCKET: {

4246 socket_ipc =

4247 - nd_thread_create("ebpf_socket_ipc", NETDATA_THREAD_OPTION_DEFAULT, ebpf_socket_thread_ipc, NULL);

4247 + nd_thread_create("ebpf_socket_ipc", NETDATA_THREAD_OPTION_JOINABLE, ebpf_socket_thread_ipc, NULL);

4248 break;

4249 }

4250 case NETDATA_EBPF_INTEGRATION_SHM:

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

4389 cgroup_integration_thread.start_routine = ebpf_cgroup_integration;

4390

4391 cgroup_integration_thread.thread =

4392 - nd_thread_create(cgroup_integration_thread.name, NETDATA_THREAD_OPTION_DEFAULT, ebpf_cgroup_integration, NULL);

4392 + nd_thread_create(cgroup_integration_thread.name, NETDATA_THREAD_OPTION_JOINABLE, ebpf_cgroup_integration, NULL);

4393

4394 ebpf_initialize_data_sharing();

4395

src/collectors/ebpf.plugin/ebpf_cachestat.c

+1 -1

@@ -1760,7 +1760,7 @@ void *ebpf_cachestat_thread(void *ptr)

1760 pthread_mutex_unlock(&lock);

1761

1762 ebpf_read_cachestat.thread =

1763 - nd_thread_create(ebpf_read_cachestat.name, NETDATA_THREAD_OPTION_DEFAULT, ebpf_read_cachestat_thread, em);

1763 + nd_thread_create(ebpf_read_cachestat.name, NETDATA_THREAD_OPTION_JOINABLE, ebpf_read_cachestat_thread, em);

1764

1765 cachestat_collector(em);

1766

src/collectors/ebpf.plugin/ebpf_dcstat.c

+1 -1

@@ -1532,7 +1532,7 @@ void *ebpf_dcstat_thread(void *ptr)

1532 pthread_mutex_unlock(&lock);

1533

1534 ebpf_read_dcstat.thread =

1535 - nd_thread_create(ebpf_read_dcstat.name, NETDATA_THREAD_OPTION_DEFAULT, ebpf_read_dcstat_thread, em);

1535 + nd_thread_create(ebpf_read_dcstat.name, NETDATA_THREAD_OPTION_JOINABLE, ebpf_read_dcstat_thread, em);

1536

1537 dcstat_collector(em);

1538

src/collectors/ebpf.plugin/ebpf_fd.c

+1 -1

@@ -1558,7 +1558,7 @@ void *ebpf_fd_thread(void *ptr)

1558

1559 pthread_mutex_unlock(&lock);

1560

1561 - ebpf_read_fd.thread = nd_thread_create(ebpf_read_fd.name, NETDATA_THREAD_OPTION_DEFAULT, ebpf_read_fd_thread, em);

1561 + ebpf_read_fd.thread = nd_thread_create(ebpf_read_fd.name, NETDATA_THREAD_OPTION_JOINABLE, ebpf_read_fd_thread, em);

1562

1563 fd_collector(em);

1564

src/collectors/ebpf.plugin/ebpf_functions.c

+1 -1

@@ -31,7 +31,7 @@ static int ebpf_function_start_thread(ebpf_module_t *em, int period)

31 netdata_log_info("Starting thread %s with lifetime = %d", em->info.thread_name, period);

32 #endif

34 - st->thread = nd_thread_create(st->name, NETDATA_THREAD_OPTION_DEFAULT, st->start_routine, em);

34 + st->thread = nd_thread_create(st->name, NETDATA_THREAD_OPTION_JOINABLE, st->start_routine, em);

35 return st->thread ? 0 : 1;

36 }

src/collectors/ebpf.plugin/ebpf_shm.c

+1 -1

@@ -1395,7 +1395,7 @@ void *ebpf_shm_thread(void *ptr)

1395 pthread_mutex_unlock(&lock);

1396

1397 ebpf_read_shm.thread =

1398 - nd_thread_create(ebpf_read_shm.name, NETDATA_THREAD_OPTION_DEFAULT, ebpf_read_shm_thread, em);

1398 + nd_thread_create(ebpf_read_shm.name, NETDATA_THREAD_OPTION_JOINABLE, ebpf_read_shm_thread, em);

1399

1400 shm_collector(em);

1401

src/collectors/ebpf.plugin/ebpf_socket.c

+1 -1

@@ -3051,7 +3051,7 @@ void *ebpf_socket_thread(void *ptr)

3051 NETDATA_MAX_SOCKET_VECTOR);

3052

3053 ebpf_read_socket.thread =

3054 - nd_thread_create(ebpf_read_socket.name, NETDATA_THREAD_OPTION_DEFAULT, ebpf_read_socket_thread, em);

3054 + nd_thread_create(ebpf_read_socket.name, NETDATA_THREAD_OPTION_JOINABLE, ebpf_read_socket_thread, em);

3055

3056 pthread_mutex_lock(&lock);

3057 ebpf_socket_create_global_charts(em);

src/collectors/ebpf.plugin/ebpf_swap.c

+1 -1

@@ -1221,7 +1221,7 @@ void *ebpf_swap_thread(void *ptr)

1221 pthread_mutex_unlock(&lock);

1222

1223 ebpf_read_swap.thread =

1224 - nd_thread_create(ebpf_read_swap.name, NETDATA_THREAD_OPTION_DEFAULT, ebpf_read_swap_thread, em);

1224 + nd_thread_create(ebpf_read_swap.name, NETDATA_THREAD_OPTION_JOINABLE, ebpf_read_swap_thread, em);

1225

1226 swap_collector(em);

1227

src/collectors/ebpf.plugin/ebpf_vfs.c

+1 -1

@@ -2958,7 +2958,7 @@ void *ebpf_vfs_thread(void *ptr)

2958 pthread_mutex_unlock(&lock);

2959

2960 ebpf_read_vfs.thread =

2961 - nd_thread_create(ebpf_read_vfs.name, NETDATA_THREAD_OPTION_DEFAULT, ebpf_read_vfs_thread, em);

2961 + nd_thread_create(ebpf_read_vfs.name, NETDATA_THREAD_OPTION_JOINABLE, ebpf_read_vfs_thread, em);

2962

2963 vfs_collector(em);

2964

src/collectors/freeipmi.plugin/freeipmi_plugin.c

+2 -2

@@ -1982,9 +1982,9 @@ int main (int argc, char **argv) {

1982 },

1983 };

1984

1985 - nd_thread_create("IPMI[sensors]", NETDATA_THREAD_OPTION_DONT_LOG, netdata_ipmi_collection_thread, &sensors_data);

1985 + nd_thread_create("IPMI[sensors]", NETDATA_THREAD_OPTION_DONT_LOG | NETDATA_THREAD_OPTION_JOINABLE, netdata_ipmi_collection_thread, &sensors_data);

1986 if(netdata_do_sel)

1987 - nd_thread_create("IPMI[sel]", NETDATA_THREAD_OPTION_DONT_LOG, netdata_ipmi_collection_thread, &sel_data);

1987 + nd_thread_create("IPMI[sel]", NETDATA_THREAD_OPTION_DONT_LOG | NETDATA_THREAD_OPTION_JOINABLE, netdata_ipmi_collection_thread, &sel_data);

1988

1989 // ------------------------------------------------------------------------

1990 // the main loop

src/collectors/statsd.plugin/statsd.c

+1 -1

@@ -2613,7 +2613,7 @@ void *statsd_main(void *ptr) {

2613 char tag[NETDATA_THREAD_TAG_MAX + 1];

2614 snprintfz(tag, NETDATA_THREAD_TAG_MAX, "STATSD_IN[%d]", i + 1);

2615 spinlock_init(&statsd.collection_threads_status[i].spinlock);

2616 - statsd.collection_threads_status[i].thread = nd_thread_create(tag, NETDATA_THREAD_OPTION_DEFAULT,

2616 + statsd.collection_threads_status[i].thread = nd_thread_create(tag, NETDATA_THREAD_OPTION_JOINABLE,

2617 statsd_collector_thread, &statsd.collection_threads_status[i]);

2618 }

2619

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

+1 -1

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

67 // ------------------------------------------------------------------------

68 // watcher thread

70 - nd_thread_create("SDWATCH", NETDATA_THREAD_OPTION_DONT_LOG, nd_journal_watcher_main, NULL);

70 + nd_thread_create("SDWATCH", NETDATA_THREAD_OPTION_DONT_LOG | NETDATA_THREAD_OPTION_JOINABLE, nd_journal_watcher_main, NULL);

72 // ------------------------------------------------------------------------

73 // the event loop for functions

src/daemon/daemon-shutdown.c

+7 -27

@@ -70,8 +70,7 @@ void cancel_main_threads(void) {

70 if (!static_threads)

71 return;

73 - int i, found = 0;

74 - usec_t max = 5 * USEC_PER_SEC, step = 100000;

73 + int i;

74 for (i = 0; static_threads[i].name != NULL ; i++) {

75 if (static_threads[i].enabled == NETDATA_MAIN_THREAD_RUNNING) {

76 if (static_threads[i].thread) {

@@ -81,35 +80,15 @@ void cancel_main_threads(void) {

80 netdata_log_info("EXIT: No thread running (marking as EXITED): %s", static_threads[i].name);

81 static_threads[i].enabled = NETDATA_MAIN_THREAD_EXITED;

82 }

84 - found++;

83 }

84 }

88 - while(found && max > 0) {

89 - max -= step;

90 - netdata_log_info("Waiting %d threads to finish...", found);

91 - sleep_usec(step);

92 - found = 0;

93 - for (i = 0; static_threads[i].name != NULL ; i++) {

94 - if (static_threads[i].enabled == NETDATA_MAIN_THREAD_EXITED)

95 - continue;

96 -

97 - // Don't wait ourselves.

98 - if (nd_thread_is_me(static_threads[i].thread))

99 - continue;

100 -

101 - found++;

102 - }

103 - }

104 -

105 - if(found) {

106 - for (i = 0; static_threads[i].name != NULL ; i++) {

107 - if (static_threads[i].enabled != NETDATA_MAIN_THREAD_EXITED)

108 - netdata_log_error("Main thread %s takes too long to exit. Giving up...", static_threads[i].name);

109 - }

86 + for (i = 0; static_threads[i].name != NULL ; i++) {

87 + struct netdata_static_thread *st = &static_threads[i];

88 + if(st->thread && !nd_thread_is_me(static_threads[i].thread))

89 + nd_thread_join(st->thread);

90 }

111 - else

112 - netdata_log_info("All threads finished.");

91 + netdata_log_info("All threads finished.");

93 freez(static_threads);

94 static_threads = NULL;

@@ -315,6 +294,7 @@ static void netdata_cleanup_and_exit(EXIT_REASON reason, bool abnormal, bool exi

294 if (!abnormal)

295 add_agent_event(EVENT_AGENT_SHUTDOWN_TIME, (int64_t)(now_monotonic_usec() - shutdown_start_time));

296

297 + nd_thread_join_threads();

298 sqlite_close_databases();

299 watcher_step_complete(WATCHER_STEP_ID_CLOSE_SQL_DATABASES);

300 sqlite_library_shutdown();

src/daemon/main.c

+2 -2

@@ -1069,7 +1069,7 @@ int netdata_main(int argc, char **argv) {

1069

1070 if(st->enabled) {

1071 netdata_log_debug(D_SYSTEM, "Starting thread %s.", st->name);

1072 - st->thread = nd_thread_create(st->name, NETDATA_THREAD_OPTION_DEFAULT, st->start_routine, st);

1072 + st->thread = nd_thread_create(st->name, NETDATA_THREAD_OPTION_JOINABLE, st->start_routine, st);

1073 }

1074 else

1075 netdata_log_debug(D_SYSTEM, "Not starting thread %s.", st->name);

@@ -1120,7 +1120,7 @@ int netdata_main(int argc, char **argv) {

1120 struct netdata_static_thread *st = &static_threads[i];

1121 st->enabled = 1;

1122 netdata_log_debug(D_SYSTEM, "Starting thread %s.", st->name);

1123 - st->thread = nd_thread_create(st->name, NETDATA_THREAD_OPTION_DEFAULT, st->start_routine, st);

1123 + st->thread = nd_thread_create(st->name, NETDATA_THREAD_OPTION_JOINABLE, st->start_routine, st);

1124 }

1125 }

1126 }

src/exporting/exporting_engine.c

@@ -222,6 +222,7 @@ void *exporting_main(void *ptr)

222 return NULL;

223 #endif

224 }

225 + service_exits();

226

227 cleanup:

228 return NULL;

src/libnetdata/functions_evloop/functions_evloop.c

+2 -2

@@ -362,12 +362,12 @@ struct functions_evloop_globals *functions_evloop_init(size_t worker_threads, co

362

363 char tag_buffer[NETDATA_THREAD_TAG_MAX + 1];

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

365 - wg->reader_thread = nd_thread_create(tag_buffer, NETDATA_THREAD_OPTION_DONT_LOG,

365 + wg->reader_thread = nd_thread_create(tag_buffer, NETDATA_THREAD_OPTION_DONT_LOG | NETDATA_THREAD_OPTION_JOINABLE,

366 rrd_functions_worker_globals_reader_main, wg);

367

368 for(size_t i = 0; i < wg->workers ; i++) {

369 snprintfz(tag_buffer, NETDATA_THREAD_TAG_MAX, "%s_WORK[%zu]", wg->tag, i+1);

370 - wg->worker_threads[i] = nd_thread_create(tag_buffer, NETDATA_THREAD_OPTION_DONT_LOG,

370 + wg->worker_threads[i] = nd_thread_create(tag_buffer, NETDATA_THREAD_OPTION_DONT_LOG | NETDATA_THREAD_OPTION_JOINABLE,

371 rrd_functions_worker_globals_worker_main, wg);

372 }

373

src/libnetdata/threads/threads.c

+38 -26

@@ -261,28 +261,25 @@ void query_target_free(void);

261 void service_exits(void);

262 void rrd_collector_finished(void);

263

264 -static void nd_thread_join_exited_detached_threads(void) {

265 - while(1) {

264 +void nd_thread_join_threads()

265 +{

266 + ND_THREAD *nti;

267 + do {

268 spinlock_lock(&threads_globals.exited.spinlock);

269

268 - ND_THREAD *nti = threads_globals.exited.list;

269 - while (nti && nd_thread_status_check(nti, NETDATA_THREAD_OPTION_JOINABLE) == 0)

270 - nti = nti->next;

270 + nti = threads_globals.exited.list;

271

272 - if(nti) {

272 + if (nti) {

273 DOUBLE_LINKED_LIST_REMOVE_ITEM_UNSAFE(threads_globals.exited.list, nti, prev, next);

274 nti->list = ND_THREAD_LIST_NONE;

275 }

276

277 spinlock_unlock(&threads_globals.exited.spinlock);

278

279 - if(nti) {

280 - nd_log(NDLS_DAEMON, NDLP_INFO, "Joining detached thread '%s', tid %d", nti->tag, nti->tid);

281 - nd_thread_join(nti);

282 - }

283 - else

284 - break;

285 - }

279 + // handles null

280 + nd_thread_join(nti);

281 +

282 + } while (nti);

283 }

284

285 static void nd_thread_exit(ND_THREAD *nti) {

@@ -342,12 +339,12 @@ static void nd_thread_exit(ND_THREAD *nti) {

339 }

340 spinlock_unlock(&threads_globals.running.spinlock);

341

345 - if (nd_thread_status_check(nti, NETDATA_THREAD_OPTION_JOINABLE) != NETDATA_THREAD_OPTION_JOINABLE) {

346 - spinlock_lock(&threads_globals.exited.spinlock);

347 - DOUBLE_LINKED_LIST_APPEND_ITEM_UNSAFE(threads_globals.exited.list, nti, prev, next);

348 - nti->list = ND_THREAD_LIST_EXITED;

349 - spinlock_unlock(&threads_globals.exited.spinlock);

350 - }

342 + //if (nd_thread_status_check(nti, NETDATA_THREAD_OPTION_JOINABLE) != NETDATA_THREAD_OPTION_JOINABLE) {

343 + spinlock_lock(&threads_globals.exited.spinlock);

344 + DOUBLE_LINKED_LIST_APPEND_ITEM_UNSAFE(threads_globals.exited.list, nti, prev, next);

345 + nti->list = ND_THREAD_LIST_EXITED;

346 + spinlock_unlock(&threads_globals.exited.spinlock);

347 + //}

348 }

349

350 static void *nd_thread_starting_point(void *ptr) {

@@ -388,16 +385,18 @@ bool nd_thread_is_me(ND_THREAD *nti) {

385 return nti && nti->thread == pthread_self();

386 }

387

391 -ND_THREAD *nd_thread_create(const char *tag, NETDATA_THREAD_OPTIONS options, void *(*start_routine)(void *), void *arg) {

392 - nd_thread_join_exited_detached_threads();

393 -

388 +ND_THREAD *nd_thread_create(const char *tag, NETDATA_THREAD_OPTIONS options, void *(*start_routine)(void *), void *arg)

389 +{

390 ND_THREAD *nti = callocz(1, sizeof(*nti));

391 spinlock_init(&nti->canceller.spinlock);

392 nti->arg = arg;

393 nti->start_routine = start_routine;

398 - nti->options = options & NETDATA_THREAD_OPTIONS_ALL;

394 + nti->options = (options & NETDATA_THREAD_OPTIONS_ALL);

395 strncpyz(nti->tag, tag, ND_THREAD_TAG_MAX);

396

397 + if ((options & NETDATA_THREAD_OPTION_JOINABLE) == 0)

398 + nd_log_daemon(NDLP_INFO, "WARNING: Creating detached thread '%s'", tag);

399 +

400 int ret = pthread_create(&nti->thread, &threads_globals.attr, nd_thread_starting_point, nti);

401 if(ret != 0) {

402 nd_log(NDLS_DAEMON, NDLP_ERR,

@@ -447,7 +446,15 @@ int nd_thread_join(ND_THREAD *nti) {

446 if(!nti)

447 return ESRCH;

448

450 - int ret = pthread_join(nti->thread, NULL);

449 + if(nd_thread_status_check(nti, NETDATA_THREAD_STATUS_JOINED)) {

450 + freez(nti);

451 + return 0;

452 + }

453 +

454 + int ret = 0;

455 + bool joinable = nd_thread_status_check(nti, NETDATA_THREAD_OPTION_JOINABLE);

456 + if (joinable)

457 + ret = pthread_join(nti->thread, NULL);

458 if(ret != 0) {

459 // we can't join the thread

460

@@ -457,8 +464,10 @@ int nd_thread_join(ND_THREAD *nti) {

464 }

465 else {

466 // we successfully joined the thread

467 + if (joinable)

468 + nd_log(NDLS_DAEMON, NDLP_DEBUG, "Joining thread '%s', tid %d", nti->tag, nti->tid);

469

461 - nd_thread_status_set(nti, NETDATA_THREAD_STATUS_JOINED);

470 + nd_thread_status_set(nti, NETDATA_THREAD_STATUS_JOINED);

471

472 spinlock_lock(&threads_globals.running.spinlock);

473 if(nti->list == ND_THREAD_LIST_RUNNING) {

@@ -474,7 +483,10 @@ int nd_thread_join(ND_THREAD *nti) {

483 }

484 spinlock_unlock(&threads_globals.exited.spinlock);

485

477 - freez(nti);

486 + if (joinable)

487 + freez(nti);

488 + else

489 + nti->thread = 0;

490 }

491

492 return ret;

src/libnetdata/threads/threads.h

@@ -116,6 +116,7 @@ void nd_thread_rwspinlock_write_unlocked(void);

116 #endif

117

118 void nd_thread_can_run_sql(bool exclude);

119 +void nd_thread_join_threads();

120 bool nd_thread_runs_sql(void);

121

122 #endif //NETDATA_THREADS_H

src/plugins.d/plugins_d.c

+3 -1

@@ -119,6 +119,7 @@ static void pluginsd_worker_thread_handle_error(struct plugind *cd, int worker_r

119

120 #undef SERIAL_FAILURES_THRESHOLD

121

122 +

123 static void *pluginsd_worker_thread(void *arg) {

124 struct plugind *cd = (struct plugind *) arg;

125

@@ -399,7 +400,7 @@ void *pluginsd_main(void *ptr) {

400 snprintfz(tag, NETDATA_THREAD_TAG_MAX, "PD[%s]", pluginname);

401

402 // spawn a new thread for it

402 - cd->unsafe.thread = nd_thread_create(tag, NETDATA_THREAD_OPTION_DEFAULT | NETDATA_THREAD_OPTION_JOINABLE,

403 + cd->unsafe.thread = nd_thread_create(tag, NETDATA_THREAD_OPTION_JOINABLE,

404 pluginsd_worker_thread, cd);

405 }

406 }

@@ -411,5 +412,6 @@ void *pluginsd_main(void *ptr) {

412 pluginsd_sleep(scan_frequency);

413 }

414

415 + service_exits();

416 return NULL;

417 }

src/streaming/stream-connector.c

+1 -1

@@ -684,7 +684,7 @@ bool stream_connector_init(struct sender_state *s) {

684 snprintfz(tag, NETDATA_THREAD_TAG_MAX, THREAD_TAG_STREAM_SENDER "-CN" "[%d]",

685 sc->id);

686

687 - sc->thread = nd_thread_create(tag, NETDATA_THREAD_OPTION_DEFAULT, stream_connector_thread, sc);

687 + sc->thread = nd_thread_create(tag, NETDATA_THREAD_OPTION_JOINABLE, stream_connector_thread, sc);

688 if (!sc->thread)

689 nd_log_daemon(NDLP_ERR,

690 "STREAM CONNECT '%s': failed to create new thread for client.",

src/streaming/stream-thread.c

+1 -1

@@ -728,7 +728,7 @@ static struct stream_thread * stream_thread_assign_and_start(RRDHOST *host) {

728 char tag[NETDATA_THREAD_TAG_MAX + 1];

729 snprintfz(tag, NETDATA_THREAD_TAG_MAX, THREAD_TAG_STREAM "[%zu]", sth->id);

730

731 - sth->thread = nd_thread_create(tag, NETDATA_THREAD_OPTION_DEFAULT, stream_thread, sth);

731 + sth->thread = nd_thread_create(tag, NETDATA_THREAD_OPTION_JOINABLE, stream_thread, sth);

732 if (!sth->thread)

733 nd_log(NDLS_DAEMON, NDLP_ERR, "STREAM THREAD[%zu]: failed to create new thread for client.", sth->id);

734 }

src/web/server/static/static-threaded.c

+1 -1

@@ -384,7 +384,7 @@ void *socket_listen_main_static_threaded(void *ptr) {

384 char tag[50 + 1];

385 snprintfz(tag, sizeof(tag) - 1, "WEB[%d]", i+1);

386

387 - static_workers_private_data[i].thread = nd_thread_create(tag, NETDATA_THREAD_OPTION_DEFAULT,

387 + static_workers_private_data[i].thread = nd_thread_create(tag, NETDATA_THREAD_OPTION_JOINABLE,

388 socket_listen_main_static_threaded_worker,

389 (void *)&static_workers_private_data[i]);

390 }