1#include <compiler/intrinsic.h>
2#include <sch/sched.h>
3#include <thread/apc.h>
4#include <thread/thread.h>
5
6#include "sch/internal.h"
7
8/* Lock ordering here:
9 *
10 * wait_lock -> one wait_header.lock at a time, runqueues -> thread
11 *
12 * APCs can run a little bit when we're going to complete a wait start,
13 * and alerts only unlink interruptible registrations
14 */
15static inline uint64_t wait_epoch(struct thread *t) {
16 return atomic_load_rlx(&t->wait_epoch);
17}
18
19static void unlink_blocks(struct thread *t, uint16_t got) {
20 for (size_t i = 0; i < t->wait_block_count; i++) {
21 struct thread_wait_block *b = &t->active_wait_blocks[i];
22
23 if (b->state != THREAD_WAIT_BLOCK_ACTIVE)
24 continue;
25
26 enum irql irql = spin_lock_high(&b->header->lock);
27
28 list_node_delete_init(n: &b->obj_node);
29 b->state =
30 i == got ? THREAD_WAIT_BLOCK_SATISFIED : THREAD_WAIT_BLOCK_INACTIVE;
31
32 spin_unlock(&b->header->lock, irql);
33 }
34}
35
36static inline enum thread_resume_reason manual_reason(struct thread *t) {
37 return t->wait_state == THREAD_STATE_SLEEPING
38 ? THREAD_WAKE_REASON_SLEEP_MANUAL
39 : THREAD_WAKE_REASON_BLOCKING_MANUAL;
40}
41
42static void deliver_alert_locked(struct thread *t) {
43 if (!t->alert_pending || !t->active_wait_blocks || t->wait_done ||
44 atomic_load_rlx(&t->wait_type) != THREAD_WAIT_INTERRUPTIBLE)
45 return;
46
47 unlink_blocks(t, THREAD_WAIT_KEY_NONE);
48
49 t->alert_pending = false;
50 t->wait_done = true;
51 t->wait_status = THREAD_WAIT_ALERTED;
52 t->wait_key = THREAD_WAIT_KEY_NONE;
53 t->wait_completion_reason = manual_reason(t);
54
55 scheduler_complete_object_wait(t, reason: t->wait_completion_reason);
56}
57
58/* If an object already got signaled at registration,
59 * we'll need to shrink the count so unlink_blocks
60 * doesn't read the rest, because only blocks [0, key)
61 * would be linked right now */
62static void satisfy_at_prepare(struct thread *t, uint16_t key) {
63 SPINLOCK_ASSERT_HELD(&t->wait_lock);
64 t->active_wait_blocks[key].state = THREAD_WAIT_BLOCK_INACTIVE;
65 t->wait_block_count = key + 1;
66 unlink_blocks(t, got: key);
67 t->active_wait_blocks[key].state = THREAD_WAIT_BLOCK_SATISFIED;
68
69 t->wait_key = key;
70 t->wait_done = true;
71 t->wait_status = THREAD_WAIT_SATISFIED;
72 t->wait_completion_reason = manual_reason(t);
73
74 scheduler_complete_object_wait(t, reason: t->wait_completion_reason);
75}
76
77static void prepare_wait(const struct thread_wait_object *objects, size_t count,
78 struct thread_wait_block *storage,
79 enum thread_wait_type type, thread_act_reason_t reason,
80 enum thread_state state) {
81 enforce_precond(count <= UINT16_MAX, CAPACITY_EXCEEDED);
82 enforce_precond(type != THREAD_WAIT_NONE, "wait type must be provided");
83 enforce_precond(storage || count <= THREAD_NUM_WAIT_BLOCKS,
84 CAPACITY_EXCEEDED);
85
86 struct thread *t = thread_get_current();
87 if (!storage)
88 storage = t->wait_blocks; /* TODO: establish identities for the 4
89 * wait blocks, we'll take 0 for now */
90
91 enum irql irql = spin_lock_high(&t->wait_lock);
92 enforce(!t->active_wait_blocks);
93 t->active_wait_blocks = storage;
94 t->wait_block_count = count;
95 t->wait_key = THREAD_WAIT_KEY_NONE;
96 t->wait_done = false;
97 t->wait_status = THREAD_WAIT_SATISFIED;
98 atomic_inc_rlx(&t->wait_epoch);
99
100 t->wait_state = state;
101
102 enum irql tirql = spin_lock_high(&t->lock);
103
104 enforce(!t->object_wait);
105 t->object_wait = true;
106
107 atomic_store_rlx(&t->wait_type, type);
108 thread_clear_flag(t, flag: THREAD_FLAG_YIELDED);
109
110 thread_diag_record_arm(t, site: "wait_blocks", ci_return_address(0),
111 object: count ? objects[0].object : NULL, state, type,
112 reason);
113
114 /* Registration visible now, but we need to stay runnable so callers
115 * can do their work before completeing */
116 if (state == THREAD_STATE_SLEEPING) {
117 thread_add_sleep_reason(t, reason);
118 } else {
119 thread_add_block_reason(t, reason);
120 }
121
122 thread_update_runtime_buckets(thread: t, time: time_get_ms());
123
124 spin_unlock(&t->lock, tirql);
125
126 for (size_t i = 0; i < count; i++) {
127 struct thread_wait_block *b = &storage[i];
128 const struct thread_wait_object *obj = &objects[i];
129
130 enforce(obj->header);
131 *b = (struct thread_wait_block){.thread = t,
132 .object = obj->object,
133 .key = i,
134 .state = THREAD_WAIT_BLOCK_ACTIVE,
135 .header = objects[i].header};
136
137 /* We put the checking and linking into one section here so
138 * signalers are only able to see us linked or we see their signal */
139 enum irql hirql = spin_lock_high(&b->header->lock);
140
141 bool acquired = obj->try_acquire && obj->try_acquire(obj->object);
142
143 if (!acquired)
144 list_head_push_back(h: &b->header->waiters, n: &b->obj_node);
145
146 spin_unlock(&b->header->lock, hirql);
147
148 if (acquired) {
149 satisfy_at_prepare(t, key: i);
150 break;
151 }
152 }
153
154 deliver_alert_locked(t);
155 spin_unlock(&t->wait_lock, irql);
156}
157
158void thread_wait_prepare(const struct thread_wait_object *objs, size_t count,
159 struct thread_wait_block *storage,
160 enum thread_wait_type type,
161 enum thread_block_reason reason) {
162 prepare_wait(objects: objs, count, storage, type, reason, state: THREAD_STATE_BLOCKED);
163}
164
165void thread_wait_prepare_to_sleep(struct thread_wait_header *hdr, void *object,
166 enum thread_wait_type type) {
167 const struct thread_wait_object obj = {.header = hdr, .object = object};
168 prepare_wait(objects: &obj, /* count = */ 1, NULL, type, reason: THREAD_SLEEP_REASON_MANUAL,
169 state: THREAD_STATE_SLEEPING);
170}
171
172void thread_wait_prepare_one(struct thread_wait_header *hdr, void *object,
173 enum thread_wait_type type,
174 enum thread_block_reason reason) {
175 const struct thread_wait_object obj = {.header = hdr, .object = object};
176 thread_wait_prepare(objs: &obj, /* count = */ 1, NULL, type, reason);
177}
178
179static bool satisfy(struct thread *t, uint64_t epoch, uint16_t key,
180 enum thread_resume_reason reason,
181 void (*callback)(struct thread *)) {
182 if (!t->active_wait_blocks || wait_epoch(t) != epoch ||
183 key >= t->wait_block_count ||
184 t->active_wait_blocks[key].state != THREAD_WAIT_BLOCK_ACTIVE)
185 return false;
186
187 SPINLOCK_ASSERT_HELD(&t->wait_lock);
188 unlink_blocks(t, got: key);
189 t->wait_key = key;
190 t->wait_done = true;
191 t->wait_status = THREAD_WAIT_SATISFIED;
192 t->wait_completion_reason = reason;
193
194 if (callback)
195 callback(t);
196
197 scheduler_complete_object_wait(t, reason);
198 return true;
199}
200
201bool thread_wait_satisfy(struct thread_wait_block *b,
202 enum thread_resume_reason reason,
203 void (*cb)(struct thread *)) {
204 struct thread *t = b->thread;
205
206 enum irql irql = spin_lock_high(&t->wait_lock);
207
208 uint16_t key = b->key;
209
210 /* Has to be a block in the thread's storage */
211 enforce(t->active_wait_blocks);
212 enforce(b->state != THREAD_WAIT_BLOCK_SATISFIED);
213
214 bool curr = key < t->wait_block_count && b == &t->active_wait_blocks[key];
215
216 bool won = curr && satisfy(t, epoch: wait_epoch(t), key, reason, callback: cb);
217
218 spin_unlock(&t->wait_lock, irql);
219
220 return won;
221}
222
223bool thread_wait_satisfy_epoch(struct thread_wait_block *b, uint64_t epoch,
224 enum thread_resume_reason reason,
225 void (*callback)(struct thread *)) {
226 struct thread *t = b->thread;
227
228 enum irql irql = spin_lock_high(&t->wait_lock);
229 bool won = satisfy(t, epoch, key: b->key, reason, callback);
230 spin_unlock(&t->wait_lock, irql);
231
232 return won;
233}
234
235struct thread *thread_wait_header_satisfy_one(struct thread_wait_header *hdr,
236 enum thread_resume_reason reason,
237 void (*callback)(struct thread *),
238 void (*empty_locked)(void *),
239 void *arg) {
240 while (true) {
241 enum irql irql = spin_lock_high(&hdr->lock);
242
243 if (list_head_empty(h: &hdr->waiters)) {
244 if (empty_locked)
245 empty_locked(arg);
246
247 spin_unlock(&hdr->lock, irql);
248 return NULL;
249 }
250
251 struct thread_wait_block *b =
252 list_first_entry(&hdr->waiters, struct thread_wait_block, obj_node);
253
254 /* b is linked, this is b's wait */
255 struct thread *t = b->thread;
256 uint64_t epoch = wait_epoch(t);
257 uint16_t key = b->key;
258
259 enforce(thread_getref(t), STALE_REFERENCE);
260 spin_unlock(&hdr->lock, irql);
261
262 irql = spin_lock_high(&t->wait_lock);
263 bool won = satisfy(t, epoch, key, reason, callback);
264 spin_unlock(&t->wait_lock, irql);
265
266 thread_putref(t);
267 if (won)
268 return t;
269 }
270}
271
272struct thread *thread_wait_header_satisfy(struct thread_wait_header *hdr,
273 enum thread_resume_reason reason,
274 void (*callback)(struct thread *)) {
275 return thread_wait_header_satisfy_one(hdr, reason, callback, NULL, NULL);
276}
277
278size_t thread_wait_header_satisfy_all(struct thread_wait_header *hdr,
279 enum thread_resume_reason reason,
280 void (*cb)(struct thread *),
281 void (*before_lock)(void *), void *arg) {
282 struct list_head snapshot;
283 list_head_init(h: &snapshot);
284
285 enum irql irql = spin_lock_high(&hdr->lock);
286
287 if (before_lock)
288 before_lock(arg);
289
290 list_head_splice_back_init(dst: &snapshot, src: &hdr->waiters);
291
292 /* We have to still keep hdr->lock held until unlock so
293 * other unlinks are correct while snapshot is on our stack */
294 size_t woken = 0;
295 while (!list_head_empty(h: &snapshot)) {
296
297 struct thread_wait_block *b =
298 list_first_entry(&snapshot, struct thread_wait_block, obj_node);
299
300 struct thread *t = b->thread;
301 uint64_t epoch = wait_epoch(t);
302 uint16_t key = b->key;
303
304 list_node_delete_init(n: &b->obj_node);
305 bool last = list_head_empty(h: &snapshot);
306
307 enforce(thread_getref(t), STALE_REFERENCE);
308 spin_unlock(&hdr->lock, irql);
309
310 enum irql wirql = spin_lock_high(&t->wait_lock);
311
312 if (satisfy(t, epoch, key, reason, callback: cb))
313 woken++;
314
315 spin_unlock(&t->wait_lock, wirql);
316 thread_putref(t);
317
318 if (last)
319 return woken;
320
321 irql = spin_lock_high(&hdr->lock);
322 }
323
324 spin_unlock(&hdr->lock, irql);
325 return woken;
326}
327
328static void finish_locked(struct thread *t) {
329 t->active_wait_blocks = NULL;
330 t->wait_block_count = 0;
331
332 enum irql irql = spin_lock_high(&t->lock);
333
334 t->object_wait = false;
335 atomic_store_rlx(&t->wait_type, THREAD_WAIT_NONE);
336 enum thread_state state = thread_get_state(t);
337
338 if (state == THREAD_STATE_BLOCKED || state == THREAD_STATE_SLEEPING)
339 thread_set_state(t, state: THREAD_STATE_RUNNING);
340
341 spin_unlock(&t->lock, irql);
342}
343
344struct thread_wait_result thread_wait_complete(void) {
345 struct thread *t = thread_get_current();
346
347 /* Whether the previous loop ran any APCs. We need this because
348 * a delivery pass would leave the thread RUNNING. The
349 * pass after one that delivers has to go actually block. */
350 bool delivered = false;
351
352 while (true) {
353 enum irql irql = spin_lock_high(&t->wait_lock);
354 enforce(t->active_wait_blocks);
355
356 if (t->wait_done) {
357 struct thread_wait_result result = {.status = t->wait_status,
358 .key = t->wait_key,
359 .reason =
360 t->wait_completion_reason};
361 finish_locked(t);
362
363 /* This is where we tunnel out of the wait.
364 *
365 * TODO: I really need to consider refactoring these random odd
366 * hooks into a set of generic functions that I can gradually expand */
367 wroost_post_wait(curr: t);
368 spin_unlock(&t->wait_lock, irql);
369 return result;
370 }
371
372 enum irql tirql = spin_lock_high(&t->lock);
373
374 bool apc =
375 atomic_load_rlx(&t->wait_type) == THREAD_WAIT_INTERRUPTIBLE &&
376 !delivered && atomic_load(&t->apc_pending_mask) != 0;
377
378 thread_clear_flag(t, flag: THREAD_FLAG_YIELDED);
379
380 thread_set_state(t, state: apc ? THREAD_STATE_RUNNING : t->wait_state);
381
382 thread_update_runtime_buckets(thread: t, time: time_get_ms());
383
384 spin_unlock(&t->lock, tirql);
385 spin_unlock(&t->wait_lock, irql);
386
387 if (apc)
388 apc_check_and_deliver(t);
389
390 delivered = apc;
391
392 scheduler_yield();
393 }
394}
395
396static void wait_timeout_send(struct timer *timer) {
397 struct thread *t = container_of(timer, struct thread, wait_timer);
398
399 enum irql irql = spin_lock_high(&t->wait_lock);
400
401 if (!t->wait_done && wait_epoch(t) == t->wait_timer_epoch) {
402 unlink_blocks(t, THREAD_WAIT_KEY_NONE);
403 t->wait_done = true;
404 t->wait_status = THREAD_WAIT_TIMED_OUT;
405 t->wait_key = THREAD_WAIT_KEY_NONE;
406 t->wait_completion_reason = THREAD_WAKE_REASON_SLEEP_TIMEOUT;
407 scheduler_complete_object_wait(t, reason: t->wait_completion_reason);
408 }
409
410 spin_unlock(&t->wait_lock, irql);
411}
412
413struct thread_wait_result thread_wait_complete_timeout(time_ms_t timeout_ms) {
414 struct thread *t = thread_get_current();
415
416 enum irql irql = spin_lock_high(&t->wait_lock);
417 enforce(t->active_wait_blocks);
418
419 bool armed = !t->wait_done;
420 t->wait_timer_epoch = atomic_load_rlx(&t->wait_epoch);
421
422 spin_unlock(&t->wait_lock, irql);
423
424 if (armed) {
425 timer_init(timer: &t->wait_timer, func: wait_timeout_send, NULL);
426
427 /* TOPC_NONE because it is not imperative that we
428 * have the timer run on this CPU */
429 t->wait_timer.flags = TIMER_FLAG_IRQ | TIMER_FLAG_PINNED |
430 TIMER_FLAG_CPU(smp_id(TOPC_NONE));
431
432 timer_modify(timer: &t->wait_timer, new: timer_delta_us(MS_TO_US(timeout_ms)));
433 }
434
435 struct thread_wait_result result = thread_wait_complete();
436
437 /* Gotta remove it since new waits will reuse this timer */
438 if (armed)
439 timer_delete_sync(timer: &t->wait_timer);
440
441 return result;
442}
443
444void thread_wait_cancel(void) {
445 struct thread *t = thread_get_current();
446 enum irql irql = spin_lock_high(&t->wait_lock);
447 enforce(t->active_wait_blocks);
448 unlink_blocks(t, THREAD_WAIT_KEY_NONE);
449 finish_locked(t);
450 spin_unlock(&t->wait_lock, irql);
451}
452
453void thread_alert(struct thread *cc_nonnull_ptr t) {
454 enforce_precond(t);
455 enum irql irql = spin_lock_high(&t->wait_lock);
456
457 if (!thread_test_flag(t, flag: THREAD_FLAG_DYING)) {
458 t->alert_pending = true;
459 deliver_alert_locked(t);
460 }
461
462 spin_unlock(&t->wait_lock, irql);
463}
464
465void thread_park(void) {
466 prepare_wait(/* objects = */ NULL, /* count = */ 0, /* storage = */ NULL,
467 type: THREAD_WAIT_INTERRUPTIBLE, reason: THREAD_SLEEP_REASON_MANUAL,
468 state: THREAD_STATE_SLEEPING);
469 thread_wait_complete();
470}
471
472void thread_wait_header_init(struct thread_wait_header *header) {
473 spinlock_init(&header->lock);
474 list_head_init(h: &header->waiters);
475}
476
477void thread_wait_init(struct thread *t) {
478 spinlock_init(&t->wait_lock);
479
480 t->active_wait_blocks = NULL;
481 t->wait_block_count = 0;
482 atomic_store_rlx(&t->wait_epoch, 0);
483 t->wait_key = THREAD_WAIT_KEY_NONE;
484 t->object_wait = false;
485 t->alert_pending = false;
486
487 for (size_t i = 0; i < THREAD_NUM_WAIT_BLOCKS; i++) {
488 memset(&t->wait_blocks[i], 0, sizeof(struct thread_wait_block));
489 list_node_init(n: &t->wait_blocks[i].obj_node);
490 }
491}
492