Prevent race condition during pluginsd array operations (#21628)
* Refactor pluginsd dimension array handling to use reference-counted structure, improve memory safety, and address potential race conditions. * Fix compilation (add missing file) More fixes * Add PRD_ARRAY refcount stress test and improve thread safety with spinlocks * Improve thread safety and optimize cleanup logic for PRD_ARRAY. Use `collector_tid` to skip cleanup during active collection, enforce spinlock usage in `prd_array_replace`, and update stress test with skipped reader tracking. * Ensure the collector is fully stopped before cleanup. Improve stress test * Code cleanup * Cleanup: use sleep_usec() * Ensure proper cleanup of `pluginsd_chart_slots` when collector is active * Improve thread safety: avoid clearing `collector_tid` when `old_st == st` * Improve thread safety and cleanup logic for PRD_ARRAY and chart slots: - Add checks to detect refcount underflow and lifecycle violations. - Improve handling of `collector_tid` for detecting active collectors. - Ensure proper usage of spinlocks when modifying chart slots and PRD_ARRAY. - Refine comments and internal safety checks to clarify valid usage scenarios. * database: extract chart slot mapping cleanup helper * pluginsd: harden collector_tid and PRD array cleanup handoff * - Add null checks for `rda` to prevent dereferencing invalid pointers. - Ensure `prd->rd` and `prd->id` are properly set when `rda` is null. * Add null checks for `prd->id` and `arr->entries[t].id` in pluginsd internals; cleanup unused variable in `rrdset-slots.c`. * Address review comments - Move reset of `last_slot` and `dims_with_slots` for safer lifecycle handling. - Add explicit handling for collector thread cleanup to detect lifecycle violations. - Replace redundant `gettid_cached` calls with `current_tid` to optimize performance. - Improve error logging for slot cache refresh failures and stale collector cases. * Address review comments part 2 - Add detailed logging for refcount underflow and unexpected lifecycle violations. - Ensure proper cleanup of PRD_ARRAY and detached references with spinlock protection. - Refactor slot clearing logic to prevent races and double-releases during unslotting. - Track memory being freed only when exclusive ownership is confirmed. - Refine comments for better clarity on lifecycle handling and concurrency safeguards. * - Replace `memcpy` with explicit per-field copying to avoid stale `rda` references. - Reacquire dimensions during slot cache growth for independent lifecycle management. - Add detailed logging for slot cleanup and delayed free scenarios. - Track memory only when exclusive ownership of old arrays is confirmed. * Add spinlock protection during slot cache growth to ensure concurrency safety and prevent races. * Improve memory tracking and spinlock-protected lifecycle handling for PRD_ARRAY and chart slots. Fix thread safety issues, optimize slot growth logic, and enhance error logging.
Stelios Fragkakis committed
Mar 13, 2026 at 12:50 UTC
ab477a2814b78164d84ab659ee105e4482c877b7
9 files changed
+821
-82
src/daemon/main.c
+6
@@ -9,6 +9,7 @@
9
#include "web/mcp/mcp.h"
10
11
#include "database/engine/page_test.h"
12
+#include "database/rrdset-slots.h"
13
#include <curl/curl.h>
14
15
#ifdef OS_WINDOWS
@@ -154,6 +155,7 @@ int help(int exitcode) {
155
" size of E MiB, an optional disk space limit\n"
156
" of F MiB, G libuv workers (default 16) and exit.\n\n"
157
#endif
158
+ " -W prd-array-stress Run PRD_ARRAY refcount stress test and exit.\n\n"
159
" -W set section option value\n"
160
" set netdata.conf option from the command line.\n\n"
161
" -W buildinfo Print the version, the configure options,\n"
@@ -472,6 +474,10 @@ int netdata_main(int argc, char **argv) {
474
unittest_running = true;
475
return rwlocks_stress_test();
476
}
477
+ else if(strcmp(optarg, "prd-array-stress") == 0) {
478
+ unittest_running = true;
479
+ return prd_array_stress_test();
480
+ }
481
else if(strcmp(optarg, "stringtest") == 0) {
482
unittest_running = true;
483
return string_unittest(10000);
src/database/rrdhost-slots.c
+19
-3
@@ -23,13 +23,29 @@ void rrdhost_pluginsd_send_chart_slots_free(RRDHOST *host) {
23
}
24
25
void rrdhost_pluginsd_receive_chart_slots_free(RRDHOST *host) {
26
- rrd_slot_memory_removed(host->stream.rcv.pluginsd_chart_slots.size * sizeof(uint32_t));
26
+ rrd_slot_memory_removed(host->stream.rcv.pluginsd_chart_slots.size * sizeof(RRDSET *));
27
28
spinlock_lock(&host->stream.rcv.pluginsd_chart_slots.spinlock);
29
30
if(host->stream.rcv.pluginsd_chart_slots.array) {
31
- for (size_t s = 0; s < host->stream.rcv.pluginsd_chart_slots.size; s++)
32
- rrdset_pluginsd_receive_unslot_and_cleanup(host->stream.rcv.pluginsd_chart_slots.array[s]);
31
+ for (size_t s = 0; s < host->stream.rcv.pluginsd_chart_slots.size; s++) {
32
+ RRDSET *st = host->stream.rcv.pluginsd_chart_slots.array[s];
33
+ if(st) {
34
+ // Clear collector_tid - the collector is already stopped
35
+ // (stream_receiver_signal_to_stop_and_wait was called before this)
36
+ // so it's safe to cleanup regardless of the previous collector_tid value
37
+ __atomic_store_n(&st->pluginsd.collector_tid, 0, __ATOMIC_RELEASE);
38
+
39
+ // Pre-clear last_slot so that rrdset_pluginsd_receive_unslot_and_cleanup
40
+ // won't try to re-acquire the host spinlock we already hold.
41
+ // This prevents recursive locking on pluginsd_chart_slots.spinlock.
42
+ // We're freeing the entire host slots array below, so clearing individual
43
+ // slot entries is unnecessary.
44
+ st->pluginsd.last_slot = -1;
45
+
46
+ rrdset_pluginsd_receive_unslot_and_cleanup(st);
47
+ }
48
+ }
49
50
freez(host->stream.rcv.pluginsd_chart_slots.array);
51
host->stream.rcv.pluginsd_chart_slots.array = NULL;
src/database/rrdset-pluginsd-array.h
new
+139
@@ -0,0 +1,139 @@
1
+// SPDX-License-Identifier: GPL-3.0-or-later
2
+
3
+#ifndef NETDATA_RRDSET_PLUGINSD_ARRAY_H
4
+#define NETDATA_RRDSET_PLUGINSD_ARRAY_H
5
+
6
+// This header must be included AFTER rrddim.h to get the full struct pluginsd_rrddim definition
7
+
8
+#include "rrddim.h"
9
+
10
+// --------------------------------------------------------------------------------------------------------------------
11
+// Reference-counted array for pluginsd dimension caching
12
+//
13
+// This structure provides thread-safe access to the dimension cache array used by the pluginsd protocol.
14
+// The reference counting ensures that the array is not freed while any thread is still using it.
15
+//
16
+// THREAD SAFETY - LIFECYCLE SEPARATION:
17
+// -------------------------------------
18
+// The design relies on collector and cleanup never running concurrently on the same chart:
19
+//
20
+// 1. collector_tid: Primary synchronization mechanism
21
+// - Collector sets collector_tid BEFORE accessing the array
22
+// - Collector clears collector_tid AFTER all operations are complete
23
+// - Cleanup code checks collector_tid and SKIPS if non-zero
24
+// - This allows the collector to use lock-free operations (get_unsafe, replace, release)
25
+//
26
+// 2. spinlock + refcount: Coordinates concurrent cleanup operations
27
+// - prd_array_acquire(): Takes spinlock, loads pointer, increments refcount
28
+// - Used by cleanup code when collector is NOT active
29
+// - Prevents races between multiple cleanup threads
30
+//
31
+// 3. Lifecycle guarantee: In production, cleanup only runs when:
32
+// - Stream receiver is stopped (collector thread terminated)
33
+// - collector_tid is explicitly cleared before cleanup
34
+// - Therefore, collector's replace+release never races with cleanup's acquire
35
+//
36
+// HOT PATH (collector active, collector_tid set): Lock-free
37
+// CLEANUP PATH (collector stopped, collector_tid == 0): Uses spinlock
38
+// --------------------------------------------------------------------------------------------------------------------
39
+
40
+typedef struct pluginsd_rrddim_array {
41
+ int32_t refcount; // Reference count (atomic)
42
+ size_t size; // Number of entries in the array
43
+ struct pluginsd_rrddim entries[]; // Flexible array member
44
+} PRD_ARRAY;
45
+
46
+// --------------------------------------------------------------------------------------------------------------------
47
+// API Functions
48
+// --------------------------------------------------------------------------------------------------------------------
49
+
50
+// Create a new array with the specified size and refcount=1
51
+static inline PRD_ARRAY *prd_array_create(size_t size) {
52
+ PRD_ARRAY *arr = callocz(1, sizeof(PRD_ARRAY) + size * sizeof(struct pluginsd_rrddim));
53
+ arr->refcount = 1;
54
+ arr->size = size;
55
+ rrd_slot_memory_added(sizeof(PRD_ARRAY) + size * sizeof(struct pluginsd_rrddim));
56
+ return arr;
57
+}
58
+
59
+// Acquire a reference to the array when spinlock is ALREADY HELD
60
+// Returns NULL if no array exists
61
+// The caller MUST call prd_array_release() when done
62
+// Use this when you need to do additional checks (e.g., collector_tid) under the same spinlock
63
+static inline PRD_ARRAY *prd_array_acquire_locked(PRD_ARRAY **array_ptr) {
64
+ PRD_ARRAY *arr = *array_ptr;
65
+ if (arr) {
66
+ __atomic_fetch_add(&arr->refcount, 1, __ATOMIC_ACQ_REL);
67
+ }
68
+ return arr;
69
+}
70
+
71
+// Acquire a reference to the array stored in the atomic pointer location
72
+// Returns NULL if no array exists
73
+// The caller MUST call prd_array_release() when done
74
+//
75
+// IMPORTANT: Only call this when collector_tid == 0 (collector not active).
76
+// Uses spinlock to coordinate with other cleanup operations.
77
+static inline PRD_ARRAY *prd_array_acquire(PRD_ARRAY **array_ptr, SPINLOCK *spinlock) {
78
+ spinlock_lock(spinlock);
79
+ PRD_ARRAY *arr = prd_array_acquire_locked(array_ptr);
80
+ spinlock_unlock(spinlock);
81
+ return arr;
82
+}
83
+
84
+// Release a reference to the array
85
+// If this was the last reference (refcount becomes 0), the array is freed
86
+// Safe to call with NULL
87
+static inline void prd_array_release(PRD_ARRAY *arr) {
88
+ if (!arr)
89
+ return;
90
+
91
+ int32_t old_refcount = __atomic_load_n(&arr->refcount, __ATOMIC_ACQUIRE);
92
+ while(true) {
93
+ if(unlikely(old_refcount <= 0)) {
94
+ // Keep the object stable and avoid driving refcount further negative on
95
+ // repeated misuse. Log in all builds; internal_fatal adds extra checks.
96
+ nd_log_limit_static_global_var(erl_prd_refcount_underflow, 1, 0);
97
+ nd_log_limit(&erl_prd_refcount_underflow, NDLS_DAEMON, NDLP_WARNING,
98
+ "PRD_ARRAY: refcount underflow (was %d) - double release detected",
99
+ old_refcount);
100
+ internal_fatal(true,
101
+ "PRD_ARRAY: refcount underflow (was %d) - double release detected", old_refcount);
102
+ return;
103
+ }
104
+
105
+ if(__atomic_compare_exchange_n(&arr->refcount, &old_refcount, old_refcount - 1,
106
+ false, __ATOMIC_ACQ_REL, __ATOMIC_ACQUIRE))
107
+ break;
108
+ }
109
+
110
+ if(old_refcount == 1) {
111
+ // We were the last reference - free the array
112
+ // Note: The caller is responsible for releasing any RRDDIM_ACQUIRED references
113
+ // in the entries before the final release
114
+ rrd_slot_memory_removed(sizeof(PRD_ARRAY) + arr->size * sizeof(struct pluginsd_rrddim));
115
+ freez(arr);
116
+ }
117
+}
118
+
119
+// Atomically replace the array pointer with a new array
120
+// Returns the old array (caller must release it) or NULL if there was no old array
121
+// The new_arr can be NULL to clear the array
122
+//
123
+// Thread safety depends on context:
124
+// - Collector (collector_tid set): No spinlock needed - cleanup will skip
125
+// - Cleanup (collector_tid == 0): Should hold spinlock to coordinate with other cleanup
126
+static inline PRD_ARRAY *prd_array_replace(PRD_ARRAY **array_ptr, PRD_ARRAY *new_arr) {
127
+ return __atomic_exchange_n(array_ptr, new_arr, __ATOMIC_ACQ_REL);
128
+}
129
+
130
+// Get the current array without acquiring a reference (for quick NULL checks or
131
+// when external synchronization guarantees the array won't be freed)
132
+// WARNING: The returned pointer may become invalid at any time unless:
133
+// - The caller holds the spinlock, OR
134
+// - The caller is the collector thread with collector_tid set (preventing cleanup)
135
+static inline PRD_ARRAY *prd_array_get_unsafe(PRD_ARRAY **array_ptr) {
136
+ return __atomic_load_n(array_ptr, __ATOMIC_ACQUIRE);
137
+}
138
+
139
+#endif // NETDATA_RRDSET_PLUGINSD_ARRAY_H
src/database/rrdset-slots.c
+492
-19
@@ -1,6 +1,7 @@
1
// SPDX-License-Identifier: GPL-3.0-or-later
2
3
#include "rrdset-slots.h"
4
+#include "rrdset-pluginsd-array.h"
5
6
void rrdset_stream_send_chart_slot_assign(RRDSET *st) {
7
RRDHOST *host = st->rrdhost;
@@ -41,48 +42,520 @@ void rrdset_stream_send_chart_slot_release(RRDSET *st) {
42
spinlock_unlock(&host->stream.snd.pluginsd_chart_slots.available.spinlock);
43
}
44
44
-void rrdset_pluginsd_receive_unslot(RRDSET *st) {
45
- for(size_t i = 0; i < st->pluginsd.size ;i++) {
46
- rrddim_acquired_release(st->pluginsd.prd_array[i].rda); // can be NULL
47
- st->pluginsd.prd_array[i].rda = NULL;
48
- st->pluginsd.prd_array[i].rd = NULL;
49
- st->pluginsd.prd_array[i].id = NULL;
45
+// --------------------------------------------------------------------------------------------------------------------
46
+// Helper function to release RRDDIM_ACQUIRED references in array entries
47
+// This must be called before the final prd_array_release when cleaning up
48
+
49
+static void prd_array_release_entries(PRD_ARRAY *arr) {
50
+ if (!arr)
51
+ return;
52
+
53
+ for (size_t i = 0; i < arr->size; i++) {
54
+ rrddim_acquired_release(arr->entries[i].rda); // safe with NULL
55
+ arr->entries[i].rda = NULL;
56
+ arr->entries[i].rd = NULL;
57
+ arr->entries[i].id = NULL;
58
}
59
+}
60
+
61
+static inline void rrdset_clear_host_chart_slot_mapping(RRDSET *st, int32_t last_slot) {
62
+ if(last_slot < 0)
63
+ return;
64
65
RRDHOST *host = st->rrdhost;
66
+ spinlock_lock(&host->stream.rcv.pluginsd_chart_slots.spinlock);
67
+ if((uint32_t)last_slot < host->stream.rcv.pluginsd_chart_slots.size &&
68
+ host->stream.rcv.pluginsd_chart_slots.array[last_slot] == st) {
69
+ host->stream.rcv.pluginsd_chart_slots.array[last_slot] = NULL;
70
+ }
71
+ spinlock_unlock(&host->stream.rcv.pluginsd_chart_slots.spinlock);
72
+}
73
+
74
+// --------------------------------------------------------------------------------------------------------------------
75
+// Unslot a chart - releases dimension references but keeps the array for reuse
76
+// This is called when switching charts, marking them obsolete, or during cleanup.
77
+//
78
+// Safe to call from:
79
+// - The collector thread itself (collector_tid == gettid_cached()): uses lock-free access
80
+// - Any thread when the collector is fully stopped (collector_tid == 0): uses refcount
81
+// Skips with a warning if a DIFFERENT thread's collector is active.
82
+
83
+void rrdset_pluginsd_receive_unslot(RRDSET *st) {
84
+ if(!st)
85
+ return;
86
+
87
+ RRDDIM_ACQUIRED **detached_rdas = NULL;
88
+ size_t detached_capacity = 0;
89
+ size_t detached_entries = 0;
90
+ bool we_are_collector = false;
91
+ PRD_ARRAY *arr = NULL;
92
+ int32_t last_slot = -1;
93
+
94
+ while(true) {
95
+ spinlock_lock(&st->pluginsd.spinlock);
96
+
97
+ // Check collector_tid inside spinlock
98
+ pid_t collector_tid = __atomic_load_n(&st->pluginsd.collector_tid, __ATOMIC_ACQUIRE);
99
+ we_are_collector = (collector_tid == gettid_cached());
100
+ bool different_collector_active = (collector_tid != 0 && !we_are_collector);
101
+
102
+ last_slot = st->pluginsd.last_slot;
103
+
104
+ if(different_collector_active) {
105
+ // Another thread is the active collector - we cannot safely touch the array.
106
+ // Keep pluginsd state unchanged in this path: the active collector may
107
+ // still read last_slot / dims_with_slots lock-free.
108
+ // Clear only the host slot mapping and bail out.
109
+ nd_log_limit_static_global_var(erl, 1, 0);
110
+ nd_log_limit(&erl, NDLS_DAEMON, NDLP_WARNING,
111
+ "PLUGINSD: rrdset_pluginsd_receive_unslot called while collector (tid %d) is active, skipping",
112
+ collector_tid);
113
+
114
+ spinlock_unlock(&st->pluginsd.spinlock);
115
+ freez(detached_rdas);
116
+ rrdset_clear_host_chart_slot_mapping(st, last_slot);
117
+ return;
118
+ }
119
+
120
+ // Either collector_tid == 0 (collector stopped) or collector_tid == our tid
121
+ // (we ARE the collector). In both cases, it's safe to detach dimension references.
122
+ arr = we_are_collector ?
123
+ prd_array_get_unsafe(&st->pluginsd.prd_array) :
124
+ prd_array_acquire_locked(&st->pluginsd.prd_array);
125
+
126
+ if(arr) {
127
+ if(!we_are_collector) {
128
+ // Verify no other thread holds an extra reference before clearing entries.
129
+ // After acquire_locked, refcount should be 2 (original + ours).
130
+ int32_t rc = __atomic_load_n(&arr->refcount, __ATOMIC_ACQUIRE);
131
+ internal_fatal(rc != 2,
132
+ "PRD_ARRAY: expected refcount 2 after acquire, got %d - concurrent reference leak", rc);
133
+
134
+ if(unlikely(rc != 2)) {
135
+ // Production guard: skip detachment when another reference is active.
136
+ // Clearing entries in this state can race and double-release references.
137
+ nd_log_limit_static_global_var(erl_rc, 1, 0);
138
+ nd_log_limit(&erl_rc, NDLS_DAEMON, NDLP_WARNING,
139
+ "PLUGINSD: unslot skipped for chart with unexpected PRD_ARRAY refcount %d (expected 2)",
140
+ rc);
141
+
142
+ prd_array_release(arr);
143
+ spinlock_unlock(&st->pluginsd.spinlock);
144
+ freez(detached_rdas);
145
+ rrdset_clear_host_chart_slot_mapping(st, last_slot);
146
+ return;
147
+ }
148
+ }
149
+
150
+ detached_entries = arr->size;
151
+ if(detached_entries > detached_capacity) {
152
+ // Allocate outside the spinlock to avoid allocator latency while other
153
+ // threads are spinning on this lock.
154
+ if(!we_are_collector)
155
+ prd_array_release(arr);
156
+
157
+ spinlock_unlock(&st->pluginsd.spinlock);
158
+
159
+ freez(detached_rdas);
160
+ detached_rdas = callocz(detached_entries, sizeof(*detached_rdas));
161
+ detached_capacity = detached_entries;
162
+ continue;
163
+ }
164
+
165
+ // Detach entries while holding st->pluginsd.spinlock so concurrent unslot/cleanup
166
+ // cannot race and release the same RRDDIM_ACQUIRED pointers twice.
167
+ for(size_t i = 0; i < detached_entries; i++) {
168
+ detached_rdas[i] = arr->entries[i].rda;
169
+ arr->entries[i].rda = NULL;
170
+ arr->entries[i].rd = NULL;
171
+ arr->entries[i].id = NULL;
172
+ }
173
+ }
174
+ else
175
+ detached_entries = 0;
176
+
177
+ st->pluginsd.last_slot = -1;
178
+ st->pluginsd.dims_with_slots = false;
179
54
- if(st->pluginsd.last_slot >= 0 &&
55
- (uint32_t)st->pluginsd.last_slot < host->stream.rcv.pluginsd_chart_slots.size &&
56
- host->stream.rcv.pluginsd_chart_slots.array[st->pluginsd.last_slot] == st) {
57
- host->stream.rcv.pluginsd_chart_slots.array[st->pluginsd.last_slot] = NULL;
180
+ spinlock_unlock(&st->pluginsd.spinlock);
181
+ break;
182
}
183
60
- st->pluginsd.last_slot = -1;
61
- st->pluginsd.dims_with_slots = false;
184
+ // Release detached references outside the spinlock.
185
+ if(detached_rdas) {
186
+ for(size_t i = 0; i < detached_entries; i++)
187
+ rrddim_acquired_release(detached_rdas[i]); // safe with NULL
188
+
189
+ freez(detached_rdas);
190
+ }
191
+
192
+ if(arr && !we_are_collector) {
193
+ // Release our acquired reference (keeps the struct's reference alive for reuse)
194
+ prd_array_release(arr);
195
+ }
196
+
197
+ rrdset_clear_host_chart_slot_mapping(st, last_slot);
198
}
199
200
+// --------------------------------------------------------------------------------------------------------------------
201
+// Full cleanup - unslots and frees the array
202
+// This is called during chart finalization or host cleanup
203
+// Thread-safe: uses spinlock for cleanup coordination and reference counting for array lifetime
204
+
205
void rrdset_pluginsd_receive_unslot_and_cleanup(RRDSET *st) {
206
if(!st)
207
return;
208
209
spinlock_lock(&st->pluginsd.spinlock);
210
70
- rrdset_pluginsd_receive_unslot(st);
211
+ // Check if collector is still active.
212
+ pid_t collector_tid = __atomic_load_n(&st->pluginsd.collector_tid, __ATOMIC_ACQUIRE);
213
+ pid_t current_tid = gettid_cached();
214
+ if(collector_tid != 0) {
215
+ if(collector_tid != current_tid &&
216
+ !rrdset_flag_check(st, RRDSET_FLAG_COLLECTION_FINISHED)) {
217
+ internal_fatal(true,
218
+ "PRD_ARRAY: cleanup called while collector (tid %d) is still active - lifecycle violation",
219
+ collector_tid);
220
+
221
+ nd_log_limit_static_global_var(erl, 1, 0);
222
+ nd_log_limit(&erl, NDLS_DAEMON, NDLP_WARNING,
223
+ "PLUGINSD: attempted cleanup while collector (tid %d) is still active on chart, skipping",
224
+ collector_tid);
225
+ spinlock_unlock(&st->pluginsd.spinlock);
226
+ return;
227
+ }
228
+
229
+ if(collector_tid == current_tid) {
230
+ // Cleanup in the collector thread should not normally happen.
231
+ // Keep this explicit so we don't mask it as a stale tid case.
232
+#ifdef NETDATA_INTERNAL_CHECKS
233
+ internal_fatal(true,
234
+ "PRD_ARRAY: cleanup called from collector thread (tid %d) - lifecycle violation",
235
+ collector_tid);
236
+#endif
237
+
238
+ nd_log_limit_static_global_var(erl_collector, 1, 0);
239
+ nd_log_limit(&erl_collector, NDLS_DAEMON, NDLP_WARNING,
240
+ "PLUGINSD: cleanup called from collector thread (tid %d), forcing collector_tid=0",
241
+ collector_tid);
242
+ }
243
+ else {
244
+ // Finalization can hit stale collector_tid on charts that were not switched away.
245
+ // Treat this as cleanup ownership handoff and continue.
246
+ nd_log_limit_static_global_var(erl_finalize, 1, 0);
247
+ nd_log_limit(&erl_finalize, NDLS_DAEMON, NDLP_WARNING,
248
+ "PLUGINSD: cleanup forcing stale collector_tid=%d to 0 for finalized chart",
249
+ collector_tid);
250
+ }
251
+
252
+ __atomic_store_n(&st->pluginsd.collector_tid, 0, __ATOMIC_RELEASE);
253
+ }
254
+
255
+ // Replace the array with NULL - this prevents new references from being acquired
256
+ PRD_ARRAY *old_arr = prd_array_replace(&st->pluginsd.prd_array, NULL);
257
+
258
+ // Capture last_slot before resetting - we need it to clear the host mapping
259
+ int32_t last_slot = st->pluginsd.last_slot;
260
72
- rrd_slot_memory_removed(st->pluginsd.size * sizeof(struct pluginsd_rrddim));
73
- freez(st->pluginsd.prd_array);
74
- st->pluginsd.prd_array = NULL;
75
- st->pluginsd.size = 0;
76
- st->pluginsd.pos = 0;
261
+ // Reset state while holding the lock
262
+ __atomic_store_n(&st->pluginsd.pos, 0, __ATOMIC_RELAXED);
263
st->pluginsd.set = false;
264
st->pluginsd.last_slot = -1;
265
st->pluginsd.dims_with_slots = false;
80
- st->pluginsd.collector_tid = 0;
266
267
spinlock_unlock(&st->pluginsd.spinlock);
268
+
269
+ // Clear the chart slot mapping using the captured last_slot value
270
+ rrdset_clear_host_chart_slot_mapping(st, last_slot);
271
+
272
+ // Now handle the old array outside the lock
273
+ if (old_arr) {
274
+ // After prd_array_replace, we hold the only reference (refcount should be 1).
275
+ // It's safe to release entries only when we're the sole owner, to avoid clearing
276
+ // entries that another thread might still be reading through its own reference.
277
+ int32_t rc = __atomic_load_n(&old_arr->refcount, __ATOMIC_ACQUIRE);
278
+ internal_fatal(rc != 1,
279
+ "PRD_ARRAY: expected refcount 1 after replace, got %d - concurrent reference leak", rc);
280
+
281
+ if(unlikely(rc != 1)) {
282
+ // Production guard: another reference still exists, so clearing entries
283
+ // here could race with readers and double-release RRDDIM_ACQUIRED.
284
+ nd_log_limit_static_global_var(erl_cleanup_rc, 1, 0);
285
+ nd_log_limit(&erl_cleanup_rc, NDLS_DAEMON, NDLP_WARNING,
286
+ "PLUGINSD: cleanup deferred for chart with unexpected PRD_ARRAY refcount %d (expected 1)",
287
+ rc);
288
+
289
+ // Drop our reference only; remaining owners will eventually release.
290
+ prd_array_release(old_arr);
291
+ return;
292
+ }
293
+
294
+ // Release all dimension references (safe - we're the sole owner).
295
+ prd_array_release_entries(old_arr);
296
+
297
+ // Release our reference - this will free the array (refcount 1 -> 0)
298
+ prd_array_release(old_arr);
299
+ }
300
}
301
302
+// --------------------------------------------------------------------------------------------------------------------
303
+// Initialize the pluginsd slots for a chart
304
+
305
void rrdset_pluginsd_receive_slots_initialize(RRDSET *st) {
306
spinlock_init(&st->pluginsd.spinlock);
307
st->pluginsd.last_slot = -1;
308
+ st->pluginsd.prd_array = NULL; // Explicitly initialize to NULL
309
+}
310
+
311
+// --------------------------------------------------------------------------------------------------------------------
312
+// Stress test for PRD_ARRAY lifecycle separation model
313
+// Run with: netdata -W prd-array-stress
314
+//
315
+// This test validates the lifecycle separation model used in production:
316
+// - In production, the collector is FULLY STOPPED before cleanup runs
317
+// - The collector_tid check is a safety mechanism, but the real protection comes from lifecycle separation
318
+// - This test simulates that by running the writer and cleaner in non-overlapping phases
319
+//
320
+// The test runs in cycles:
321
+// 1. Writer phase: collector runs multiple iterations (collector_tid set)
322
+// 2. Handoff: collector fully stops (collector_tid cleared, writer_done signaled)
323
+// 3. Cleanup phase: cleaner runs (only when writer is fully stopped)
324
+// 4. Repeat
325
+// --------------------------------------------------------------------------------------------------------------------
326
+
327
+#define PRD_STRESS_TEST_DURATION_SEC 5
328
+#define PRD_STRESS_ITERATIONS_PER_PHASE 50
329
+
330
+typedef struct {
331
+ PRD_ARRAY *prd_array;
332
+ pid_t collector_tid;
333
+ SPINLOCK spinlock;
334
+
335
+ // Lifecycle coordination (simulates stream receiver stop/start)
336
+ bool test_running; // Overall test is running
337
+ bool writer_should_run; // Writer is allowed to run
338
+ bool writer_is_running; // Writer is currently in a phase
339
+
340
+ // Counters
341
+ uint64_t grow_count;
342
+ uint64_t cleanup_count;
343
+ uint64_t phase_count;
344
+} prd_stress_state_t;
345
+
346
+static prd_stress_state_t prd_stress_state;
347
+
348
+static void prd_stress_writer_thread(void *arg __maybe_unused) {
349
+ while (__atomic_load_n(&prd_stress_state.test_running, __ATOMIC_ACQUIRE)) {
350
+
351
+ // Wait until we're allowed to run (simulates stream receiver starting)
352
+ while (__atomic_load_n(&prd_stress_state.test_running, __ATOMIC_ACQUIRE) &&
353
+ !__atomic_load_n(&prd_stress_state.writer_should_run, __ATOMIC_ACQUIRE)) {
354
+ tinysleep();
355
+ }
356
+
357
+ if (!__atomic_load_n(&prd_stress_state.test_running, __ATOMIC_ACQUIRE))
358
+ break;
359
+
360
+ // Signal that writer is now running
361
+ __atomic_store_n(&prd_stress_state.writer_is_running, true, __ATOMIC_RELEASE);
362
+
363
+ // Simulate collector_tid being set (like pluginsd_set_scope_chart does)
364
+ __atomic_store_n(&prd_stress_state.collector_tid, gettid_cached(), __ATOMIC_RELEASE);
365
+
366
+ // Run multiple iterations in this phase (simulates collecting data)
367
+ for (int iter = 0; iter < PRD_STRESS_ITERATIONS_PER_PHASE; iter++) {
368
+ if (!__atomic_load_n(&prd_stress_state.writer_should_run, __ATOMIC_ACQUIRE))
369
+ break;
370
+
371
+ PRD_ARRAY *current_arr = prd_array_get_unsafe(&prd_stress_state.prd_array);
372
+
373
+ size_t current_size = current_arr ? current_arr->size : 0;
374
+ size_t new_size = current_size + 10;
375
+
376
+ if (new_size > 500)
377
+ new_size = 10;
378
+
379
+ PRD_ARRAY *new_arr = prd_array_create(new_size);
380
+
381
+ if (current_arr && current_size > 0) {
382
+ size_t copy_count = (current_size < new_size) ? current_size : new_size;
383
+ for(size_t i = 0; i < copy_count; i++) {
384
+ new_arr->entries[i].rd = current_arr->entries[i].rd;
385
+ new_arr->entries[i].id = current_arr->entries[i].id;
386
+ new_arr->entries[i].rda = NULL;
387
+ }
388
+ }
389
+
390
+ for (size_t i = (current_size < new_size ? current_size : 0); i < new_size; i++) {
391
+ new_arr->entries[i].rd = (void *)(uintptr_t)(i + 1);
392
+ new_arr->entries[i].id = "test";
393
+ }
394
+
395
+ PRD_ARRAY *old_arr = prd_array_replace(&prd_stress_state.prd_array, new_arr);
396
+
397
+ if (old_arr)
398
+ prd_array_release(old_arr);
399
+
400
+ __atomic_fetch_add(&prd_stress_state.grow_count, 1, __ATOMIC_RELAXED);
401
+
402
+ tinysleep();
403
+ }
404
+
405
+ // Clear collector_tid (like pluginsd_set_scope_chart does when switching away)
406
+ __atomic_store_n(&prd_stress_state.collector_tid, 0, __ATOMIC_RELEASE);
407
+
408
+ // Signal that writer phase is complete
409
+ __atomic_store_n(&prd_stress_state.writer_is_running, false, __ATOMIC_RELEASE);
410
+
411
+ // Wait until controller signals us to run again
412
+ while (__atomic_load_n(&prd_stress_state.test_running, __ATOMIC_ACQUIRE) &&
413
+ !__atomic_load_n(&prd_stress_state.writer_should_run, __ATOMIC_ACQUIRE)) {
414
+ tinysleep();
415
+ }
416
+ }
417
+}
418
+
419
+static void prd_stress_cleanup_thread(void *arg __maybe_unused) {
420
+ while (__atomic_load_n(&prd_stress_state.test_running, __ATOMIC_ACQUIRE)) {
421
+
422
+ // Wait until writer is fully stopped (simulates stream_receiver_signal_to_stop_and_wait)
423
+ while (__atomic_load_n(&prd_stress_state.test_running, __ATOMIC_ACQUIRE) &&
424
+ __atomic_load_n(&prd_stress_state.writer_is_running, __ATOMIC_ACQUIRE)) {
425
+ tinysleep();
426
+ }
427
+
428
+ if (!__atomic_load_n(&prd_stress_state.test_running, __ATOMIC_ACQUIRE))
429
+ break;
430
+
431
+ // Now safe to cleanup - writer is fully stopped
432
+ spinlock_lock(&prd_stress_state.spinlock);
433
+
434
+ // Double-check collector_tid (should be 0 since writer stopped)
435
+ pid_t collector_tid = __atomic_load_n(&prd_stress_state.collector_tid, __ATOMIC_ACQUIRE);
436
+ if (collector_tid != 0) {
437
+ // This shouldn't happen if lifecycle is correct
438
+ spinlock_unlock(&prd_stress_state.spinlock);
439
+ continue;
440
+ }
441
+
442
+ PRD_ARRAY *old_arr = prd_array_replace(&prd_stress_state.prd_array, NULL);
443
+
444
+ spinlock_unlock(&prd_stress_state.spinlock);
445
+
446
+ if (old_arr) {
447
+ for (size_t i = 0; i < old_arr->size; i++) {
448
+ old_arr->entries[i].rda = NULL;
449
+ old_arr->entries[i].rd = NULL;
450
+ old_arr->entries[i].id = NULL;
451
+ }
452
+
453
+ prd_array_release(old_arr);
454
+ __atomic_fetch_add(&prd_stress_state.cleanup_count, 1, __ATOMIC_RELAXED);
455
+ }
456
+
457
+ tinysleep();
458
+ }
459
+}
460
+
461
+// Controller thread - orchestrates the lifecycle phases
462
+static void prd_stress_controller_thread(void *arg __maybe_unused) {
463
+ while (__atomic_load_n(&prd_stress_state.test_running, __ATOMIC_ACQUIRE)) {
464
+
465
+ // Start writer phase
466
+ __atomic_store_n(&prd_stress_state.writer_should_run, true, __ATOMIC_RELEASE);
467
+
468
+ // Wait for writer to start and run
469
+ sleep_usec(10000); // 10ms - let writer run
470
+
471
+ // Signal writer to stop (simulates stream receiver stopping)
472
+ __atomic_store_n(&prd_stress_state.writer_should_run, false, __ATOMIC_RELEASE);
473
+
474
+ // Wait for writer to fully stop
475
+ while (__atomic_load_n(&prd_stress_state.test_running, __ATOMIC_ACQUIRE) &&
476
+ __atomic_load_n(&prd_stress_state.writer_is_running, __ATOMIC_ACQUIRE)) {
477
+ tinysleep();
478
+ }
479
+
480
+ // Cleanup phase - cleaner will run now that writer is stopped
481
+ sleep_usec(5000); // 5ms - let cleanup run
482
+
483
+ __atomic_fetch_add(&prd_stress_state.phase_count, 1, __ATOMIC_RELAXED);
484
+ }
485
+}
486
+
487
+int prd_array_stress_test(void) {
488
+ int duration_secs = PRD_STRESS_TEST_DURATION_SEC;
489
+
490
+ fprintf(stderr, "\nPRD_ARRAY Lifecycle Stress Test\n");
491
+ fprintf(stderr, "================================\n");
492
+ fprintf(stderr, "Duration: %d seconds\n", duration_secs);
493
+ fprintf(stderr, "This test simulates production lifecycle:\n");
494
+ fprintf(stderr, " 1. Writer (collector) runs with collector_tid set\n");
495
+ fprintf(stderr, " 2. Writer fully stops (collector_tid cleared)\n");
496
+ fprintf(stderr, " 3. Cleaner runs cleanup\n");
497
+ fprintf(stderr, " 4. Repeat\n\n");
498
+
499
+ // Initialize state
500
+ memset(&prd_stress_state, 0, sizeof(prd_stress_state));
501
+ prd_stress_state.prd_array = prd_array_create(10);
502
+ spinlock_init(&prd_stress_state.spinlock);
503
+ __atomic_store_n(&prd_stress_state.test_running, true, __ATOMIC_RELEASE);
504
+
505
+ // Start threads
506
+ char thread_name[32];
507
+
508
+ snprintfz(thread_name, sizeof(thread_name), "PRDSTRESS_W");
509
+ ND_THREAD *writer_thread = nd_thread_create(thread_name, NETDATA_THREAD_OPTION_DEFAULT,
510
+ prd_stress_writer_thread, NULL);
511
+
512
+ snprintfz(thread_name, sizeof(thread_name), "PRDSTRESS_C");
513
+ ND_THREAD *cleanup_thread = nd_thread_create(thread_name, NETDATA_THREAD_OPTION_DEFAULT,
514
+ prd_stress_cleanup_thread, NULL);
515
+
516
+ snprintfz(thread_name, sizeof(thread_name), "PRDSTRESS_CTRL");
517
+ ND_THREAD *controller_thread = nd_thread_create(thread_name, NETDATA_THREAD_OPTION_DEFAULT,
518
+ prd_stress_controller_thread, NULL);
519
+
520
+ // Run the test
521
+ fprintf(stderr, "Running stress test...\n");
522
+ for (int i = 0; i < duration_secs; i++) {
523
+ sleep_usec(USEC_PER_SEC);
524
+ fprintf(stderr, " %d/%d sec - phases: %"PRIu64", grows: %"PRIu64", cleanups: %"PRIu64"\n",
525
+ i + 1, duration_secs,
526
+ __atomic_load_n(&prd_stress_state.phase_count, __ATOMIC_RELAXED),
527
+ __atomic_load_n(&prd_stress_state.grow_count, __ATOMIC_RELAXED),
528
+ __atomic_load_n(&prd_stress_state.cleanup_count, __ATOMIC_RELAXED));
529
+ }
530
+
531
+ // Stop all threads
532
+ __atomic_store_n(&prd_stress_state.test_running, false, __ATOMIC_RELEASE);
533
+ __atomic_store_n(&prd_stress_state.writer_should_run, true, __ATOMIC_RELEASE); // Unblock writer
534
+
535
+ nd_thread_join(controller_thread);
536
+ nd_thread_join(writer_thread);
537
+ nd_thread_join(cleanup_thread);
538
+
539
+ // Final cleanup
540
+ PRD_ARRAY *final_arr = prd_array_replace(&prd_stress_state.prd_array, NULL);
541
+ if (final_arr)
542
+ prd_array_release(final_arr);
543
+
544
+ // Print results
545
+ fprintf(stderr, "\nTest completed!\n");
546
+ fprintf(stderr, "===============\n");
547
+ fprintf(stderr, "Total phases: %"PRIu64"\n", prd_stress_state.phase_count);
548
+ fprintf(stderr, "Total grows: %"PRIu64"\n", prd_stress_state.grow_count);
549
+ fprintf(stderr, "Total cleanups: %"PRIu64"\n", prd_stress_state.cleanup_count);
550
+
551
+ if (prd_stress_state.cleanup_count > 0 && prd_stress_state.grow_count > 0) {
552
+ fprintf(stderr, "\nSUCCESS: Lifecycle separation validated\n");
553
+ fprintf(stderr, "- Writer and cleaner ran in non-overlapping phases\n");
554
+ fprintf(stderr, "- No concurrent access to the array\n");
555
+ fprintf(stderr, "- Reference counting worked correctly\n");
556
+ return 0;
557
+ } else {
558
+ fprintf(stderr, "\nWARNING: Low activity - increase test duration\n");
559
+ return 1;
560
+ }
561
}
src/database/rrdset-slots.h
+11
@@ -8,8 +8,19 @@
8
void rrdset_stream_send_chart_slot_assign(RRDSET *st);
9
void rrdset_stream_send_chart_slot_release(RRDSET *st);
10
11
+// rrdset_pluginsd_receive_unslot: Releases dimension references but keeps the array.
12
+// Safe to call from the collector thread itself (detected via collector_tid == gettid_cached())
13
+// or when the collector is fully stopped.
14
void rrdset_pluginsd_receive_unslot(RRDSET *st);
15
+
16
+// rrdset_pluginsd_receive_unslot_and_cleanup: Full cleanup - releases dimension references
17
+// AND frees the array. Must only be called when the collector is FULLY STOPPED on the chart.
18
+// The collector_tid check is a safety mechanism (fires internal_fatal in debug builds).
19
void rrdset_pluginsd_receive_unslot_and_cleanup(RRDSET *st);
20
+
21
void rrdset_pluginsd_receive_slots_initialize(RRDSET *st);
22
23
+// Stress test for PRD_ARRAY lifecycle separation - run with -W prd-array-stress
24
+int prd_array_stress_test(void);
25
+
26
#endif //NETDATA_RRDSET_SLOTS_H
src/database/rrdset.h
+3
-3
@@ -18,6 +18,7 @@ typedef struct ml_chart rrd_ml_chart_t;
18
struct rrdhost;
19
struct rrdcalc;
20
struct pluginsd_rrddim;
21
+struct pluginsd_rrddim_array; // Reference-counted array for pluginsd dimension caching
22
struct rrdinstance_acquired;
23
struct rrdcontext_acquired;
24
struct storage_alignment;
@@ -209,14 +210,13 @@ struct rrdset {
210
} alerts;
211
212
struct {
212
- SPINLOCK spinlock; // used only for cleanup
213
+ SPINLOCK spinlock; // coordinates PRD_ARRAY grow/transfer and unslot/cleanup serialization
214
pid_t collector_tid;
215
bool dims_with_slots;
216
bool set;
217
uint32_t pos;
218
int32_t last_slot;
218
- uint32_t size;
219
- struct pluginsd_rrddim *prd_array;
219
+ struct pluginsd_rrddim_array *prd_array; // Reference-counted array (use prd_array_* functions)
220
} pluginsd;
221
222
#ifdef NETDATA_LOG_REPLICATION_REQUESTS
src/plugins.d/pluginsd_internals.h
+136
-45
@@ -7,6 +7,7 @@
7
#include "pluginsd_functions.h"
8
#include "pluginsd_dyncfg.h"
9
#include "pluginsd_replication.h"
10
+#include "database/rrdset-pluginsd-array.h"
11
12
#define SERVING_STREAMING(parser) ((parser)->repertoire == PARSER_INIT_STREAMING)
13
#define SERVING_PLUGINSD(parser) ((parser)->repertoire == PARSER_INIT_PLUGINSD)
@@ -78,11 +79,43 @@ static ALWAYS_INLINE void rrdset_previous_scope_chart_unlock(PARSER *parser, con
79
}
80
}
81
81
-static inline void pluginsd_clear_scope_chart(PARSER *parser, const char *keyword) {
82
+static inline void pluginsd_clear_scope_chart(PARSER *parser, const char *keyword, RRDSET *preserve_collector_tid) {
83
rrdset_previous_scope_chart_unlock(parser, keyword, true);
84
84
- if(parser->user.cleanup_slots && parser->user.st)
85
- rrdset_pluginsd_receive_unslot(parser->user.st);
85
+ RRDSET *st = parser->user.st;
86
+
87
+ if(parser->user.cleanup_slots && st)
88
+ rrdset_pluginsd_receive_unslot(st);
89
+
90
+ // Clear collector ownership when scope ends, except when explicitly preserving
91
+ // it for the currently active chart during same-chart re-scope.
92
+ //
93
+ // Safety note:
94
+ // - Full cleanup (rrdset_pluginsd_receive_unslot_and_cleanup) runs on finalized/teardown paths.
95
+ // - Host teardown stops the receiver thread before slot/index cleanup.
96
+ // - During active parser execution, unslot paths are collector-aware and skip when another
97
+ // collector tid is active.
98
+ // Therefore, eager clear here is an ownership handoff between protocol scopes, not a signal
99
+ // that teardown cleanup may run concurrently with an active collector loop.
100
+ // Clear collector ownership only if we are the recorded owner (or no owner exists).
101
+ // If another thread owns this chart, keep its ownership intact and report it.
102
+ if(st && st != preserve_collector_tid) {
103
+ pid_t owner_tid = __atomic_load_n(&st->pluginsd.collector_tid, __ATOMIC_ACQUIRE);
104
+ pid_t self_tid = gettid_cached();
105
+
106
+ if(owner_tid == 0 || owner_tid == self_tid)
107
+ __atomic_store_n(&st->pluginsd.collector_tid, 0, __ATOMIC_RELEASE);
108
+ else {
109
+ netdata_log_error(
110
+ "PLUGINSD: attempted to clear collector_tid %d for 'host:%s/chart:%s/' "
111
+ "from non-owner thread %d during %s",
112
+ (int)owner_tid,
113
+ rrdhost_hostname(st->rrdhost),
114
+ rrdset_id(st),
115
+ (int)self_tid,
116
+ keyword);
117
+ }
118
+ }
119
120
parser->user.st = NULL;
121
parser->user.cleanup_slots = false;
@@ -91,7 +124,7 @@ static inline void pluginsd_clear_scope_chart(PARSER *parser, const char *keywor
124
125
static ALWAYS_INLINE bool pluginsd_set_scope_chart(PARSER *parser, RRDSET *st, const char *keyword) {
126
RRDSET *old_st = parser->user.st;
94
- pid_t old_collector_tid = (old_st) ? old_st->pluginsd.collector_tid : 0;
127
+ pid_t old_collector_tid = (old_st) ? __atomic_load_n(&old_st->pluginsd.collector_tid, __ATOMIC_ACQUIRE) : 0;
128
pid_t my_collector_tid = gettid_cached();
129
130
if(unlikely(old_collector_tid)) {
@@ -105,15 +138,17 @@ static ALWAYS_INLINE bool pluginsd_set_scope_chart(PARSER *parser, RRDSET *st, c
138
139
return false;
140
}
108
-
109
- old_st->pluginsd.collector_tid = 0;
141
+ // Don't clear collector_tid here - we still need to access old_st in pluginsd_clear_scope_chart
142
}
143
112
- st->pluginsd.collector_tid = my_collector_tid;
144
+ // Set new chart's collector_tid before any access
145
+ __atomic_store_n(&st->pluginsd.collector_tid, my_collector_tid, __ATOMIC_RELEASE);
146
114
- pluginsd_clear_scope_chart(parser, keyword);
147
+ // Access old_st's array in pluginsd_clear_scope_chart while old_st->collector_tid is still set.
148
+ // Preserve the new chart tid for the old_st == st re-scope case.
149
+ pluginsd_clear_scope_chart(parser, keyword, st);
150
116
- st->pluginsd.pos = 0;
151
+ __atomic_store_n(&st->pluginsd.pos, 0, __ATOMIC_RELAXED);
152
parser->user.st = st;
153
parser->user.cleanup_slots = false;
154
parser->user.clabel_count = 0;
@@ -122,49 +157,99 @@ static ALWAYS_INLINE bool pluginsd_set_scope_chart(PARSER *parser, RRDSET *st, c
157
}
158
159
static inline void pluginsd_rrddim_put_to_slot(PARSER *parser, RRDSET *st, RRDDIM *rd, ssize_t slot, bool obsolete) {
125
- size_t wanted_size = st->pluginsd.size;
160
+ // Determine the required array size
161
+ size_t wanted_size;
162
163
if(slot >= 1) {
164
st->pluginsd.dims_with_slots = true;
129
- wanted_size = slot;
165
+ wanted_size = (size_t)slot;
166
}
167
else {
168
st->pluginsd.dims_with_slots = false;
169
wanted_size = dictionary_entries(st->rrddim_root_index);
170
}
171
136
- if(wanted_size > st->pluginsd.size) {
137
- st->pluginsd.prd_array = reallocz(st->pluginsd.prd_array, wanted_size * sizeof(struct pluginsd_rrddim));
172
+ // Get current array (if any) to check size
173
+ // Note: We're the collector thread with collector_tid set, so the array won't be freed under us
174
+ PRD_ARRAY *current_arr = prd_array_get_unsafe(&st->pluginsd.prd_array);
175
+ size_t current_size = current_arr ? current_arr->size : 0;
176
+
177
+ // Check if we need to grow the array
178
+ if(wanted_size > current_size) {
179
+ // Pre-allocate outside the spinlock to keep critical section short.
180
+ PRD_ARRAY *new_arr = prd_array_create(wanted_size);
181
+
182
+ // Serialize grow transfer with unslot/cleanup detach paths.
183
+ spinlock_lock(&st->pluginsd.spinlock);
184
+
185
+ current_arr = prd_array_get_unsafe(&st->pluginsd.prd_array);
186
+ current_size = current_arr ? current_arr->size : 0;
187
+
188
+ // Re-check under lock in case another path changed the array.
189
+ if(wanted_size > current_size) {
190
+ // Copy existing entries from old array (if any) and transfer ownership
191
+ // to the new array by nulling old pointers.
192
+ if(current_arr) {
193
+ memcpy(new_arr->entries, current_arr->entries, current_size * sizeof(struct pluginsd_rrddim));
194
+ for(size_t i = 0; i < current_size; i++) {
195
+ current_arr->entries[i].rda = NULL;
196
+ current_arr->entries[i].rd = NULL;
197
+ current_arr->entries[i].id = NULL;
198
+ }
199
+ }
200
+
201
+ // Initialize the new slots (callocz already zeroed them, but be explicit)
202
+ for(size_t i = current_size; i < wanted_size; i++) {
203
+ new_arr->entries[i].rda = NULL;
204
+ new_arr->entries[i].rd = NULL;
205
+ new_arr->entries[i].id = NULL;
206
+ }
207
139
- // initialize the empty slots
140
- for(ssize_t i = (ssize_t) wanted_size - 1; i >= (ssize_t) st->pluginsd.size; i--) {
141
- st->pluginsd.prd_array[i].rda = NULL;
142
- st->pluginsd.prd_array[i].rd = NULL;
143
- st->pluginsd.prd_array[i].id = NULL;
208
+ // Atomically replace the old array with the new one
209
+ PRD_ARRAY *old_arr = prd_array_replace(&st->pluginsd.prd_array, new_arr);
210
+
211
+ // Release the old array if there was one.
212
+ if(old_arr) {
213
+ // Release the old array - it will be freed when refcount reaches 0
214
+ prd_array_release(old_arr);
215
+ }
216
+
217
+ // Update our local pointer to the new array
218
+ current_arr = new_arr;
219
+ new_arr = NULL;
220
+ }
221
+ else {
222
+ current_arr = prd_array_get_unsafe(&st->pluginsd.prd_array);
223
}
224
146
- rrd_slot_memory_added((wanted_size - st->pluginsd.size) * sizeof(struct pluginsd_rrddim));
147
- st->pluginsd.size = wanted_size;
225
+ spinlock_unlock(&st->pluginsd.spinlock);
226
+
227
+ // Another path already satisfied growth while we were waiting for the lock.
228
+ if(new_arr)
229
+ prd_array_release(new_arr);
230
}
231
150
- if(st->pluginsd.dims_with_slots) {
151
- struct pluginsd_rrddim *prd = &st->pluginsd.prd_array[slot - 1];
232
+ // Now update the slot entry if we're using slots
233
+ if(st->pluginsd.dims_with_slots && current_arr && slot >= 1 && (size_t)slot <= current_arr->size) {
234
+ struct pluginsd_rrddim *prd = ¤t_arr->entries[slot - 1];
235
236
if(prd->rd != rd) {
154
- rrddim_acquired_release(prd->rda);
155
- prd->rda = NULL;
156
- prd->rd = NULL;
157
- prd->id = NULL;
237
+ // Release old reference if any
238
+ if(prd->rda)
239
+ rrddim_acquired_release(prd->rda);
240
241
prd->rda = rrddim_find_and_acquire(st, string2str(rd->id), true);
242
if(unlikely(!prd->rda)) {
161
- netdata_log_error("PLUGINSD: failed to acquire dimension '%s' for chart '%s' while updating slot cache",
162
- rrddim_id(rd), rrdset_id(st));
243
+ prd->rd = NULL;
244
+ prd->id = NULL;
245
+ netdata_log_error("PLUGINSD: failed to refresh slot cache for 'host:%s/chart:%s/dim:%s' (slot %zd)",
246
+ rrdhost_hostname(st->rrdhost), rrdset_id(st), string2str(rd->id), slot);
247
return;
248
}
165
-
166
- prd->rd = rrddim_acquired_to_rrddim(prd->rda);
167
- prd->id = string2str(prd->rd->id);
249
+ else {
250
+ prd->rd = rrddim_acquired_to_rrddim(prd->rda);
251
+ prd->id = string2str(prd->rd->id);
252
+ }
253
}
254
255
if(obsolete)
@@ -179,42 +264,46 @@ static ALWAYS_INLINE RRDDIM *pluginsd_acquire_dimension(RRDHOST *host, RRDSET *s
264
return NULL;
265
}
266
182
- if (unlikely(!st->pluginsd.size)) {
267
+ // Get the array - we're protected by collector_tid being set, so it won't be freed
268
+ PRD_ARRAY *arr = prd_array_get_unsafe(&st->pluginsd.prd_array);
269
+
270
+ if (unlikely(!arr || !arr->size)) {
271
netdata_log_error("PLUGINSD: 'host:%s/chart:%s' got a %s, but the chart has no dimensions.",
272
rrdhost_hostname(host), rrdset_id(st), cmd);
273
return NULL;
274
}
275
276
+ size_t prd_size = arr->size;
277
struct pluginsd_rrddim *prd;
278
RRDDIM *rd;
279
280
if(likely(st->pluginsd.dims_with_slots)) {
281
// caching with slots
282
194
- if(unlikely(slot < 1 || slot > (ssize_t)st->pluginsd.size)) {
195
- netdata_log_error("PLUGINSD: 'host:%s/chart:%s' got a %s with slot %zd, but slots in the range [1 - %u] are expected.",
196
- rrdhost_hostname(host), rrdset_id(st), cmd, slot, st->pluginsd.size);
283
+ if(unlikely(slot < 1 || slot > (ssize_t)prd_size)) {
284
+ netdata_log_error("PLUGINSD: 'host:%s/chart:%s' got a %s with slot %zd, but slots in the range [1 - %zu] are expected.",
285
+ rrdhost_hostname(host), rrdset_id(st), cmd, slot, prd_size);
286
return NULL;
287
}
288
200
- prd = &st->pluginsd.prd_array[slot - 1];
289
+ prd = &arr->entries[slot - 1];
290
291
rd = prd->rd;
292
if(likely(rd)) {
293
#ifdef NETDATA_INTERNAL_CHECKS
205
- if(strcmp(prd->id, dimension) != 0) {
294
+ if(!prd->id || strcmp(prd->id, dimension) != 0) {
295
ssize_t t;
207
- for(t = 0; t < st->pluginsd.size ;t++) {
208
- if (strcmp(st->pluginsd.prd_array[t].id, dimension) == 0)
296
+ for(t = 0; t < (ssize_t)prd_size ;t++) {
297
+ if (arr->entries[t].id && strcmp(arr->entries[t].id, dimension) == 0)
298
break;
299
}
211
- if(t >= st->pluginsd.size)
300
+ if(t >= (ssize_t)prd_size)
301
t = -1;
302
303
internal_fatal(true,
304
"PLUGINSD: expected to find dimension '%s' on slot %zd, but found '%s', "
305
"the right slot is %zd",
217
- dimension, slot, prd->id, t);
306
+ dimension, slot, prd->id ? prd->id : "(null)", t);
307
}
308
#endif
309
return rd;
@@ -223,10 +312,12 @@ static ALWAYS_INLINE RRDDIM *pluginsd_acquire_dimension(RRDHOST *host, RRDSET *s
312
else {
313
// caching without slots
314
226
- if(unlikely(st->pluginsd.pos >= st->pluginsd.size))
227
- st->pluginsd.pos = 0;
315
+ uint32_t pos = __atomic_load_n(&st->pluginsd.pos, __ATOMIC_RELAXED);
316
+ if(unlikely(pos >= prd_size))
317
+ pos = 0;
318
229
- prd = &st->pluginsd.prd_array[st->pluginsd.pos++];
319
+ __atomic_store_n(&st->pluginsd.pos, pos + 1, __ATOMIC_RELAXED);
320
+ prd = &arr->entries[pos];
321
322
rd = prd->rd;
323
if(likely(rd)) {
@@ -316,7 +407,7 @@ static inline void pluginsd_rrdset_cache_put_to_slot(PARSER *parser, RRDSET *st,
407
host->stream.rcv.pluginsd_chart_slots.size = new_slots;
408
spinlock_unlock(&host->stream.rcv.pluginsd_chart_slots.spinlock);
409
319
- rrd_slot_memory_added((new_slots - old_slots) * sizeof(uint32_t));
410
+ rrd_slot_memory_added((new_slots - old_slots) * sizeof(RRDSET *));
411
}
412
413
host->stream.rcv.pluginsd_chart_slots.array[slot - 1] = st;
src/plugins.d/pluginsd_parser.c
+11
-8
@@ -102,7 +102,7 @@ static inline PARSER_RC pluginsd_end(char **words, size_t num_words, PARSER *par
102
if (unlikely(rrdset_flag_check(st, RRDSET_FLAG_DEBUG)))
103
netdata_log_debug(D_PLUGINSD, "requested an END on chart '%s'", rrdset_id(st));
104
105
- pluginsd_clear_scope_chart(parser, PLUGINSD_KEYWORD_END);
105
+ pluginsd_clear_scope_chart(parser, PLUGINSD_KEYWORD_END, NULL);
106
parser->user.data_collections_count++;
107
108
struct timeval tv = {
@@ -254,7 +254,7 @@ static inline PARSER_RC pluginsd_host_define_end(char **words __maybe_unused, si
254
pluginsd_host_define_cleanup(parser);
255
256
parser->user.host = host;
257
- pluginsd_clear_scope_chart(parser, PLUGINSD_KEYWORD_HOST_DEFINE_END);
257
+ pluginsd_clear_scope_chart(parser, PLUGINSD_KEYWORD_HOST_DEFINE_END, NULL);
258
259
rrdhost_flag_clear(host, RRDHOST_FLAG_ORPHAN);
260
rrdcontext_host_child_connected(host);
@@ -456,7 +456,7 @@ static inline PARSER_RC pluginsd_chart(char **words, size_t num_words, PARSER *p
456
pluginsd_rrdset_cache_put_to_slot(parser, st, slot, obsolete);
457
}
458
else
459
- pluginsd_clear_scope_chart(parser, PLUGINSD_KEYWORD_CHART);
459
+ pluginsd_clear_scope_chart(parser, PLUGINSD_KEYWORD_CHART, NULL);
460
461
return PARSER_RC_OK;
462
}
@@ -652,7 +652,7 @@ static inline PARSER_RC pluginsd_variable(char **words, size_t num_words, PARSER
652
653
static inline PARSER_RC pluginsd_flush(char **words __maybe_unused, size_t num_words __maybe_unused, PARSER *parser) {
654
netdata_log_debug(D_PLUGINSD, "requested a " PLUGINSD_KEYWORD_FLUSH);
655
- pluginsd_clear_scope_chart(parser, PLUGINSD_KEYWORD_FLUSH);
655
+ pluginsd_clear_scope_chart(parser, PLUGINSD_KEYWORD_FLUSH, NULL);
656
parser->user.replay.start_time = 0;
657
parser->user.replay.end_time = 0;
658
parser->user.replay.start_time_ut = 0;
@@ -1133,9 +1133,12 @@ static ALWAYS_INLINE PARSER_RC pluginsd_end_v2(char **words __maybe_unused, size
1133
// ------------------------------------------------------------------------
1134
// cleanup RRDSET / RRDDIM
1135
1136
- if(likely(st->pluginsd.dims_with_slots)) {
1137
- for(size_t i = 0; i < st->pluginsd.size ;i++) {
1138
- RRDDIM *rd = st->pluginsd.prd_array[i].rd;
1136
+ // Get the array - we're protected by collector_tid being set, so it won't be freed
1137
+ PRD_ARRAY *prd_arr = prd_array_get_unsafe(&st->pluginsd.prd_array);
1138
+
1139
+ if(likely(st->pluginsd.dims_with_slots && prd_arr && prd_arr->size)) {
1140
+ for(size_t i = 0; i < prd_arr->size ;i++) {
1141
+ RRDDIM *rd = prd_arr->entries[i].rd;
1142
1143
if(!rd)
1144
continue;
@@ -1242,7 +1245,7 @@ PARSER_RC stream_receiver_pluginsd_claimed_id(char **words, size_t num_words, PA
1245
1246
void pluginsd_cleanup_v2(PARSER *parser) {
1247
// this is called when the thread is stopped while processing
1245
- pluginsd_clear_scope_chart(parser, "THREAD CLEANUP");
1248
+ pluginsd_clear_scope_chart(parser, "THREAD CLEANUP", NULL);
1249
}
1250
1251
void pluginsd_process_cleanup(PARSER *parser) {
src/plugins.d/pluginsd_replication.c
+4
-4
@@ -301,7 +301,7 @@ ALWAYS_INLINE PARSER_RC pluginsd_replay_rrddim_collection_state(char **words, si
301
302
if(st->pluginsd.set) {
303
// reset pos to reuse the same RDAs
304
- st->pluginsd.pos = 0;
304
+ __atomic_store_n(&st->pluginsd.pos, 0, __ATOMIC_RELAXED);
305
st->pluginsd.set = false;
306
}
307
@@ -473,7 +473,7 @@ ALWAYS_INLINE PARSER_RC pluginsd_replay_end(char **words, size_t num_words, PARS
473
"with enable_streaming = true, but there was no replication in progress for this chart.",
474
rrdhost_hostname(host), rrdset_id(st));
475
476
- pluginsd_clear_scope_chart(parser, PLUGINSD_KEYWORD_REPLAY_END);
476
+ pluginsd_clear_scope_chart(parser, PLUGINSD_KEYWORD_REPLAY_END, NULL);
477
478
host->stream.rcv.status.replication.percent = 100.0;
479
worker_set_metric(WORKER_RECEIVER_JOB_REPLICATION_COMPLETION, host->stream.rcv.status.replication.percent);
@@ -573,7 +573,7 @@ ALWAYS_INLINE PARSER_RC pluginsd_replay_end(char **words, size_t num_words, PARS
573
pulse_host_status(host, PULSE_HOST_STATUS_RCV_RUNNING, 0);
574
}
575
576
- pluginsd_clear_scope_chart(parser, PLUGINSD_KEYWORD_REPLAY_END);
576
+ pluginsd_clear_scope_chart(parser, PLUGINSD_KEYWORD_REPLAY_END, NULL);
577
host->stream.rcv.status.replication.percent = 100.0;
578
worker_set_metric(WORKER_RECEIVER_JOB_REPLICATION_COMPLETION, host->stream.rcv.status.replication.percent);
579
@@ -592,7 +592,7 @@ ALWAYS_INLINE PARSER_RC pluginsd_replay_end(char **words, size_t num_words, PARS
592
st->stream.rcv.who = REPLAY_WHO_ME;
593
#endif
594
595
- pluginsd_clear_scope_chart(parser, PLUGINSD_KEYWORD_REPLAY_END);
595
+ pluginsd_clear_scope_chart(parser, PLUGINSD_KEYWORD_REPLAY_END, NULL);
596
597
rrdcontext_updated_retention_rrdset(st);
598