spinlocks now timeout at 10 minutes, to reveal deadlocks (#20197)
Costa Tsaousis committed
Apr 28, 2025 at 14:53 UTC
1b0445021a26f704f941fc235f74c763bd1fad00
4 files changed
+52
src/libnetdata/locks/rw-spinlock.c
+14
@@ -35,6 +35,7 @@ ALWAYS_INLINE bool rw_spinlock_tryread_lock_with_trace(RW_SPINLOCK *rw_spinlock,
35
ALWAYS_INLINE void rw_spinlock_read_lock_with_trace(RW_SPINLOCK *rw_spinlock, const char *func) {
36
size_t spins = 0;
37
usec_t usec = 1;
38
+ usec_t deadlock_timestamp = 0;
39
40
while (true) {
41
// Optimistically increment reader count
@@ -52,6 +53,12 @@ ALWAYS_INLINE void rw_spinlock_read_lock_with_trace(RW_SPINLOCK *rw_spinlock, co
53
__atomic_sub_fetch(&rw_spinlock->counter, 1, __ATOMIC_RELEASE);
54
55
spins++;
56
+
57
+ // Check for deadlock every SPINS_BEFORE_DEADLOCK_CHECK iterations
58
+ if ((spins % SPINS_BEFORE_DEADLOCK_CHECK) == 0) {
59
+ spinlock_deadlock_detect(&deadlock_timestamp);
60
+ }
61
+
62
microsleep(usec);
63
usec = usec >= MAX_USEC ? MAX_USEC : usec * 2;
64
}
@@ -88,6 +95,7 @@ ALWAYS_INLINE bool rw_spinlock_trywrite_lock_with_trace(RW_SPINLOCK *rw_spinlock
95
ALWAYS_INLINE void rw_spinlock_write_lock_with_trace(RW_SPINLOCK *rw_spinlock, const char *func) {
96
size_t spins = 0;
97
usec_t usec = 1;
98
+ usec_t deadlock_timestamp = 0;
99
100
while (1) {
101
// Optimistically set writer bit
@@ -111,6 +119,12 @@ ALWAYS_INLINE void rw_spinlock_write_lock_with_trace(RW_SPINLOCK *rw_spinlock, c
119
}
120
121
spins++;
122
+
123
+ // Check for deadlock every SPINS_BEFORE_DEADLOCK_CHECK iterations
124
+ if ((spins % SPINS_BEFORE_DEADLOCK_CHECK) == 0) {
125
+ spinlock_deadlock_detect(&deadlock_timestamp);
126
+ }
127
+
128
microsleep(usec);
129
usec = usec >= MAX_USEC ? MAX_USEC : usec * 2;
130
}
src/libnetdata/locks/spinlock.c
+25
@@ -4,6 +4,24 @@
4
5
#define MAX_USEC 512 // Maximum backoff limit in microseconds
6
7
+// ----------------------------------------------------------------------------
8
+// Deadlock detection function
9
+void spinlock_deadlock_detect(usec_t *timestamp) {
10
+ if (!*timestamp) {
11
+ // First time checking - initialize the timestamp
12
+ *timestamp = now_monotonic_high_precision_usec();
13
+ return;
14
+ }
15
+
16
+ // Check if we've exceeded the timeout
17
+ usec_t now = now_monotonic_high_precision_usec();
18
+ if (now - *timestamp >= SPINLOCK_DEADLOCK_TIMEOUT_SEC * USEC_PER_SEC) {
19
+ // We've been spinning for too long - likely deadlock
20
+ fatal("DEADLOCK DETECTED: spinlock could not be acquired for %d seconds",
21
+ SPINLOCK_DEADLOCK_TIMEOUT_SEC);
22
+ }
23
+}
24
+
25
// ----------------------------------------------------------------------------
26
// spinlock implementation
27
// https://www.youtube.com/watch?v=rmGJc9PXpuE&t=41s
@@ -17,6 +35,7 @@ ALWAYS_INLINE void spinlock_init_with_trace(SPINLOCK *spinlock, const char *func
35
ALWAYS_INLINE void spinlock_lock_with_trace(SPINLOCK *spinlock, const char *func) {
36
size_t spins = 0;
37
usec_t usec = 1;
38
+ usec_t deadlock_timestamp = 0;
39
40
while (true) {
41
if (!__atomic_load_n(&spinlock->locked, __ATOMIC_RELAXED) &&
@@ -27,6 +46,12 @@ ALWAYS_INLINE void spinlock_lock_with_trace(SPINLOCK *spinlock, const char *func
46
47
// Backoff strategy with exponential growth
48
spins++;
49
+
50
+ // Check for deadlock every SPINS_BEFORE_DEADLOCK_CHECK iterations
51
+ if ((spins % SPINS_BEFORE_DEADLOCK_CHECK) == 0) {
52
+ spinlock_deadlock_detect(&deadlock_timestamp);
53
+ }
54
+
55
microsleep(usec);
56
usec = usec >= MAX_USEC ? MAX_USEC : usec * 2;
57
}
src/libnetdata/locks/spinlock.h
+6
@@ -35,6 +35,12 @@ typedef struct netdata_spinlock
35
#define SPINLOCK_INITIALIZER { .locked = false }
36
#endif
37
38
+#define SPINLOCK_DEADLOCK_TIMEOUT_SEC 600 // Number of seconds to wait before declaring a deadlock
39
+#define SPINS_BEFORE_DEADLOCK_CHECK 20000 // Check for deadlock every 20000 spins (approx. once per second)
40
+
41
+// Helper function to detect deadlocks
42
+void spinlock_deadlock_detect(usec_t *timestamp);
43
+
44
void spinlock_init_with_trace(SPINLOCK *spinlock, const char *func);
45
#define spinlock_init(spinlock) spinlock_init_with_trace(spinlock, __FUNCTION__)
46
src/libnetdata/locks/waitq.c
+7
@@ -74,6 +74,7 @@ ALWAYS_INLINE void waitq_acquire_with_trace(WAITQ *waitq, WAITQ_PRIORITY priorit
74
75
size_t spins = 0;
76
usec_t usec = 1;
77
+ usec_t deadlock_timestamp = 0;
78
79
while(true) {
80
while (write_our_priority(waitq, our_order)) {
@@ -88,6 +89,12 @@ ALWAYS_INLINE void waitq_acquire_with_trace(WAITQ *waitq, WAITQ_PRIORITY priorit
89
90
// Back off
91
spins++;
92
+
93
+ // Check for deadlock every SPINS_BEFORE_DEADLOCK_CHECK iterations
94
+ if ((spins % SPINS_BEFORE_DEADLOCK_CHECK) == 0) {
95
+ spinlock_deadlock_detect(&deadlock_timestamp);
96
+ }
97
+
98
microsleep(usec);
99
usec = usec >= MAX_USEC ? MAX_USEC : usec * 2;
100
}