@cryptotaxi247 / netdata-1 / commits / 71e08dba6

Improve agent shutdown (#20280)

* Join web threads on shutdown Join cgroups discover thread No capabilities on LWT Free statsd structure / join threads Signal more services on shutdown Exporting engine free instance Join statsd collection threads * Message doing shutdown for pending threads to join * Add missing lock * Do not join main WEB thread * Attempt to join only exited threads * Switch to sleep_usec * Make sure to finalize statements before setting thread exit status * Revert some of the signalling for now Fix condition for CTXLOAD thread creation failure

Stelios Fragkakis committed May 16, 2025 at 10:33 UTC 71e08dba6bfcdb478aaf5872a7b340ec09524f0d

9 files changed +55 -30

src/aclk/aclk_tx_msgs.c

+1 -1

@@ -204,7 +204,7 @@ uint16_t aclk_send_agent_connection_update(mqtt_wss_client client, int reachable

204 .reachable = (reachable ? 1 : 0),

205 .lwt = 0,

206 .session_id = aclk_session_newarch,

207 - .capabilities = aclk_get_agent_capas()

207 + .capabilities = aclk_get_agent_capas(),

208 };

209

210 CLAIM_ID claim_id = claim_id_get();

src/collectors/cgroups.plugin/sys_fs_cgroup.c

@@ -1342,6 +1342,9 @@ static void cgroup_main_cleanup(void *pptr) {

1342 sleep_usec(step);

1343 }

1344 }

1345 + // We should be done, but just in case, avoid blocking shutdown

1346 + if (__atomic_load_n(&discovery_thread.exited, __ATOMIC_RELAXED))

1347 + (void) nd_thread_join(discovery_thread.thread);

1348

1349 static_thread->enabled = NETDATA_MAIN_THREAD_EXITED;

1350 }

src/collectors/statsd.plugin/statsd.c

+14 -16

@@ -234,7 +234,7 @@ typedef struct statsd_app {

234

235 struct collection_thread_status {

236 SPINLOCK spinlock;

237 - bool running;

237 + bool initializing;

238 uint32_t max_sockets;

239

240 ND_THREAD *thread;

@@ -1083,10 +1083,6 @@ void statsd_collector_thread_cleanup(void *pptr) {

1083 struct statsd_udp *d = CLEANUP_FUNCTION_GET_PTR(pptr);

1084 if(!d) return;

1085

1086 - spinlock_lock(&d->status->spinlock);

1087 - d->status->running = false;

1088 - spinlock_unlock(&d->status->spinlock);

1089 -

1086 #ifdef HAVE_RECVMMSG

1087 size_t i;

1088 for (i = 0; i < d->size; i++)

@@ -1107,7 +1103,7 @@ static bool statsd_should_stop(void) {

1103 void *statsd_collector_thread(void *ptr) {

1104 struct collection_thread_status *status = ptr;

1105 spinlock_lock(&status->spinlock);

1110 - status->running = true;

1106 + status->initializing = false;

1107 spinlock_unlock(&status->spinlock);

1108

1109 worker_register("STATSD");

@@ -2402,17 +2398,18 @@ static void statsd_main_cleanup(void *pptr) {

2398 if (statsd.collection_threads_status) {

2399 int i;

2400 for (i = 0; i < statsd.threads; i++) {

2405 - spinlock_lock(&statsd.collection_threads_status[i].spinlock);

2406 -

2407 - if(statsd.collection_threads_status[i].running) {

2408 - collector_info("STATSD: signalling data collection thread %d to stop...", i + 1);

2409 - nd_thread_signal_cancel(statsd.collection_threads_status[i].thread);

2410 - }

2411 - else

2412 - collector_info("STATSD: data collection thread %d found stopped.", i + 1);

2413 -

2414 - spinlock_unlock(&statsd.collection_threads_status[i].spinlock);

2401 + bool initializing;

2402 + do {

2403 + spinlock_lock(&statsd.collection_threads_status[i].spinlock);

2404 + initializing = statsd.collection_threads_status[i].initializing;

2405 + spinlock_unlock(&statsd.collection_threads_status[i].spinlock);

2406 + if (unlikely(initializing))

2407 + sleep_usec(1000);

2408 + } while(initializing);

2409 +

2410 + (void) nd_thread_join(statsd.collection_threads_status[i].thread);

2411 }

2412 + freez(statsd.collection_threads_status);

2413 }

2414

2415 collector_info("STATSD: closing sockets...");

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

2610 char tag[NETDATA_THREAD_TAG_MAX + 1];

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

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

2613 + statsd.collection_threads_status[i].initializing = true;

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

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

2616 }

src/daemon/daemon-shutdown.c

+8 -4

@@ -85,9 +85,10 @@ void cancel_main_threads(void) {

85 }

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

88 - struct netdata_static_thread *st = &static_threads[i];

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

90 - nd_thread_join(st->thread);

88 + if(static_threads[i].thread && !nd_thread_is_me(static_threads[i].thread)) {

89 + if (static_threads[i].enabled == NETDATA_MAIN_THREAD_EXITED)

90 + nd_thread_join(static_threads[i].thread);

91 + }

92 }

93 netdata_log_info("All threads finished.");

@@ -197,8 +198,11 @@ static void netdata_cleanup_and_exit(EXIT_REASON reason, bool abnormal, bool exi

198 webrtc_close_all_connections();

199 watcher_step_complete(WATCHER_STEP_ID_CLOSE_WEBRTC_CONNECTIONS);

200

200 - service_signal_exit(SERVICE_MAINTENANCE | ABILITY_DATA_QUERIES | ABILITY_WEB_REQUESTS |

201 + service_signal_exit(SERVICE_MAINTENANCE | ABILITY_DATA_QUERIES | ABILITY_WEB_REQUESTS | SERVICE_ACLK |

202 ABILITY_STREAMING_CONNECTIONS | SERVICE_SYSTEMD);

203 +

204 + service_signal_exit(SERVICE_EXPORTERS | SERVICE_HEALTH | SERVICE_WEB_SERVER | SERVICE_HTTPD);

205 +

206 watcher_step_complete(WATCHER_STEP_ID_DISABLE_MAINTENANCE_NEW_QUERIES_NEW_WEB_REQUESTS_NEW_STREAMING_CONNECTIONS);

207

208 service_wait_exit(SERVICE_MAINTENANCE | SERVICE_SYSTEMD, 5 * USEC_PER_SEC);

src/database/sqlite/sqlite_metadata.c

+2 -1

@@ -1935,7 +1935,8 @@ static void ctx_hosts_load(uv_work_t *req)

1935 __atomic_store_n(&hclt[thread_index].busy, true, __ATOMIC_RELAXED);

1936 hclt[thread_index].host = host;

1937 hclt[thread_index].thread = nd_thread_create("CTXLOAD", NETDATA_THREAD_OPTION_DEFAULT, restore_host_context, &hclt[thread_index]);

1938 - async_exec += (hclt[thread_index].thread != NULL);

1938 + rc = (hclt[thread_index].thread == NULL);

1939 + async_exec += (rc == 0);

1940 // if it failed, mark the thread slot as free

1941 if (rc)

1942 __atomic_store_n(&hclt[thread_index].busy, false, __ATOMIC_RELAXED);

src/exporting/exporting_engine.c

@@ -106,6 +106,7 @@ static void exporting_clean_engine()

106 instance = instance->next;

107

108 clean_instance(current_instance);

109 + freez(current_instance);

110 }

111

112 freez((void *)engine->config.hostname);

src/health/health_event_loop.c

+1 -2

@@ -702,9 +702,8 @@ static void health_main_cleanup(void *pptr) {

702

703 worker_unregister();

704 static_thread->enabled = NETDATA_MAIN_THREAD_EXITING;

705 - static_thread->enabled = NETDATA_MAIN_THREAD_EXITED;

706 -

705 finalize_self_prepared_sql_statements();

706 + static_thread->enabled = NETDATA_MAIN_THREAD_EXITED;

707 nd_log(NDLS_DAEMON, NDLP_DEBUG, "Health thread ended.");

708 }

709

src/libnetdata/threads/threads.c

+2 -3

@@ -272,6 +272,7 @@ void nd_thread_join_threads()

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 + nd_log_daemon(NDLP_DEBUG, "nd_thread_join_threads: Joining thread with id %d (%s) during shutdown", nti->tid, nti->tag);

276 }

277

278 spinlock_unlock(&threads_globals.exited.spinlock);

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

458 if(!nti)

459 return ESRCH;

460

460 - if(nd_thread_status_check(nti, NETDATA_THREAD_STATUS_JOINED)) {

461 - freez(nti);

461 + if(nd_thread_status_check(nti, NETDATA_THREAD_STATUS_JOINED))

462 return 0;

463 - }

463

464 int ret;

465 if((ret = uv_thread_join(&nti->thread))) {

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

+23 -3

@@ -64,7 +64,8 @@ struct web_server_static_threaded_worker {

64 ND_THREAD *thread;

66 int id;

67 - int running;

67 + bool initializing;

68 + SPINLOCK spinlock;

70 size_t max_sockets;

@@ -274,7 +275,6 @@ static void socket_listen_main_static_threaded_worker_cleanup(void *pptr) {

275 worker_private->sends

276 );

277

277 - worker_private->running = 0;

278 worker_unregister();

279 }

280

@@ -284,7 +284,9 @@ static bool web_server_should_stop(void) {

284

285 void *socket_listen_main_static_threaded_worker(void *ptr) {

286 worker_private = ptr;

287 - worker_private->running = 1;

287 + spinlock_lock(&worker_private->spinlock);

288 + worker_private->initializing = false;

289 + spinlock_unlock(&worker_private->spinlock);

290 worker_register("WEB");

291 worker_register_job_name(WORKER_JOB_ADD_CONNECTION, "connect");

292 worker_register_job_name(WORKER_JOB_DEL_COLLECTION, "disconnect");

@@ -361,6 +363,20 @@ static void socket_listen_main_static_threaded_cleanup(void *pptr) {

363 listen_sockets_close(&api_sockets);

364

365 netdata_log_info("all static web threads stopped.");

366 +

367 + // Lets join all threads

368 + for (int i = 1; i < static_threaded_workers_count; i++) {

369 + bool initializing;

370 + do {

371 + spinlock_lock(&static_workers_private_data[i].spinlock);

372 + initializing = static_workers_private_data[i].initializing;

373 + spinlock_unlock(&static_workers_private_data[i].spinlock);

374 + if (unlikely(initializing))

375 + sleep_usec(1000);

376 + } while(initializing);

377 + (void) nd_thread_join(static_workers_private_data[i].thread);

378 + }

379 +

380 static_thread->enabled = NETDATA_MAIN_THREAD_EXITED;

381 }

382

@@ -387,6 +403,8 @@ void *socket_listen_main_static_threaded(void *ptr) {

403 sizeof(struct web_server_static_threaded_worker));

404

405 int i;

406 + spinlock_init(&static_workers_private_data[0].spinlock);

407 + static_workers_private_data[0].initializing = true;

408 for (i = 1; i < static_threaded_workers_count; i++) {

409 static_workers_private_data[i].id = i;

410 static_workers_private_data[i].max_sockets = max_sockets / static_threaded_workers_count;

@@ -394,6 +412,8 @@ void *socket_listen_main_static_threaded(void *ptr) {

412 char tag[50 + 1];

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

414

415 + spinlock_init(&static_workers_private_data[i].spinlock);

416 + static_workers_private_data[i].initializing = true;

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

418 socket_listen_main_static_threaded_worker,

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