@cryptotaxi247 / netdata
Code Commits Issues Pulls Sessions CI/CD
master
master
c 284 lines 9.2 KB
Raw
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 ALWAYS_INLINE uint64_t make_order(WAITQ_PRIORITY priority, uint64_t seqno) {
11 return ((uint64_t)priority << PRIORITY_SHIFT) + seqno;
12 }
13
14 static ALWAYS_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 ALWAYS_INLINE void waitq_init(WAITQ *waitq) {
20 spinlock_init(&waitq->spinlock);
21 waitq->current_priority = 0;
22 waitq->last_seqno = 0;
23 }
24
25 ALWAYS_INLINE void waitq_destroy(WAITQ *wq __maybe_unused) { ; }
26
27 static ALWAYS_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 != NO_PRIORITY && current < our_order)
34 return false;
35
36 } while(!__atomic_compare_exchange_n(
37 &waitq->current_priority,
38 &current,
39 our_order,
40 false,
41 __ATOMIC_RELAXED,
42 __ATOMIC_RELAXED));
43
44 return true;
45 }
46
47 static ALWAYS_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 ALWAYS_INLINE bool waitq_try_acquire_with_trace(WAITQ *waitq, WAITQ_PRIORITY priority, const char *func __maybe_unused) {
61 // Fast path for no contention - try to get the lock immediately without a sequence number
62 if (__atomic_load_n(&waitq->current_priority, __ATOMIC_RELAXED) == NO_PRIORITY &&
63 spinlock_trylock(&waitq->spinlock)) {
64 waitq->writer = gettid_cached();
65 return true;
66 }
67
68 // Normal path with queuing if contention exists
69 uint64_t our_order = get_our_order(waitq, priority);
70
71 bool rc = write_our_priority(waitq, our_order) && spinlock_trylock(&waitq->spinlock);
72 if(rc)
73 waitq->writer = gettid_cached();
74
75 clear_our_priority(waitq, our_order);
76
77 return rc;
78 }
79
80 ALWAYS_INLINE void waitq_acquire_with_trace(WAITQ *waitq, WAITQ_PRIORITY priority, const char *func) {
81 // Fast path for no contention - try to get the lock immediately without a sequence number
82 if (__atomic_load_n(&waitq->current_priority, __ATOMIC_RELAXED) == NO_PRIORITY &&
83 spinlock_trylock(&waitq->spinlock)) {
84 waitq->writer = gettid_cached();
85 return;
86 }
87
88 // Normal path with queuing if contention exists
89 uint64_t our_order = get_our_order(waitq, priority);
90
91 size_t spins = 0;
92 usec_t usec = 1;
93 usec_t deadlock_timestamp = 0;
94
95 while(true) {
96 while (write_our_priority(waitq, our_order)) {
97 if(spinlock_trylock(&waitq->spinlock)) {
98 waitq->writer = gettid_cached();
99 clear_our_priority(waitq, our_order);
100 worker_spinlock_contention(func, spins);
101 return;
102 }
103 yield_the_processor();
104 }
105
106 // Back off
107 spins++;
108
109 // Check for deadlock every SPINS_BEFORE_DEADLOCK_CHECK iterations
110 if ((spins % SPINS_BEFORE_DEADLOCK_CHECK) == 0) {
111 spinlock_deadlock_detect(&deadlock_timestamp, "waitq", func);
112 }
113
114 microsleep(usec);
115 usec = usec >= MAX_USEC ? MAX_USEC : usec * 2;
116 }
117 }
118
119 ALWAYS_INLINE void waitq_release(WAITQ *waitq) {
120 spinlock_unlock(&waitq->spinlock);
121 }
122
123 // --------------------------------------------------------------------------------------------------------------------
124
125 #define THREADS_PER_PRIORITY 2
126 #define TEST_DURATION_SEC 2
127
128 // For stress test statistics
129 typedef struct thread_stats {
130 WAITQ_PRIORITY priority;
131 size_t executions; // how many times we got through
132 usec_t total_wait_time; // total time spent waiting
133 usec_t max_wait_time; // maximum time spent waiting
134 } THREAD_STATS;
135
136 struct thread_args {
137 THREAD_STATS *stats;
138 WAITQ *wq;
139 bool with_sleep;
140 bool *stop_flag;
141 };
142
143 static const char *priority_to_string(WAITQ_PRIORITY p) {
144 switch(p) {
145 case WAITQ_PRIO_URGENT: return "URGENT";
146 case WAITQ_PRIO_HIGH: return "HIGH";
147 case WAITQ_PRIO_NORMAL: return "NORMAL";
148 case WAITQ_PRIO_LOW: return "LOW";
149 default: return "UNKNOWN";
150 }
151 }
152
153 static void stress_thread(void *arg) {
154 struct thread_args *args = arg;
155
156 THREAD_STATS *stats = args->stats;
157 WAITQ *wq = args->wq;
158 bool with_sleep = args->with_sleep;
159 bool *stop_flag = args->stop_flag;
160
161 while(!__atomic_load_n(stop_flag, __ATOMIC_ACQUIRE)) {
162 usec_t waiting_since_ut = now_monotonic_usec();
163 waitq_acquire(wq, stats->priority);
164 usec_t wait_time = now_monotonic_usec() - waiting_since_ut;
165 stats->executions++;
166 stats->total_wait_time += wait_time;
167 if(wait_time > stats->max_wait_time)
168 stats->max_wait_time = wait_time;
169
170 if(with_sleep)
171 tinysleep();
172
173 waitq_release(wq);
174 }
175 }
176
177 static void print_thread_stats(THREAD_STATS *stats, size_t count, usec_t duration) {
178 fprintf(stderr, "\n%-8s %12s %12s %12s %12s %12s\n",
179 "PRIORITY", "EXECUTIONS", "EXEC/SEC", "AVG WAIT", "MAX WAIT", "% WAITING");
180
181 size_t total_execs = 0;
182 for(size_t i = 0; i < count; i++)
183 total_execs += stats[i].executions;
184
185 double total_time_sec = duration / (double)USEC_PER_SEC;
186
187 for(size_t i = 0; i < count; i++) {
188 double execs_per_sec = stats[i].executions / total_time_sec;
189 double avg_wait = stats[i].executions ? (double)stats[i].total_wait_time / stats[i].executions : 0;
190 double percent_waiting = stats[i].total_wait_time * 100.0 / duration;
191
192 fprintf(stderr, "%-8s %12zu %12.1f %12.1f %12"PRIu64" %12.1f%%\n",
193 priority_to_string(stats[i].priority),
194 stats[i].executions,
195 execs_per_sec,
196 avg_wait,
197 stats[i].max_wait_time,
198 percent_waiting);
199 }
200 }
201
202 static int unittest_stress(void) {
203 int errors = 0;
204 fprintf(stderr, "\nStress testing waiting queue...\n");
205
206 WAITQ wq = WAITQ_INITIALIZER;
207 enum {
208 WAITQ_STRESS_NUM_PRIORITIES = 4,
209 WAITQ_STRESS_TOTAL_THREADS = WAITQ_STRESS_NUM_PRIORITIES * THREADS_PER_PRIORITY
210 };
211
212 // Test both with and without sleep
213 for(int test = 0; test < 2; test++) {
214 bool with_sleep = (test == 1);
215 bool stop_flag = false;
216
217 fprintf(stderr, "\nRunning %ds stress test %s sleep:\n",
218 TEST_DURATION_SEC, with_sleep ? "with" : "without");
219
220 // Prepare thread stats and args
221 THREAD_STATS stats[WAITQ_STRESS_TOTAL_THREADS];
222 struct thread_args thread_args[WAITQ_STRESS_TOTAL_THREADS];
223 ND_THREAD *threads[WAITQ_STRESS_TOTAL_THREADS];
224
225 fprintf(stderr, "Starting %zu threads for %ds test %s sleep...\n",
226 (size_t)WAITQ_STRESS_TOTAL_THREADS,
227 TEST_DURATION_SEC,
228 with_sleep ? "with" : "without");
229
230 // Initialize stats and create threads
231 size_t thread_idx = 0;
232 for(int prio = WAITQ_PRIO_URGENT; prio <= WAITQ_PRIO_LOW; prio++) {
233 for(int t = 0; t < THREADS_PER_PRIORITY; t++) {
234 stats[thread_idx] = (THREAD_STATS){
235 .priority = prio,
236 .executions = 0,
237 .total_wait_time = 0,
238 .max_wait_time = 0
239 };
240 thread_args[thread_idx] = (struct thread_args){
241 .stats = &stats[thread_idx], // Pass pointer to stats
242 .wq = &wq,
243 .with_sleep = with_sleep,
244 .stop_flag = &stop_flag
245 };
246
247 char thread_name[32];
248 snprintf(thread_name, sizeof(thread_name), "STRESS%d-%d", prio, t);
249 threads[thread_idx] = nd_thread_create(
250 thread_name, NETDATA_THREAD_OPTION_DONT_LOG, stress_thread, &thread_args[thread_idx]);
251 thread_idx++;
252 }
253 }
254
255 // Let it run
256 time_t start = now_monotonic_sec();
257 fprintf(stderr, "Running...");
258 while(now_monotonic_sec() - start < TEST_DURATION_SEC) {
259 fprintf(stderr, ".");
260 sleep_usec(500000); // Print a dot every 0.5 seconds
261 }
262 fprintf(stderr, "\n");
263
264
265 fprintf(stderr, "Stopping threads...\n");
266 __atomic_store_n(&stop_flag, true, __ATOMIC_RELEASE);
267
268 // Wait for threads and collect stats
269 fprintf(stderr, "Waiting for %zu threads to finish...\n", (size_t)WAITQ_STRESS_TOTAL_THREADS);
270 for(size_t i = 0; i < WAITQ_STRESS_TOTAL_THREADS; i++)
271 nd_thread_join(threads[i]);
272
273 // Print stats
274 print_thread_stats(stats, WAITQ_STRESS_TOTAL_THREADS, TEST_DURATION_SEC * USEC_PER_SEC);
275 }
276
277 waitq_destroy(&wq);
278 return errors;
279 }
280
281 int unittest_waiting_queue(void) {
282 int errors = unittest_stress();
283 return errors;
284 }