1#include <console/panic.h>
2#include <container_of.h>
3#include <global.h>
4#include <mem/alloc.h>
5#include <sch/sched.h>
6#include <smp/core.h>
7#include <stdarg.h>
8#include <string.h>
9#include <test/fleet.h>
10#include <thread/thread.h>
11#include <thread/thread_diag.h>
12#include <time/time.h>
13
14#ifdef TEST_ENABLED
15
16static cpu_id_t test_core_modulo(size_t index) {
17 size_t count = global.core_count ? global.core_count : 1;
18 return (cpu_id_t) (index % count);
19}
20
21void *test_alloc(struct test_context *ctx, size_t size) {
22 if (cc_unlikely(!ctx))
23 panic("test_alloc: NULL test_context");
24
25 size_t total;
26 if (cc_unlikely(
27 ci_add_overflow(sizeof(struct test_alloc_node), size, &total)))
28 panic(OVERFLOW, "test_alloc: %zu bytes", size);
29
30 struct test_alloc_node *a = valloc(total, ALLOC_ZERO);
31 if (cc_unlikely(!a))
32 panic(OUT_OF_MEMORY, "test_alloc: %zu bytes", size);
33
34 a->next = ctx->allocs;
35 ctx->allocs = a;
36 return (void *) (a + 1);
37}
38
39static void test_alloc_teardown(struct test_context *ctx) {
40 if (!ctx)
41 return;
42
43 struct test_alloc_node *a = ctx->allocs;
44 while (a) {
45 struct test_alloc_node *next = a->next;
46 vfree(a);
47 a = next;
48 }
49 ctx->allocs = NULL;
50}
51
52void *test_fleet_alloc(struct test_fleet *f, size_t size) {
53 if (cc_unlikely(!f || !f->ctx))
54 panic("test_fleet_alloc: NULL fleet or context");
55 return test_alloc(ctx: f->ctx, size);
56}
57
58struct thread *test_spawn_internal(struct test_context *ctx,
59 void (*entry)(void *),
60 const struct thread_create_params *params,
61 const char *name) {
62 if (!ctx)
63 return NULL;
64
65 if (ctx->helper_count >= TEST_MAX_HELPERS) {
66 test_err("test_spawn: exceeded TEST_MAX_HELPERS (%d)",
67 TEST_MAX_HELPERS);
68 return NULL;
69 }
70
71 struct thread_create_params p =
72 params ? *params : (struct thread_create_params){0};
73 p.joinable = true;
74
75 struct thread *t = thread_spawn_internal(entry, params: p, fmt: "%s", name);
76 if (!t)
77 return NULL;
78
79 enforce(thread_getref(t));
80 ctx->helpers[ctx->helper_count++] = t;
81 return t;
82}
83
84static bool test_helpers_teardown(struct test_context *ctx) {
85 if (!ctx || ctx->helper_count == 0)
86 return false;
87
88 bool leaked = false;
89 for (size_t i = 0; i < ctx->helper_count; i++) {
90 struct thread *t = ctx->helpers[i];
91 if (!t)
92 continue;
93 ctx->helpers[i] = NULL;
94 if (!(thread_get_flags(t) & THREAD_FLAG_JOINED)) {
95 leaked = true;
96 thread_alert(thread: t);
97 thread_join(t);
98 }
99 thread_putref(t);
100 }
101 ctx->helper_count = 0;
102 return leaked;
103}
104
105struct test_run_on_core_ctx {
106 void (*fn)(void *);
107 void *arg;
108};
109
110void test_fleet_report(struct test_fleet *f, const char *file, uint32_t line,
111 const char *fmt, ...) {
112
113 if (!once_token_claim(&f->failed)) {
114 test_fleet_stop(f, reason: TEST_STOP_FINDING);
115 return;
116 }
117
118 f->fail_file = file;
119 f->fail_line = line;
120
121 if (f->ctx) {
122 va_list args;
123 va_start(args, fmt);
124 vsnprintf(f->ctx->fail_msg, (int) sizeof(f->ctx->fail_msg), fmt, args);
125 va_end(args);
126 }
127
128 test_fleet_stop(f, reason: TEST_STOP_FINDING);
129}
130
131/* We first wait for the completion to bother us, then run */
132static void test_fleet_worker_main(void *arg) {
133 struct test_conc_worker *self = arg;
134
135 struct test_conc_worker *base = self - self->index;
136 struct test_fleet *f = container_of(base, struct test_fleet, workers[0]);
137
138 event_wait(e: &f->conc.start, type: THREAD_WAIT_UNINTERRUPTIBLE,
139 reason: THREAD_BLOCK_REASON_MANUAL);
140 atomic_inc_release(&f->started);
141
142 if (!test_conc_must_stop(c: &f->conc)) {
143 test_worker_fn body = f->bodies[self->index];
144 if (body)
145 body(f, self);
146 }
147
148 test_conc_unpark_self(c: &f->conc, w: self);
149 atomic_inc_release(&f->finished);
150}
151
152static void test_fleet_deadline(struct timer *timer) {
153 struct test_fleet *f = container_of(timer, struct test_fleet, hard_timer);
154
155 test_conc_publish_stop(c: &f->conc, reason: TEST_STOP_STALL);
156 panic("test fleet did not drain within deadline");
157}
158
159struct test_fleet *test_fleet_init_full(struct test_context *ctx,
160 const struct test_fleet_opts opts) {
161 if (cc_unlikely(!ctx))
162 panic("test_fleet_init: NULL test_context");
163
164 struct test_fleet *f = vnew(f, ALLOC_ZERO);
165 if (cc_unlikely(!f))
166 panic(OUT_OF_MEMORY, "test_fleet_init");
167
168 once_token_init(&f->failed);
169 f->ctx = ctx;
170 f->opts = opts;
171
172 if (!f->opts.soft_ms)
173 f->opts.soft_ms = TEST_FLEET_SOFT_MS;
174
175 if (!f->opts.drain_ms)
176 f->opts.drain_ms = TEST_FLEET_DRAIN_MS;
177
178 test_conc_init(c: &f->conc, workers: f->workers, worker_count: 0);
179
180 ctx->fleet = f;
181
182 return f;
183}
184
185static bool test_fleet_perturb_body(struct test_fleet *f,
186 struct test_conc_worker *self);
187
188static struct test_conc_worker *test_fleet_claim(struct test_fleet *f,
189 const char *role,
190 test_worker_fn body,
191 void *arg) {
192 /* Perturbers target workers[0, subject_count) */
193 enforce(!f->has_perturbers || body == test_fleet_perturb_body,
194 "fleet subjects must be spawned before perturbers");
195
196 if (f->count >= TEST_FLEET_MAX_WORKERS) {
197 f->spawn_failed = true;
198 return NULL;
199 }
200
201 struct test_conc_worker *w = &f->workers[f->count];
202 w->index = f->count;
203 w->role = role;
204 w->arg = arg;
205 w->rng.state = test_rng_seed_for(base_seed: f->ctx ? f->ctx->seed : 0, index: w->index);
206 f->bodies[f->count] = body;
207
208 f->count++;
209 f->conc.worker_count = f->count;
210 return w;
211}
212
213static bool test_fleet_attach(struct test_fleet *f, struct test_conc_worker *w,
214 struct thread *t) {
215 if (!t) {
216 f->spawn_failed = true;
217 f->count--;
218 f->conc.worker_count = f->count;
219 return false;
220 }
221
222 enforce(thread_getref(t));
223 atomic_store_release(&w->th, t);
224 return true;
225}
226
227struct test_conc_worker *
228test_fleet_spawn_full(struct test_fleet *f, const char *role,
229 test_worker_fn body,
230 const struct test_fleet_worker_opts opts) {
231 void *arg = opts.arg;
232 struct test_conc_worker *w = test_fleet_claim(f, role, body, arg);
233 if (!w)
234 return NULL;
235
236 enum thread_flags flags = opts.flags;
237 enum thread_prio_class prio_class = opts.prio_class;
238 cpu_id_t core_id = CPU_ID_MAX;
239
240 core_id = test_core_modulo(index: opts.core_id);
241
242 if (core_id != CPU_ID_MAX) {
243 struct thread *t = thread_create(
244 ("test_%s_%zu", role, w->index), test_fleet_worker_main, .arg = w,
245 .joinable = true, .allowed_cpus = cpu_mask_of(core_id),
246 .flags = flags, .pinned = f->opts.pin || opts.pinned,
247 .base_prio_class = prio_class);
248 if (!t) {
249 f->spawn_failed = true;
250 f->count--;
251 f->conc.worker_count = f->count;
252 return NULL;
253 }
254
255 enforce(thread_getref(t));
256 atomic_store_release(&w->th, t);
257 thread_enqueue_on_core(t, core_id);
258 return w;
259 }
260
261 struct thread *t = thread_spawn(
262 ("test_%s_%zu", role, w->index), test_fleet_worker_main, .arg = w,
263 .joinable = true, .flags = flags, .base_prio_class = prio_class);
264
265 return test_fleet_attach(f, w, t) ? w : NULL;
266}
267
268struct thread *test_fleet_worker_thread(struct test_fleet *f,
269 const char *role) {
270 if (!f || !role)
271 return NULL;
272
273 for (size_t i = 0; i < f->count; i++) {
274 if (f->workers[i].role && strcmp(str1: f->workers[i].role, str2: role) == 0)
275 return atomic_load_acq(&f->workers[i].th);
276 }
277
278 return NULL;
279}
280
281size_t test_fleet_spawn_per_core(struct test_fleet *f, const char *role,
282 test_worker_fn body, void *arg,
283 size_t per_core) {
284 size_t spawned = 0;
285
286 for (size_t rep = 0; rep < per_core; rep++) {
287 for (size_t cpu = 0; cpu < global.core_count; cpu++) {
288 if (!test_fleet_spawn(f, role, body, .arg = arg, .core_id = cpu,
289 .pinned = true))
290 return spawned;
291 spawned++;
292 }
293 }
294
295 return spawned;
296}
297
298static bool test_fleet_perturb_body(struct test_fleet *f,
299 struct test_conc_worker *self) {
300 const struct test_perturb_desc *desc = self->arg;
301 desc->fn(&f->perturb_env, self);
302 return true;
303}
304
305size_t test_fleet_spawn_perturbers(struct test_fleet *f,
306 const char *const *names,
307 int64_t victim_cpu) {
308 if (!f->has_perturbers) {
309 /* One victim for the whole fleet */
310 if (victim_cpu == TEST_PERTURB_VICTIM_ANY) {
311 uint64_t seed = test_rng_seed_for(base_seed: f->ctx ? f->ctx->seed : 0,
312 TEST_FLEET_MAX_WORKERS);
313 victim_cpu = (int64_t) (seed % global.core_count);
314 }
315
316 f->subject_count = f->count;
317 f->perturb_env = (struct test_perturb_env){
318 .conc = &f->conc,
319 .subject_count = f->subject_count,
320 .victim_cpu = victim_cpu,
321 };
322 f->has_perturbers = true;
323 }
324
325 size_t spawned = 0;
326 for (const char *const *name = names; *name; name++) {
327 const struct test_perturb_desc *desc = test_perturb_lookup(name: *name);
328 enforce(desc, "no such perturber");
329
330 test_fleet_spawn(f, desc->name, test_fleet_perturb_body,
331 .arg = (void *) desc);
332 spawned++;
333 }
334
335 return spawned;
336}
337
338void test_fleet_start_all(struct test_fleet *f) {
339 atomic_store_release(&f->conc.active, true);
340
341 if (f->opts.soft_ms)
342 test_liveness_start(state: &f->liveness, conc: &f->conc, threshold_ms: f->opts.soft_ms,
343 policy: TEST_ON_STALL_REPORT);
344
345 event_set(e: &f->conc.start);
346}
347
348#define TEST_FLEET_DUMP_ARMS 8
349
350static void test_fleet_dump_worker(const struct test_conc_worker *w) {
351 struct thread *t = atomic_load_acq(&w->th);
352
353 if (!t) {
354 test_err(" %s[%zu]: <not spawned>", w->role ? w->role : "worker",
355 w->index);
356 return;
357 }
358
359 struct scheduler *rq = atomic_load_acq(&t->scheduler);
360
361 test_err(" %s[%zu] '%s': state=%d flags=0x%x wait_type=%d parked=%d "
362 "rq=%ld",
363 w->role ? w->role : "worker", w->index, t->name,
364 (int) thread_get_state(t), (unsigned) thread_get_flags(t),
365 (int) thread_get_wait_type(t),
366 (int) atomic_load(&((struct test_conc_worker *) w)->parked),
367 rq ? (long) rq->core_id : -1L);
368
369 if (thread_get_state(t) == THREAD_STATE_ZOMBIE)
370 return;
371
372#ifdef DEBUG_THREAD
373 test_err(" %s[%zu] apcs_pending=0x%x apcs_ran=%u ctxsw=%zu",
374 w->role ? w->role : "worker", w->index,
375 (unsigned) atomic_load(&t->apc_pending_mask), t->total_apcs_ran,
376 t->context_switches);
377#endif
378
379 thread_dump_wait_trace(t, role: w->role ? w->role : "worker", idx: w->index,
380 TEST_FLEET_DUMP_ARMS);
381}
382
383static void test_fleet_dump(struct test_fleet *f,
384 const struct test_stall_evidence *ev) {
385 if (ev)
386 test_err("fleet stalled: silent_ms=%lu progress=%lu",
387 (unsigned long) ev->silent_ms, (unsigned long) ev->progress);
388
389 test_err("fleet: %zu worker(s), started=%zu finished=%zu stop=%s", f->count,
390 atomic_load(&f->started), atomic_load(&f->finished),
391 test_stop_to_str(atomic_load(&f->conc.stop)));
392
393 if (f->has_perturbers)
394 test_err("fleet: %zu subject(s), perturbers from [%zu], victim cpu %ld",
395 f->subject_count, f->subject_count,
396 (long) f->perturb_env.victim_cpu);
397
398 for (size_t i = 0; i < f->count; i++)
399 test_fleet_dump_worker(w: &f->workers[i]);
400}
401
402static void test_fleet_arm_hard_deadline(struct test_fleet *f) {
403 timer_init(timer: &f->hard_timer, func: test_fleet_deadline, NULL);
404 f->hard_timer.flags = TIMER_FLAG_IRQ;
405 f->hard_timer.expiration_us = time_get_us() + MS_TO_US(f->opts.drain_ms);
406 timer_add_global(timer: &f->hard_timer);
407 f->hard_armed = true;
408}
409
410static void test_fleet_disarm_hard_deadline(struct test_fleet *f) {
411 if (f->hard_armed) {
412 timer_shutdown_sync(timer: &f->hard_timer);
413 f->hard_armed = false;
414 }
415}
416
417static void test_fleet_join_range(struct test_fleet *f, size_t from,
418 size_t to) {
419 for (size_t i = from; i < to; i++) {
420 struct thread *t = atomic_load_acq(&f->workers[i].th);
421
422 if (t)
423 thread_join(t);
424 }
425}
426
427static void test_fleet_drain(struct test_fleet *f, enum test_stop reason) {
428 test_conc_publish_stop(c: &f->conc, reason);
429 test_fleet_arm_hard_deadline(f);
430
431 size_t subjects = f->has_perturbers ? f->subject_count : f->count;
432 test_fleet_join_range(f, from: 0, to: subjects);
433
434 /* Perturbers run until told to stop */
435 if (subjects < f->count) {
436 test_conc_publish_stop(c: &f->conc, reason: TEST_STOP_BUDGET);
437 test_fleet_join_range(f, from: subjects, to: f->count);
438 }
439
440 test_fleet_disarm_hard_deadline(f);
441}
442
443static void test_fleet_thread_putref(struct test_fleet *f) {
444 for (size_t i = 0; i < f->count; i++) {
445 struct thread *t = atomic_xchg_acq_rel(&f->workers[i].th, NULL);
446 if (t)
447 thread_putref(t);
448 }
449
450 atomic_store_release(&f->conc.active, false);
451}
452
453struct test_verdict test_fleet_join(struct test_fleet *f) {
454 if (!atomic_load_acq(&f->conc.active))
455 test_fleet_start_all(f);
456
457 bool spawn_failed = f->spawn_failed;
458
459 test_fleet_drain(f, reason: f->opts.run_until_stop ? TEST_STOP_BUDGET : TEST_RUN);
460
461 struct test_stall_evidence ev;
462 bool stalled = test_liveness_take(state: &f->liveness, out: &ev);
463 test_liveness_stop(state: &f->liveness);
464
465 bool failed = once_token_claimed(&f->failed);
466 const char *fail_file = f->fail_file;
467 uint32_t fail_line = f->fail_line;
468
469 if (stalled || failed || spawn_failed)
470 test_fleet_dump(f, ev: stalled ? &ev : NULL);
471
472 test_fleet_thread_putref(f);
473 f->joined = true;
474
475 if (spawn_failed)
476 return TEST_FAIL("fleet could not spawn all workers");
477
478 if (failed) {
479 test_err("worker failure at %s:%u: %s", fail_file ? fail_file : "?",
480 fail_line, f->ctx ? f->ctx->fail_msg : "");
481 return TEST_FAIL(f->ctx ? f->ctx->fail_msg : "worker failed");
482 }
483
484 if (stalled)
485 return TEST_FAIL("fleet stalled: no progress before the soft deadline");
486
487 return TEST_SUCCESS;
488}
489
490bool test_fleet_teardown(struct test_context *ctx) {
491 struct test_fleet *f = ctx->fleet;
492 if (!f)
493 return false;
494
495 ctx->fleet = NULL;
496
497 bool leaked = !f->joined && f->count > 0;
498 if (leaked) {
499 if (!atomic_load_acq(&f->conc.active))
500 test_fleet_start_all(f);
501
502 test_fleet_drain(f, reason: TEST_STOP_BUDGET);
503 test_liveness_stop(state: &f->liveness);
504 test_fleet_thread_putref(f);
505 }
506
507 vfree(f);
508 return leaked;
509}
510
511bool test_context_teardown(struct test_context *ctx) {
512 if (!ctx)
513 return false;
514
515 bool leaked_fleet = test_fleet_teardown(ctx);
516 bool leaked_helpers = test_helpers_teardown(ctx);
517 test_alloc_teardown(ctx);
518 return leaked_fleet || leaked_helpers;
519}
520
521static void test_run_on_core_main(void *arg) {
522 struct test_run_on_core_ctx *c = arg;
523 c->fn(c->arg);
524}
525
526/* TODO: there is probably a nicer way to go about a "bother random
527 * CPU to run arbitrary pinned code in thread context and wait for
528 * it to finish". We have system workqueues per-CPU, but we have no
529 * flush_work() equivalent. Once we get that, this will be much
530 * cleaner and can just be a workqueue call wrapper */
531bool test_run_on_core(cpu_id_t core_id, void (*fn)(void *), void *arg) {
532 struct test_run_on_core_ctx c = {.fn = fn, .arg = arg};
533
534 struct thread *t = thread_spawn(
535 ("test_on_core_%zu", core_id), test_run_on_core_main, .arg = &c,
536 .joinable = true, .allowed_cpus = cpu_mask_of(core_id), .pinned = true,
537 .on_cpu = core_id);
538 if (!t)
539 return false;
540
541 thread_join(t);
542 return true;
543}
544
545#endif /* TEST_ENABLED */
546