1#ifdef TEST_RCU
2
3#include <crypto/prng.h>
4#include <mem/alloc.h>
5#include <sch/sched.h>
6#include <string.h>
7#include <sync/rcu.h>
8#include <test.h>
9#include <thread/thread.h>
10#include <thread/workqueue.h>
11#include <time/spin_sleep.h>
12
13#define NUM_RCU_READERS (global.core_count)
14#define RCU_TEST_DURATION_MS 50
15
16struct rcu_test_data {
17 int value;
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 *) {
25 uint64_t end = time_get_ms() + RCU_TEST_DURATION_MS;
26
27 while (time_get_ms() < end) {
28 rcu_read_lock();
29
30 struct rcu_test_data *p = rcu_dereference(shared_ptr);
31 if (p) {
32 int v = p->value;
33 if (v != 42 && v != 43) {
34 atomic_store(&rcu_test_failed, true);
35 test_info("RCU reader saw invalid value");
36 test_info("%d", v);
37 }
38 }
39
40 rcu_read_unlock();
41
42 scheduler_yield();
43 }
44
45 atomic_fetch_add(&rcu_reads_done, 1);
46}
47
48static atomic_bool volatile rcu_deferred_freed = false;
49
50static void rcu_free_fn(struct rcu_cb *cb, void *ptr) {
51 kfree(ptr);
52 atomic_store(&rcu_deferred_freed, true);
53 kfree(cb);
54}
55
56static void rcu_writer_thread(void *) {
57 sleep_spin_ms(msec: 30);
58
59 struct rcu_test_data *old = shared_ptr;
60
61 struct rcu_test_data *new = kmalloc(sizeof(*new), ALLOC_FLAGS_ZERO);
62 new->value = 43;
63 rcu_assign_pointer(shared_ptr, new);
64
65 rcu_synchronize();
66 rcu_defer(kmalloc(sizeof(struct rcu_cb), ALLOC_FLAGS_ZERO), fn: rcu_free_fn,
67 arg: old);
68}
69
70TEST_DECLARE(rcu_test, .tier = TEST_TIER_UNIT) {
71 struct rcu_test_data *initial = kmalloc(sizeof(*initial), ALLOC_FLAGS_ZERO);
72 initial->value = 42;
73 shared_ptr = initial;
74
75 for (uint64_t i = 0; i < NUM_RCU_READERS; i++)
76 thread_spawn(name: "rcu_reader_test", entry: rcu_reader_thread, NULL);
77
78 thread_spawn(name: "rcu_writer_test", entry: rcu_writer_thread, NULL);
79
80 while (atomic_load(&rcu_reads_done) < NUM_RCU_READERS) {
81 scheduler_yield();
82 }
83
84 for (int i = 0; i < 100 && !atomic_load(&rcu_deferred_freed); i++)
85 sleep_spin_ms(msec: 1);
86
87 TEST_ASSERT(!atomic_load(&rcu_test_failed));
88
89 return TEST_SUCCESS;
90}
91
92#define STRESS_NUM_READERS (global.core_count * 8)
93#define STRESS_NUM_WRITERS (global.core_count)
94#define STRESS_DURATION_MS 2000
95#define STRESS_PRINT_MS 1000
96
97struct rcu_stress_node {
98 uint64_t seq; /* monotonic sequence number (for debugging) */
99 int value;
100 size_t freed_gen, enqueued_on;
101};
102
103static _Atomic(struct rcu_stress_node *) stress_shared = NULL;
104
105/* book-keeping for the test */
106static atomic_bool stress_stop = false;
107static atomic_bool stress_failed = false;
108static _Atomic uint32_t stress_readers_done = 0;
109static _Atomic uint32_t stress_writers_done = 0;
110static _Atomic uint32_t stress_deferred_freed = 0;
111static _Atomic uint32_t stress_replacements = 0;
112static atomic_size_t gen_freed = 0;
113
114/* deferred free callback */
115static void stress_free_cb(struct rcu_cb *cb, void *ptr) {
116 atomic_store(&gen_freed, cb->gen_when_called);
117 struct rcu_stress_node *n = ptr;
118 /* optional debug trace */
119 n->value = 34;
120 n->freed_gen = cb->gen_when_called;
121 n->enqueued_on = cb->enqueued_waiting_on_gen;
122 atomic_fetch_add(&stress_deferred_freed, 1);
123 kfree(cb);
124 kfree(n);
125}
126
127/* reader thread: very tight loop, yields frequently */
128static void rcu_stress_reader(void *arg) {
129 (void) arg;
130
131 time_t last_print = time_get_ms();
132 size_t iter = 0;
133 while (!atomic_load(&stress_stop)) {
134 rcu_read_lock();
135
136 struct rcu_stress_node *p = rcu_dereference(stress_shared);
137 if (p) {
138 int v = p->value;
139 if (v != 42 && v != 43) {
140 atomic_store(&stress_failed, true);
141 test_err("RCU stress reader saw invalid value");
142 test_err("RCU stress reader observed invalid value %d, "
143 "freed during gen %zu enqueued_on %zu currently "
144 "started gen %zu quiescent for gen %zu\nat a nesting "
145 "depth of %zu",
146 v, p->freed_gen, p->enqueued_on,
147 thread_get_current()->rcu_start_gen,
148 thread_get_current()->rcu_quiescent_gen,
149 thread_get_current()->rcu_nesting);
150 break;
151 }
152 volatile uint64_t seq = p->seq;
153 (void) seq;
154 }
155
156 rcu_read_unlock();
157
158 if (time_get_ms() - last_print > STRESS_PRINT_MS) {
159 last_print = time_get_ms();
160 test_info(
161 "\'%-20s\' on iteration %7zu w/ %7zu replacements and %7zu "
162 "frees",
163 thread_get_current()->name, iter, stress_replacements,
164 stress_deferred_freed);
165 }
166
167 /* yield to exercise scheduler preemption and context switching */
168 scheduler_yield();
169 iter++;
170 }
171
172 test_info("RCU stress reader %s left, %u remaining",
173 thread_get_current()->name,
174 STRESS_NUM_READERS - stress_readers_done - 1);
175
176 atomic_fetch_add(&stress_readers_done, 1);
177}
178
179/* writer thread: continuously replace the pointer, sometimes synchronize */
180static void rcu_stress_writer(void *arg) {
181 (void) arg;
182 uint64_t local_iter = 0;
183
184 while (!atomic_load(&stress_stop)) {
185 struct rcu_stress_node *new = kmalloc(sizeof(*new), ALLOC_FLAGS_ZERO);
186 if (!new) {
187 /* allocation failure — mark as failure and exit */
188 atomic_store(&stress_failed, true);
189 test_info("RCU stress writer kmalloc failed");
190 break;
191 }
192
193 new->seq = (uint64_t) atomic_fetch_add(&stress_replacements, 1) + 1;
194 /* alternate values to ensure readers see both */
195 new->value = (local_iter & 1) ? 43 : 42;
196 local_iter++;
197
198 struct rcu_stress_node *old =
199 atomic_exchange_explicit(&stress_shared, new, memory_order_acq_rel);
200
201 /*
202 * Defer freeing the old pointer. We deliberately create a backlog by
203 * deferring every single old pointer; later we wait for them to be
204 * freed to assert correctness.
205 */
206 if (old)
207 rcu_defer(kmalloc(sizeof(struct rcu_cb), ALLOC_FLAGS_ZERO),
208 fn: stress_free_cb, arg: old);
209
210 /*
211 * Occasionally force a synchronize call to exercise explicit grace
212 * period advancement (do this about once every ~32 replacements).
213 */
214 if ((local_iter & 0x1f) == 0) {
215 rcu_synchronize();
216 }
217
218 scheduler_yield();
219 }
220
221 atomic_fetch_add(&stress_writers_done, 1);
222}
223
224/* a reclaimer thread that also calls synchronize periodically to help drain */
225static void rcu_stress_reclaimer(void *arg) {
226 (void) arg;
227 while (!atomic_load(&stress_stop)) {
228 /* attempt to shrink deferred backlog by forcing grace periods */
229 rcu_synchronize();
230 /* small backoff between synchronizations */
231 sleep_spin_ms(msec: 5);
232 }
233}
234
235/* Test registration */
236TEST_DECLARE(rcu_stress_test, .tier = TEST_TIER_UNIT, .print_logs = true, ) {
237 /* initial object */
238 struct rcu_stress_node *initial =
239 kmalloc(sizeof(*initial), ALLOC_FLAGS_ZERO);
240 TEST_ASSERT(initial != NULL);
241 initial->seq = 1;
242 initial->value = 42;
243
244 atomic_store(&stress_stop, false);
245 atomic_store(&stress_failed, false);
246 atomic_store(&stress_readers_done, 0);
247 atomic_store(&stress_writers_done, 0);
248 atomic_store(&stress_deferred_freed, 0);
249 atomic_store(&stress_replacements, 0);
250 stress_shared = initial;
251
252 /* spawn readers (more than cores) */
253 for (uint32_t i = 0; i < STRESS_NUM_READERS; ++i) {
254 thread_spawn(name: "rcu_str_reader_%u", entry: rcu_stress_reader, NULL, i);
255 }
256
257 /* spawn writers */
258 for (uint32_t i = 0; i < STRESS_NUM_WRITERS; ++i) {
259 thread_spawn(name: "rcu_str_writer_%u", entry: rcu_stress_writer, NULL, i);
260 }
261
262 /* spawn one reclaimer to periodically call synchronize */
263 thread_spawn(name: "rcu_str_reclaimer", entry: rcu_stress_reclaimer, NULL);
264
265 /* run for the configured duration */
266 uint64_t stop_at = time_get_ms() + STRESS_DURATION_MS;
267 while (time_get_ms() < stop_at) {
268 if (atomic_load(&stress_failed)) {
269 test_info("RCU stress test failed early due to detection");
270 break;
271 }
272 /* let other threads run */
273 scheduler_yield();
274 }
275
276 /* signal stop to all readers/writers/reclaimer */
277 atomic_store(&stress_stop, true);
278
279 /* wait for readers to finish */
280 while (atomic_load(&stress_readers_done) < STRESS_NUM_READERS) {
281 scheduler_yield();
282 }
283
284 while (atomic_load(&stress_writers_done) < STRESS_NUM_WRITERS) {
285 scheduler_yield();
286 }
287
288 /* wait up to a reasonable timeout for deferred frees to run */
289 for (int i = 0; i < 100 && atomic_load(&stress_deferred_freed) <
290 atomic_load(&stress_replacements);
291 i++) {
292 /* call synchronize here to help force callbacks */
293 rcu_synchronize();
294 sleep_spin_ms(msec: 1);
295 }
296
297 test_info("RCU stress test: replacements=%u freed=%u",
298 (unsigned) atomic_load(&stress_replacements),
299 (unsigned) atomic_load(&stress_deferred_freed));
300
301 /* checks */
302 TEST_ASSERT(!atomic_load(&stress_failed));
303
304 /*
305 * We expect at least some frees to have occurred. On very constrained
306 * implementations it may be possible not all deferred callbacks have
307 * yet run; fail only if zero frees happened or if obviously fewer frees
308 * than replacements exist (tunable).
309 */
310 TEST_ASSERT(atomic_load(&stress_deferred_freed) > 0);
311
312 /* finally, free the last published pointer (if any) from test */
313 struct rcu_stress_node *last = stress_shared;
314 if (last) {
315 /* old-style: synchronize then free directly */
316 rcu_synchronize();
317 kfree(last);
318 atomic_fetch_add(&stress_deferred_freed, 1);
319 }
320
321 return TEST_SUCCESS;
322}
323
324#endif
325