Improve ML host access (#22559)
* Use atomic operations to ensure safe access and updates of `ml_host` in ML code paths. * Remove unused `queue()` accessor in `ml_dimension` and handle `ml_host` access concurrency in `ml.cc`. * Use atomic exchange for safe detachment of `ml_host` and `ml_chart` ensuring strict ordering for concurrent readers.
Stelios Fragkakis committed
May 26, 2026 at 08:02 UTC
04d4a59d7c31e422198c2b64ce77242653605732
3 files changed
+54
-35
src/ml/ml.cc
+15
-5
@@ -633,12 +633,21 @@ ml_dimension_deserialize_kmeans(const char *json_str)
633
return true;
634
}
635
636
+ // ml_host may have been unpublished by ml_host_delete() concurrently;
637
+ // the acquired RRDHOST keeps RH alive but not RH->ml_host.
638
+ ml_host_t *host = AcqDim.host();
639
+ if (!host) {
640
+ pulse_ml_models_ignored();
641
+ json_object_put(root);
642
+ return true;
643
+ }
644
+
645
ml_queue_item_t item;
646
item.type = ML_QUEUE_ITEM_TYPE_ADD_EXISTING_MODEL;
647
item.add_existing_model = {
648
DLI, inlined_km
649
};
641
- ml_queue_push(AcqDim.queue(), item);
650
+ ml_queue_push(host->queue, item);
651
652
json_object_put(root);
653
return true;
@@ -723,7 +732,7 @@ static bool ml_dimension_update_models(ml_worker_t *worker, ml_dimension_t *dim,
732
733
spinlock_lock(&dim->slock);
734
726
- ml_host_t *host = (ml_host_t *) dim->rd->rrdset->rrdhost->ml_host;
735
+ ml_host_t *host = (ml_host_t *) __atomic_load_n(&dim->rd->rrdset->rrdhost->ml_host, __ATOMIC_ACQUIRE);
736
if (!ml_should_publish_model_update(host && host->ml_running,
737
dim->reset_generation,
738
expected_generation,
@@ -1174,13 +1183,14 @@ void ml_detect_main(void *arg)
1183
RRDHOST *rh;
1184
rrd_rdlock();
1185
rrdhost_foreach_read(rh) {
1177
- if (!rh->ml_host)
1186
+ ml_host_t *host = (ml_host_t *) __atomic_load_n(&rh->ml_host, __ATOMIC_ACQUIRE);
1187
+ if (!host)
1188
continue;
1189
1190
if (!service_running(SERVICE_COLLECTORS))
1191
break;
1192
1183
- ml_host_detect_once((ml_host_t *) rh->ml_host, detect_owa);
1193
+ ml_host_detect_once(host, detect_owa);
1194
}
1195
rrd_rdunlock();
1196
@@ -1282,7 +1292,7 @@ static enum ml_worker_result ml_worker_add_existing_model(ml_worker_t *worker, m
1292
return ML_WORKER_RESULT_OK;
1293
}
1294
1285
- ml_host_t *host = (ml_host_t *) Dim->rd->rrdset->rrdhost->ml_host;
1295
+ ml_host_t *host = (ml_host_t *) __atomic_load_n(&Dim->rd->rrdset->rrdhost->ml_host, __ATOMIC_ACQUIRE);
1296
if (!host || !host->ml_running) {
1297
pulse_ml_models_ignored();
1298
return ML_WORKER_RESULT_OK;
src/ml/ml_dimension.h
+1
-6
@@ -134,12 +134,7 @@ public:
134
ml_host_t *host() const {
135
assert(acquired());
136
RRDHOST *RH = rrdhost_acquired_to_rrdhost(AcqRH);
137
- return reinterpret_cast<ml_host_t *>(RH->ml_host);
138
- }
139
-
140
- ml_queue_t *queue() const {
141
- assert(acquired());
142
- return host()->queue;
137
+ return reinterpret_cast<ml_host_t *>(__atomic_load_n(&RH->ml_host, __ATOMIC_ACQUIRE));
138
}
139
140
ml_dimension_t *dimension() const {
src/ml/ml_public.cc
+38
-24
@@ -22,7 +22,7 @@ static void ml_host_clear_context_anomaly_rate(ml_host_t *host)
22
23
static void ml_dimension_enqueue_create_model(RRDHOST *rh, RRDDIM *rd)
24
{
25
- ml_host_t *host = (ml_host_t *) rh->ml_host;
25
+ ml_host_t *host = (ml_host_t *) __atomic_load_n(&rh->ml_host, __ATOMIC_ACQUIRE);
26
if (!host)
27
return;
28
@@ -95,12 +95,26 @@ void ml_host_new(RRDHOST *rh)
95
spinlock_init(&host->context_anomaly_rate_spinlock);
96
97
host->ml_running = false;
98
- rh->ml_host = (rrd_ml_host_t *) host;
98
+
99
+ // Publish with release semantics so readers that load rh->ml_host with
100
+ // acquire semantics observe the host's `rh`, `ml_running`, `mutex`,
101
+ // `queue`, etc. as fully initialized. Without this, the C++ compiler
102
+ // may reorder field stores after the publish store of rh->ml_host, and
103
+ // a concurrent reader would see host != NULL with partially-initialized
104
+ // fields, producing SIGSEGV faults inside ml_dimension_is_anomalous and
105
+ // similar readers.
106
+ __atomic_store_n(&rh->ml_host, (rrd_ml_host_t *)host, __ATOMIC_RELEASE);
107
}
108
109
void ml_host_delete(RRDHOST *rh)
110
{
103
- ml_host_t *host = (ml_host_t *) rh->ml_host;
111
+ // Atomically detach `rh->ml_host` and obtain the previous pointer in a
112
+ // single RMW. Using exchange (rather than separate load + store) keeps
113
+ // the unpublish and the freeing on this thread strictly ordered: no
114
+ // store/operation that follows can be reordered before the unpublish,
115
+ // so concurrent readers observe either the live host or NULL -- never
116
+ // the freed host memory.
117
+ ml_host_t *host = (ml_host_t *) __atomic_exchange_n(&rh->ml_host, (rrd_ml_host_t *)NULL, __ATOMIC_ACQ_REL);
118
if (!host)
119
return;
120
@@ -108,11 +122,10 @@ void ml_host_delete(RRDHOST *rh)
122
netdata_mutex_destroy(&host->mutex);
123
124
delete host;
111
- rh->ml_host = NULL;
125
}
126
127
void ml_host_start(RRDHOST *rh) {
115
- ml_host_t *host = (ml_host_t *) rh->ml_host;
128
+ ml_host_t *host = (ml_host_t *) __atomic_load_n(&rh->ml_host, __ATOMIC_ACQUIRE);
129
if (!host)
130
return;
131
@@ -146,7 +159,7 @@ void ml_host_start(RRDHOST *rh) {
159
}
160
161
void ml_host_stop(RRDHOST *rh) {
149
- ml_host_t *host = (ml_host_t *) rh->ml_host;
162
+ ml_host_t *host = (ml_host_t *) __atomic_load_n(&rh->ml_host, __ATOMIC_ACQUIRE);
163
if (!host || !host->ml_running)
164
return;
165
@@ -210,7 +223,7 @@ void ml_host_stop(RRDHOST *rh) {
223
224
void ml_host_get_info(RRDHOST *rh, BUFFER *wb)
225
{
213
- ml_host_t *host = (ml_host_t *) rh->ml_host;
226
+ ml_host_t *host = (ml_host_t *) __atomic_load_n(&rh->ml_host, __ATOMIC_ACQUIRE);
227
if (!host) {
228
buffer_json_member_add_boolean(wb, "enabled", false);
229
return;
@@ -243,7 +256,7 @@ void ml_host_get_info(RRDHOST *rh, BUFFER *wb)
256
257
void ml_host_get_detection_info(RRDHOST *rh, BUFFER *wb)
258
{
246
- ml_host_t *host = (ml_host_t *) rh->ml_host;
259
+ ml_host_t *host = (ml_host_t *) __atomic_load_n(&rh->ml_host, __ATOMIC_ACQUIRE);
260
if (!host)
261
return;
262
@@ -261,7 +274,7 @@ void ml_host_get_detection_info(RRDHOST *rh, BUFFER *wb)
274
}
275
276
bool ml_host_get_host_status(RRDHOST *rh, struct ml_metrics_statistics *mlm) {
264
- ml_host_t *host = (ml_host_t *) rh->ml_host;
277
+ ml_host_t *host = (ml_host_t *) __atomic_load_n(&rh->ml_host, __ATOMIC_ACQUIRE);
278
if (!host) {
279
memset(mlm, 0, sizeof(*mlm));
280
return false;
@@ -281,7 +294,7 @@ bool ml_host_get_host_status(RRDHOST *rh, struct ml_metrics_statistics *mlm) {
294
}
295
296
bool ml_host_running(RRDHOST *rh) {
284
- ml_host_t *host = (ml_host_t *) rh->ml_host;
297
+ ml_host_t *host = (ml_host_t *) __atomic_load_n(&rh->ml_host, __ATOMIC_ACQUIRE);
298
if(!host)
299
return false;
300
@@ -299,7 +312,7 @@ void ml_host_get_models(RRDHOST *rh, BUFFER *wb)
312
313
void ml_chart_new(RRDSET *rs)
314
{
302
- ml_host_t *host = (ml_host_t *) rs->rrdhost->ml_host;
315
+ ml_host_t *host = (ml_host_t *) __atomic_load_n(&rs->rrdhost->ml_host, __ATOMIC_ACQUIRE);
316
if (!host)
317
return;
318
@@ -320,16 +333,17 @@ void ml_chart_new(RRDSET *rs)
333
334
void ml_chart_delete(RRDSET *rs)
335
{
323
- ml_host_t *host = (ml_host_t *) rs->rrdhost->ml_host;
336
+ ml_host_t *host = (ml_host_t *) __atomic_load_n(&rs->rrdhost->ml_host, __ATOMIC_ACQUIRE);
337
if (!host)
338
return;
339
327
- ml_chart_t *chart = (ml_chart_t *) __atomic_load_n(&rs->ml_chart, __ATOMIC_ACQUIRE);
328
-
329
- // Unpublish BEFORE freeing so a concurrent reader that loads rs->ml_chart
330
- // observes either the live chart (with chart->rs set) or NULL -- never
331
- // the freed chart memory.
332
- __atomic_store_n(&rs->ml_chart, (rrd_ml_chart_t *)NULL, __ATOMIC_RELEASE);
340
+ // Atomically detach `rs->ml_chart` and obtain the previous pointer in a
341
+ // single RMW. Using exchange (rather than separate load + store) keeps
342
+ // the unpublish and the freeing on this thread strictly ordered: no
343
+ // store/operation that follows can be reordered before the unpublish,
344
+ // so concurrent readers observe either the live chart (with chart->rs
345
+ // set) or NULL -- never the freed chart memory.
346
+ ml_chart_t *chart = (ml_chart_t *) __atomic_exchange_n(&rs->ml_chart, (rrd_ml_chart_t *)NULL, __ATOMIC_ACQ_REL);
347
delete chart;
348
}
349
@@ -389,7 +403,7 @@ void ml_dimension_new(RRDDIM *rd)
403
// will sweep all untrained dimensions and enqueue them when it runs.
404
// This avoids double-enqueueing the same dim from both paths.
405
RRDHOST *rh = rd->rrdset->rrdhost;
392
- ml_host_t *host = (ml_host_t *) rh->ml_host;
406
+ ml_host_t *host = (ml_host_t *) __atomic_load_n(&rh->ml_host, __ATOMIC_ACQUIRE);
407
if (host && host->ml_running)
408
ml_dimension_enqueue_create_model(rh, rd);
409
}
@@ -431,8 +445,8 @@ ALWAYS_INLINE_ONLY void ml_dimension_received_anomaly(RRDDIM *rd, bool is_anomal
445
if (!dim)
446
return;
447
434
- ml_host_t *host = (ml_host_t *) rd->rrdset->rrdhost->ml_host;
435
- if (!host->ml_running)
448
+ ml_host_t *host = (ml_host_t *) __atomic_load_n(&rd->rrdset->rrdhost->ml_host, __ATOMIC_ACQUIRE);
449
+ if (!host || !host->ml_running)
450
return;
451
452
ml_chart_t *chart = (ml_chart_t *) __atomic_load_n(&rd->rrdset->ml_chart, __ATOMIC_ACQUIRE);
@@ -450,8 +464,8 @@ bool ml_dimension_is_anomalous(RRDDIM *rd, time_t curr_time, double value, bool
464
if (!dim)
465
return false;
466
453
- ml_host_t *host = (ml_host_t *) rd->rrdset->rrdhost->ml_host;
454
- if (!host->ml_running)
467
+ ml_host_t *host = (ml_host_t *) __atomic_load_n(&rd->rrdset->rrdhost->ml_host, __ATOMIC_ACQUIRE);
468
+ if (!host || !host->ml_running)
469
return false;
470
471
ml_chart_t *chart = (ml_chart_t *) __atomic_load_n(&rd->rrdset->ml_chart, __ATOMIC_ACQUIRE);
@@ -622,7 +636,7 @@ bool ml_model_received_from_child(RRDHOST *host, const char *json)
636
}
637
638
void ml_host_disconnected(RRDHOST *rh) {
625
- ml_host_t *host = (ml_host_t *) rh->ml_host;
639
+ ml_host_t *host = (ml_host_t *) __atomic_load_n(&rh->ml_host, __ATOMIC_ACQUIRE);
640
if (!host)
641
return;
642