Context loading priority to vnodes (#21040)
* host context loading priority to vnodes * Fix reset_host_context_load_flag, check if barrier is actually needed * Fix metric delete race condition * Use semaphores to better handle a timeout so as the agent startup is not blocked
Stelios Fragkakis committed
Sep 24, 2025 at 12:21 UTC
fdedec476e1be1bbe6b0be6126a033810f23b6f6
3 files changed
+102
-42
src/database/engine/mrg-internals.h
+14
-9
@@ -232,16 +232,21 @@ static bool metric_release(MRG *mrg, METRIC *metric) {
232
233
if (refcount_release(&metric->refcount) == 0) {
234
// we are the last user
235
- bool already_deleted = __atomic_load_n(&metric->deleted, __ATOMIC_ACQUIRE);
236
- if (already_deleted || !acquired_metric_has_retention(mrg, metric)) {
237
- if (!already_deleted) {
238
- acquired_for_deletion_metric_delete(mrg, metric);
235
+ if (!acquired_metric_has_retention(mrg, metric)) {
236
+ // This metric is eligible for deletion.
237
+ // Atomically check and set the 'deleted' flag.
238
+ // If __atomic_test_and_set returns 'true', it means the flag was already set.
239
+ if (!__atomic_test_and_set(&metric->deleted, __ATOMIC_ACQ_REL)) {
240
+ // We won the race. The flag was 'false' and we set it to 'true'.
241
+ // We are now responsible for deletion.
242
+ acquired_for_deletion_metric_delete(mrg, metric);
243
+ uuidmap_free(metric->uuid);
244
+ aral_freez(mrg->index[partition].aral, metric);
245
+ __atomic_sub_fetch(&mrg->index[partition].stats.entries_acquired, 1, __ATOMIC_RELAXED);
246
+ __atomic_sub_fetch(&mrg->index[partition].stats.current_references, 1, __ATOMIC_RELAXED);
247
+ return true;
248
}
240
- uuidmap_free(metric->uuid);
241
- aral_freez(mrg->index[partition].aral, metric);
242
- __atomic_sub_fetch(&mrg->index[partition].stats.entries_acquired, 1, __ATOMIC_RELAXED);
243
- __atomic_sub_fetch(&mrg->index[partition].stats.current_references, 1, __ATOMIC_RELAXED);
244
- return true;
249
+ // Another thread is already deleting it. nothing to do
250
}
251
}
252
src/database/sqlite/sqlite_aclk.c
+48
-6
@@ -92,9 +92,14 @@ enum {
92
IDX_IS_REGISTERED,
93
};
94
95
+struct children {
96
+ int vnodes;
97
+ int normal;
98
+};
99
+
100
static int create_host_callback(void *data, int argc, char **argv, char **column)
101
{
97
- int *number_of_chidren = data;
102
+ struct children *node_data = data;
103
UNUSED(argc);
104
UNUSED(column);
105
@@ -175,7 +180,10 @@ static int create_host_callback(void *data, int argc, char **argv, char **column
180
pulse_host_status(host, 0, 0); // this will detect the receiver status
181
}
182
178
- (*number_of_chidren)++;
183
+ if (IS_VIRTUAL_HOST_OS(host))
184
+ node_data->vnodes++;
185
+ else
186
+ node_data->normal++;
187
188
#ifdef NETDATA_INTERNAL_CHECKS
189
char node_str[UUID_STR_LEN] = "<none>";
@@ -973,26 +981,41 @@ void create_aclk_config(RRDHOST *host, nd_uuid_t *host_uuid __maybe_unused, nd_u
981
"SELECT ni.host_id, ni.node_id FROM host h, node_instance ni " \
982
"WHERE h.host_id = ni.host_id AND ni.node_id IS NOT NULL"
983
984
+
985
+uv_sem_t ctx_sem;
986
+
987
void aclk_synchronization_init(void)
988
{
989
char *err_msg = NULL;
990
int rc;
991
992
nd_log_daemon(NDLP_INFO, "Creating archived hosts");
982
- int number_of_children = 0;
983
- rc = sqlite3_exec_monitored(db_meta, SQL_FETCH_ALL_HOSTS, create_host_callback, &number_of_children, &err_msg);
993
+ struct children node_data = { 0, 0};
994
+
995
+ rc = sqlite3_exec_monitored(db_meta, SQL_FETCH_ALL_HOSTS, create_host_callback, &node_data, &err_msg);
996
997
if (rc != SQLITE_OK) {
998
nd_log_daemon(NDLP_ERR, "SQLite error when loading archived hosts, rc = %d (%s)", rc, err_msg);
999
sqlite3_free(err_msg);
1000
}
1001
990
- nd_log_daemon(NDLP_INFO, "Created %d archived hosts", number_of_children);
1002
+ nd_log_daemon(
1003
+ NDLP_INFO,
1004
+ "Created %d archived hosts (%d children and %d vnodes)",
1005
+ node_data.normal + node_data.vnodes,
1006
+ node_data.normal,
1007
+ node_data.vnodes);
1008
+
1009
+ bool sem_init = true;
1010
+ uv_sem_init(&ctx_sem, 0);
1011
+
1012
// Trigger host context load for hosts that have been created
1013
if (unlikely(!metadata_queue_load_host_context())) {
1014
nd_log_daemon(NDLP_WARNING, "Failed to queue command to load contexts for archived hosts");
1015
// Reset context load flag so that contexts will be loaded on demand
1016
reset_host_context_load_flag();
1017
+ uv_sem_destroy(&ctx_sem);
1018
+ sem_init = false;
1019
}
1020
1021
rc = sqlite3_exec_monitored(db_meta, SQL_FETCH_ALL_INSTANCES, aclk_config_parameters, NULL, &err_msg);
@@ -1004,9 +1027,28 @@ void aclk_synchronization_init(void)
1027
1028
aclk_initialize_event_loop();
1029
1007
- if (!number_of_children)
1030
+ if (!(node_data.normal + node_data.vnodes))
1031
aclk_queue_node_info(localhost, true);
1032
1033
+ if (sem_init) {
1034
+ int finished_vnodes = 0;
1035
+ time_t deadline = now_realtime_sec() + 60; // hard timeput to avoid infinite block
1036
+ while (finished_vnodes < node_data.vnodes) {
1037
+ if (uv_sem_trywait(&ctx_sem) == 0) {
1038
+ finished_vnodes++;
1039
+ continue;
1040
+ }
1041
+
1042
+ if (now_realtime_sec() >= deadline) {
1043
+ nd_log_daemon(NDLP_WARNING, "Vnodes context load still in progress, continue with agent start");
1044
+ break;
1045
+ }
1046
+ sleep_usec(100 * USEC_PER_MS);
1047
+ }
1048
+ if (finished_vnodes == node_data.vnodes) {
1049
+ uv_sem_destroy(&ctx_sem);
1050
+ }
1051
+ }
1052
nd_log_daemon(NDLP_INFO, "ACLK sync initialization completed");
1053
}
1054
src/database/sqlite/sqlite_metadata.c
+40
-27
@@ -1741,6 +1741,7 @@ __thread sqlite3 *db_meta_thread = NULL;
1741
__thread sqlite3 *db_context_thread = NULL;
1742
__thread bool main_context_thread = false;
1743
1744
+extern uv_sem_t ctx_sem;
1745
static void restore_host_context(void *arg)
1746
{
1747
struct host_context_load_thread *hclt = arg;
@@ -1787,6 +1788,10 @@ static void restore_host_context(void *arg)
1788
1789
aclk_queue_node_info(host, false);
1790
1791
+ if (IS_VIRTUAL_HOST_OS(host)) {
1792
+ uv_sem_post(&ctx_sem);
1793
+ }
1794
+
1795
// Check and clear the thread local variables
1796
if (!main_context_thread) {
1797
db_meta_thread = NULL;
@@ -1847,7 +1852,7 @@ void reset_host_context_load_flag()
1852
RRDHOST *host;
1853
dfe_start_reentrant(rrdhost_root_index, host)
1854
{
1850
- rrdhost_flag_set(host, RRDHOST_FLAG_PENDING_CONTEXT_LOAD);
1855
+ rrdhost_flag_clear(host, RRDHOST_FLAG_PENDING_CONTEXT_LOAD);
1856
}
1857
dfe_done(host);
1858
}
@@ -1876,36 +1881,44 @@ static void ctx_hosts_load(uv_work_t *req)
1881
size_t host_count = 0;
1882
size_t sync_exec = 0;
1883
size_t async_exec = 0;
1879
- dfe_start_reentrant(rrdhost_root_index, host) {
1880
- if (!rrdhost_flag_check(host, RRDHOST_FLAG_PENDING_CONTEXT_LOAD))
1881
- continue;
1884
1883
- if (unlikely(SHUTDOWN_REQUESTED(config)))
1884
- break;
1885
+ for (int pass=0 ; pass < 2 ; pass++) {
1886
+ dfe_start_reentrant(rrdhost_root_index, host) {
1887
+ // pass 0 will do vnodes (skip the rest)
1888
+ // pass 1 will do the rest (skip vnodes)
1889
+ if (pass == IS_VIRTUAL_HOST_OS(host))
1890
+ continue;
1891
1886
- nd_log_daemon(NDLP_DEBUG, "Loading context for host %s", rrdhost_hostname(host));
1887
-
1888
- int rc = 0;
1889
- bool thread_found = cleanup_finished_threads(hclt, max_threads, false, &thread_index);
1890
- if (thread_found) {
1891
- __atomic_store_n(&hclt[thread_index].busy, true, __ATOMIC_RELAXED);
1892
- hclt[thread_index].host = host;
1893
- hclt[thread_index].thread = nd_thread_create("CTXLOAD", NETDATA_THREAD_OPTION_DEFAULT, restore_host_context, &hclt[thread_index]);
1894
- rc = (hclt[thread_index].thread == NULL);
1895
- async_exec += (rc == 0);
1896
- // if it failed, mark the thread slot as free
1897
- if (rc)
1898
- __atomic_store_n(&hclt[thread_index].busy, false, __ATOMIC_RELAXED);
1899
- }
1900
- // if single thread, thread creation failure or failure tofind slot
1901
- if (rc || !thread_found) {
1902
- sync_exec++;
1903
- struct host_context_load_thread hclt_sync = {.host = host};
1904
- restore_host_context(&hclt_sync);
1892
+ if (!rrdhost_flag_check(host, RRDHOST_FLAG_PENDING_CONTEXT_LOAD))
1893
+ continue;
1894
+
1895
+ if (unlikely(SHUTDOWN_REQUESTED(config)))
1896
+ break;
1897
+
1898
+ nd_log_daemon(NDLP_DEBUG, "Loading context for host %s", rrdhost_hostname(host));
1899
+
1900
+ int rc = 0;
1901
+ bool thread_found = cleanup_finished_threads(hclt, max_threads, false, &thread_index);
1902
+ if (thread_found) {
1903
+ __atomic_store_n(&hclt[thread_index].busy, true, __ATOMIC_RELAXED);
1904
+ hclt[thread_index].host = host;
1905
+ hclt[thread_index].thread = nd_thread_create("CTXLOAD", NETDATA_THREAD_OPTION_DEFAULT, restore_host_context, &hclt[thread_index]);
1906
+ rc = (hclt[thread_index].thread == NULL);
1907
+ async_exec += (rc == 0);
1908
+ // if it failed, mark the thread slot as free
1909
+ if (rc)
1910
+ __atomic_store_n(&hclt[thread_index].busy, false, __ATOMIC_RELAXED);
1911
+ }
1912
+ // if single thread, thread creation failure or failure tofind slot
1913
+ if (rc || !thread_found) {
1914
+ sync_exec++;
1915
+ struct host_context_load_thread hclt_sync = {.host = host};
1916
+ restore_host_context(&hclt_sync);
1917
+ }
1918
+ host_count++;
1919
}
1906
- host_count++;
1920
+ dfe_done(host);
1921
}
1908
- dfe_done(host);
1922
1923
bool should_clean_threads = cleanup_finished_threads(hclt, max_threads, true, NULL);
1924