1#include "sync/tests/test_internal.h"
2
3TEST_GROUP_DECLARE(rcu, .intensity_desc = {
4 .curve = SCALE_PIECEWISE_LOG,
5 .unit = "ms",
6 });
7
8#define NUM_RCU_READERS (global.core_count)
9static size_t rcu_test_duration_ms = 50;
10
11struct rcu_test_data {
12 int value;
13};
14
15static _Atomic(struct rcu_test_data *) shared_ptr = NULL;
16static atomic_bool rcu_test_failed = false;
17static _Atomic uint32_t rcu_reads_done = 0;
18
19static void rcu_reader_thread(void *) {
20 uint64_t end = time_get_ms() + rcu_test_duration_ms;
21
22 while (time_get_ms() < end) {
23 rcu_read_lock();
24
25 struct rcu_test_data *p = rcu_dereference(shared_ptr);
26 if (p) {
27 int v = p->value;
28 if (v != 42 && v != 43) {
29 atomic_store(&rcu_test_failed, true);
30 test_info("RCU reader saw invalid value");
31 test_info("%d", v);
32 }
33 }
34
35 rcu_read_unlock();
36
37 scheduler_yield();
38 }
39
40 atomic_fetch_add(&rcu_reads_done, 1);
41}
42
43static atomic_bool volatile rcu_deferred_freed = false;
44
45static void rcu_free_fn(struct rcu_cb *cb, void *ptr) {
46 kfree(ptr);
47 atomic_store(&rcu_deferred_freed, true);
48 kfree(cb);
49}
50
51static void rcu_writer_thread(void *) {
52 sleep_spin_ms(msec: 30);
53
54 struct rcu_test_data *old = shared_ptr;
55
56 struct rcu_test_data *new = kmalloc(sizeof(*new), ALLOC_FLAGS_ZERO);
57 new->value = 43;
58 rcu_assign_pointer(shared_ptr, new);
59
60 rcu_synchronize();
61 rcu_defer(kmalloc(sizeof(struct rcu_cb), ALLOC_FLAGS_ZERO), fn: rcu_free_fn,
62 arg: old);
63}
64
65TEST_DECLARE_INTEGRATION(rcu, basic, TEST_INTENSITY(40, 50, 200)) {
66 rcu_test_duration_ms = ctx->intensity_val ? ctx->intensity_val : 50;
67 if (rcu_test_duration_ms < 40)
68 rcu_test_duration_ms = 40;
69
70 atomic_store(&rcu_test_failed, false);
71 atomic_store(&rcu_reads_done, 0);
72 atomic_store(&rcu_deferred_freed, false);
73
74 struct rcu_test_data *initial = kmalloc(sizeof(*initial), ALLOC_FLAGS_ZERO);
75 initial->value = 42;
76 shared_ptr = initial;
77
78 struct thread *readers[NUM_RCU_READERS];
79 for (uint64_t i = 0; i < NUM_RCU_READERS; i++)
80 readers[i] =
81 thread_spawn_joinable(name: "rcu_reader_test", entry: rcu_reader_thread, NULL);
82
83 struct thread *writer =
84 thread_spawn_joinable(name: "rcu_writer_test", entry: rcu_writer_thread, NULL);
85
86 for (uint64_t i = 0; i < NUM_RCU_READERS; i++) {
87 if (readers[i])
88 thread_join(t: readers[i]);
89 }
90
91 if (writer)
92 thread_join(t: writer);
93
94 TEST_ASSERT_EQ(atomic_load(&rcu_reads_done), NUM_RCU_READERS);
95
96 for (int i = 0; i < 100 && !atomic_load(&rcu_deferred_freed); i++)
97 sleep_spin_ms(msec: 1);
98
99 TEST_ASSERT(!atomic_load(&rcu_test_failed));
100
101 return TEST_SUCCESS;
102}
103
104#define STRESS_NUM_READERS (global.core_count * 8)
105#define STRESS_NUM_WRITERS (global.core_count)
106static size_t rcu_stress_duration_ms = 2000;
107#define STRESS_PRINT_MS 1000
108
109struct rcu_stress_node {
110 uint64_t seq; /* monotonic sequence number (for debugging) */
111 int value;
112 size_t freed_gen, enqueued_on;
113};
114
115static _Atomic(struct rcu_stress_node *) stress_shared = NULL;
116
117static atomic_bool stress_stop = false;
118static atomic_bool stress_failed = false;
119static _Atomic uint32_t stress_readers_done = 0;
120static _Atomic uint32_t stress_writers_done = 0;
121static _Atomic uint32_t stress_deferred_freed = 0;
122static _Atomic uint32_t stress_replacements = 0;
123static atomic_size_t gen_freed = 0;
124
125static void stress_free_cb(struct rcu_cb *cb, void *ptr) {
126 atomic_store(&gen_freed, cb->gen_when_called);
127 struct rcu_stress_node *n = ptr;
128 n->value = 34;
129 n->freed_gen = cb->gen_when_called;
130 n->enqueued_on = cb->enqueued_waiting_on_gen;
131 atomic_fetch_add(&stress_deferred_freed, 1);
132 kfree(cb);
133 kfree(n);
134}
135
136static void rcu_stress_reader(void *arg) {
137 (void) arg;
138
139 time_ms_t last_print = time_get_ms();
140 size_t iter = 0;
141 while (!atomic_load(&stress_stop)) {
142 rcu_read_lock();
143
144 struct rcu_stress_node *p = rcu_dereference(stress_shared);
145 if (p) {
146 int v = p->value;
147 if (v != 42 && v != 43) {
148 atomic_store(&stress_failed, true);
149 test_err("RCU stress reader saw invalid value");
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("\'%-17s\' iter %7zu w/ %7zu rplace and %7zu free",
161 thread_get_current()->name, iter,
162 (size_t) atomic_load(&stress_replacements),
163 (size_t) atomic_load(&stress_deferred_freed));
164 }
165
166 scheduler_yield();
167 iter++;
168 }
169
170 atomic_fetch_add(&stress_readers_done, 1);
171}
172
173static void rcu_stress_writer(void *arg) {
174 (void) arg;
175 uint64_t local_iter = 0;
176
177 while (!atomic_load(&stress_stop)) {
178 struct rcu_stress_node *new = kmalloc(sizeof(*new), ALLOC_FLAGS_ZERO);
179 if (!new) {
180 atomic_store(&stress_failed, true);
181 test_info("RCU stress writer kmalloc failed");
182 break;
183 }
184
185 new->seq = (uint64_t) atomic_fetch_add(&stress_replacements, 1) + 1;
186 new->value = (local_iter & 1) ? 43 : 42;
187 local_iter++;
188
189 struct rcu_stress_node *old =
190 atomic_exchange_explicit(&stress_shared, new, memory_order_acq_rel);
191
192 if (old)
193 rcu_defer(kmalloc(sizeof(struct rcu_cb), ALLOC_FLAGS_ZERO),
194 fn: stress_free_cb, arg: old);
195
196 if ((local_iter & 0x1f) == 0) {
197 rcu_synchronize();
198 }
199
200 scheduler_yield();
201 }
202
203 atomic_fetch_add(&stress_writers_done, 1);
204}
205
206static void rcu_stress_reclaimer(void *arg) {
207 (void) arg;
208 while (!atomic_load(&stress_stop)) {
209 rcu_synchronize();
210 sleep_spin_ms(msec: 5);
211 }
212}
213
214TEST_DECLARE_INTEGRATION(rcu, stress, TEST_INTENSITY(200, 2000, 10000)) {
215 rcu_stress_duration_ms = ctx->intensity_val ? ctx->intensity_val : 2000;
216 atomic_store(&stress_stop, false);
217 atomic_store(&stress_failed, false);
218 atomic_store(&stress_readers_done, 0);
219 atomic_store(&stress_writers_done, 0);
220 atomic_store(&stress_deferred_freed, 0);
221 atomic_store(&stress_replacements, 0);
222 atomic_store(&gen_freed, 0);
223
224 struct rcu_stress_node *initial =
225 kmalloc(sizeof(*initial), ALLOC_FLAGS_ZERO);
226 initial->seq = 0;
227 initial->value = 42;
228 stress_shared = initial;
229
230 struct thread *readers[STRESS_NUM_READERS];
231 struct thread *writers[STRESS_NUM_WRITERS];
232
233 for (uint32_t i = 0; i < STRESS_NUM_READERS; ++i) {
234 readers[i] =
235 thread_spawn_joinable(name: "rcu_stread_%u", entry: rcu_stress_reader, NULL, i);
236 }
237
238 for (uint32_t i = 0; i < STRESS_NUM_WRITERS; ++i) {
239 writers[i] =
240 thread_spawn_joinable(name: "rcu_strite_%u", entry: rcu_stress_writer, NULL, i);
241 }
242
243 struct thread *reclaimer =
244 thread_spawn_joinable(name: "rcu_streclaim", entry: rcu_stress_reclaimer, NULL);
245
246 uint64_t stop_at = time_get_ms() + rcu_stress_duration_ms;
247 while (time_get_ms() < stop_at) {
248 if (atomic_load(&stress_failed)) {
249 test_info("RCU stress test failed early due to detection");
250 break;
251 }
252 scheduler_yield();
253 }
254
255 atomic_store(&stress_stop, true);
256
257 for (uint32_t i = 0; i < STRESS_NUM_READERS; ++i) {
258 if (readers[i])
259 thread_join(t: readers[i]);
260 }
261
262 for (uint32_t i = 0; i < STRESS_NUM_WRITERS; ++i) {
263 if (writers[i])
264 thread_join(t: writers[i]);
265 }
266
267 if (reclaimer)
268 thread_join(t: reclaimer);
269
270 for (int i = 0; i < 100 && atomic_load(&stress_deferred_freed) <
271 atomic_load(&stress_replacements);
272 i++) {
273 rcu_synchronize();
274 sleep_spin_ms(msec: 1);
275 }
276
277 test_info("RCU stress test: replacements=%u freed=%u",
278 (unsigned) atomic_load(&stress_replacements),
279 (unsigned) atomic_load(&stress_deferred_freed));
280 test_info(
281 " [RCU STATS] Completed %u replacements, %u deferred frees across "
282 "64 readers & 8 writers\n",
283 (unsigned) atomic_load(&stress_replacements),
284 (unsigned) atomic_load(&stress_deferred_freed));
285
286 TEST_ASSERT(!atomic_load(&stress_failed));
287 TEST_ASSERT_GT(atomic_load(&stress_deferred_freed), 0);
288 TEST_ASSERT_EQ(atomic_load(&stress_deferred_freed),
289 atomic_load(&stress_replacements));
290
291 struct rcu_stress_node *last = stress_shared;
292 if (last) {
293 rcu_synchronize();
294 kfree(last);
295 atomic_fetch_add(&stress_deferred_freed, 1);
296 }
297
298 return TEST_SUCCESS;
299}
300