1#include <sch/sched.h>
2#include <sync/mutex.h>
3#include <test/fleet.h>
4#include <thread/thread.h>
5#include <time/spin_sleep.h>
6
7TEST_GROUP_DEFINE(fleet);
8
9struct counting_fix {
10 atomic_uint32_t ran;
11 atomic_uint32_t saw_gate_open;
12};
13
14static bool counting_worker(struct test_fleet *f, struct test_conc_worker *w) {
15 struct counting_fix *fix = w->arg;
16 cc_ignore(f);
17
18 atomic_inc(&fix->ran);
19 return true;
20}
21
22TEST_DEFINE_INTEGRATION(fleet, spawn_join_clean,
23 TEST_INTENSITY_LINEAR(2, 8, 64, "workers")) {
24 struct test_fleet *fleet = test_fleet_init(ctx);
25
26 struct counting_fix *fix = test_new(fix);
27
28 size_t n = ctx->intensity_val ? ctx->intensity_val : 8;
29 for (size_t i = 0; i < n; i++)
30 test_fleet_spawn(fleet, "count", counting_worker, .arg = fix);
31
32 test_fleet_start_all(f: fleet);
33
34 struct test_verdict v = test_fleet_join(f: fleet);
35 if (v.result != TEST_RESULT_OK)
36 return v;
37
38 TEST_ASSERT_EQ(atomic_load(&fix->ran), n);
39 return TEST_SUCCESS;
40}
41
42struct gate_fix {
43 atomic_bool released;
44 atomic_uint32_t jumped_early;
45 atomic_uint32_t ran;
46};
47
48static bool gate_worker(struct test_fleet *f, struct test_conc_worker *w) {
49 struct gate_fix *fix = w->arg;
50 cc_ignore(f);
51
52 if (!atomic_load(&fix->released))
53 atomic_inc(&fix->jumped_early);
54
55 atomic_inc(&fix->ran);
56 return true;
57}
58
59TEST_DEFINE_INTEGRATION(fleet, start_gate_holds,
60 TEST_INTENSITY_LINEAR(2, 16, 64, "workers")) {
61 struct test_fleet *fleet = test_fleet_init(ctx);
62
63 struct gate_fix *fix = test_new(fix);
64
65 size_t n = ctx->intensity_val ? ctx->intensity_val : 16;
66 for (size_t i = 0; i < n; i++)
67 test_fleet_spawn(fleet, "gate", gate_worker, .arg = fix);
68
69 sleep_spin_ms(msec: 20);
70 TEST_ASSERT_EQ(atomic_load(&fix->ran), 0);
71
72 atomic_store(&fix->released, true);
73 test_fleet_start_all(f: fleet);
74
75 struct test_verdict v = test_fleet_join(f: fleet);
76 if (v.result != TEST_RESULT_OK)
77 return v;
78
79 TEST_ASSERT_EQ(atomic_load(&fix->jumped_early), 0);
80 TEST_ASSERT_EQ(atomic_load(&fix->ran), n);
81 return TEST_SUCCESS;
82}
83
84static bool failing_worker(struct test_fleet *f, struct test_conc_worker *w) {
85 cc_ignore(w);
86 TEST_WORKER_CHECK_EQ(f, 2 + 2, 5);
87 return true;
88}
89
90TEST_DEFINE_INTEGRATION(fleet, worker_failure_folds) {
91 struct test_fleet *fleet = test_fleet_init(ctx);
92
93 test_fleet_spawn(fleet, "fail", failing_worker);
94 test_fleet_start_all(f: fleet);
95
96 struct test_verdict v = test_fleet_join(f: fleet);
97
98 TEST_ASSERT_EQ(v.result, TEST_RESULT_FAILED);
99 TEST_ASSERT_NONNULL(v.msg);
100 TEST_ASSERT(strstr(v.msg, "2 + 2") != NULL);
101 TEST_ASSERT(strstr(v.msg, "5") != NULL);
102 return TEST_SUCCESS;
103}
104
105struct lock_fix {
106 struct mutex lock;
107 atomic_bool second_acquired;
108};
109
110static bool lock_failing_worker(struct test_fleet *f,
111 struct test_conc_worker *w) {
112 struct lock_fix *fix = w->arg;
113
114 mutex_lock(&fix->lock);
115 TEST_WORKER_CHECK_GOTO(f, false, out);
116 mutex_unlock(&fix->lock);
117 return true;
118
119out:
120 mutex_unlock(&fix->lock);
121 return false;
122}
123
124TEST_DEFINE_INTEGRATION(fleet, worker_check_releases_lock) {
125 struct test_fleet *fleet = test_fleet_init(ctx);
126
127 struct lock_fix *fix = test_new(fix);
128 mutex_init(&fix->lock);
129
130 test_fleet_spawn(fleet, "locker", lock_failing_worker, .arg = fix);
131 test_fleet_start_all(f: fleet);
132
133 struct test_verdict v = test_fleet_join(f: fleet);
134 TEST_ASSERT_EQ(v.result, TEST_RESULT_FAILED);
135
136 mutex_lock(&fix->lock);
137 atomic_store(&fix->second_acquired, true);
138 mutex_unlock(&fix->lock);
139
140 TEST_ASSERT(atomic_load(&fix->second_acquired));
141 return TEST_SUCCESS;
142}
143
144static bool racing_failer(struct test_fleet *f, struct test_conc_worker *w) {
145 TEST_WORKER_CHECK_MSG(f, false, "worker %zu", w->index);
146 return true;
147}
148
149TEST_DEFINE_INTEGRATION(fleet, first_failure_wins,
150 TEST_INTENSITY_LINEAR(2, 16, 64, "workers")) {
151 struct test_fleet *fleet = test_fleet_init(ctx);
152
153 size_t n = ctx->intensity_val ? ctx->intensity_val : 16;
154 for (size_t i = 0; i < n; i++)
155 test_fleet_spawn(fleet, "racer", racing_failer);
156
157 test_fleet_start_all(f: fleet);
158 struct test_verdict v = test_fleet_join(f: fleet);
159
160 TEST_ASSERT_EQ(v.result, TEST_RESULT_FAILED);
161 TEST_ASSERT_NONNULL(v.msg);
162 TEST_ASSERT(strstr(v.msg, "worker ") != NULL);
163 return TEST_SUCCESS;
164}
165
166struct alloc_fix {
167 uint64_t canary;
168 atomic_uint32_t seen;
169};
170
171#define FLEET_CANARY UINT64_C(0x5A5AC0FFEE5A5A11)
172
173static bool canary_worker(struct test_fleet *f, struct test_conc_worker *w) {
174 struct alloc_fix *fix = w->arg;
175
176 for (size_t i = 0; i < 64; i++) {
177 TEST_WORKER_CHECK_EQ(f, fix->canary, FLEET_CANARY);
178 scheduler_yield();
179 }
180 atomic_inc(&fix->seen);
181 return true;
182}
183
184TEST_DEFINE_INTEGRATION(fleet, alloc_outlives_body,
185 TEST_INTENSITY_LINEAR(2, 8, 32, "workers")) {
186 struct test_fleet *fleet = test_fleet_init(ctx);
187
188 struct alloc_fix *fix = test_new(fix);
189 fix->canary = FLEET_CANARY;
190
191 size_t n = ctx->intensity_val ? ctx->intensity_val : 8;
192 for (size_t i = 0; i < n; i++)
193 test_fleet_spawn(fleet, "canary", canary_worker, .arg = fix);
194
195 test_fleet_start_all(f: fleet);
196 struct test_verdict v = test_fleet_join(f: fleet);
197 if (v.result != TEST_RESULT_OK)
198 return v;
199
200 TEST_ASSERT_EQ(atomic_load(&fix->seen), n);
201 TEST_ASSERT_EQ(fix->canary, FLEET_CANARY);
202 return TEST_SUCCESS;
203}
204
205/* ==================== affinity ==================== */
206
207struct affinity_fix {
208 atomic_uint32_t wrong_core;
209 atomic_uint32_t checked;
210};
211
212static void affinity_probe(void *arg) {
213 struct affinity_fix *fix = arg;
214 atomic_inc(&fix->checked);
215}
216
217TEST_DEFINE_INTEGRATION(fleet, run_on_core, .min_cores = 2) {
218 struct affinity_fix fix = {0};
219
220 for (size_t cpu = 0; cpu < global.core_count && cpu < 4; cpu++)
221 TEST_ASSERT(test_run_on_core(cpu, affinity_probe, &fix));
222
223 size_t expected = global.core_count < 4 ? global.core_count : 4;
224 TEST_ASSERT_EQ(atomic_load(&fix.checked), expected);
225 TEST_ASSERT_EQ(atomic_load(&fix.wrong_core), 0);
226 return TEST_SUCCESS;
227}
228
229static bool pinned_worker(struct test_fleet *f, struct test_conc_worker *w) {
230 struct affinity_fix *fix = w->arg;
231
232 TEST_WORKER_CHECK(f, thread_is_pinned(thread_get_current()));
233 atomic_inc(&fix->checked);
234 return true;
235}
236
237TEST_DEFINE_INTEGRATION(fleet, spawn_on_core_pins, .min_cores = 2) {
238 struct test_fleet *fleet = test_fleet_init(ctx);
239
240 struct affinity_fix *fix = test_new(fix);
241
242 size_t n =
243 test_fleet_spawn_per_core(f: fleet, role: "pinned", body: pinned_worker, arg: fix, per_core: 1);
244 TEST_ASSERT_EQ(n, global.core_count);
245
246 test_fleet_start_all(f: fleet);
247 struct test_verdict v = test_fleet_join(f: fleet);
248 if (v.result != TEST_RESULT_OK)
249 return v;
250
251 TEST_ASSERT_EQ(atomic_load(&fix->checked), n);
252 return TEST_SUCCESS;
253}
254