Waiting Queue (#19302)
* add strict checks to waiting queue * waiting queue implementation purely using atomics * trylock should insist if we are the potential winner
Costa Tsaousis committed
Dec 31, 2024 at 10:13 UTC
a801fd3dcaa892593033d3108bb6d4eba7db6a5d
13 files changed
+377
-595
CMakeLists.txt
+2
-2
@@ -949,8 +949,8 @@ set(LIBNETDATA_FILES
949
src/libnetdata/locks/rw-spinlock.h
950
src/libnetdata/atomics/atomic_flags.h
951
src/libnetdata/atomics/atomics.h
952
- src/libnetdata/waiting-queue/waiting-queue.c
953
- src/libnetdata/waiting-queue/waiting-queue.h
952
+ src/libnetdata/locks/waitq.c
953
+ src/libnetdata/locks/waitq.h
954
src/libnetdata/object-state/object-state.c
955
src/libnetdata/object-state/object-state.h
956
)
src/collectors/systemd-journal.plugin/systemd-journal-annotations.c
-1
@@ -611,7 +611,6 @@ static void netdata_systemd_journal_message_ids_init(void) {
611
msgid_into_dict("9ce0cb58ab8b44df82c4bf1ad9ee22de", "Netdata alert transition");
612
msgid_into_dict("6db0018e83e34320ae2a659d78019fb7", "Netdata alert notification");
613
msgid_into_dict("23e93dfccbf64e11aac858b9410d8a82", "Netdata fatal message");
614
-
614
msgid_into_dict("8ddaf5ba33a74078b609250db1e951f3", "Sensor state transition");
615
}
616
src/database/engine/cache.c
+31
-31
@@ -22,8 +22,7 @@ typedef int32_t REFCOUNT;
22
#define PGC_WITH_ARAL 1
23
#endif
24
25
-// unfortunately waiting queue is significantly slower than spinlock
26
-// #define PGC_QUEUE_LOCK_AS_WAITING_QUEUE 1
25
+#define PGC_QUEUE_LOCK_AS_WAITING_QUEUE 1
26
27
typedef enum __attribute__ ((__packed__)) {
28
// mutually exclusive flags
@@ -75,7 +74,7 @@ struct pgc_page {
74
75
struct pgc_queue {
76
#if defined(PGC_QUEUE_LOCK_AS_WAITING_QUEUE)
78
- WAITING_QUEUE *wq;
77
+ WAITQ wq;
78
#else
79
SPINLOCK spinlock;
80
#endif
@@ -246,17 +245,17 @@ static inline size_t pgc_indexing_partition(PGC *cache, Word_t metric_id) {
245
_result; \
246
})
247
249
-#define PGC_QUEUE_LOCK_PRIO_COLLECTORS WAITING_QUEUE_PRIO_URGENT
250
-#define PGC_QUEUE_LOCK_PRIO_EVICTORS WAITING_QUEUE_PRIO_HIGH
251
-#define PGC_QUEUE_LOCK_PRIO_FLUSHERS WAITING_QUEUE_PRIO_NORMAL
252
-#define PGC_QUEUE_LOCK_PRIO_OTHERS WAITING_QUEUE_PRIO_LOW
248
+#define PGC_QUEUE_LOCK_PRIO_COLLECTORS WAITQ_PRIO_URGENT
249
+#define PGC_QUEUE_LOCK_PRIO_EVICTORS WAITQ_PRIO_HIGH
250
+#define PGC_QUEUE_LOCK_PRIO_FLUSHERS WAITQ_PRIO_NORMAL
251
+#define PGC_QUEUE_LOCK_PRIO_LOW WAITQ_PRIO_LOW
252
253
#if defined(PGC_QUEUE_LOCK_AS_WAITING_QUEUE)
255
-#define pgc_queue_trylock(cache, ll) waiting_queue_try_acquire((ll)->wq)
256
-#define pgc_queue_lock(cache, ll, prio) waiting_queue_acquire((ll)->wq, prio)
257
-#define pgc_queue_unlock(cache, ll) waiting_queue_release((ll)->wq)
254
+#define pgc_queue_trylock(cache, ll, prio) waitq_try_acquire(&((ll)->wq), prio)
255
+#define pgc_queue_lock(cache, ll, prio) waitq_acquire(&((ll)->wq), prio)
256
+#define pgc_queue_unlock(cache, ll) waitq_release(&((ll)->wq))
257
#else
259
-#define pgc_queue_trylock(cache, ll) spinlock_trylock(&((ll)->spinlock))
258
+#define pgc_queue_trylock(cache, ll, prio) spinlock_trylock(&((ll)->spinlock))
259
#define pgc_queue_lock(cache, ll, prio) spinlock_lock(&((ll)->spinlock))
260
#define pgc_queue_unlock(cache, ll) spinlock_unlock(&((ll)->spinlock))
261
#endif
@@ -608,7 +607,7 @@ static inline void pgc_stats_index_judy_change(PGC *cache, size_t mem_before_jud
607
}
608
}
609
611
-static void pgc_queue_add(PGC *cache __maybe_unused, struct pgc_queue *q, PGC_PAGE *page, bool having_lock, WAITING_QUEUE_PRIORITY prio __maybe_unused) {
610
+static void pgc_queue_add(PGC *cache __maybe_unused, struct pgc_queue *q, PGC_PAGE *page, bool having_lock, WAITQ_PRIORITY prio __maybe_unused) {
611
if(!having_lock)
612
pgc_queue_lock(cache, q, prio);
613
@@ -677,7 +676,8 @@ static void pgc_queue_add(PGC *cache __maybe_unused, struct pgc_queue *q, PGC_PA
676
pgc_size_histogram_add(cache, &q->stats->size_histogram, page);
677
}
678
680
-static void pgc_queue_del(PGC *cache __maybe_unused, struct pgc_queue *q, PGC_PAGE *page, bool having_lock, WAITING_QUEUE_PRIORITY prio __maybe_unused) {
679
+static void pgc_queue_del(PGC *cache __maybe_unused, struct pgc_queue *q, PGC_PAGE *page, bool having_lock,
680
+ WAITQ_PRIORITY prio __maybe_unused) {
681
if(cache->config.stats)
682
pgc_size_histogram_del(cache, &q->stats->size_histogram, page);
683
@@ -736,7 +736,7 @@ static inline void page_has_been_accessed(PGC *cache, PGC_PAGE *page) {
736
__atomic_add_fetch(&page->accesses, 1, __ATOMIC_RELAXED);
737
738
if (flags & PGC_PAGE_CLEAN) {
739
- if(pgc_queue_trylock(cache, &cache->clean)) {
739
+ if(pgc_queue_trylock(cache, &cache->clean, PGC_QUEUE_LOCK_PRIO_EVICTORS)) {
740
DOUBLE_LINKED_LIST_REMOVE_ITEM_UNSAFE(cache->clean.base, page, link.prev, link.next);
741
DOUBLE_LINKED_LIST_APPEND_ITEM_UNSAFE(cache->clean.base, page, link.prev, link.next);
742
pgc_queue_unlock(cache, &cache->clean);
@@ -752,7 +752,7 @@ static inline void page_has_been_accessed(PGC *cache, PGC_PAGE *page) {
752
// ----------------------------------------------------------------------------
753
// state transitions
754
755
-static inline void page_set_clean(PGC *cache, PGC_PAGE *page, bool having_transition_lock, bool having_clean_lock, WAITING_QUEUE_PRIORITY prio) {
755
+static inline void page_set_clean(PGC *cache, PGC_PAGE *page, bool having_transition_lock, bool having_clean_lock, WAITQ_PRIORITY prio) {
756
if(!having_transition_lock)
757
page_transition_lock(cache, page);
758
@@ -777,7 +777,7 @@ static inline void page_set_clean(PGC *cache, PGC_PAGE *page, bool having_transi
777
page_transition_unlock(cache, page);
778
}
779
780
-static inline void page_set_dirty(PGC *cache, PGC_PAGE *page, bool having_hot_lock, WAITING_QUEUE_PRIORITY prio) {
780
+static inline void page_set_dirty(PGC *cache, PGC_PAGE *page, bool having_hot_lock, WAITQ_PRIORITY prio) {
781
if(!having_hot_lock)
782
// to avoid deadlocks, we have to get the hot lock before the page transition
783
// since this is what all_hot_to_dirty() does
@@ -819,7 +819,7 @@ static inline void page_set_dirty(PGC *cache, PGC_PAGE *page, bool having_hot_lo
819
page_transition_unlock(cache, page);
820
}
821
822
-static inline void page_set_hot(PGC *cache, PGC_PAGE *page, WAITING_QUEUE_PRIORITY prio) {
822
+static inline void page_set_hot(PGC *cache, PGC_PAGE *page, WAITQ_PRIORITY prio) {
823
page_transition_lock(cache, page);
824
825
PGC_PAGE_FLAGS flags = page_get_status_flags(page);
@@ -1222,7 +1222,7 @@ static bool evict_pages_with_filter(PGC *cache, size_t max_skip, size_t max_evic
1222
timing_dbengine_evict_init();
1223
1224
if(!all_of_them && !wait) {
1225
- if(!pgc_queue_trylock(cache, &cache->clean)) {
1225
+ if(!pgc_queue_trylock(cache, &cache->clean, PGC_QUEUE_LOCK_PRIO_EVICTORS)) {
1226
stopped_before_finishing = true;
1227
goto premature_exit;
1228
}
@@ -1759,7 +1759,7 @@ static bool flush_pages(PGC *cache, size_t max_flushes, Word_t section, bool wai
1759
// we have been called from a data collection thread
1760
// let's not waste its time...
1761
1762
- if(!pgc_queue_trylock(cache, &cache->dirty)) {
1762
+ if(!pgc_queue_trylock(cache, &cache->dirty, PGC_QUEUE_LOCK_PRIO_FLUSHERS)) {
1763
// we would block, so give up...
1764
return false;
1765
}
@@ -1942,7 +1942,7 @@ static bool flush_pages(PGC *cache, size_t max_flushes, Word_t section, bool wai
1942
, "DBENGINE CACHE: flushing pages mismatch");
1943
1944
if(!all_of_them && !wait) {
1945
- if(pgc_queue_trylock(cache, &cache->dirty))
1945
+ if(pgc_queue_trylock(cache, &cache->dirty, PGC_QUEUE_LOCK_PRIO_FLUSHERS))
1946
have_dirty_lock = true;
1947
1948
else {
@@ -2099,9 +2099,9 @@ PGC *pgc_create(const char *name,
2099
2100
2101
#if defined(PGC_QUEUE_LOCK_AS_WAITING_QUEUE)
2102
- cache->hot.wq = waiting_queue_create();
2103
- cache->dirty.wq = waiting_queue_create();
2104
- cache->clean.wq = waiting_queue_create();
2102
+ waitq_init(&cache->hot.wq);
2103
+ waitq_init(&cache->dirty.wq);
2104
+ waitq_init(&cache->clean.wq);
2105
#else
2106
spinlock_init(&cache->hot.spinlock);
2107
spinlock_init(&cache->dirty.spinlock);
@@ -2174,9 +2174,9 @@ void pgc_destroy(PGC *cache) {
2174
}
2175
2176
#if defined(PGC_QUEUE_LOCK_AS_WAITING_QUEUE)
2177
- waiting_queue_destroy(cache->hot.wq);
2178
- waiting_queue_destroy(cache->dirty.wq);
2179
- waiting_queue_destroy(cache->clean.wq);
2177
+ waitq_destroy(&cache->hot.wq);
2178
+ waitq_destroy(&cache->dirty.wq);
2179
+ waitq_destroy(&cache->clean.wq);
2180
#endif
2181
freez(cache->index);
2182
freez(cache);
@@ -2453,7 +2453,7 @@ void pgc_open_cache_to_journal_v2(PGC *cache, Word_t section, unsigned datafile_
2453
__atomic_add_fetch(&rrdeng_cache_efficiency_stats.journal_v2_indexing_started, 1, __ATOMIC_RELAXED);
2454
p2_add_fetch(&cache->stats.p2_workers_jv2_flush, 1);
2455
2456
- pgc_queue_lock(cache, &cache->hot, PGC_QUEUE_LOCK_PRIO_OTHERS);
2456
+ pgc_queue_lock(cache, &cache->hot, PGC_QUEUE_LOCK_PRIO_LOW);
2457
2458
Pvoid_t JudyL_metrics = NULL;
2459
Pvoid_t JudyL_extents_pos = NULL;
@@ -2584,7 +2584,7 @@ void pgc_open_cache_to_journal_v2(PGC *cache, Word_t section, unsigned datafile_
2584
}
2585
2586
yield_the_processor(); // do not lock too aggressively
2587
- pgc_queue_lock(cache, &cache->hot, PGC_QUEUE_LOCK_PRIO_OTHERS);
2587
+ pgc_queue_lock(cache, &cache->hot, PGC_QUEUE_LOCK_PRIO_LOW);
2588
}
2589
2590
spinlock_unlock(&sp->migration_to_v2_spinlock);
@@ -2608,7 +2608,7 @@ void pgc_open_cache_to_journal_v2(PGC *cache, Word_t section, unsigned datafile_
2608
2609
// balance-parents: transition from hot to clean directly
2610
yield_the_processor(); // do not lock too aggressively
2611
- page_set_clean(cache, pi->page, true, false, PGC_QUEUE_LOCK_PRIO_OTHERS);
2611
+ page_set_clean(cache, pi->page, true, false, PGC_QUEUE_LOCK_PRIO_LOW);
2612
page_transition_unlock(cache, pi->page);
2613
page_release(cache, pi->page, true);
2614
@@ -2659,7 +2659,7 @@ void pgc_open_evict_clean_pages_of_datafile(PGC *cache, struct rrdengine_datafil
2659
size_t pgc_count_clean_pages_having_data_ptr(PGC *cache, Word_t section, void *ptr) {
2660
size_t found = 0;
2661
2662
- pgc_queue_lock(cache, &cache->clean, PGC_QUEUE_LOCK_PRIO_OTHERS);
2662
+ pgc_queue_lock(cache, &cache->clean, PGC_QUEUE_LOCK_PRIO_LOW);
2663
for(PGC_PAGE *page = cache->clean.base; page ;page = page->link.next)
2664
found += (page->data == ptr && page->section == section) ? 1 : 0;
2665
pgc_queue_unlock(cache, &cache->clean);
@@ -2670,7 +2670,7 @@ size_t pgc_count_clean_pages_having_data_ptr(PGC *cache, Word_t section, void *p
2670
size_t pgc_count_hot_pages_having_data_ptr(PGC *cache, Word_t section, void *ptr) {
2671
size_t found = 0;
2672
2673
- pgc_queue_lock(cache, &cache->hot, PGC_QUEUE_LOCK_PRIO_OTHERS);
2673
+ pgc_queue_lock(cache, &cache->hot, PGC_QUEUE_LOCK_PRIO_LOW);
2674
Pvoid_t *section_pages_pptr = JudyLGet(cache->hot.sections_judy, section, PJE0);
2675
if(section_pages_pptr) {
2676
struct section_pages *sp = *section_pages_pptr;
src/libnetdata/libnetdata.h
+1
-1
@@ -125,7 +125,7 @@ extern const char *netdata_configured_host_prefix;
125
#include "locks/spinlock.h"
126
#include "locks/rw-spinlock.h"
127
#include "completion/completion.h"
128
-#include "waiting-queue/waiting-queue.h"
128
+#include "libnetdata/locks/waitq.h"
129
#include "clocks/clocks.h"
130
#include "simple_pattern/simple_pattern.h"
131
#include "libnetdata/log/nd_log.h"
src/libnetdata/locks/spinlock.c
+1
-1
@@ -28,7 +28,7 @@ void spinlock_lock_with_trace(SPINLOCK *spinlock, const char *func) {
28
// Backoff strategy with exponential growth
29
spins++;
30
microsleep(usec);
31
- usec = usec > MAX_USEC ? MAX_USEC : usec * 2;
31
+ usec = usec >= MAX_USEC ? MAX_USEC : usec * 2;
32
}
33
34
#ifdef NETDATA_INTERNAL_CHECKS
src/libnetdata/locks/waitq.c
new
+264
@@ -0,0 +1,264 @@
1
+// SPDX-License-Identifier: GPL-3.0-or-later
2
+
3
+#include "waitq.h"
4
+
5
+#define MAX_USEC 512 // Maximum backoff limit in microseconds
6
+
7
+#define PRIORITY_SHIFT 32
8
+#define NO_PRIORITY 0
9
+
10
+static inline uint64_t make_order(WAITQ_PRIORITY priority, uint64_t seqno) {
11
+ return ((uint64_t)priority << PRIORITY_SHIFT) + seqno;
12
+}
13
+
14
+static inline uint64_t get_our_order(WAITQ *waitq, WAITQ_PRIORITY priority) {
15
+ uint64_t seqno = __atomic_add_fetch(&waitq->last_seqno, 1, __ATOMIC_RELAXED);
16
+ return make_order(priority, seqno);
17
+}
18
+
19
+void waitq_init(WAITQ *waitq) {
20
+ spinlock_init(&waitq->spinlock);
21
+ waitq->current_priority = 0;
22
+ waitq->last_seqno = 0;
23
+}
24
+
25
+void waitq_destroy(WAITQ *wq __maybe_unused) { ; }
26
+
27
+static inline bool write_our_priority(WAITQ *waitq, uint64_t our_order) {
28
+ uint64_t current = __atomic_load_n(&waitq->current_priority, __ATOMIC_RELAXED);
29
+ if(current == our_order) return true;
30
+
31
+ do {
32
+
33
+ if(current > our_order)
34
+ return false;
35
+
36
+ } while(!__atomic_compare_exchange_n(
37
+ &waitq->current_priority,
38
+ ¤t,
39
+ our_order,
40
+ false,
41
+ __ATOMIC_RELAXED,
42
+ __ATOMIC_RELAXED));
43
+
44
+ return true;
45
+}
46
+
47
+static inline bool clear_our_priority(WAITQ *waitq, uint64_t our_order) {
48
+ uint64_t expected = our_order;
49
+
50
+ return
51
+ __atomic_compare_exchange_n(
52
+ &waitq->current_priority,
53
+ &expected,
54
+ NO_PRIORITY,
55
+ false,
56
+ __ATOMIC_RELAXED,
57
+ __ATOMIC_RELAXED);
58
+}
59
+
60
+bool waitq_try_acquire_with_trace(WAITQ *waitq, WAITQ_PRIORITY priority, const char *func __maybe_unused) {
61
+ uint64_t our_order = get_our_order(waitq, priority);
62
+
63
+ bool rc = write_our_priority(waitq, our_order) && spinlock_trylock(&waitq->spinlock);
64
+ if(rc)
65
+ waitq->writer = gettid_cached();
66
+
67
+ clear_our_priority(waitq, our_order);
68
+
69
+ return rc;
70
+}
71
+
72
+void waitq_acquire_with_trace(WAITQ *waitq, WAITQ_PRIORITY priority, const char *func) {
73
+ uint64_t our_order = get_our_order(waitq, priority);
74
+
75
+ size_t spins = 0;
76
+ usec_t usec = 1;
77
+
78
+ while(true) {
79
+ while (write_our_priority(waitq, our_order)) {
80
+ if(spinlock_trylock(&waitq->spinlock)) {
81
+ waitq->writer = gettid_cached();
82
+ clear_our_priority(waitq, our_order);
83
+ worker_spinlock_contention(func, spins);
84
+ return;
85
+ }
86
+ tinysleep();
87
+ }
88
+
89
+ // Back off
90
+ spins++;
91
+ microsleep(usec);
92
+ usec = usec >= MAX_USEC ? MAX_USEC : usec * 2;
93
+ }
94
+}
95
+
96
+void waitq_release(WAITQ *waitq) {
97
+ spinlock_unlock(&waitq->spinlock);
98
+}
99
+
100
+// --------------------------------------------------------------------------------------------------------------------
101
+
102
+#define THREADS_PER_PRIORITY 2
103
+#define TEST_DURATION_SEC 2
104
+
105
+// For stress test statistics
106
+typedef struct thread_stats {
107
+ WAITQ_PRIORITY priority;
108
+ size_t executions; // how many times we got through
109
+ usec_t total_wait_time; // total time spent waiting
110
+ usec_t max_wait_time; // maximum time spent waiting
111
+} THREAD_STATS;
112
+
113
+struct thread_args {
114
+ THREAD_STATS *stats;
115
+ WAITQ *wq;
116
+ bool with_sleep;
117
+ bool *stop_flag;
118
+};
119
+
120
+static const char *priority_to_string(WAITQ_PRIORITY p) {
121
+ switch(p) {
122
+ case WAITQ_PRIO_URGENT: return "URGENT";
123
+ case WAITQ_PRIO_HIGH: return "HIGH";
124
+ case WAITQ_PRIO_NORMAL: return "NORMAL";
125
+ case WAITQ_PRIO_LOW: return "LOW";
126
+ default: return "UNKNOWN";
127
+ }
128
+}
129
+
130
+static void *stress_thread(void *arg) {
131
+ struct thread_args *args = arg;
132
+
133
+ THREAD_STATS *stats = args->stats;
134
+ WAITQ *wq = args->wq;
135
+ bool with_sleep = args->with_sleep;
136
+ bool *stop_flag = args->stop_flag;
137
+
138
+ while(!__atomic_load_n(stop_flag, __ATOMIC_ACQUIRE)) {
139
+ usec_t waiting_since_ut = now_monotonic_usec();
140
+ waitq_acquire(wq, stats->priority);
141
+ usec_t wait_time = now_monotonic_usec() - waiting_since_ut;
142
+ stats->executions++;
143
+ stats->total_wait_time += wait_time;
144
+ if(wait_time > stats->max_wait_time)
145
+ stats->max_wait_time = wait_time;
146
+
147
+ if(with_sleep)
148
+ tinysleep();
149
+
150
+ waitq_release(wq);
151
+ }
152
+
153
+ return NULL;
154
+}
155
+
156
+static void print_thread_stats(THREAD_STATS *stats, size_t count, usec_t duration) {
157
+ fprintf(stderr, "\n%-8s %12s %12s %12s %12s %12s\n",
158
+ "PRIORITY", "EXECUTIONS", "EXEC/SEC", "AVG WAIT", "MAX WAIT", "% WAITING");
159
+
160
+ size_t total_execs = 0;
161
+ for(size_t i = 0; i < count; i++)
162
+ total_execs += stats[i].executions;
163
+
164
+ double total_time_sec = duration / (double)USEC_PER_SEC;
165
+
166
+ for(size_t i = 0; i < count; i++) {
167
+ double execs_per_sec = stats[i].executions / total_time_sec;
168
+ double avg_wait = stats[i].executions ? (double)stats[i].total_wait_time / stats[i].executions : 0;
169
+ double percent_waiting = stats[i].total_wait_time * 100.0 / duration;
170
+
171
+ fprintf(stderr, "%-8s %12zu %12.1f %12.1f %12"PRIu64" %12.1f%%\n",
172
+ priority_to_string(stats[i].priority),
173
+ stats[i].executions,
174
+ execs_per_sec,
175
+ avg_wait,
176
+ stats[i].max_wait_time,
177
+ percent_waiting);
178
+ }
179
+}
180
+
181
+static int unittest_stress(void) {
182
+ int errors = 0;
183
+ fprintf(stderr, "\nStress testing waiting queue...\n");
184
+
185
+ WAITQ wq = WAITQ_INITIALIZER;
186
+ const size_t num_priorities = 4;
187
+ const size_t total_threads = num_priorities * THREADS_PER_PRIORITY;
188
+
189
+ // Test both with and without sleep
190
+ for(int test = 0; test < 2; test++) {
191
+ bool with_sleep = (test == 1);
192
+ bool stop_flag = false;
193
+
194
+ fprintf(stderr, "\nRunning %ds stress test %s sleep:\n",
195
+ TEST_DURATION_SEC, with_sleep ? "with" : "without");
196
+
197
+ // Prepare thread stats and args
198
+ THREAD_STATS stats[total_threads];
199
+ struct thread_args thread_args[total_threads];
200
+ ND_THREAD *threads[total_threads];
201
+
202
+ fprintf(stderr, "Starting %zu threads for %ds test %s sleep...\n",
203
+ total_threads,
204
+ TEST_DURATION_SEC,
205
+ with_sleep ? "with" : "without");
206
+
207
+ // Initialize stats and create threads
208
+ size_t thread_idx = 0;
209
+ for(int prio = WAITQ_PRIO_URGENT; prio >= WAITQ_PRIO_LOW; prio--) {
210
+ for(int t = 0; t < THREADS_PER_PRIORITY; t++) {
211
+ stats[thread_idx] = (THREAD_STATS){
212
+ .priority = prio,
213
+ .executions = 0,
214
+ .total_wait_time = 0,
215
+ .max_wait_time = 0
216
+ };
217
+ thread_args[thread_idx] = (struct thread_args){
218
+ .stats = &stats[thread_idx], // Pass pointer to stats
219
+ .wq = &wq,
220
+ .with_sleep = with_sleep,
221
+ .stop_flag = &stop_flag
222
+ };
223
+
224
+ char thread_name[32];
225
+ snprintf(thread_name, sizeof(thread_name), "STRESS%d-%d", prio, t);
226
+ threads[thread_idx] = nd_thread_create(
227
+ thread_name,
228
+ NETDATA_THREAD_OPTION_DONT_LOG | NETDATA_THREAD_OPTION_JOINABLE,
229
+ stress_thread,
230
+ &thread_args[thread_idx]);
231
+ thread_idx++;
232
+ }
233
+ }
234
+
235
+ // Let it run
236
+ time_t start = now_monotonic_sec();
237
+ fprintf(stderr, "Running...");
238
+ while(now_monotonic_sec() - start < TEST_DURATION_SEC) {
239
+ fprintf(stderr, ".");
240
+ sleep_usec(500000); // Print a dot every 0.5 seconds
241
+ }
242
+ fprintf(stderr, "\n");
243
+
244
+
245
+ fprintf(stderr, "Stopping threads...\n");
246
+ __atomic_store_n(&stop_flag, true, __ATOMIC_RELEASE);
247
+
248
+ // Wait for threads and collect stats
249
+ fprintf(stderr, "Waiting for %zu threads to finish...\n", total_threads);
250
+ for(size_t i = 0; i < total_threads; i++)
251
+ nd_thread_join(threads[i]);
252
+
253
+ // Print stats
254
+ print_thread_stats(stats, total_threads, TEST_DURATION_SEC * USEC_PER_SEC);
255
+ }
256
+
257
+ waitq_destroy(&wq);
258
+ return errors;
259
+}
260
+
261
+int unittest_waiting_queue(void) {
262
+ int errors = unittest_stress();
263
+ return errors;
264
+}
src/libnetdata/locks/waitq.h
new
+64
@@ -0,0 +1,64 @@
1
+// SPDX-License-Identifier: GPL-3.0-or-later
2
+
3
+#ifndef NETDATA_WAITQ_H
4
+#define NETDATA_WAITQ_H
5
+
6
+#include "libnetdata/libnetdata.h"
7
+
8
+/*
9
+ * WAITING QUEUE
10
+ * Like a spinlock, but:
11
+ *
12
+ * 1. Waiters get a sequence number (FIFO)
13
+ * 2. FIFO is respected within each priority
14
+ * 3. Higher priority threads get in first
15
+ *
16
+ * This is equivalent to 3 atomic operations for lock, and 1 for unlock.
17
+ *
18
+ * As lightweight and fast as it can be.
19
+ * About 3M thread switches/s per WAITING QUEUE, on modern hardware.
20
+ *
21
+ * Be careful: higher priority threads can starve the rest!
22
+ *
23
+ */
24
+
25
+typedef enum __attribute__((packed)) {
26
+ WAITQ_PRIO_LOW = 0, // will be last
27
+ WAITQ_PRIO_NORMAL, // will be third
28
+ WAITQ_PRIO_HIGH, // will be second
29
+ WAITQ_PRIO_URGENT, // will be first
30
+
31
+ // terminator
32
+ WAITQ_PRIO_MAX,
33
+} WAITQ_PRIORITY;
34
+
35
+typedef struct waiting_queue {
36
+ SPINLOCK spinlock; // protects the actual resource
37
+ pid_t writer; // the pid the thread currently holding the lock
38
+ uint64_t current_priority; // current highest priority attempting to acquire
39
+ uint64_t last_seqno; // for FIFO ordering within same priority
40
+} WAITQ;
41
+
42
+#define WAITQ_INITIALIZER (WAITQ){ .spinlock = SPINLOCK_INITIALIZER, .current_priority = 0, .last_seqno = 0, }
43
+
44
+// Initialize a waiting queue
45
+void waitq_init(WAITQ *waitq);
46
+
47
+// Destroy a waiting queue - must be empty
48
+void waitq_destroy(WAITQ *wq);
49
+
50
+// Returns true when the queue is acquired
51
+bool waitq_try_acquire_with_trace(WAITQ *waitq, WAITQ_PRIORITY priority, const char *func);
52
+#define waitq_try_acquire(waitq, priority) waitq_try_acquire_with_trace(waitq, priority, __FUNCTION__)
53
+
54
+// Returns when it is our turn to run
55
+// Returns time spent waiting in microseconds
56
+void waitq_acquire_with_trace(WAITQ *waitq, WAITQ_PRIORITY priority, const char *func);
57
+#define waitq_acquire(waitq, priority) waitq_acquire_with_trace(waitq, priority, __FUNCTION__)
58
+
59
+// Mark that we are done - wakes up the next in line
60
+void waitq_release(WAITQ *waitq);
61
+
62
+int unittest_waiting_queue(void);
63
+
64
+#endif // NETDATA_WAITQ_H
src/libnetdata/log/nd_log.c
+3
-2
@@ -415,8 +415,9 @@ void netdata_logger_fatal(const char *file, const char *function, const unsigned
415
size_t recursion = __atomic_add_fetch(&already_in_fatal, 1, __ATOMIC_SEQ_CST);
416
if(recursion > 1) {
417
// exit immediately, nothing more to be done
418
- fprintf(stderr, "RECURSIVE FATAL STATEMENTS, latest from %lu@%s() of %s, EXITING NOW!\n",
419
- line, function, file);
418
+ sleep(2); // give the first fatal the chance to be written
419
+ fprintf(stderr, "\nRECURSIVE FATAL STATEMENTS, latest from %s() of %lu@%s, EXITING NOW! 23e93dfccbf64e11aac858b9410d8a82\n",
420
+ function, line, file);
421
fflush(stderr);
422
_exit(1);
423
}
src/libnetdata/waiting-queue/waiting-queue.c
deleted
-485
@@ -1,485 +0,0 @@
1
-// SPDX-License-Identifier: GPL-3.0-or-later
2
-
3
-#include "waiting-queue.h"
4
-
5
-// #define WAITING_QUEUE_USE_FUTEX 1
6
-
7
-#if defined(WAITING_QUEUE_USE_FUTEX)
8
-#include <linux/futex.h>
9
-#include <sys/syscall.h>
10
-
11
-typedef int WQ_COND;
12
-typedef SPINLOCK WQ_MUTEX;
13
-
14
-static inline int futex_wait(int *uaddr, int val) {
15
- return syscall(SYS_futex, uaddr, FUTEX_WAIT_PRIVATE, val, NULL, NULL, 0);
16
-}
17
-
18
-static inline int futex_wake(int *uaddr, int n) {
19
- return syscall(SYS_futex, uaddr, FUTEX_WAKE_PRIVATE, n, NULL, NULL, 0);
20
-}
21
-
22
-static int WQ_COND_init(WQ_COND *cond) { *(cond) = 0; return 0; }
23
-#define WQ_COND_destroy(cond) debug_dummy()
24
-
25
-static inline int WQ_COND_wait(WQ_COND *cond, WQ_MUTEX *mutex) {
26
- int value = *cond;
27
- spinlock_unlock(mutex);
28
- futex_wait(cond, value);
29
- spinlock_lock(mutex);
30
- return 0;
31
-}
32
-
33
-static inline int WQ_COND_signal(WQ_COND *cond) {
34
- __atomic_add_fetch(cond, 1, __ATOMIC_SEQ_CST);
35
- futex_wake(cond, 1);
36
- return 0;
37
-}
38
-
39
-#define WQ_MUTEX_init(mutex) ({ spinlock_init(mutex) ; 0; })
40
-#define WQ_MUTEX_destroy(mutex) debug_dummy()
41
-#define WQ_MUTEX_lock(mutex) spinlock_lock(mutex)
42
-#define WQ_MUTEX_unlock(mutex) spinlock_unlock(mutex)
43
-
44
-#else // !USE_FUTEX
45
-
46
-typedef uv_cond_t WQ_COND;
47
-typedef uv_mutex_t WQ_MUTEX;
48
-
49
-#define WQ_COND_init(cond) uv_cond_init(cond)
50
-#define WQ_COND_destroy(cond) uv_cond_destroy(cond)
51
-#define WQ_COND_wait(cond, mutex) uv_cond_wait(cond, mutex)
52
-#define WQ_COND_signal(cond) uv_cond_signal(cond)
53
-
54
-#define WQ_MUTEX_init(mutex) uv_mutex_init(mutex)
55
-#define WQ_MUTEX_destroy(mutex) uv_mutex_destroy(mutex)
56
-#define WQ_MUTEX_lock(mutex) uv_mutex_lock(mutex)
57
-#define WQ_MUTEX_unlock(mutex) uv_mutex_unlock(mutex)
58
-
59
-#endif // USE_FUTEX
60
-
61
-typedef struct waiting_thread {
62
- WQ_COND cond; // condition variable for this thread
63
- usec_t waiting_since_ut; // when we started waiting
64
- Word_t priority;
65
- struct waiting_thread *prev, *next;
66
-} WAITING_THREAD;
67
-
68
-struct waiting_queue {
69
- WQ_MUTEX mutex; // ensure acquirers and releasers are synchronized
70
- Word_t last_seqno; // incrementing sequence counter
71
- SPINLOCK spinlock; // ensures there is only 1 runner at a time
72
- REFCOUNT running; // number of threads, including the one holding the lock
73
- pid_t writer;
74
- WAITING_THREAD *list; // the list of threads waiting, not including the 1 holding the lock
75
-};
76
-
77
-// Determine available bits based on system word size
78
-#if SIZEOF_VOID_P == 8
79
-#define PRIORITY_SHIFT 62ULL
80
-#define SEQNO_MASK ((1ULL << PRIORITY_SHIFT) - 1)
81
-#else
82
-#define PRIORITY_SHIFT 30U
83
-#define SEQNO_MASK ((1U << PRIORITY_SHIFT) - 1)
84
-#endif
85
-
86
-static inline Word_t make_key(WAITING_QUEUE_PRIORITY priority, Word_t seqno) {
87
- return ((Word_t)priority << PRIORITY_SHIFT) | (seqno & SEQNO_MASK);
88
-}
89
-
90
-static inline WAITING_QUEUE_PRIORITY key_get_priority(Word_t key) {
91
- return (WAITING_QUEUE_PRIORITY)(key >> PRIORITY_SHIFT);
92
-}
93
-
94
-static inline Word_t key_get_seqno(Word_t key) {
95
- return key & SEQNO_MASK;
96
-}
97
-
98
-WAITING_QUEUE *waiting_queue_create(void) {
99
- WAITING_QUEUE *wq = callocz(1, sizeof(WAITING_QUEUE));
100
-
101
- int ret = WQ_MUTEX_init(&wq->mutex);
102
- if(ret != 0) {
103
- freez(wq);
104
- return NULL;
105
- }
106
-
107
- spinlock_init(&wq->spinlock);
108
-
109
- wq->running = 0;
110
- return wq;
111
-}
112
-
113
-void waiting_queue_destroy(WAITING_QUEUE *wq) {
114
- if(!wq) return;
115
-
116
- if(wq->running)
117
- fatal("WAITING_QUEUE: destroying waiting queue that still has %d threads running/waiting", wq->running);
118
-
119
- WQ_MUTEX_destroy(&wq->mutex);
120
- freez(wq);
121
-}
122
-
123
-static inline void WAITERS_SET(WAITING_QUEUE *wq, WAITING_THREAD *wt) {
124
- for(WAITING_THREAD *t = wq->list ; t ;t = t->next) {
125
- if(wt->priority < t->priority) {
126
- DOUBLE_LINKED_LIST_INSERT_ITEM_BEFORE_UNSAFE(wq->list, t, wt, prev, next);
127
- return;
128
- }
129
- }
130
- DOUBLE_LINKED_LIST_APPEND_ITEM_UNSAFE(wq->list, wt, prev, next);
131
-}
132
-
133
-static inline void WAITERS_DEL(WAITING_QUEUE *wq, WAITING_THREAD *wt) {
134
- DOUBLE_LINKED_LIST_REMOVE_ITEM_UNSAFE(wq->list, wt, prev, next);
135
-}
136
-
137
-static inline WAITING_THREAD *WAITERS_FIRST(WAITING_QUEUE *wq) {
138
- return wq->list;
139
-}
140
-
141
-static inline void WAITING_THREAD_init(WAITING_QUEUE *wq, WAITING_THREAD *wt, WAITING_QUEUE_PRIORITY priority) {
142
- Word_t seqno = __atomic_add_fetch(&wq->last_seqno, 1, __ATOMIC_RELAXED);
143
- wt->priority = make_key(priority, seqno);
144
- wt->waiting_since_ut = now_monotonic_usec();
145
- wt->prev = wt->next = NULL;
146
-
147
- int ret = WQ_COND_init(&wt->cond);
148
- if(ret != 0)
149
- fatal("WAITING_QUEUE: cannot initialize condition variable");
150
-}
151
-
152
-static inline void WAITING_THREAD_cleanup(WAITING_QUEUE *wq __maybe_unused, WAITING_THREAD *wt __maybe_unused) {
153
- WQ_COND_destroy(&wt->cond);
154
-}
155
-
156
-bool waiting_queue_try_acquire(WAITING_QUEUE *wq) {
157
- if(__atomic_add_fetch(&wq->running, 1, __ATOMIC_RELAXED) == 1 &&
158
- spinlock_trylock(&wq->spinlock)) {
159
- return true;
160
- }
161
-
162
- __atomic_sub_fetch(&wq->running, 1, __ATOMIC_RELAXED);
163
- return false;
164
-}
165
-
166
-usec_t waiting_queue_acquire(WAITING_QUEUE *wq, WAITING_QUEUE_PRIORITY priority) {
167
- // Try fast path first - if we're the only one, just go
168
- if(__atomic_add_fetch(&wq->running, 1, __ATOMIC_RELAXED) == 1 &&
169
- spinlock_trylock(&wq->spinlock)) {
170
- return 0;
171
- }
172
-
173
- // Slow path - need to wait
174
-
175
- WAITING_THREAD wt;
176
- WAITING_THREAD_init(wq, &wt, priority);
177
-
178
- WQ_MUTEX_lock(&wq->mutex);
179
- WAITERS_SET(wq, &wt);
180
-
181
- // Wait for our turn
182
- do {
183
- if (WAITERS_FIRST(wq) == &wt && spinlock_trylock(&wq->spinlock))
184
- break;
185
- else
186
- WQ_COND_wait(&wt.cond, &wq->mutex);
187
- } while(true);
188
-
189
- wq->writer = gettid_cached();
190
- WAITERS_DEL(wq, &wt);
191
- WQ_MUTEX_unlock(&wq->mutex);
192
- WAITING_THREAD_cleanup(wq, &wt);
193
-
194
- return now_monotonic_usec() - wt.waiting_since_ut;
195
-}
196
-
197
-void waiting_queue_release(WAITING_QUEUE *wq) {
198
- wq->writer = 0;
199
- spinlock_unlock(&wq->spinlock);
200
-
201
- // Fast path if we're alone
202
- if(__atomic_sub_fetch(&wq->running, 1, __ATOMIC_RELAXED) == 0)
203
- return;
204
-
205
- // Slow path - need to signal next in line
206
- WQ_MUTEX_lock(&wq->mutex);
207
-
208
- // Wake up next in line if any
209
- if(wq->list)
210
- WQ_COND_signal(&wq->list->cond);
211
-
212
- WQ_MUTEX_unlock(&wq->mutex);
213
-}
214
-
215
-size_t waiting_queue_waiting(WAITING_QUEUE *wq) {
216
- return __atomic_load_n(&wq->running, __ATOMIC_RELAXED);
217
-}
218
-
219
-
220
-// --------------------------------------------------------------------------------------------------------------------
221
-
222
-// For stress test statistics
223
-typedef struct thread_stats {
224
- WAITING_QUEUE_PRIORITY priority;
225
- size_t executions; // how many times we got through
226
- usec_t total_wait_time; // total time spent waiting
227
- usec_t max_wait_time; // maximum time spent waiting
228
-} THREAD_STATS;
229
-
230
-struct thread_args {
231
- THREAD_STATS *stats;
232
- WAITING_QUEUE *wq;
233
- bool with_sleep;
234
- bool *stop_flag;
235
-};
236
-
237
-static const char *priority_to_string(WAITING_QUEUE_PRIORITY p) {
238
- switch(p) {
239
- case WAITING_QUEUE_PRIO_URGENT: return "URGENT";
240
- case WAITING_QUEUE_PRIO_HIGH: return "HIGH";
241
- case WAITING_QUEUE_PRIO_NORMAL: return "NORMAL";
242
- case WAITING_QUEUE_PRIO_LOW: return "LOW";
243
- default: return "UNKNOWN";
244
- }
245
-}
246
-
247
-static int unittest_functional(void) {
248
- int errors = 0;
249
- fprintf(stderr, "\nTesting waiting queue...\n");
250
-
251
- WAITING_QUEUE *wq = waiting_queue_create();
252
-
253
- // Test 1: Fast path should work with no contention
254
- fprintf(stderr, " Test 1: Fast path - no contention: ");
255
- usec_t wait_time = waiting_queue_acquire(wq, WAITING_QUEUE_PRIO_NORMAL);
256
- waiting_queue_release(wq);
257
- if(wait_time != 0) {
258
- fprintf(stderr, "FAILED (waited %"PRIu64" usec)\n", wait_time);
259
- errors++;
260
- }
261
- else
262
- fprintf(stderr, "OK\n");
263
-
264
- // Test 2: Priorities should be respected
265
- fprintf(stderr, " Test 2: Priority ordering: ");
266
- WAITING_THREAD threads[100];
267
- for(size_t t = 0; t < _countof(threads); t++) {
268
- __atomic_add_fetch(&wq->running, 1, __ATOMIC_RELAXED);
269
- WAITING_THREAD_init(wq, &threads[t], os_random(WAITING_QUEUE_PRIO_MAX));
270
- WAITERS_SET(wq, &threads[t]);
271
- }
272
-
273
- bool failed = false;
274
- size_t prio_counts[WAITING_QUEUE_PRIO_MAX] = { 0 };
275
- WAITING_QUEUE_PRIORITY last_prio = WAITING_QUEUE_PRIO_URGENT;
276
- Word_t last_seqno = 0;
277
- for(size_t t = 0; t < _countof(threads); t++) {
278
- WAITING_THREAD *wt = WAITERS_FIRST(wq);
279
- WAITERS_DEL(wq, wt);
280
- __atomic_sub_fetch(&wq->running, 1, __ATOMIC_RELAXED);
281
-
282
- WAITING_QUEUE_PRIORITY prio = key_get_priority(wt->priority);
283
- Word_t seqno = key_get_seqno(wt->priority);
284
-
285
- prio_counts[prio]++;
286
- if(prio < last_prio) {
287
- if(!failed)
288
- fprintf(stderr, "FAILED\n");
289
-
290
- fprintf(stderr, " > ERROR: prio %u is before prio %u\n", prio, last_prio);
291
- errors++;
292
- failed = true;
293
- }
294
- else if(prio == last_prio && seqno < last_seqno) {
295
- if(!failed)
296
- fprintf(stderr, "FAILED\n");
297
-
298
- fprintf(stderr, " > ERROR: seqno %lu is before seqno %lu\n", seqno, last_seqno);
299
- errors++;
300
- failed = true;
301
- }
302
-
303
- last_seqno = seqno;
304
- last_prio = prio;
305
- WAITING_THREAD_cleanup(wq, wt);
306
- }
307
-
308
- if(!failed)
309
- fprintf(stderr, "OK\n");
310
-
311
- for(size_t p = 0; p < WAITING_QUEUE_PRIO_MAX ;p++)
312
- fprintf(stderr, " > prio %zu got %zu waiters\n", p, prio_counts[p]);
313
-
314
- // Test 3: Queue stats should be accurate
315
- fprintf(stderr, " Test 3: Queue statistics: ");
316
- size_t waiting = waiting_queue_waiting(wq);
317
- if(waiting != 0) {
318
- fprintf(stderr, "FAILED (queue shows %zu waiting)\n", waiting);
319
- errors++;
320
- }
321
- else
322
- fprintf(stderr, "OK\n");
323
-
324
- waiting_queue_destroy(wq);
325
- return errors;
326
-}
327
-
328
-static void *stress_thread(void *arg) {
329
- struct thread_args *args = arg;
330
-
331
- THREAD_STATS *stats = args->stats;
332
- WAITING_QUEUE *wq = args->wq;
333
- bool with_sleep = args->with_sleep;
334
- bool *stop_flag = args->stop_flag;
335
-
336
- while(!__atomic_load_n(stop_flag, __ATOMIC_ACQUIRE)) {
337
- usec_t wait_time = waiting_queue_acquire(wq, stats->priority);
338
- stats->executions++;
339
- stats->total_wait_time += wait_time;
340
- if(wait_time > stats->max_wait_time)
341
- stats->max_wait_time = wait_time;
342
-
343
- if(with_sleep)
344
- tinysleep();
345
-
346
- waiting_queue_release(wq);
347
- }
348
-
349
- return NULL;
350
-}
351
-
352
-static void print_thread_stats(THREAD_STATS *stats, size_t count, usec_t duration) {
353
- fprintf(stderr, "\n%-8s %12s %12s %12s %12s %12s\n",
354
- "PRIORITY", "EXECUTIONS", "EXEC/SEC", "AVG WAIT", "MAX WAIT", "% WAITING");
355
-
356
- size_t total_execs = 0;
357
- for(size_t i = 0; i < count; i++)
358
- total_execs += stats[i].executions;
359
-
360
- double total_time_sec = duration / (double)USEC_PER_SEC;
361
-
362
- for(size_t i = 0; i < count; i++) {
363
- double execs_per_sec = stats[i].executions / total_time_sec;
364
- double avg_wait = stats[i].executions ? (double)stats[i].total_wait_time / stats[i].executions : 0;
365
- double percent_waiting = stats[i].total_wait_time * 100.0 / duration;
366
-
367
- fprintf(stderr, "%-8s %12zu %12.1f %12.1f %12"PRIu64" %12.1f%%\n",
368
- priority_to_string(stats[i].priority),
369
- stats[i].executions,
370
- execs_per_sec,
371
- avg_wait,
372
- stats[i].max_wait_time,
373
- percent_waiting);
374
- }
375
-}
376
-
377
-#define THREADS_PER_PRIORITY 2
378
-#define TEST_DURATION_SEC 5
379
-
380
-static int unittest_stress(void) {
381
- int errors = 0;
382
- fprintf(stderr, "\nStress testing waiting queue...\n");
383
-
384
- WAITING_QUEUE *wq = waiting_queue_create();
385
- const size_t num_priorities = 4;
386
- const size_t total_threads = num_priorities * THREADS_PER_PRIORITY;
387
-
388
- // Test both with and without sleep
389
- for(int test = 0; test < 2; test++) {
390
- bool with_sleep = (test == 1);
391
- bool stop_flag = false;
392
-
393
- fprintf(stderr, "\nRunning %ds stress test %s sleep:\n",
394
- TEST_DURATION_SEC, with_sleep ? "with" : "without");
395
-
396
- // Prepare thread stats and args
397
- THREAD_STATS stats[total_threads];
398
- struct thread_args thread_args[total_threads];
399
- ND_THREAD *threads[total_threads];
400
-
401
- fprintf(stderr, "Starting %zu threads for %ds test %s sleep...\n",
402
- total_threads,
403
- TEST_DURATION_SEC,
404
- with_sleep ? "with" : "without");
405
-
406
- // Initialize stats and create threads
407
- size_t thread_idx = 0;
408
- for(int prio = WAITING_QUEUE_PRIO_URGENT; prio <= WAITING_QUEUE_PRIO_LOW; prio++) {
409
- for(int t = 0; t < THREADS_PER_PRIORITY; t++) {
410
- stats[thread_idx] = (THREAD_STATS){
411
- .priority = prio,
412
- .executions = 0,
413
- .total_wait_time = 0,
414
- .max_wait_time = 0
415
- };
416
- thread_args[thread_idx] = (struct thread_args){
417
- .stats = &stats[thread_idx], // Pass pointer to stats
418
- .wq = wq,
419
- .with_sleep = with_sleep,
420
- .stop_flag = &stop_flag
421
- };
422
-
423
- char thread_name[32];
424
- snprintf(thread_name, sizeof(thread_name), "STRESS%d-%d", prio, t);
425
- threads[thread_idx] = nd_thread_create(
426
- thread_name,
427
- NETDATA_THREAD_OPTION_DONT_LOG | NETDATA_THREAD_OPTION_JOINABLE,
428
- stress_thread,
429
- &thread_args[thread_idx]);
430
- thread_idx++;
431
- }
432
- }
433
-
434
- // Let it run
435
- time_t start = now_monotonic_sec();
436
- fprintf(stderr, "Running...");
437
- while(now_monotonic_sec() - start < TEST_DURATION_SEC) {
438
- fprintf(stderr, ".");
439
- sleep_usec(500000); // Print a dot every 0.5 seconds
440
- }
441
- fprintf(stderr, "\n");
442
-
443
-
444
- fprintf(stderr, "Stopping threads...\n");
445
- __atomic_store_n(&stop_flag, true, __ATOMIC_RELEASE);
446
-
447
- // Wait for threads and collect stats
448
- fprintf(stderr, "Waiting for %zu threads to finish...\n", total_threads);
449
- for(size_t i = 0; i < total_threads; i++)
450
- nd_thread_join(threads[i]);
451
-
452
- // Print stats
453
- print_thread_stats(stats, total_threads, TEST_DURATION_SEC * USEC_PER_SEC);
454
-
455
-// // Basic validation
456
-// for(size_t i = 0; i < total_threads - THREADS_PER_PRIORITY; i++) {
457
-// if(stats[i].executions < stats[i + THREADS_PER_PRIORITY].executions) {
458
-// fprintf(stderr, "ERROR: Higher priority thread got fewer executions!\n");
459
-// errors++;
460
-// }
461
-// }
462
-//
463
-// // Check fairness within same priority
464
-// for(size_t i = 0; i < total_threads; i += THREADS_PER_PRIORITY) {
465
-// for(size_t j = i + 1; j < i + THREADS_PER_PRIORITY; j++) {
466
-// double diff = (double)(stats[i].executions - stats[j].executions) /
467
-// (double)(stats[i].executions + stats[j].executions);
468
-// if(fabs(diff) > 0.1) { // allow 10% difference
469
-// fprintf(stderr, "ERROR: Unfair distribution within same priority!\n");
470
-// errors++;
471
-// }
472
-// }
473
-// }
474
- }
475
-
476
- waiting_queue_destroy(wq);
477
- return errors;
478
-}
479
-
480
-int unittest_waiting_queue(void) {
481
- int errors = unittest_functional();
482
- errors += unittest_stress();
483
-
484
- return errors;
485
-}
src/libnetdata/waiting-queue/waiting-queue.h
deleted
-60
@@ -1,60 +0,0 @@
1
-// SPDX-License-Identifier: GPL-3.0-or-later
2
-
3
-#ifndef NETDATA_WAITING_QUEUE_H
4
-#define NETDATA_WAITING_QUEUE_H
5
-
6
-#include "libnetdata/libnetdata.h"
7
-
8
-/*
9
- * WAITING QUEUE
10
- * Like a mutex, or a spinlock, but:
11
- *
12
- * 1. Waiters get a sequence number (FIFO)
13
- * 2. FIFO is respected within each priority
14
- * 3. Higher priority threads get in first
15
- * 4. No wasting of resources, there are no spins
16
- *
17
- * When there are no other waiters, this is equivalent to 2 atomic
18
- * operations for lock, and 2 for unlock.
19
- *
20
- * As lightweight and fast as it can be.
21
- * About 0.5M thread switches/s per WAITING QUEUE, on modern hardware.
22
- *
23
- * Be careful: higher priority threads can starve the rest!
24
- *
25
- */
26
-
27
-typedef struct waiting_queue WAITING_QUEUE;
28
-
29
-typedef enum __attribute__((packed)) {
30
- WAITING_QUEUE_PRIO_URGENT = 0, // will be first
31
- WAITING_QUEUE_PRIO_HIGH, // will be second
32
- WAITING_QUEUE_PRIO_NORMAL, // will be third
33
- WAITING_QUEUE_PRIO_LOW, // will be last
34
-
35
- // terminator
36
- WAITING_QUEUE_PRIO_MAX,
37
-} WAITING_QUEUE_PRIORITY;
38
-
39
-// Initialize a waiting queue
40
-WAITING_QUEUE *waiting_queue_create(void);
41
-
42
-// Destroy a waiting queue - must be empty
43
-void waiting_queue_destroy(WAITING_QUEUE *wq);
44
-
45
-// Returns true when the queue is acquired
46
-bool waiting_queue_try_acquire(WAITING_QUEUE *wq);
47
-
48
-// Returns when it is our turn to run
49
-// Returns time spent waiting in microseconds
50
-usec_t waiting_queue_acquire(WAITING_QUEUE *wq, WAITING_QUEUE_PRIORITY priority);
51
-
52
-// Mark that we are done - wakes up the next in line
53
-void waiting_queue_release(WAITING_QUEUE *wq);
54
-
55
-// Return the number of threads currently waiting
56
-size_t waiting_queue_waiting(WAITING_QUEUE *wq);
57
-
58
-int unittest_waiting_queue(void);
59
-
60
-#endif // NETDATA_WAITING_QUEUE_H
src/streaming/stream-sender-api.c
+2
-3
@@ -34,7 +34,7 @@ void stream_sender_structures_init(RRDHOST *host, bool stream, STRING *parents,
34
host->sender->connector.id = -1;
35
host->sender->host = host;
36
host->sender->scb = stream_circular_buffer_create();
37
- host->sender->wait_queue = waiting_queue_create();
37
+ waitq_init(&host->sender->waitq);
38
host->sender->capabilities = stream_our_capabilities(host, true);
39
40
nd_sock_init(&host->sender->sock, netdata_ssl_streaming_sender_ctx, netdata_ssl_validate_certificate_sender);
@@ -65,8 +65,7 @@ void stream_sender_structures_free(struct rrdhost *host) {
65
stream_sender_signal_to_stop_and_wait(host, STREAM_HANDSHAKE_DISCONNECT_HOST_CLEANUP, true);
66
stream_circular_buffer_destroy(host->sender->scb);
67
host->sender->scb = NULL;
68
- waiting_queue_destroy(host->sender->wait_queue);
69
- host->sender->wait_queue = NULL;
68
+ waitq_destroy(&host->sender->waitq);
69
stream_compressor_destroy(&host->sender->compressor);
70
71
replication_sender_cleanup(host->sender);
src/streaming/stream-sender-commit.c
+8
-8
@@ -68,11 +68,11 @@ void sender_buffer_commit(struct sender_state *s, BUFFER *wb, struct sender_buff
68
if (unlikely(!src || !src_len))
69
return;
70
71
- waiting_queue_acquire(
72
- s->wait_queue,
71
+ waitq_acquire(
72
+ &s->waitq,
73
(s->host->stream.rcv.status.tid == gettid_cached() || s->host->stream.snd.status.tid == gettid_cached()) ?
74
- WAITING_QUEUE_PRIO_HIGH :
75
- WAITING_QUEUE_PRIO_NORMAL);
74
+ WAITQ_PRIO_HIGH :
75
+ WAITQ_PRIO_NORMAL);
76
stream_sender_lock(s);
77
78
// copy the sequence number of sender buffer recreates, while having our lock
@@ -87,7 +87,7 @@ void sender_buffer_commit(struct sender_state *s, BUFFER *wb, struct sender_buff
87
sender_buffer_destroy(commit);
88
89
stream_sender_unlock(s);
90
- waiting_queue_release(s->wait_queue);
90
+ waitq_release(&s->waitq);
91
return;
92
}
93
@@ -183,7 +183,7 @@ void sender_buffer_commit(struct sender_state *s, BUFFER *wb, struct sender_buff
183
msg = s->thread.msg;
184
185
stream_sender_unlock(s);
186
- waiting_queue_release(s->wait_queue);
186
+ waitq_release(&s->waitq);
187
188
if (enable_sending) {
189
msg.opcode = STREAM_OPCODE_SENDER_POLLOUT;
@@ -195,7 +195,7 @@ void sender_buffer_commit(struct sender_state *s, BUFFER *wb, struct sender_buff
195
overflow_with_lock: {
196
msg = s->thread.msg;
197
stream_sender_unlock(s);
198
- waiting_queue_release(s->wait_queue);
198
+ waitq_release(&s->waitq);
199
msg.opcode = STREAM_OPCODE_SENDER_BUFFER_OVERFLOW;
200
stream_sender_send_opcode(s, msg);
201
nd_log_limit_static_global_var(erl, 1, 0);
@@ -211,7 +211,7 @@ compression_failed_with_lock: {
211
stream_compression_deactivate(s);
212
msg = s->thread.msg;
213
stream_sender_unlock(s);
214
- waiting_queue_release(s->wait_queue);
214
+ waitq_release(&s->waitq);
215
msg.opcode = STREAM_OPCODE_SENDER_RECONNECT_WITHOUT_COMPRESSION;
216
stream_sender_send_opcode(s, msg);
217
nd_log_limit_static_global_var(erl, 1, 0);
src/streaming/stream-sender-internals.h
+1
-1
@@ -59,7 +59,7 @@ struct sender_state {
59
char remote_ip[CONNECTED_TO_SIZE + 1]; // We don't know which proxy we connect to, passed back from socket.c
60
time_t last_state_since_t; // the timestamp of the last state (online/offline) change
61
62
- WAITING_QUEUE *wait_queue;
62
+ WAITQ waitq;
63
STREAM_CIRCULAR_BUFFER *scb;
64
65
struct {