1#include "sync/tests/test_internal.h"
2#include <acpi/lapic.h>
3#include <irq/ipi.h>
4#include <irq/irq.h>
5#include <mem/nofail.h>
6
7TEST_GROUP_DEFINE(rcu, .intensity_desc = {
8 .curve = SCALE_PIECEWISE_LOG,
9 .unit = "ms",
10 });
11
12#define NUM_RCU_READERS (global.core_count)
13static size_t rcu_test_duration_ms = 50;
14
15struct rcu_test_data {
16 int value;
17 struct rcu_cb rcu;
18};
19
20static atomic(struct rcu_test_data *) shared_ptr = NULL;
21static atomic_bool rcu_test_failed = false;
22static atomic_uint32_t rcu_reads_done = 0;
23
24static void rcu_reader_thread(void *arg) {
25 cc_ignore(arg);
26 uint64_t end = time_get_ms() + rcu_test_duration_ms;
27
28 while (time_get_ms() < end) {
29 rcu_read_lock();
30
31 struct rcu_test_data *p = rcu_dereference(shared_ptr);
32 if (p) {
33 int v = p->value;
34 if (v != 42 && v != 43) {
35 atomic_store(&rcu_test_failed, true);
36 test_info("RCU reader saw invalid value");
37 test_info("%d", v);
38 }
39 }
40
41 rcu_read_unlock();
42
43 scheduler_yield();
44 }
45
46 atomic_inc(&rcu_reads_done);
47}
48
49static atomic_bool volatile rcu_deferred_freed = false;
50
51static void rcu_free_fn(struct rcu_cb *cb) {
52 vfree(container_of(cb, struct rcu_test_data, rcu));
53 atomic_store(&rcu_deferred_freed, true);
54}
55
56static void rcu_writer_thread(void *arg) {
57 cc_ignore(arg);
58 sleep_spin_ms(msec: 30);
59
60 struct rcu_test_data *old = atomic_load_rlx(&shared_ptr);
61
62 struct rcu_test_data *new = nofail_vnew(new, ALLOC_ZERO);
63 new->value = 43;
64 rcu_assign_pointer(shared_ptr, new);
65
66 rcu_synchronize();
67 rcu_defer(cb: &old->rcu, func: rcu_free_fn);
68}
69
70TEST_DEFINE_INTEGRATION(rcu, mt_readers_during_replace,
71 TEST_INTENSITY(40, 50, 200)) {
72 rcu_test_duration_ms = ctx->intensity_val ? ctx->intensity_val : 50;
73 if (rcu_test_duration_ms < 40)
74 rcu_test_duration_ms = 40;
75
76 atomic_store(&rcu_test_failed, false);
77 atomic_store(&rcu_reads_done, 0);
78 atomic_store(&rcu_deferred_freed, false);
79
80 struct rcu_test_data *initial = vnew(initial, ALLOC_ZERO);
81 TEST_ASSERT_NONNULL(initial);
82 initial->value = 42;
83 rcu_assign_pointer(shared_ptr, initial);
84
85 struct thread *readers[NUM_RCU_READERS];
86 memset(readers, 0, sizeof(readers));
87 for (uint64_t i = 0; i < NUM_RCU_READERS; i++)
88 readers[i] = thread_spawn("rcu_reader_test", rcu_reader_thread,
89 .joinable = true);
90
91 struct thread *writer =
92 thread_spawn("rcu_writer_test", rcu_writer_thread, .joinable = true);
93
94 for (uint64_t i = 0; i < NUM_RCU_READERS; i++) {
95 if (readers[i])
96 thread_join(t: readers[i]);
97 }
98
99 if (writer)
100 thread_join(t: writer);
101
102 TEST_ASSERT_EQ(atomic_load(&rcu_reads_done), NUM_RCU_READERS);
103
104 for (int i = 0; i < 100 && !atomic_load(&rcu_deferred_freed); i++)
105 sleep_spin_ms(msec: 1);
106
107 TEST_ASSERT(!atomic_load(&rcu_test_failed));
108
109 return TEST_SUCCESS;
110}
111
112#define STRESS_NUM_READERS (global.core_count * 8)
113#define STRESS_NUM_WRITERS (global.core_count)
114static size_t rcu_stress_duration_ms = 2000;
115#define STRESS_PRINT_MS 1000
116
117struct rcu_stress_node {
118 uint64_t seq; /* monotonic sequence number (for debugging) */
119 int value;
120 size_t freed_gen, enqueued_on;
121 struct rcu_cb rcu;
122};
123
124static atomic(struct rcu_stress_node *) stress_shared = NULL;
125
126static atomic_bool stress_stop = false;
127static atomic_bool stress_failed = false;
128static atomic_uint32_t stress_readers_done = 0;
129static atomic_uint32_t stress_writers_done = 0;
130static atomic_uint32_t stress_deferred_freed = 0;
131static atomic_uint32_t stress_replacements = 0;
132static atomic_size_t gen_freed = 0;
133
134static void stress_free_cb(struct rcu_cb *cb) {
135 atomic_store(&gen_freed, cb->gen_when_called);
136 struct rcu_stress_node *n = container_of(cb, struct rcu_stress_node, rcu);
137 n->value = 34;
138 n->freed_gen = cb->gen_when_called;
139 n->enqueued_on = cb->enqueued_waiting_on_gen;
140 atomic_inc(&stress_deferred_freed);
141 vfree(n);
142}
143
144static void rcu_stress_reader(void *arg) TSA_NO_ANALYSIS {
145 cc_ignore(arg);
146
147 time_ms_t last_print = time_get_ms();
148 size_t iter = 0;
149 while (!atomic_load(&stress_stop)) {
150 rcu_read_lock();
151
152 struct rcu_stress_node *p = rcu_dereference(stress_shared);
153 if (p) {
154 int v = p->value;
155 if (v != 42 && v != 43) {
156 atomic_store(&stress_failed, true);
157 test_err("RCU stress reader saw invalid value");
158 break;
159 }
160 volatile uint64_t seq = p->seq;
161 cc_ignore(seq);
162 }
163
164 rcu_read_unlock();
165
166 if (time_get_ms() - last_print > STRESS_PRINT_MS) {
167 last_print = time_get_ms();
168 test_info("\'%-17s\' iter %7zu w/ %7zu rplace and %7zu free",
169 thread_get_current()->name, iter,
170 (size_t) atomic_load(&stress_replacements),
171 (size_t) atomic_load(&stress_deferred_freed));
172 }
173
174 scheduler_yield();
175 iter++;
176 }
177
178 atomic_inc(&stress_readers_done);
179}
180
181static void rcu_stress_writer(void *arg) {
182 cc_ignore(arg);
183 uint64_t local_iter = 0;
184
185 while (!atomic_load(&stress_stop)) {
186 struct rcu_stress_node *new = vnew(new, ALLOC_ZERO);
187 if (!new) {
188 atomic_store(&stress_failed, true);
189 test_info("RCU stress writer valloc failed");
190 break;
191 }
192
193 new->seq = (uint64_t) atomic_fetch_add(&stress_replacements, 1) + 1;
194 new->value = (local_iter & 1) ? 43 : 42;
195 local_iter++;
196
197 struct rcu_stress_node *old = atomic_xchg_acq_rel(&stress_shared, new);
198
199 if (old)
200 rcu_defer(cb: &old->rcu, func: stress_free_cb);
201
202 if ((local_iter & 0x1f) == 0) {
203 rcu_synchronize();
204 }
205
206 scheduler_yield();
207 }
208
209 atomic_inc(&stress_writers_done);
210}
211
212static void rcu_stress_reclaimer(void *arg) {
213 cc_ignore(arg);
214 while (!atomic_load(&stress_stop)) {
215 rcu_synchronize();
216 sleep_spin_ms(msec: 5);
217 }
218}
219
220TEST_DEFINE_INTEGRATION(rcu, mt_stress, TEST_INTENSITY(200, 2000, 10000)) {
221 rcu_stress_duration_ms = ctx->intensity_val ? ctx->intensity_val : 2000;
222 atomic_store(&stress_stop, false);
223 atomic_store(&stress_failed, false);
224 atomic_store(&stress_readers_done, 0);
225 atomic_store(&stress_writers_done, 0);
226 atomic_store(&stress_deferred_freed, 0);
227 atomic_store(&stress_replacements, 0);
228 atomic_store(&gen_freed, 0);
229
230 struct rcu_stress_node *initial = vnew(initial, ALLOC_ZERO);
231 TEST_ASSERT_NONNULL(initial);
232 initial->seq = 0;
233 initial->value = 42;
234 rcu_assign_pointer(stress_shared, initial);
235
236 struct thread *readers[STRESS_NUM_READERS];
237 struct thread *writers[STRESS_NUM_WRITERS];
238 memset(readers, 0, sizeof(readers));
239 memset(writers, 0, sizeof(writers));
240
241 for (uint32_t i = 0; i < STRESS_NUM_READERS; ++i) {
242 readers[i] = thread_spawn(("rcu_stread_%u", i), rcu_stress_reader,
243 .joinable = true);
244 }
245
246 for (uint32_t i = 0; i < STRESS_NUM_WRITERS; ++i) {
247 writers[i] = thread_spawn(("rcu_strite_%u", i), rcu_stress_writer,
248 .joinable = true);
249 }
250
251 struct thread *reclaimer =
252 thread_spawn("rcu_streclaim", rcu_stress_reclaimer, .joinable = true);
253
254 uint64_t stop_at = time_get_ms() + rcu_stress_duration_ms;
255 while (time_get_ms() < stop_at) {
256 if (atomic_load(&stress_failed)) {
257 test_info("RCU stress test failed early due to detection");
258 break;
259 }
260 scheduler_yield();
261 }
262
263 atomic_store(&stress_stop, true);
264
265 for (uint32_t i = 0; i < STRESS_NUM_READERS; ++i) {
266 if (readers[i])
267 thread_join(t: readers[i]);
268 }
269
270 for (uint32_t i = 0; i < STRESS_NUM_WRITERS; ++i) {
271 if (writers[i])
272 thread_join(t: writers[i]);
273 }
274
275 if (reclaimer)
276 thread_join(t: reclaimer);
277
278 for (int i = 0; i < 100 && atomic_load(&stress_deferred_freed) <
279 atomic_load(&stress_replacements);
280 i++) {
281 rcu_synchronize();
282 sleep_spin_ms(msec: 1);
283 }
284
285 test_info("RCU stress test: replacements=%u freed=%u",
286 (unsigned) atomic_load(&stress_replacements),
287 (unsigned) atomic_load(&stress_deferred_freed));
288 test_info(
289 " [RCU STATS] Completed %u replacements, %u deferred frees across "
290 "64 readers & 8 writers\n",
291 (unsigned) atomic_load(&stress_replacements),
292 (unsigned) atomic_load(&stress_deferred_freed));
293
294 TEST_ASSERT(!atomic_load(&stress_failed));
295 TEST_ASSERT_GT(atomic_load(&stress_deferred_freed), 0);
296 TEST_ASSERT_EQ(atomic_load(&stress_deferred_freed),
297 atomic_load(&stress_replacements));
298
299 struct rcu_stress_node *last = atomic_load_rlx(&stress_shared);
300 if (last) {
301 rcu_synchronize();
302 vfree(last);
303 atomic_inc(&stress_deferred_freed);
304 }
305
306 return TEST_SUCCESS;
307}
308
309#define RCU_IRQ_HOLD_MS 20
310#define RCU_IRQ_WAIT_MS 1000
311#define RCU_IRQ_NODE_LIVE 0x5a5a
312#define RCU_IRQ_NODE_DEAD 0xdead
313
314struct rcu_irq_node {
315 atomic_int value;
316 struct rcu_cb rcu;
317};
318
319static struct rcu_irq_node rcu_irq_nodes[2];
320static atomic(struct rcu_irq_node *) rcu_irq_shared = NULL;
321static atomic_bool rcu_irq_in_section = false;
322static atomic_bool rcu_irq_done = false;
323static atomic_bool rcu_irq_freed = false;
324static atomic_bool rcu_irq_failed = false;
325static irq_t rcu_irq_vector;
326static bool rcu_irq_registered = false;
327
328static void rcu_irq_free_cb(struct rcu_cb *cb) {
329 struct rcu_irq_node *n = container_of(cb, struct rcu_irq_node, rcu);
330 atomic_store_release(&n->value, RCU_IRQ_NODE_DEAD);
331 atomic_store_release(&rcu_irq_freed, true);
332}
333
334static enum irq_result rcu_irq_reader(struct irq_context *ictx, void *ctx) {
335 cc_ignore(ctx, ictx);
336
337 rcu_read_lock();
338 struct rcu_irq_node *p = rcu_dereference(rcu_irq_shared);
339 atomic_store_release(&rcu_irq_in_section, true);
340
341 /* Long enough for the writer to swap, defer, and a GP to complete if
342 * this CPU is (wrongly) considered quiescent */
343 time_ms_t end = time_get_ms() + RCU_IRQ_HOLD_MS;
344 while (time_get_ms() < end) {
345 if (atomic_load_acq(&p->value) != RCU_IRQ_NODE_LIVE) {
346 atomic_store(&rcu_irq_failed, true);
347 break;
348 }
349 cpu_pause();
350 }
351
352 rcu_read_unlock();
353 atomic_store_release(&rcu_irq_done, true);
354 return IRQ_HANDLED;
355}
356
357static bool rcu_irq_wait_for(atomic_bool *flag) {
358 time_ms_t end = time_get_ms() + RCU_IRQ_WAIT_MS;
359 while (!atomic_load_acq(flag)) {
360 if (time_get_ms() >= end)
361 return false;
362
363 scheduler_yield();
364 }
365
366 return true;
367}
368
369TEST_DEFINE_INTEGRATION(rcu, irq_reader_on_idle_cpu, TEST_INTENSITY(5, 10, 100),
370 .min_cores = 4) {
371 size_t rounds = ctx->intensity_val ? ctx->intensity_val : 10;
372
373 if (!rcu_irq_registered) {
374 rcu_irq_vector = irq_alloc_entry();
375 irq_register(name: "rcu_irq_reader_test", vector: rcu_irq_vector, handler: rcu_irq_reader,
376 NULL, flags: IRQ_FLAG_NONE);
377 irq_set_chip(vector: rcu_irq_vector, chip: lapic_get_chip(), NULL);
378 rcu_irq_registered = true;
379 }
380
381 struct thread *self_thread = thread_get_current();
382 bool was_pinned = thread_pin(t: self_thread);
383 cpu_id_t self = smp_id(cond: TOPC_PINNED);
384 cpu_id_t target = (self + global.core_count / 2) % global.core_count;
385
386 atomic_store(&rcu_irq_failed, false);
387
388 size_t cur = 0;
389 atomic_store(&rcu_irq_nodes[cur].value, RCU_IRQ_NODE_LIVE);
390 rcu_assign_pointer(rcu_irq_shared, &rcu_irq_nodes[cur]);
391
392 bool timed_out = false;
393 for (size_t r = 0; r < rounds && !timed_out; r++) {
394 size_t next = cur ^ 1;
395 atomic_store(&rcu_irq_nodes[next].value, RCU_IRQ_NODE_LIVE);
396
397 atomic_store(&rcu_irq_in_section, false);
398 atomic_store(&rcu_irq_done, false);
399 atomic_store(&rcu_irq_freed, false);
400
401 /* Let the target settle back into its idle loop */
402 sleep_spin_ms(msec: 1);
403 ipi_send(cpu: target, vector: rcu_irq_vector);
404
405 if (!rcu_irq_wait_for(flag: &rcu_irq_in_section)) {
406 timed_out = true;
407 break;
408 }
409
410 rcu_assign_pointer(rcu_irq_shared, &rcu_irq_nodes[next]);
411 rcu_defer(cb: &rcu_irq_nodes[cur].rcu, func: rcu_irq_free_cb);
412
413 if (!rcu_irq_wait_for(flag: &rcu_irq_done)) {
414 timed_out = true;
415 break;
416 }
417
418 rcu_synchronize();
419 if (!rcu_irq_wait_for(flag: &rcu_irq_freed)) {
420 timed_out = true;
421 break;
422 }
423
424 cur = next;
425 if (atomic_load(&rcu_irq_failed))
426 break;
427 }
428
429 if (!was_pinned)
430 thread_unpin(t: self_thread);
431
432 TEST_ASSERT(!timed_out);
433 TEST_ASSERT(!atomic_load(&rcu_irq_failed));
434
435 return TEST_SUCCESS;
436}
437