Fix ACLK (#22051)
* freez: Fix SIGABRT * freez: Add watchdog * freez: Address review * freez: Address review (p2) * freez: Address review (p3)
thiagoftsm committed
Mar 27, 2026 at 21:00 UTC
fa854401ac211a5eb648ab47255b7c4c86f449c1
2 files changed
+53
-13
src/database/sqlite/sqlite_aclk.c
+52
-12
@@ -418,7 +418,11 @@ static void aclk_run_query_job(uv_work_t *req)
418
struct aclk_sync_config_s *config = worker->config;
419
aclk_query_t *query = (aclk_query_t *)worker->payload;
420
421
- aclk_run_query(config, query);
421
+ // aclk_run_query() frees the query; if we're shutting down we must still free it here
422
+ if (unlikely(__atomic_load_n(&config->shutdown_requested, __ATOMIC_RELAXED)))
423
+ aclk_query_free(query);
424
+ else
425
+ aclk_run_query(config, query);
426
worker_is_idle();
427
}
428
@@ -449,7 +453,12 @@ static void aclk_execute_batch(uv_work_t *req)
453
continue;
454
455
aclk_query_t *query = *Pvalue;
452
- aclk_run_query(config, query);
456
+ // Shutdown may be requested while this batch is already running, so
457
+ // re-check before each query instead of relying on a stale snapshot.
458
+ if (unlikely(__atomic_load_n(&config->shutdown_requested, __ATOMIC_RELAXED)))
459
+ aclk_query_free(query);
460
+ else
461
+ aclk_run_query(config, query);
462
}
463
464
(void) JudyLFreeArray(&aclk_query_batch->JudyL, PJE0);
@@ -514,10 +523,16 @@ static void after_start_alert_push(uv_work_t *req, int status __maybe_unused)
523
}
524
525
// Worker thread to scan hosts for pending metadata to store
517
-static void start_alert_push(uv_work_t *req __maybe_unused)
526
+static void start_alert_push(uv_work_t *req)
527
{
528
register_libuv_worker_jobs();
529
530
+ struct worker_data *worker = req->data;
531
+ struct aclk_sync_config_s *config = worker->config;
532
+
533
+ if (unlikely(__atomic_load_n(&config->shutdown_requested, __ATOMIC_RELAXED)))
534
+ return;
535
+
536
worker_is_busy(UV_EVENT_ACLK_NODE_INFO);
537
aclk_check_node_info_and_collectors();
538
worker_is_idle();
@@ -586,8 +601,8 @@ static void timer_cb(uv_timer_t *handle)
601
config->run_query_batch = true;
602
}
603
589
-#define MAX_SHUTDOWN_TIMEOUT_SECONDS (5)
604
#define SHUTDOWN_SLEEP_INTERVAL_MS (100)
605
+#define ACLK_SHUTDOWN_WATCHDOG_TIMEOUT_SECONDS (15)
606
#define CMD_POOL_SIZE (2048)
607
608
#define ACLK_JOBS_ARE_RUNNING \
@@ -641,11 +656,11 @@ static void aclk_synchronization_event_loop(void *arg)
656
Pvoid_t *Pvalue;
657
worker_data_t *worker;
658
644
- config->shutdown_requested = false;
659
+ __atomic_store_n(&config->shutdown_requested, false, __ATOMIC_RELAXED);
660
config->initialized = true;
661
completion_mark_complete(&config->start_stop_complete);
662
648
- while (likely(config->shutdown_requested == false)) {
663
+ while (likely(!__atomic_load_n(&config->shutdown_requested, __ATOMIC_RELAXED))) {
664
enum aclk_database_opcode opcode;
665
RRDHOST *host;
666
struct aclk_sync_cfg_t *aclk_host_config;
@@ -879,7 +894,7 @@ static void aclk_synchronization_event_loop(void *arg)
894
}
895
break;
896
case ACLK_SYNC_SHUTDOWN:
882
- config->shutdown_requested = true;
897
+ __atomic_store_n(&config->shutdown_requested, true, __ATOMIC_RELAXED);
898
mark_pending_req_cancel_all();
899
break;
900
default:
@@ -899,13 +914,38 @@ static void aclk_synchronization_event_loop(void *arg)
914
uv_close((uv_handle_t *)&config->async, NULL);
915
uv_walk(loop, libuv_close_callback, NULL);
916
902
- size_t loop_count = (MAX_SHUTDOWN_TIMEOUT_SECONDS * MSEC_PER_SEC) / SHUTDOWN_SLEEP_INTERVAL_MS;
917
+ size_t shutdown_wait_iterations = 0;
918
+ const size_t log_every_iterations = (10 * MSEC_PER_SEC) / SHUTDOWN_SLEEP_INTERVAL_MS;
919
+ const size_t watchdog_iterations = (ACLK_SHUTDOWN_WATCHDOG_TIMEOUT_SECONDS * MSEC_PER_SEC) / SHUTDOWN_SLEEP_INTERVAL_MS;
920
+
921
+ while (ACLK_JOBS_ARE_RUNNING || uv_loop_alive(loop)) {
922
+ (void)uv_run(loop, UV_RUN_NOWAIT);
923
+
924
+ shutdown_wait_iterations++;
925
+
926
+ if (shutdown_wait_iterations >= watchdog_iterations) {
927
+ nd_log_daemon(
928
+ NDLP_ERR,
929
+ "ACLK: shutdown watchdog timeout (%d seconds) exceeded, abandoning outstanding libuv jobs "
930
+ "(queries_running=%d, alert_push_running=%d, batch_job_running=%d)",
931
+ ACLK_SHUTDOWN_WATCHDOG_TIMEOUT_SECONDS,
932
+ config->aclk_queries_running,
933
+ config->alert_push_running,
934
+ config->aclk_batch_job_is_running);
935
+ break;
936
+ }
937
+
938
+ if ((shutdown_wait_iterations % log_every_iterations) == 0) {
939
+ nd_log_daemon(
940
+ NDLP_WARNING,
941
+ "ACLK: waiting for outstanding libuv jobs during shutdown "
942
+ "(queries_running=%d, alert_push_running=%d, batch_job_running=%d)",
943
+ config->aclk_queries_running,
944
+ config->alert_push_running,
945
+ config->aclk_batch_job_is_running);
946
+ }
947
904
- while (ACLK_JOBS_ARE_RUNNING && loop_count > 0) {
905
- if (!uv_run(loop, UV_RUN_NOWAIT))
906
- break; // No pending callbacks
948
sleep_usec(SHUTDOWN_SLEEP_INTERVAL_MS * USEC_PER_MS);
908
- loop_count--;
949
}
950
951
(void) uv_loop_close(loop);
src/database/sqlite/sqlite_metadata.c
+1
-1
@@ -2829,7 +2829,7 @@ static void metadata_event_loop(void *arg)
2829
// are we waiting for callbacks?
2830
bool callbacks_pending = (config->metadata_running || config->ctx_load_running);
2831
2832
- while ((config->metadata_running || config->ctx_load_running) && loop_count > 0) {
2832
+ while (((config->metadata_running || config->ctx_load_running) || uv_loop_alive(loop)) && loop_count > 0) {
2833
callbacks_pending = uv_run(loop, UV_RUN_NOWAIT);
2834
if (!callbacks_pending)
2835
break; // No pending callbacks