Locks Improvements (#19314)
* benchmark locks; new rw-spinlock implementation; performance improvements on waitq * benchmark and stress r/w locks
Costa Tsaousis committed
Jan 3, 2025 at 07:46 UTC
b2c0b848280486a722108d4577493cdba06eabb4
10 files changed
+978
-102
CMakeLists.txt
+4
@@ -953,6 +953,10 @@ set(LIBNETDATA_FILES
953
src/libnetdata/locks/waitq.h
954
src/libnetdata/object-state/object-state.c
955
src/libnetdata/object-state/object-state.h
956
+ src/libnetdata/locks/benchmark.c
957
+ src/libnetdata/locks/benchmark.h
958
+ src/libnetdata/locks/benchmark-rw.c
959
+ src/libnetdata/locks/benchmark-rw.h
960
)
961
962
set(LIBH2O_FILES
src/daemon/main.c
+8
@@ -403,6 +403,14 @@ int netdata_main(int argc, char **argv) {
403
unittest_running = true;
404
return unittest_waiting_queue();
405
}
406
+ else if(strcmp(optarg, "lockstest") == 0) {
407
+ unittest_running = true;
408
+ return locks_stress_test();
409
+ }
410
+ else if(strcmp(optarg, "rwlockstest") == 0) {
411
+ unittest_running = true;
412
+ return rwlocks_stress_test();
413
+ }
414
else if(strcmp(optarg, "stringtest") == 0) {
415
unittest_running = true;
416
return string_unittest(10000);
src/libnetdata/libnetdata.h
+2
-1
@@ -16,7 +16,8 @@ extern "C" {
16
17
#include "atomics/atomics.h"
18
#include "libjudy/judy-malloc.h"
19
-
19
+#include "locks/benchmark.h"
20
+#include "locks/benchmark-rw.h"
21
#include "object-state/object-state.h"
22
#include "storage-point.h"
23
#include "paths/paths.h"
src/libnetdata/locks/benchmark-rw.c
new
+439
@@ -0,0 +1,439 @@
1
+// SPDX-License-Identifier: GPL-3.0-or-later
2
+
3
+#include "benchmark-rw.h"
4
+
5
+#define MAX_THREADS 64
6
+#define TEST_DURATION_SEC 1
7
+#define STOP_SIGNAL UINT64_MAX
8
+#define MAX_CONFIGS 10
9
+
10
+// Structure to store summary statistics
11
+typedef struct {
12
+ double ops_per_sec[2][MAX_CONFIGS]; // [lock_type][config_index], Total ops/sec
13
+ double reader_ops_per_sec[2][MAX_CONFIGS]; // [lock_type][config_index], Reader ops/sec
14
+ double writer_ops_per_sec[2][MAX_CONFIGS]; // [lock_type][config_index], Writer ops/sec
15
+ int readers[MAX_CONFIGS]; // Number of readers for each config
16
+ int writers[MAX_CONFIGS]; // Number of writers for each config
17
+ int config_count; // Number of configurations tested
18
+} summary_stats_t;
19
+
20
+typedef struct {
21
+ // Protected state to validate reader/writer mutual exclusion
22
+ volatile int readers; // Number of active readers
23
+ volatile int writers; // Number of active writers
24
+ volatile uint64_t violations; // Counter for reader/writer violations
25
+
26
+ // Protected counter for actual work
27
+ uint64_t counter;
28
+
29
+ // Statistics per thread
30
+ struct {
31
+ uint64_t operations; // Number of read/write operations
32
+ usec_t test_time; // Time spent in test
33
+ volatile int ready; // Thread completed flag
34
+ } stats[MAX_THREADS];
35
+
36
+ // Per-thread control
37
+ struct {
38
+ pthread_cond_t cond; // Thread start condition
39
+ pthread_mutex_t cond_mutex; // Mutex for condition
40
+ uint64_t run_flag; // Thread run control
41
+ } thread_controls[MAX_THREADS];
42
+} rwlock_control_t;
43
+
44
+typedef enum {
45
+ THREAD_READER,
46
+ THREAD_WRITER
47
+} thread_type_t;
48
+
49
+typedef struct {
50
+ int thread_id;
51
+ thread_type_t type;
52
+ void *lock; // Points to either pthread_rwlock_t or RW_SPINLOCK
53
+ bool is_spinlock; // true for RW_SPINLOCK, false for pthread_rwlock_t
54
+ rwlock_control_t *control;
55
+ ND_THREAD *thread;
56
+} thread_context_t;
57
+
58
+static inline void verify_no_violations(rwlock_control_t *control) {
59
+ if(__atomic_load_n(&control->violations, __ATOMIC_RELAXED) > 0) {
60
+ fprintf(stderr, "\nFATAL ERROR: Detected %"PRIu64" read/write violations!\n"
61
+ "This indicates readers and writers were concurrently inside the lock.\n",
62
+ control->violations);
63
+ exit(1);
64
+ }
65
+}
66
+
67
+static inline void check_access_safety(rwlock_control_t *control, thread_type_t type) {
68
+ if(type == THREAD_READER) {
69
+ // Reader entering critical section
70
+ __atomic_add_fetch(&control->readers, 1, __ATOMIC_RELAXED);
71
+
72
+ // Check if we have any writers - this would be a violation
73
+ if(__atomic_load_n(&control->writers, __ATOMIC_RELAXED) > 0) {
74
+ __atomic_add_fetch(&control->violations, 1, __ATOMIC_RELAXED);
75
+ }
76
+ }
77
+ else {
78
+ // Writer entering critical section
79
+ int writers = __atomic_add_fetch(&control->writers, 1, __ATOMIC_RELAXED);
80
+
81
+ // Check for other writers - violation!
82
+ if(writers > 1) {
83
+ __atomic_add_fetch(&control->violations, 1, __ATOMIC_RELAXED);
84
+ }
85
+
86
+ // Check if we have any readers - this would be a violation
87
+ if(__atomic_load_n(&control->readers, __ATOMIC_RELAXED) > 0) {
88
+ __atomic_add_fetch(&control->violations, 1, __ATOMIC_RELAXED);
89
+ }
90
+ }
91
+}
92
+
93
+static void release_access(rwlock_control_t *control, thread_type_t type) {
94
+ if(type == THREAD_READER) {
95
+ __atomic_sub_fetch(&control->readers, 1, __ATOMIC_RELAXED);
96
+ }
97
+ else {
98
+ __atomic_sub_fetch(&control->writers, 1, __ATOMIC_RELAXED);
99
+ }
100
+}
101
+
102
+static void wait_for_start(pthread_cond_t *cond, pthread_mutex_t *mutex, uint64_t *flag) {
103
+ pthread_mutex_lock(mutex);
104
+ while (*flag == 0)
105
+ pthread_cond_wait(cond, mutex);
106
+ pthread_mutex_unlock(mutex);
107
+}
108
+
109
+static void* benchmark_thread(void *arg) {
110
+ thread_context_t *ctx = (thread_context_t *)arg;
111
+ rwlock_control_t *control = ctx->control;
112
+
113
+ while(1) {
114
+ // Wait for start signal
115
+ wait_for_start(&control->thread_controls[ctx->thread_id].cond,
116
+ &control->thread_controls[ctx->thread_id].cond_mutex,
117
+ &control->thread_controls[ctx->thread_id].run_flag);
118
+
119
+ if (control->thread_controls[ctx->thread_id].run_flag == STOP_SIGNAL)
120
+ break;
121
+
122
+ usec_t start = now_monotonic_high_precision_usec();
123
+ uint64_t operations = 0;
124
+
125
+ while (control->thread_controls[ctx->thread_id].run_flag) {
126
+ if(ctx->is_spinlock) {
127
+ RW_SPINLOCK *spinlock = ctx->lock;
128
+ if(ctx->type == THREAD_READER) {
129
+ rw_spinlock_read_lock(spinlock);
130
+ check_access_safety(control, THREAD_READER);
131
+ control->counter++; // Just to do some work
132
+ release_access(control, THREAD_READER);
133
+ rw_spinlock_read_unlock(spinlock);
134
+ }
135
+ else {
136
+ rw_spinlock_write_lock(spinlock);
137
+ check_access_safety(control, THREAD_WRITER);
138
+ control->counter++;
139
+ release_access(control, THREAD_WRITER);
140
+ rw_spinlock_write_unlock(spinlock);
141
+ }
142
+ }
143
+ else {
144
+ pthread_rwlock_t *rwlock = ctx->lock;
145
+ if(ctx->type == THREAD_READER) {
146
+ pthread_rwlock_rdlock(rwlock);
147
+ check_access_safety(control, THREAD_READER);
148
+ control->counter++;
149
+ release_access(control, THREAD_READER);
150
+ pthread_rwlock_unlock(rwlock);
151
+ }
152
+ else {
153
+ pthread_rwlock_wrlock(rwlock);
154
+ check_access_safety(control, THREAD_WRITER);
155
+ control->counter++;
156
+ release_access(control, THREAD_WRITER);
157
+ pthread_rwlock_unlock(rwlock);
158
+ }
159
+ }
160
+ operations++;
161
+ }
162
+
163
+ // Store results
164
+ usec_t test_time = now_monotonic_high_precision_usec() - start;
165
+ __atomic_store_n(&control->stats[ctx->thread_id].test_time, test_time, __ATOMIC_RELEASE);
166
+ __atomic_store_n(&control->stats[ctx->thread_id].operations, operations, __ATOMIC_RELEASE);
167
+ __atomic_store_n(&control->stats[ctx->thread_id].ready, 1, __ATOMIC_RELEASE);
168
+ }
169
+
170
+ return NULL;
171
+}
172
+
173
+static void print_summary(const summary_stats_t *summary) {
174
+ fprintf(stderr, "\n=== Performance Summary (Million operations/sec) ===\n\n");
175
+ fprintf(stderr, "%-16s %-8s %-8s %-16s %-16s\n",
176
+ "Lock Type", "Readers", "Writers", "Reader Ops/s", "Writer Ops/s");
177
+ fprintf(stderr, "----------------------------------------------------------------------\n");
178
+
179
+ const char *lock_names[] = {"pthread_rwlock", "rw_spinlock"};
180
+
181
+ for (int config = 0; config < summary->config_count; config++) {
182
+ for (int lock_type = 0; lock_type < 2; lock_type++) {
183
+ // double total_ops = summary->ops_per_sec[lock_type][config];
184
+ int readers = summary->readers[config];
185
+ int writers = summary->writers[config];
186
+
187
+ // Get the actual reader and writer operations
188
+ double reader_ops = readers > 0 ? summary->reader_ops_per_sec[lock_type][config] : 0;
189
+ double writer_ops = writers > 0 ? summary->writer_ops_per_sec[lock_type][config] : 0;
190
+
191
+ fprintf(stderr, "%-16s %-8d %-8d %-16.2f %-16.2f\n",
192
+ lock_names[lock_type],
193
+ readers,
194
+ writers,
195
+ reader_ops / 1000000.0,
196
+ writer_ops / 1000000.0);
197
+ }
198
+ // Add a separator between configurations
199
+ if (config < summary->config_count - 1)
200
+ fprintf(stderr, "----------------------------------------------------------------------\n");
201
+ }
202
+ fprintf(stderr, "\n");
203
+}
204
+
205
+static void print_thread_stats(const char *test_name, int readers, int writers,
206
+ thread_context_t *contexts, rwlock_control_t *control,
207
+ summary_stats_t *summary, int config_idx, int lock_type) {
208
+ fprintf(stderr, "\n%-20s (readers: %d, writers: %d)\n", test_name, readers, writers);
209
+ fprintf(stderr, "%4s %8s %12s %12s %12s\n",
210
+ "THR", "TYPE", "OPS", "OPS/SEC", "TIME (ms)");
211
+
212
+ uint64_t total_ops = 0;
213
+ double total_ops_per_sec = 0;
214
+ double reader_ops_per_sec = 0;
215
+ double writer_ops_per_sec = 0;
216
+
217
+ for(int i = 0; i < readers + writers; i++) {
218
+ uint64_t ops = __atomic_load_n(&control->stats[i].operations, __ATOMIC_RELAXED);
219
+ usec_t time = __atomic_load_n(&control->stats[i].test_time, __ATOMIC_RELAXED);
220
+ double ops_per_sec = (double)ops * USEC_PER_SEC / time;
221
+
222
+ fprintf(stderr, "%4d %8s %12"PRIu64" %12.0f %12.2f\n",
223
+ i,
224
+ contexts[i].type == THREAD_READER ? "READER" : "WRITER",
225
+ ops,
226
+ ops_per_sec,
227
+ (double)time / 1000.0);
228
+
229
+ total_ops += ops;
230
+ total_ops_per_sec += ops_per_sec;
231
+
232
+ if (contexts[i].type == THREAD_READER) {
233
+ reader_ops_per_sec += ops_per_sec;
234
+ } else {
235
+ writer_ops_per_sec += ops_per_sec;
236
+ }
237
+ }
238
+
239
+ fprintf(stderr, "%4s %8s %12"PRIu64" %12.0f\n",
240
+ "TOT", "", total_ops, total_ops_per_sec);
241
+
242
+ // Store in summary
243
+ summary->ops_per_sec[lock_type][config_idx] = total_ops_per_sec;
244
+ summary->reader_ops_per_sec[lock_type][config_idx] = reader_ops_per_sec;
245
+ summary->writer_ops_per_sec[lock_type][config_idx] = writer_ops_per_sec;
246
+ summary->readers[config_idx] = readers;
247
+ summary->writers[config_idx] = writers;
248
+
249
+ verify_no_violations(control);
250
+}
251
+
252
+
253
+static void run_test(const char *name, int readers, int writers,
254
+ thread_context_t *contexts, rwlock_control_t *control,
255
+ summary_stats_t *summary, int config_idx, int lock_type) {
256
+ fprintf(stderr, "\nRunning test: %s with %d readers and %d writers...\n",
257
+ name, readers, writers);
258
+
259
+ // Reset all stats and control
260
+ memset(&control->stats, 0, sizeof(control->stats));
261
+ control->counter = 0;
262
+ control->readers = 0;
263
+ control->writers = 0;
264
+ control->violations = 0;
265
+
266
+ int total_threads = readers + writers;
267
+
268
+ // Signal threads to start
269
+ for(int i = 0; i < total_threads; i++) {
270
+ pthread_mutex_lock(&control->thread_controls[i].cond_mutex);
271
+ control->thread_controls[i].run_flag = 1;
272
+ pthread_cond_signal(&control->thread_controls[i].cond);
273
+ pthread_mutex_unlock(&control->thread_controls[i].cond_mutex);
274
+ }
275
+
276
+ // Wait for test duration
277
+ sleep_usec(TEST_DURATION_SEC * USEC_PER_SEC);
278
+
279
+ // Signal threads to stop
280
+ for(int i = 0; i < total_threads; i++) {
281
+ __atomic_store_n(&control->thread_controls[i].run_flag, 0, __ATOMIC_RELEASE);
282
+ }
283
+
284
+ // Wait for threads to report results
285
+ for(int i = 0; i < total_threads; i++) {
286
+ while(!__atomic_load_n(&control->stats[i].ready, __ATOMIC_ACQUIRE))
287
+ sleep_usec(10);
288
+ }
289
+
290
+ print_thread_stats(name, readers, writers, contexts, control, summary, config_idx, lock_type);
291
+}
292
+
293
+int rwlocks_stress_test(void) {
294
+ pthread_rwlock_t pthread_rwlock = PTHREAD_RWLOCK_INITIALIZER;
295
+ RW_SPINLOCK rw_spinlock = RW_SPINLOCK_INITIALIZER;
296
+ summary_stats_t summary = {0};
297
+
298
+ // Initialize control structures
299
+ rwlock_control_t pthread_control = { 0 };
300
+ rwlock_control_t spinlock_control = { 0 };
301
+
302
+ // Initialize per-thread controls for both locks
303
+ for(int i = 0; i < MAX_THREADS; i++) {
304
+ pthread_control.thread_controls[i].cond = (pthread_cond_t)PTHREAD_COND_INITIALIZER;
305
+ pthread_control.thread_controls[i].cond_mutex = (pthread_mutex_t)PTHREAD_MUTEX_INITIALIZER;
306
+ pthread_control.thread_controls[i].run_flag = 0;
307
+
308
+ spinlock_control.thread_controls[i].cond = (pthread_cond_t)PTHREAD_COND_INITIALIZER;
309
+ spinlock_control.thread_controls[i].cond_mutex = (pthread_mutex_t)PTHREAD_MUTEX_INITIALIZER;
310
+ spinlock_control.thread_controls[i].run_flag = 0;
311
+ }
312
+
313
+ // Create thread contexts
314
+ thread_context_t pthread_contexts[MAX_THREADS];
315
+ thread_context_t spinlock_contexts[MAX_THREADS];
316
+
317
+ fprintf(stderr, "\nStarting RW locks benchmark...\n");
318
+
319
+ // Test configurations: [readers, writers]
320
+ int configs[][2] = {
321
+ {1, 0}, // Single reader
322
+ {0, 1}, // Single writer
323
+ {1, 1}, // One reader + one writer
324
+ {2, 1}, // Two readers + one writer
325
+ {1, 2}, // One reader + two writers
326
+ {2, 2}, // Two readers + two writers
327
+ {4, 1}, // Four readers + one writer
328
+ {1, 4}, // One reader + four writers
329
+ {4, 4}, // Four readers + four writers
330
+ };
331
+
332
+ const int num_configs = sizeof(configs) / sizeof(configs[0]);
333
+ summary.config_count = num_configs;
334
+
335
+ // Create all threads
336
+ for(int i = 0; i < MAX_THREADS; i++) {
337
+ char thr_name[32];
338
+
339
+ // Initialize pthread contexts
340
+ pthread_contexts[i] = (thread_context_t){
341
+ .thread_id = i,
342
+ .type = i % 2 == 0 ? THREAD_READER :THREAD_WRITER,
343
+ .lock = &pthread_rwlock,
344
+ .is_spinlock = false,
345
+ .control = &pthread_control
346
+ };
347
+
348
+ snprintf(thr_name, sizeof(thr_name), "pthread_rw%d", i);
349
+ pthread_contexts[i].thread = nd_thread_create(
350
+ thr_name,
351
+ NETDATA_THREAD_OPTION_DONT_LOG | NETDATA_THREAD_OPTION_JOINABLE,
352
+ benchmark_thread,
353
+ &pthread_contexts[i]);
354
+
355
+ // Initialize spinlock contexts
356
+ spinlock_contexts[i] = (thread_context_t){
357
+ .thread_id = i,
358
+ .type = i % 2 == 0 ? THREAD_READER : THREAD_WRITER,
359
+ .lock = &rw_spinlock,
360
+ .is_spinlock = true,
361
+ .control = &spinlock_control
362
+ };
363
+
364
+ snprintf(thr_name, sizeof(thr_name), "spin_rw%d", i);
365
+ spinlock_contexts[i].thread = nd_thread_create(
366
+ thr_name,
367
+ NETDATA_THREAD_OPTION_DONT_LOG | NETDATA_THREAD_OPTION_JOINABLE,
368
+ benchmark_thread,
369
+ &spinlock_contexts[i]);
370
+ }
371
+
372
+ // Run all configurations
373
+ for(int i = 0; i < num_configs; i++) {
374
+ int readers = configs[i][0];
375
+ int writers = configs[i][1];
376
+
377
+ // Create all threads
378
+ int thread_idx = 0;
379
+
380
+ // First assign reader threads
381
+ for(int r = 0; r < readers; r++) {
382
+ pthread_contexts[thread_idx].type = THREAD_READER;
383
+ spinlock_contexts[thread_idx].type = THREAD_READER;
384
+ thread_idx++;
385
+ }
386
+
387
+ // Then assign writer threads
388
+ for(int w = 0; w < writers; w++) {
389
+ pthread_contexts[thread_idx].type = THREAD_WRITER;
390
+ spinlock_contexts[thread_idx].type = THREAD_WRITER;
391
+ thread_idx++;
392
+ }
393
+
394
+ char test_name[64];
395
+ snprintf(test_name, sizeof(test_name), "pthread_rwlock %dR/%dW", readers, writers);
396
+ run_test(test_name, readers, writers, pthread_contexts, &pthread_control, &summary, i, 0);
397
+
398
+ snprintf(test_name, sizeof(test_name), "rw_spinlock %dR/%dW", readers, writers);
399
+ run_test(test_name, readers, writers, spinlock_contexts, &spinlock_control, &summary, i, 1);
400
+ }
401
+
402
+ // Print the summary table
403
+ print_summary(&summary);
404
+
405
+ // Stop all threads
406
+ fprintf(stderr, "\nStopping threads...\n");
407
+ for(int i = 0; i < MAX_THREADS; i++) {
408
+ // Signal pthread threads
409
+ pthread_mutex_lock(&pthread_control.thread_controls[i].cond_mutex);
410
+ pthread_control.thread_controls[i].run_flag = STOP_SIGNAL;
411
+ pthread_cond_signal(&pthread_control.thread_controls[i].cond);
412
+ pthread_mutex_unlock(&pthread_control.thread_controls[i].cond_mutex);
413
+
414
+ // Signal spinlock threads
415
+ pthread_mutex_lock(&spinlock_control.thread_controls[i].cond_mutex);
416
+ spinlock_control.thread_controls[i].run_flag = STOP_SIGNAL;
417
+ pthread_cond_signal(&spinlock_control.thread_controls[i].cond);
418
+ pthread_mutex_unlock(&spinlock_control.thread_controls[i].cond_mutex);
419
+ }
420
+
421
+ // Join all threads
422
+ fprintf(stderr, "\nWaiting for threads to exit...\n");
423
+ for(int i = 0; i < MAX_THREADS; i++) {
424
+ nd_thread_join(pthread_contexts[i].thread);
425
+ nd_thread_join(spinlock_contexts[i].thread);
426
+ }
427
+
428
+ // Cleanup condition variables and mutexes
429
+ for(int i = 0; i < MAX_THREADS; i++) {
430
+ pthread_cond_destroy(&pthread_control.thread_controls[i].cond);
431
+ pthread_mutex_destroy(&pthread_control.thread_controls[i].cond_mutex);
432
+ pthread_cond_destroy(&spinlock_control.thread_controls[i].cond);
433
+ pthread_mutex_destroy(&spinlock_control.thread_controls[i].cond_mutex);
434
+ }
435
+
436
+ pthread_rwlock_destroy(&pthread_rwlock);
437
+
438
+ return 0;
439
+}
src/libnetdata/locks/benchmark-rw.h
new
+10
@@ -0,0 +1,10 @@
1
+// SPDX-License-Identifier: GPL-3.0-or-later
2
+
3
+#ifndef NETDATA_BENCHMARK_RW_H
4
+#define NETDATA_BENCHMARK_RW_H
5
+
6
+#include "../libnetdata.h"
7
+
8
+int rwlocks_stress_test(void);
9
+
10
+#endif //NETDATA_BENCHMARK_RW_H
src/libnetdata/locks/benchmark.c
new
+434
@@ -0,0 +1,434 @@
1
+// SPDX-License-Identifier: GPL-3.0-or-later
2
+
3
+#include "benchmark.h"
4
+
5
+#define MAX_THREADS 64
6
+#define TEST_DURATION_SEC 1
7
+#define STOP_SIGNAL UINT64_MAX
8
+#define NUM_LOCK_TYPES 5
9
+
10
+// Structure to store summary stats
11
+typedef struct {
12
+ double locks_per_sec[NUM_LOCK_TYPES][7]; // [lock_type][thread_count_index]
13
+} summary_stats_t;
14
+
15
+typedef struct {
16
+ uint64_t locks;
17
+ usec_t test_time;
18
+ volatile int ready;
19
+} thread_stats_t;
20
+
21
+typedef struct {
22
+ pthread_cond_t cond; // Individual condition for each thread
23
+ pthread_mutex_t cond_mutex; // Individual mutex for each thread
24
+ uint64_t run_flag; // Individual run flag for each thread
25
+} thread_control_t;
26
+
27
+typedef struct {
28
+ uint64_t protected_counter;
29
+ thread_stats_t stats[MAX_THREADS];
30
+ thread_control_t thread_controls[MAX_THREADS]; // Array of per-thread controls
31
+} lock_control_t;
32
+
33
+typedef enum {
34
+ LOCK_MUTEX,
35
+ LOCK_RWLOCK,
36
+ LOCK_SPINLOCK,
37
+ LOCK_RW_SPINLOCK,
38
+ LOCK_WAITQ
39
+} lock_type_t;
40
+
41
+typedef struct {
42
+ int thread_id;
43
+ lock_type_t type;
44
+ WAITQ_PRIORITY priority; // For waitq only
45
+ lock_control_t *control;
46
+ void *lock; // Points to the actual lock
47
+ ND_THREAD *thread;
48
+} thread_context_t;
49
+
50
+static const char *lock_names[] = {
51
+ "Mutex",
52
+ "RWLock",
53
+ "Spinlock",
54
+ "RW Spinlock",
55
+ "WaitQ"
56
+};
57
+
58
+static const char *priority_to_string(WAITQ_PRIORITY p) {
59
+ switch(p) {
60
+ case WAITQ_PRIO_URGENT: return "URGENT";
61
+ case WAITQ_PRIO_HIGH: return "HIGH";
62
+ case WAITQ_PRIO_NORMAL: return "NORMAL";
63
+ case WAITQ_PRIO_LOW: return "LOW";
64
+ default: return "UNKNOWN";
65
+ }
66
+}
67
+
68
+static void print_summary(const summary_stats_t *summary) {
69
+ fprintf(stderr, "\n=== Performance Summary (Million locks/sec) ===\n\n");
70
+ fprintf(stderr, "%-12s %8s %8s %8s %8s %8s %8s %8s\n",
71
+ "Lock Type", "1", "2", "4", "8", "16", "32", "64");
72
+ fprintf(stderr, "------------------------------------------------------------------------------\n");
73
+
74
+ for(int type = 0; type < NUM_LOCK_TYPES; type++) {
75
+ fprintf(stderr, "%-12s", lock_names[type]);
76
+ for(int i = 0; i < 7; i++) { // 6 different thread counts
77
+ fprintf(stderr, " %8.2f", summary->locks_per_sec[type][i] / 1000000.0);
78
+ }
79
+ fprintf(stderr, "\n");
80
+ }
81
+ fprintf(stderr, "\n");
82
+}
83
+
84
+static void wait_for_signal(pthread_cond_t *cond, pthread_mutex_t *mutex, uint64_t *flag) {
85
+ pthread_mutex_lock(mutex);
86
+ while (*flag == 0)
87
+ pthread_cond_wait(cond, mutex);
88
+ pthread_mutex_unlock(mutex);
89
+}
90
+
91
+static void* benchmark_thread(void *arg) {
92
+ thread_context_t *ctx = (thread_context_t *)arg;
93
+ thread_stats_t *stats = &ctx->control->stats[ctx->thread_id];
94
+ thread_control_t *thread_control = &ctx->control->thread_controls[ctx->thread_id];
95
+
96
+ while(1) {
97
+ wait_for_signal(&thread_control->cond, &thread_control->cond_mutex, &thread_control->run_flag);
98
+
99
+ if (thread_control->run_flag == STOP_SIGNAL)
100
+ break;
101
+
102
+ usec_t start = now_monotonic_high_precision_usec();
103
+ uint64_t local_counter = 0;
104
+
105
+ switch(ctx->type) {
106
+ case LOCK_MUTEX: {
107
+ pthread_mutex_t *mutex = ctx->lock;
108
+ while (thread_control->run_flag) {
109
+ pthread_mutex_lock(mutex);
110
+ ctx->control->protected_counter++;
111
+ pthread_mutex_unlock(mutex);
112
+ local_counter++;
113
+ }
114
+ break;
115
+ }
116
+
117
+ case LOCK_RWLOCK: {
118
+ pthread_rwlock_t *rwlock = ctx->lock;
119
+ while (thread_control->run_flag) {
120
+ pthread_rwlock_wrlock(rwlock);
121
+ ctx->control->protected_counter++;
122
+ pthread_rwlock_unlock(rwlock);
123
+ local_counter++;
124
+ }
125
+ break;
126
+ }
127
+
128
+ case LOCK_SPINLOCK: {
129
+ SPINLOCK *spinlock = ctx->lock;
130
+ while (thread_control->run_flag) {
131
+ spinlock_lock(spinlock);
132
+ ctx->control->protected_counter++;
133
+ spinlock_unlock(spinlock);
134
+ local_counter++;
135
+ }
136
+ break;
137
+ }
138
+
139
+ case LOCK_RW_SPINLOCK: {
140
+ RW_SPINLOCK *rw_spinlock = ctx->lock;
141
+ while (thread_control->run_flag) {
142
+ rw_spinlock_write_lock(rw_spinlock);
143
+ ctx->control->protected_counter++;
144
+ rw_spinlock_write_unlock(rw_spinlock);
145
+ local_counter++;
146
+ }
147
+ break;
148
+ }
149
+
150
+ case LOCK_WAITQ: {
151
+ WAITQ *waitq = ctx->lock;
152
+ WAITQ_PRIORITY priority = ctx->priority;
153
+ while (thread_control->run_flag) {
154
+ waitq_acquire(waitq, priority);
155
+ ctx->control->protected_counter++;
156
+ waitq_release(waitq);
157
+ local_counter++;
158
+ }
159
+ break;
160
+ }
161
+ }
162
+
163
+ // Store results atomically
164
+ usec_t test_time = now_monotonic_high_precision_usec() - start;
165
+ __atomic_store_n(&stats->test_time, test_time, __ATOMIC_RELEASE);
166
+ __atomic_store_n(&stats->locks, local_counter, __ATOMIC_RELEASE);
167
+ __atomic_store_n(&stats->ready, 1, __ATOMIC_RELEASE);
168
+ }
169
+
170
+ return NULL;
171
+}
172
+
173
+static void print_thread_stats(const char *test_name, int threads, thread_context_t *contexts,
174
+ thread_stats_t *stats, uint64_t protected_counter,
175
+ summary_stats_t *summary, int thread_count_idx) {
176
+ fprintf(stderr, "\n%-20s (threads: %d)\n", test_name, threads);
177
+ if (strcmp(test_name, "WaitQ") == 0) {
178
+ fprintf(stderr, "%4s %8s %12s %12s %12s\n",
179
+ "THR", "PRIO", "LOCKS", "LOCKS/SEC", "TIME (ms)");
180
+ }
181
+ else {
182
+ fprintf(stderr, "%4s %12s %12s %12s\n",
183
+ "THR", "LOCKS", "LOCKS/SEC", "TIME (ms)");
184
+ }
185
+
186
+ uint64_t total_locks = 0;
187
+ double total_locks_per_sec = 0;
188
+
189
+ for(int i = 0; i < threads; i++) {
190
+ uint64_t locks = __atomic_load_n(&stats[i].locks, __ATOMIC_ACQUIRE);
191
+ usec_t time = __atomic_load_n(&stats[i].test_time, __ATOMIC_ACQUIRE);
192
+ double locks_per_sec = (double)locks * USEC_PER_SEC / time;
193
+ total_locks_per_sec += locks_per_sec;
194
+
195
+ if (strcmp(test_name, "WaitQ") == 0) {
196
+ fprintf(stderr, "%4d %8s %12"PRIu64" %12.0f %12.2f\n",
197
+ i,
198
+ priority_to_string(contexts[i].priority),
199
+ locks,
200
+ locks_per_sec,
201
+ (double)time / 1000.0);
202
+ }
203
+ else {
204
+ fprintf(stderr, "%4d %12"PRIu64" %12.0f %12.2f\n",
205
+ i, locks, locks_per_sec,
206
+ (double)time / 1000.0);
207
+ }
208
+
209
+ total_locks += locks;
210
+ }
211
+
212
+ if(total_locks != protected_counter) {
213
+ fprintf(stderr, "\nERROR: Counter mismatch!\n");
214
+ fprintf(stderr, "Sum of thread counters: %"PRIu64"\n", total_locks);
215
+ fprintf(stderr, "Protected counter: %"PRIu64"\n", protected_counter);
216
+ fprintf(stderr, "Difference: %"PRIu64"\n",
217
+ total_locks > protected_counter ?
218
+ total_locks - protected_counter :
219
+ protected_counter - total_locks);
220
+
221
+ fflush(stderr);
222
+ _exit(1);
223
+ }
224
+
225
+ fprintf(stderr, "%4s %12"PRIu64"\n", "TOT", total_locks);
226
+
227
+ // Store in summary for the final table
228
+ summary->locks_per_sec[contexts[0].type][thread_count_idx] = total_locks_per_sec;
229
+}
230
+
231
+static void run_test(const char *name, int threads, thread_context_t *contexts,
232
+ lock_control_t *control, summary_stats_t *summary) {
233
+ fprintf(stderr, "\nRunning test: %s with %d threads...\n", name, threads);
234
+
235
+ // Reset stats and counter
236
+ for(int i = 0; i < threads; i++) {
237
+ __atomic_store_n(&control->stats[i].locks, 0, __ATOMIC_RELEASE);
238
+ __atomic_store_n(&control->stats[i].test_time, 0, __ATOMIC_RELEASE);
239
+ __atomic_store_n(&control->stats[i].ready, 0, __ATOMIC_RELEASE);
240
+ }
241
+ control->protected_counter = 0;
242
+
243
+ // Signal only the threads we need for this test
244
+ for(int i = 0; i < threads; i++) {
245
+ thread_control_t *thread_control = &control->thread_controls[i];
246
+ pthread_mutex_lock(&thread_control->cond_mutex);
247
+ thread_control->run_flag = 1;
248
+ pthread_cond_signal(&thread_control->cond);
249
+ pthread_mutex_unlock(&thread_control->cond_mutex);
250
+ }
251
+
252
+ // Wait for test duration
253
+ sleep_usec(TEST_DURATION_SEC * USEC_PER_SEC);
254
+
255
+ // Signal threads to stop
256
+ for(int i = 0; i < threads; i++) {
257
+ thread_control_t *thread_control = &control->thread_controls[i];
258
+ __atomic_store_n(&thread_control->run_flag, 0, __ATOMIC_RELEASE);
259
+ }
260
+
261
+ // Wait for threads to report results
262
+ for(int i = 0; i < threads; i++) {
263
+ while(!__atomic_load_n(&control->stats[i].ready, __ATOMIC_ACQUIRE))
264
+ sleep_usec(10);
265
+ }
266
+
267
+ // Get thread count index for summary
268
+ int thread_count_idx;
269
+ switch(threads) {
270
+ case 1: thread_count_idx = 0; break;
271
+ case 2: thread_count_idx = 1; break;
272
+ case 4: thread_count_idx = 2; break;
273
+ case 8: thread_count_idx = 3; break;
274
+ case 16: thread_count_idx = 4; break;
275
+ case 32: thread_count_idx = 5; break;
276
+ case 64: thread_count_idx = 6; break;
277
+ default: thread_count_idx = 0; break;
278
+ }
279
+
280
+ print_thread_stats(name, threads, contexts, control->stats, control->protected_counter,
281
+ summary, thread_count_idx);
282
+}
283
+
284
+static void set_waitq_priorities(int thread_count, thread_context_t *contexts) {
285
+ switch(thread_count) {
286
+ case 1:
287
+ contexts[0].priority = WAITQ_PRIO_URGENT;
288
+ break;
289
+
290
+ case 2:
291
+ contexts[0].priority = WAITQ_PRIO_URGENT;
292
+ contexts[1].priority = WAITQ_PRIO_HIGH;
293
+ break;
294
+
295
+ case 4:
296
+ contexts[0].priority = WAITQ_PRIO_URGENT;
297
+ contexts[1].priority = WAITQ_PRIO_HIGH;
298
+ contexts[2].priority = WAITQ_PRIO_NORMAL;
299
+ contexts[3].priority = WAITQ_PRIO_LOW;
300
+ break;
301
+
302
+ default: { // 8, 16, 32
303
+ int threads_per_priority = thread_count / 4;
304
+ int remainder = thread_count % 4;
305
+ int thread_idx = 0;
306
+
307
+ for (int prio = WAITQ_PRIO_URGENT; prio <= WAITQ_PRIO_LOW; prio++) {
308
+ int count = threads_per_priority + (remainder > 0 ? 1 : 0);
309
+ remainder--;
310
+
311
+ for (int i = 0; i < count && thread_idx < thread_count; i++) {
312
+ contexts[thread_idx++].priority = prio;
313
+ }
314
+ }
315
+ break;
316
+ }
317
+ }
318
+}
319
+
320
+int locks_stress_test(void) {
321
+ summary_stats_t summary = {0};
322
+
323
+ // Initialize actual locks
324
+ pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
325
+ pthread_rwlock_t rwlock = PTHREAD_RWLOCK_INITIALIZER;
326
+ SPINLOCK spinlock = SPINLOCK_INITIALIZER;
327
+ RW_SPINLOCK rw_spinlock = RW_SPINLOCK_INITIALIZER;
328
+ WAITQ waitq = WAITQ_INITIALIZER;
329
+
330
+ void *locks[] = {
331
+ &mutex,
332
+ &rwlock,
333
+ &spinlock,
334
+ &rw_spinlock,
335
+ &waitq
336
+ };
337
+
338
+ // Initialize control structures
339
+ lock_control_t controls[NUM_LOCK_TYPES] = { 0 };
340
+ for(int i = 0; i < NUM_LOCK_TYPES; i++) {
341
+ // Initialize per-thread condition variables and mutexes
342
+ for(int j = 0; j < MAX_THREADS; j++) {
343
+ controls[i].thread_controls[j].cond = (pthread_cond_t)PTHREAD_COND_INITIALIZER;
344
+ controls[i].thread_controls[j].cond_mutex = (pthread_mutex_t)PTHREAD_MUTEX_INITIALIZER;
345
+ controls[i].thread_controls[j].run_flag = 0;
346
+ }
347
+ }
348
+
349
+ // Initialize thread arrays
350
+ thread_context_t *threads[NUM_LOCK_TYPES];
351
+ for(int i = 0; i < NUM_LOCK_TYPES; i++) {
352
+ threads[i] = calloc(MAX_THREADS, sizeof(thread_context_t));
353
+ if(!threads[i]) {
354
+ fprintf(stderr, "Failed to allocate memory for threads\n");
355
+ return 1;
356
+ }
357
+
358
+ // Initialize thread contexts
359
+ for(int j = 0; j < MAX_THREADS; j++) {
360
+ threads[i][j] = (thread_context_t){
361
+ .thread_id = j,
362
+ .type = i,
363
+ .control = &controls[i],
364
+ .lock = locks[i]
365
+ };
366
+ }
367
+ }
368
+
369
+ // Create all threads
370
+ fprintf(stderr, "Creating threads...\n");
371
+ for(int type = 0; type < NUM_LOCK_TYPES; type++) {
372
+ for(int i = 0; i < MAX_THREADS; i++) {
373
+ char thr_name[32];
374
+ snprintf(thr_name, sizeof(thr_name), "%s%d", lock_names[type], i);
375
+ threads[type][i].thread = nd_thread_create(
376
+ thr_name,
377
+ NETDATA_THREAD_OPTION_DONT_LOG | NETDATA_THREAD_OPTION_JOINABLE,
378
+ benchmark_thread,
379
+ &threads[type][i]);
380
+ }
381
+ }
382
+
383
+ // Run tests with different thread counts
384
+ int thread_counts[] = {1, 2, 4, 8, 16, 32, 64};
385
+
386
+ // Warm up the CPU
387
+ sleep_usec(100000);
388
+
389
+ for(size_t i = 0; i < sizeof(thread_counts)/sizeof(thread_counts[0]); i++) {
390
+ int count = thread_counts[i];
391
+
392
+ // Set waitq priorities for this test
393
+ set_waitq_priorities(count, threads[LOCK_WAITQ]);
394
+
395
+ // Run test for each lock type
396
+ for(int type = 0; type < NUM_LOCK_TYPES; type++) {
397
+ run_test(lock_names[type], count, threads[type], &controls[type], &summary);
398
+ }
399
+ }
400
+
401
+ // Print the summary table
402
+ print_summary(&summary);
403
+
404
+ // Signal all threads to exit
405
+ fprintf(stderr, "\nStopping threads...\n");
406
+ for(int type = 0; type < NUM_LOCK_TYPES; type++) {
407
+ for(int i = 0; i < MAX_THREADS; i++) {
408
+ thread_control_t *thread_control = &controls[type].thread_controls[i];
409
+ pthread_mutex_lock(&thread_control->cond_mutex);
410
+ thread_control->run_flag = STOP_SIGNAL;
411
+ pthread_cond_signal(&thread_control->cond);
412
+ pthread_mutex_unlock(&thread_control->cond_mutex);
413
+ }
414
+ }
415
+
416
+ // Join all threads
417
+ fprintf(stderr, "\nWaiting for threads to exit...\n");
418
+ for(int type = 0; type < NUM_LOCK_TYPES; type++) {
419
+ for(int i = 0; i < MAX_THREADS; i++) {
420
+ nd_thread_join(threads[type][i].thread);
421
+ }
422
+ }
423
+
424
+ // Cleanup condition variables and mutexes
425
+ for(int type = 0; type < NUM_LOCK_TYPES; type++) {
426
+ for(int i = 0; i < MAX_THREADS; i++) {
427
+ pthread_cond_destroy(&controls[type].thread_controls[i].cond);
428
+ pthread_mutex_destroy(&controls[type].thread_controls[i].cond_mutex);
429
+ }
430
+ free(threads[type]);
431
+ }
432
+
433
+ return 0;
434
+}
\ No newline at end of file
src/libnetdata/locks/benchmark.h
new
+10
@@ -0,0 +1,10 @@
1
+// SPDX-License-Identifier: GPL-3.0-or-later
2
+
3
+#ifndef NETDATA_BENCHMARK_H
4
+#define NETDATA_BENCHMARK_H
5
+
6
+#include "../libnetdata.h"
7
+
8
+int locks_stress_test(void);
9
+
10
+#endif //NETDATA_BENCHMARK_H
src/libnetdata/locks/rw-spinlock.c
+69
-99
@@ -2,7 +2,10 @@
2
3
#include "libnetdata/libnetdata.h"
4
5
-#define WRITER_LOCKED (-65536)
5
+#define MAX_USEC 512 // Maximum backoff limit in microseconds
6
+
7
+#define WRITER_BIT (1U << 31)
8
+#define READER_MASK (~WRITER_BIT)
9
10
// ----------------------------------------------------------------------------
11
// rw_spinlock implementation
@@ -15,27 +18,13 @@ void rw_spinlock_init_with_trace(RW_SPINLOCK *rw_spinlock, const char *func __ma
18
bool rw_spinlock_tryread_lock_with_trace(RW_SPINLOCK *rw_spinlock, const char *func) {
19
size_t spins = 0;
20
18
- REFCOUNT expected = rw_spinlock->counter;
19
- while (true) {
20
- if(expected == WRITER_LOCKED)
21
- // writer is active
22
- return false;
23
-
24
- if(expected < 0)
25
- fatal("RW_SPINLOCK: refcount found negative, on %s(), called from %s()", __FUNCTION__, func);
26
-
27
- // increment reader count
28
- if (__atomic_compare_exchange_n(
29
- &rw_spinlock->counter,
30
- &expected,
31
- expected + 1,
32
- false, // Strong CAS
33
- __ATOMIC_ACQUIRE, // Success memory order
34
- __ATOMIC_RELAXED // Failure memory order
35
- ))
36
- break;
21
+ uint32_t val = __atomic_add_fetch(&rw_spinlock->counter, 1, __ATOMIC_ACQUIRE);
22
38
- spins++;
23
+ // Check if a writer holds the lock
24
+ if (val & WRITER_BIT) {
25
+ // Undo our increment and fail
26
+ __atomic_sub_fetch(&rw_spinlock->counter, 1, __ATOMIC_RELEASE);
27
+ return false;
28
}
29
30
worker_spinlock_contention(func, spins);
@@ -45,109 +34,90 @@ bool rw_spinlock_tryread_lock_with_trace(RW_SPINLOCK *rw_spinlock, const char *f
34
35
void rw_spinlock_read_lock_with_trace(RW_SPINLOCK *rw_spinlock, const char *func) {
36
size_t spins = 0;
48
-
49
- REFCOUNT expected = rw_spinlock->counter;
50
-
51
- // we should not increase it if it is negative (a writer holds the lock)
52
- if(expected == WRITER_LOCKED) expected = 0;
37
+ usec_t usec = 1;
38
39
while (true) {
55
- if(expected < 0)
56
- fatal("RW_SPINLOCK: refcount found negative, on %s(), called from %s()", __FUNCTION__, func);
57
-
58
- // Attempt to increment reader count
59
- if (__atomic_compare_exchange_n(
60
- &rw_spinlock->counter,
61
- &expected,
62
- expected + 1,
63
- false, // Strong CAS
64
- __ATOMIC_ACQUIRE, // Success memory order
65
- __ATOMIC_RELAXED // Failure memory order
66
- ))
67
- break;
68
-
69
- spins++;
70
-
71
- if (expected == WRITER_LOCKED) {
72
- // writer is active
40
+ // Optimistically increment reader count
41
+ uint32_t val = __atomic_add_fetch(&rw_spinlock->counter, 1, __ATOMIC_ACQUIRE);
42
+
43
+ // Check if a writer holds the lock
44
+ if (!(val & WRITER_BIT)) {
45
+ // no writer, we are in
46
+ worker_spinlock_contention(func, spins);
47
+ nd_thread_rwspinlock_read_locked();
48
+ return;
49
+ }
50
74
- // we should not increase it if it is negative (a writer holds the lock)
75
- expected = 0;
51
+ // Undo our increment and retry
52
+ __atomic_sub_fetch(&rw_spinlock->counter, 1, __ATOMIC_RELEASE);
53
77
- // wait a bit before retrying
78
- tinysleep();
79
- yield_the_processor();
80
- }
54
+ spins++;
55
+ microsleep(usec);
56
+ usec = usec >= MAX_USEC ? MAX_USEC : usec * 2;
57
}
82
-
83
- worker_spinlock_contention(func, spins);
84
- nd_thread_rwspinlock_read_locked();
58
}
59
60
void rw_spinlock_read_unlock_with_trace(RW_SPINLOCK *rw_spinlock, const char *func __maybe_unused) {
88
- REFCOUNT x = __atomic_sub_fetch(&rw_spinlock->counter, 1, __ATOMIC_RELEASE);
89
- if (x < 0)
90
- fatal("RW_SPINLOCK: readers is negative %d, on %s called from %s()", x, __FUNCTION__, func);
91
-
61
+ __atomic_sub_fetch(&rw_spinlock->counter, 1, __ATOMIC_RELEASE);
62
nd_thread_rwspinlock_read_unlocked();
63
}
64
65
bool rw_spinlock_trywrite_lock_with_trace(RW_SPINLOCK *rw_spinlock, const char *func) {
96
- REFCOUNT expected = 0;
97
-
98
- // Attempt to acquire writer lock when no readers or writers are active
99
- if (!__atomic_compare_exchange_n(
100
- &rw_spinlock->counter,
101
- &expected,
102
- WRITER_LOCKED,
103
- false, // Strong CAS
104
- __ATOMIC_ACQUIRE, // Success memory order
105
- __ATOMIC_RELAXED // Failure memory order
106
- )) {
107
- return false;
66
+ // Optimistically set writer bit
67
+ uint32_t old = __atomic_fetch_or(&rw_spinlock->counter, WRITER_BIT, __ATOMIC_ACQUIRE);
68
+
69
+ if(old == 0) {
70
+ rw_spinlock->writer = gettid_cached();
71
+ worker_spinlock_contention(func, 0);
72
+ nd_thread_rwspinlock_write_locked();
73
+ return true;
74
}
75
110
- __atomic_store_n(&rw_spinlock->writer, gettid_cached(), __ATOMIC_RELAXED);
111
- worker_spinlock_contention(func, 0);
112
- nd_thread_rwspinlock_write_locked();
113
- return true;
76
+ // Check if we were the only one
77
+ if (old & WRITER_BIT) {
78
+ // there is a writer inside (keep the writer bit there)
79
+ }
80
+ else /* if ((old & READER_MASK) != 0) */ {
81
+ // there are readers inside, remove the writer bit we added
82
+ __atomic_and_fetch(&rw_spinlock->counter, ~WRITER_BIT, __ATOMIC_RELEASE);
83
+ }
84
+
85
+ return false;
86
}
87
88
void rw_spinlock_write_lock_with_trace(RW_SPINLOCK *rw_spinlock, const char *func) {
89
size_t spins = 0;
90
+ usec_t usec = 1;
91
+
92
+ while (1) {
93
+ // Optimistically set writer bit
94
+ uint32_t old = __atomic_fetch_or(&rw_spinlock->counter, WRITER_BIT, __ATOMIC_ACQUIRE);
95
+
96
+ // Check if we were the only one
97
+ if (old == 0) {
98
+ rw_spinlock->writer = gettid_cached();
99
+ worker_spinlock_contention(func, spins);
100
+ nd_thread_rwspinlock_write_locked();
101
+ return;
102
+ }
103
119
- while (true) {
120
- REFCOUNT expected = 0;
121
-
122
- // Attempt to acquire writer lock when no readers or writers are active
123
- if (__atomic_compare_exchange_n(
124
- &rw_spinlock->counter,
125
- &expected,
126
- WRITER_LOCKED,
127
- false, // Strong CAS
128
- __ATOMIC_ACQUIRE, // Success memory order
129
- __ATOMIC_RELAXED // Failure memory order
130
- )) {
131
- break;
104
+ // Check if we were the only one
105
+ if (old & WRITER_BIT) {
106
+ // there is a writer inside (keep the writer bit there)
107
+ }
108
+ else /* if ((old & READER_MASK) != 0) */ {
109
+ // there are readers inside, remove the writer bit we added
110
+ __atomic_and_fetch(&rw_spinlock->counter, ~WRITER_BIT, __ATOMIC_RELEASE);
111
}
112
113
spins++;
135
- tinysleep();
114
+ microsleep(usec);
115
+ usec = usec >= MAX_USEC ? MAX_USEC : usec * 2;
116
}
137
-
138
- __atomic_store_n(&rw_spinlock->writer, gettid_cached(), __ATOMIC_RELAXED);
139
- worker_spinlock_contention(func, spins);
140
- nd_thread_rwspinlock_write_locked();
117
}
118
119
void rw_spinlock_write_unlock_with_trace(RW_SPINLOCK *rw_spinlock, const char *func __maybe_unused) {
144
-#ifdef NETDATA_INTERNAL_CHECKS
145
- int32_t x = __atomic_load_n(&rw_spinlock->counter, __ATOMIC_RELAXED);
146
- if (x != WRITER_LOCKED)
147
- fatal("RW_SPINLOCK: writer unlock encountered unexpected state: %d, on %s() called from %s()", x, __FUNCTION__, func);
148
-#endif
149
-
150
- __atomic_store_n(&rw_spinlock->writer, 0, __ATOMIC_RELAXED);
151
- __atomic_store_n(&rw_spinlock->counter, 0, __ATOMIC_RELEASE); // Release writer lock
120
+ rw_spinlock->writer = 0;
121
+ __atomic_and_fetch(&rw_spinlock->counter, ~WRITER_BIT, __ATOMIC_RELEASE);
122
nd_thread_rwspinlock_write_unlocked();
123
}
src/libnetdata/locks/rw-spinlock.h
+1
-1
@@ -8,7 +8,7 @@
8
9
typedef struct netdata_rw_spinlock {
10
pid_t writer;
11
- REFCOUNT counter; // positive is readers, negative is a writer
11
+ uint32_t counter;
12
} RW_SPINLOCK;
13
14
#define RW_SPINLOCK_INITIALIZER { .counter = 0, .writer = 0, }
src/libnetdata/locks/waitq.c
+1
-1
@@ -83,7 +83,7 @@ void waitq_acquire_with_trace(WAITQ *waitq, WAITQ_PRIORITY priority, const char
83
worker_spinlock_contention(func, spins);
84
return;
85
}
86
- tinysleep();
86
+ yield_the_processor();
87
}
88
89
// Back off