1#include <sch/sched.h>
2#include <smp/core.h>
3#include <test/conc.h>
4#include <thread/thread.h>
5#include <thread/wait.h>
6#include <time/time.h>
7
8#if defined(TEST_ENABLED) || defined(TEST_NIGHTMARE_ENABLED)
9
10PERCPU_DEFINE(struct test_progress_counter, test_progress_store);
11PERCPU_EXPORT_AS(test_progress, test_progress_store);
12
13uint64_t test_conc_progress_sum(void) {
14 uint64_t sum = 0;
15 struct test_progress_counter *counter;
16 percpu_for_each(counter, test_progress) {
17 sum += atomic_load_rlx(&counter->count);
18 }
19
20 return sum;
21}
22
23uint64_t test_rng_seed_for(uint64_t base_seed, size_t index) {
24 struct test_rng rng = {.state = base_seed ^
25 (PRNG_SPLITMIX64_GAMMA * (index + 1))};
26 return test_rng_next(rng: &rng);
27}
28
29void test_conc_init(struct test_conc *c, struct test_conc_worker *workers,
30 size_t worker_count) {
31 atomic_store_rlx(&c->stop, TEST_RUN);
32 atomic_store_rlx(&c->quiesce_requested, false);
33 atomic_store_rlx(&c->parked_count, 0);
34 atomic_store_rlx(&c->aux, NULL);
35 c->workers = workers;
36 c->worker_count = worker_count;
37 event_init(e: &c->start, type: EVENT_NOTIFICATION, false);
38 atomic_store_release(&c->active, false);
39}
40
41void test_conc_publish_stop(struct test_conc *c, enum test_stop reason) {
42 enum test_stop observed = atomic_load_acq(&c->stop);
43 bool advanced = false;
44
45 while (observed < reason) {
46 if (atomic_cas_weak(&c->stop, &observed, reason, mo_acq_rel,
47 mo_acquire)) {
48 advanced = true;
49 break;
50 }
51 }
52
53 if (!advanced || observed != TEST_RUN)
54 return;
55
56 for (size_t i = 0; i < c->worker_count; i++) {
57 struct thread *thread = atomic_load_acq(&c->workers[i].th);
58
59 if (thread)
60 thread_alert(thread);
61 }
62
63 struct thread *aux = atomic_load_acq(&c->aux);
64 if (aux)
65 thread_alert(thread: aux);
66}
67
68void test_conc_park(struct test_conc *c, struct test_conc_worker *w) {
69 bool was_parked = atomic_xchg_acq_rel(&w->parked, true);
70 if (!was_parked)
71 atomic_inc_release(&c->parked_count);
72
73 while (test_conc_must_park(c) && !test_conc_must_stop(c))
74 scheduler_yield();
75
76 if (!was_parked)
77 test_conc_unpark_self(c, w);
78}
79
80void test_conc_unpark_self(struct test_conc *c, struct test_conc_worker *w) {
81 if (!atomic_load_acq(&w->parked))
82 return;
83
84 atomic_dec_release(&c->parked_count);
85 atomic_store_release(&w->parked, false);
86}
87
88static bool test_liveness_eval(struct test_liveness_state *state) {
89 if (atomic_load_acq(&state->phase) != TEST_LIVE_ARMED)
90 return false;
91
92 bool quiesce = test_conc_must_park(c: state->conc);
93 uint64_t current_progress = test_conc_progress_sum();
94 time_ms_t now_ms = time_get_ms();
95 if (quiesce) {
96 state->last_progress = current_progress;
97 state->last_change_ms = now_ms;
98 state->was_quiesced = true;
99 return false;
100 }
101
102 if (state->was_quiesced) {
103 state->last_progress = current_progress;
104 state->last_change_ms = now_ms;
105 state->was_quiesced = false;
106 return false;
107 }
108
109 if (current_progress != state->last_progress) {
110 state->last_progress = current_progress;
111 state->last_change_ms = now_ms;
112 return false;
113 }
114
115 if (now_ms < state->last_change_ms ||
116 now_ms - state->last_change_ms < state->threshold_ms)
117 return false;
118
119 state->pending = (struct test_stall_evidence){
120 .silent_ms = now_ms - state->last_change_ms,
121 .progress = current_progress,
122 };
123
124 enum test_liveness_phase expected = TEST_LIVE_ARMED;
125 if (!atomic_cas_strong(&state->phase, &expected, TEST_LIVE_PENDING,
126 mo_acq_rel, mo_acquire))
127 return false;
128
129 return true;
130}
131
132static void test_liveness_watchdog_cb(struct watchdog_callback *cb) {
133 struct test_liveness_state *state =
134 container_of(cb, struct test_liveness_state, callback);
135 struct test_conc *c = state->conc;
136
137 if (!c)
138 return;
139
140 if (!atomic_load_acq(&c->active))
141 return;
142
143 if (test_conc_stop_reason(c) != TEST_RUN)
144 return;
145
146 if (test_liveness_eval(state))
147 test_conc_publish_stop(c, reason: TEST_STOP_STALL);
148}
149
150bool test_liveness_start(struct test_liveness_state *state,
151 struct test_conc *conc, time_ms_t threshold_ms,
152 enum test_on_stall policy) {
153 *state = (struct test_liveness_state){
154 .callback = {.fn = test_liveness_watchdog_cb},
155 .phase = ATOMIC_VAR_INIT(TEST_LIVE_ARMED),
156 .last_progress = test_conc_progress_sum(),
157 .last_change_ms = time_get_ms(),
158 .threshold_ms = threshold_ms,
159 .was_quiesced = false,
160 .policy = policy,
161 .coordinator_cpu = 0,
162 .registered = false,
163 .conc = conc,
164 };
165
166 watchdog_callback_add(cpu: state->coordinator_cpu, cb: &state->callback);
167 state->registered = true;
168 return true;
169}
170
171void test_liveness_stop(struct test_liveness_state *state) {
172 if (state->registered) {
173 watchdog_callback_remove(cpu: state->coordinator_cpu, cb: &state->callback);
174 state->registered = false;
175 }
176
177 atomic_store_release(&state->phase, TEST_LIVE_OFF);
178}
179
180bool test_liveness_take(struct test_liveness_state *state,
181 struct test_stall_evidence *out) {
182 enum test_liveness_phase expected = TEST_LIVE_PENDING;
183 if (!atomic_cas_strong(&state->phase, &expected, TEST_LIVE_REPORTED,
184 mo_acq_rel, mo_acquire))
185 return false;
186
187 if (out)
188 *out = state->pending;
189 return true;
190}
191
192#endif /* TEST_ENABLED || TEST_NIGHTMARE_ENABLED */
193