1#include "sync/tests/test_internal.h"
2#include <mem/nofail.h>
3
4struct chaos_state {
5 struct thread_wait_header wait;
6 struct spinlock lock;
7 bool signaled;
8 struct thread *t;
9 atomic_bool alive;
10};
11
12#define CHAOS_THREADS_MAX 64
13
14#define CHAOS_THREADS_MIN 2
15#define CHAOS_THREADS_DEFAULT 8
16#define CHAOS_THREADS_INTENSE 32
17
18static size_t chaos_threads = CHAOS_THREADS_DEFAULT;
19
20#if 0
21#define CHAOS_LOG(fmt, ...) \
22 test_info("[chaos %lu] " fmt, time_get_ms(), ##__VA_ARGS__)
23#else
24#define CHAOS_LOG(fmt, ...) ((void) 0)
25#endif
26
27#define CHAOS_ITERS_DEFAULT 200
28#define CHAOS_ITERS_MIN 16
29
30#define CHAOS_APC_PERIOD_BASE 2
31
32static size_t chaos_iters_count = CHAOS_ITERS_DEFAULT;
33static size_t chaos_apc_period = CHAOS_APC_PERIOD_BASE;
34static struct chaos_state states[CHAOS_THREADS_MAX];
35static atomic_bool chaos_stop = false;
36static atomic_bool starter_ok = false;
37static atomic_uint32_t sync_chaos_left = 0;
38
39static struct mutex chaos_fuzz_mtx = MUTEX_INIT;
40static struct rwlock chaos_fuzz_rw = RWLOCK_INIT(THREAD_PRIO_CLASS_TIMESHARE);
41static struct spinlock chaos_fuzz_spin = SPINLOCK_INIT;
42static struct qspinlock chaos_fuzz_qspin = QSPINLOCK_INIT;
43
44static atomic_uint64_t chaos_apc_lock_taken = 0;
45static atomic_uint64_t chaos_apc_lock_skips = 0;
46
47static atomic_uint64_t chaos_iters[CHAOS_THREADS_MAX];
48static atomic_uint64_t chaos_spammer_iters;
49static atomic_uint64_t chaos_waker_iters;
50
51#define CHAOS_DIAG_SPAMMER CHAOS_THREADS_MAX
52#define CHAOS_DIAG_WAKER (CHAOS_THREADS_MAX + 1)
53#define CHAOS_DIAG_COUNT (CHAOS_THREADS_MAX + 2)
54
55static struct thread_diag *chaos_diag[CHAOS_DIAG_COUNT];
56
57static void chaos_diag_attach(struct thread *t, size_t slot) {
58 if (!t || chaos_diag[slot])
59 return;
60
61 struct thread_diag *d = nofail_vnew(d, ALLOC_ZERO);
62
63 if (!thread_diag_attach(t, d)) {
64 vfree(d);
65 return;
66 }
67
68 chaos_diag[slot] = d;
69}
70
71static void chaos_diag_release(struct thread *t, size_t slot) {
72 if (t)
73 thread_diag_detach(t);
74
75 vfree(chaos_diag[slot]);
76 chaos_diag[slot] = NULL;
77}
78
79static void chaos_apc_fn(struct apc_context *ac, void *arg) {
80 cc_ignore(ac);
81 cc_ignore(arg);
82 CHAOS_LOG("apc executed on %p", thread_get_current());
83
84 enum irql irql;
85 if (!spin_trylock_high(&chaos_fuzz_spin, &irql)) {
86 atomic_inc_rlx(&chaos_apc_lock_skips);
87 return;
88 }
89
90 atomic_inc_rlx(&chaos_apc_lock_taken);
91 for (volatile int j = 0; j < 10; j++)
92 cpu_pause();
93 spin_unlock(&chaos_fuzz_spin, irql);
94}
95
96static void chaos_apc_spammer(void *arg) {
97 cc_ignore(arg);
98 CHAOS_LOG("apc spammer start");
99
100 size_t pass = 0;
101
102 while (!atomic_load(&chaos_stop)) {
103 atomic_inc_rlx(&chaos_spammer_iters);
104
105 if (pass++ % chaos_apc_period) {
106 thread_sleep_for_ms(ms: 10);
107 scheduler_yield();
108 continue;
109 }
110
111 int id = prng_next() % chaos_threads;
112
113 if (!atomic_load(&states[id].alive)) {
114 thread_sleep_for_ms(ms: 10);
115 scheduler_yield();
116 continue;
117 }
118
119 rcu_read_lock();
120 bool got = thread_getref_rcu(t: states[id].t);
121 rcu_read_unlock();
122
123 if (!got) {
124 thread_sleep_for_ms(ms: 10);
125 scheduler_yield();
126 continue;
127 }
128
129 struct apc *a = vnew(a, ALLOC_ZERO);
130 if (a) {
131 apc_init(a, fn: chaos_apc_fn, NULL, destroy: apc_destroy_free);
132 CHAOS_LOG("queue apc to %p", states[id].t);
133 apc_enqueue(t: states[id].t, a, type: APC_TYPE_KERNEL);
134 apc_putref(a);
135 }
136 thread_putref(t: states[id].t);
137
138 thread_sleep_for_ms(ms: 10);
139 scheduler_yield();
140 }
141}
142
143static void chaos_sleeper(void *arg) {
144 size_t id = (size_t) arg;
145 struct thread *t = thread_get_current();
146 states[id].t = t;
147 atomic_store(&states[id].alive, true);
148
149 while (!atomic_load(&starter_ok))
150 cpu_pause();
151
152 for (size_t i = 0; i < chaos_iters_count; i++) {
153 atomic_inc_rlx(&chaos_iters[id]);
154
155 /* Exercise mutex */
156 mutex_lock(&chaos_fuzz_mtx);
157 for (volatile int j = 0; j < (int) (prng_next() & 0xF); j++)
158 cpu_pause();
159 mutex_unlock(&chaos_fuzz_mtx);
160
161 /* Exercise rwlock */
162 if (prng_next() & 1) {
163 rw_lock(&chaos_fuzz_rw, RWLOCK_READ);
164 for (volatile int j = 0; j < (int) (prng_next() & 0xF); j++)
165 cpu_pause();
166 rw_unlock(&chaos_fuzz_rw);
167 } else {
168 rw_lock(&chaos_fuzz_rw, RWLOCK_WRITE);
169 for (volatile int j = 0; j < (int) (prng_next() & 0xF); j++)
170 cpu_pause();
171 rw_unlock(&chaos_fuzz_rw);
172 }
173
174 /* Exercise qspinlock */
175 enum irql irql = qspin_lock(&chaos_fuzz_qspin);
176 for (volatile int j = 0; j < (int) (prng_next() & 0xF); j++)
177 cpu_pause();
178 qspin_unlock(&chaos_fuzz_qspin, irql);
179
180 /* Sleep and wait for waker */
181 struct chaos_state *state = &states[id];
182 irql = spin_lock_high(&state->lock);
183 if (!state->signaled) {
184 thread_wait_prepare_to_sleep(header: &state->wait, object: state,
185 type: THREAD_WAIT_INTERRUPTIBLE);
186 spin_unlock(&state->lock, irql);
187 thread_wait_complete();
188 irql = spin_lock_high(&state->lock);
189 }
190 state->signaled = false;
191 spin_unlock(&state->lock, irql);
192 CHAOS_LOG("sleeper %zu woke up, iter %zu", id, i);
193 }
194
195 atomic_store(&states[id].alive, false);
196 atomic_dec(&sync_chaos_left);
197}
198
199static void chaos_waker(void *arg) {
200 cc_ignore(arg);
201 CHAOS_LOG("waker start");
202
203 while (!atomic_load(&chaos_stop)) {
204 atomic_inc_rlx(&chaos_waker_iters);
205 int id = prng_next() % chaos_threads;
206
207 if (!atomic_load(&states[id].alive)) {
208 scheduler_yield();
209 continue;
210 }
211
212 rcu_read_lock();
213 bool got = thread_getref_rcu(t: states[id].t);
214 rcu_read_unlock();
215
216 if (!got) {
217 scheduler_yield();
218 continue;
219 }
220
221 CHAOS_LOG("wake %p", states[id].t);
222 struct chaos_state *state = &states[id];
223 enum irql irql = spin_lock_high(&state->lock);
224 state->signaled = true;
225 thread_wait_header_satisfy(hdr: &state->wait,
226 reason: THREAD_WAKE_REASON_SLEEP_MANUAL, NULL);
227 spin_unlock(&state->lock, irql);
228 thread_putref(t: states[id].t);
229
230 scheduler_yield();
231 }
232}
233
234/* Ring depth per thread in a stall report */
235#define CHAOS_ARMS_FOCUS 16
236#define CHAOS_ARMS_LIVE 4
237
238#define CHAOS_JOIN_POLL_MS 5000
239#define CHAOS_QUIET_POLLS 3
240#define CHAOS_MAX_REPORTS 2
241
242static bool chaos_progress_moved(atomic_uint64_t *counter, uint64_t *last) {
243 uint64_t now = atomic_load_rlx(counter);
244 bool moved = now != *last;
245 *last = now;
246 return moved;
247}
248
249/* `max_arms` caps the wait-arm ring dump and 0 skips the trace */
250static void chaos_dump_thread(const char *role, size_t idx, struct thread *t,
251 uint64_t max_arms) {
252 if (!t) {
253 log_msg(LOG_ERROR, " %s[%zu]: <null>", role, idx);
254 return;
255 }
256
257 log_msg(LOG_ERROR, " %s[%zu] '%s': state=%d flags=0x%x wait_type=%d", role,
258 idx, t->name, (int) thread_get_state(t),
259 (unsigned) thread_get_flags(t), (int) thread_get_wait_type(t));
260
261#ifdef DEBUG_THREAD
262 log_msg(LOG_ERROR,
263 " %s[%zu] mask=0x%x apcs_ran=%u "
264 "ctxsw=%zu",
265 role, idx, (unsigned) atomic_load(&t->apc_pending_mask),
266 t->total_apcs_ran, t->context_switches);
267#endif
268
269 if (max_arms)
270 thread_dump_wait_trace(t, role, idx, max_arms);
271}
272
273static void chaos_report_stall(const char *waiting_on, size_t waiting_idx,
274 struct thread **threads, struct thread *spammer,
275 struct thread *waker) {
276 log_msg(LOG_ERROR,
277 "chaos: no progress while joining %s[%zu] -- left=%u stop=%d "
278 "threads=%zu apc_taken=%llu apc_skips=%llu",
279 waiting_on, waiting_idx, atomic_load(&sync_chaos_left),
280 (int) atomic_load(&chaos_stop), chaos_threads,
281 (unsigned long long) atomic_load(&chaos_apc_lock_taken),
282 (unsigned long long) atomic_load(&chaos_apc_lock_skips));
283
284 log_msg(LOG_ERROR, " spammer_iters=%llu waker_iters=%llu",
285 (unsigned long long) atomic_load(&chaos_spammer_iters),
286 (unsigned long long) atomic_load(&chaos_waker_iters));
287
288 bool stuck_sleeper = strcmp(str1: waiting_on, str2: "sleeper") == 0;
289
290 for (size_t i = 0; i < chaos_threads; i++) {
291 log_msg(LOG_ERROR, " sleeper[%zu] alive=%d iters=%llu", i,
292 (int) atomic_load(&states[i].alive),
293 (unsigned long long) atomic_load(&chaos_iters[i]));
294 uint64_t arms = 0;
295 if (stuck_sleeper && i == waiting_idx)
296 arms = CHAOS_ARMS_FOCUS;
297 else if (atomic_load(&states[i].alive))
298 arms = CHAOS_ARMS_LIVE;
299
300 chaos_dump_thread(role: "sleeper", idx: i, t: threads[i], max_arms: arms);
301 }
302
303 uint64_t helper_arms = stuck_sleeper ? CHAOS_ARMS_LIVE : CHAOS_ARMS_FOCUS;
304 chaos_dump_thread(role: "spammer", idx: 0, t: spammer, max_arms: helper_arms);
305 chaos_dump_thread(role: "waker", idx: 0, t: waker, max_arms: helper_arms);
306}
307
308static bool chaos_join_watched(struct thread *t, const char *role, size_t idx,
309 atomic_uint64_t *counter,
310 struct thread **threads, struct thread *spammer,
311 struct thread *waker) {
312 uint64_t last = atomic_load_rlx(counter);
313
314 size_t quiet = 0;
315 size_t reports = 0;
316 int status;
317
318 while (!thread_join_timeout(t, CHAOS_JOIN_POLL_MS, status_out: &status)) {
319 if (chaos_progress_moved(counter, last: &last)) {
320 quiet = 0;
321 continue;
322 }
323
324 if (++quiet < CHAOS_QUIET_POLLS)
325 continue;
326
327 quiet = 0;
328 chaos_report_stall(waiting_on: role, waiting_idx: idx, threads, spammer, waker);
329
330 if (++reports < CHAOS_MAX_REPORTS)
331 continue;
332
333 log_msg(LOG_ERROR,
334 "chaos: giving up on %s[%zu] after %zu stall report(s) -- "
335 "failing the test",
336 role, idx, reports);
337 return false;
338 }
339
340 return true;
341}
342
343TEST_DEFINE_INTEGRATION(mutex, interruptible_apc_fuzz,
344 TEST_INTENSITY(CHAOS_THREADS_MIN, CHAOS_THREADS_DEFAULT,
345 CHAOS_THREADS_INTENSE),
346 .min_cores = 2) {
347 chaos_threads =
348 ctx->intensity_val ? ctx->intensity_val : CHAOS_THREADS_DEFAULT;
349 if (chaos_threads > CHAOS_THREADS_MAX)
350 chaos_threads = CHAOS_THREADS_MAX;
351
352 chaos_iters_count =
353 CHAOS_ITERS_DEFAULT * chaos_threads / CHAOS_THREADS_DEFAULT;
354 if (chaos_iters_count < CHAOS_ITERS_MIN)
355 chaos_iters_count = CHAOS_ITERS_MIN;
356
357 chaos_apc_period =
358 CHAOS_APC_PERIOD_BASE * CHAOS_THREADS_DEFAULT / chaos_threads;
359 if (!chaos_apc_period)
360 chaos_apc_period = 1;
361
362 for (size_t i = 0; i < CHAOS_THREADS_MAX; i++) {
363 states[i].t = NULL;
364 thread_wait_header_init(header: &states[i].wait);
365 spinlock_init(&states[i].lock);
366 states[i].signaled = false;
367 atomic_store(&states[i].alive, false);
368 }
369
370 atomic_store(&sync_chaos_left, chaos_threads);
371 atomic_store(&chaos_stop, false);
372 atomic_store(&starter_ok, false);
373 atomic_store(&chaos_apc_lock_taken, 0);
374 atomic_store(&chaos_apc_lock_skips, 0);
375 atomic_store(&chaos_spammer_iters, 0);
376 atomic_store(&chaos_waker_iters, 0);
377 for (size_t i = 0; i < CHAOS_THREADS_MAX; i++)
378 atomic_store(&chaos_iters[i], 0);
379
380 struct thread *threads[CHAOS_THREADS_MAX];
381 for (size_t i = 0; i < chaos_threads; i++) {
382 threads[i] = thread_create("cs", chaos_sleeper, .arg = (void *) i,
383 .joinable = true);
384 TEST_ASSERT_NONNULL(threads[i]);
385 enforce(thread_getref(threads[i]));
386 chaos_diag_attach(t: threads[i], slot: i);
387 thread_enqueue(t: threads[i]);
388 }
389
390 struct thread *spammer =
391 thread_spawn("chaos_apc_spammer", chaos_apc_spammer, .joinable = true);
392 TEST_ASSERT_NONNULL(spammer);
393 struct thread *waker =
394 thread_spawn("chaos_waker", chaos_waker, .joinable = true);
395 TEST_ASSERT_NONNULL(waker);
396
397 chaos_diag_attach(t: spammer, CHAOS_DIAG_SPAMMER);
398 chaos_diag_attach(t: waker, CHAOS_DIAG_WAKER);
399
400 atomic_store(&starter_ok, true);
401
402 bool wedged = false;
403 for (size_t i = 0; i < chaos_threads; i++) {
404 if (!chaos_join_watched(t: threads[i], role: "sleeper", idx: i, counter: &chaos_iters[i],
405 threads, spammer, waker)) {
406 wedged = true;
407 break;
408 }
409 }
410
411 atomic_store(&chaos_stop, true);
412
413 if (!wedged) {
414 wedged =
415 !chaos_join_watched(t: spammer, role: "spammer", idx: 0, counter: &chaos_spammer_iters,
416 threads, spammer, waker);
417 if (!wedged) {
418 /* The join consumed its last caller-owned ref */
419 spammer = NULL;
420 wedged = !chaos_join_watched(t: waker, role: "waker", idx: 0, counter: &chaos_waker_iters,
421 threads, spammer, waker);
422 if (!wedged)
423 waker = NULL; /* do not touch it past this point. */
424 }
425 }
426
427 if (wedged)
428 return TEST_FAIL("chaos threads made no progress");
429
430 for (size_t i = 0; i < chaos_threads; i++) {
431 chaos_diag_release(t: threads[i], slot: i);
432 thread_putref(t: threads[i]);
433 }
434
435 chaos_diag_release(t: spammer, CHAOS_DIAG_SPAMMER);
436 chaos_diag_release(t: waker, CHAOS_DIAG_WAKER);
437
438 TEST_ASSERT_EQ(atomic_load(&sync_chaos_left), 0);
439
440 test_info("apc lock: %llu taken, %llu skipped",
441 (unsigned long long) atomic_load(&chaos_apc_lock_taken),
442 (unsigned long long) atomic_load(&chaos_apc_lock_skips));
443
444 return TEST_SUCCESS;
445}
446