xco.c (49997B)
1 /* 2 * xco.c — implementation for xco.h. 3 * 4 * Two parts: 5 * - Event substrate (runtime, latch, semaphore, select/allof, queue 6 * (chan is a typedef alias for the cap=0+BLOCK case), broadcast, 7 * notify, timer, pairing heap, timeout, ticker, task group). Each 8 * event type is a small struct with a static vtable. 9 * Waitlists are intrusive linked lists; fire-all detaches the whole 10 * list before iterating so callbacks can do anything, including 11 * re-park or unpark sibling waiters, without iterator hazards. 12 * 13 * - Stack-switching coroutines (xco). The platform layer 14 * (arch/<name>/xco_arch.c) supplies the register save/restore 15 * primitive (xco_platform_switch) and the initial-context setup 16 * (xco_platform_init). Everything else — the state machine, the 17 * resume chain, the xco_trampoline wrapping user fn entry/exit, and 18 * the xco_self() TLS pointer — lives here. 19 * 20 * This translation unit is architecture-neutral. The platform context 21 * type is forward-declared (in xco_platform.h) and only ever referred 22 * to by pointer, so its actual size and layout never cross into this 23 * file. We reserve raw space for it inside xco_impl_t using 24 * XCO__CTX_SIZE / XCO__CTX_ALIGN from the arch's xco_arch.h, and cast 25 * to xco_platform_ctx_t * when calling into the platform layer. 26 */ 27 28 #include "xco.h" 29 #include "xco_platform_internal.h" 30 31 #include <assert.h> 32 #include <stddef.h> 33 #include <stdint.h> 34 35 /* ==================================================================== 36 * Thread (XCO_MT) 37 * 38 * The Vyukov MPSC inbox rides the waiter's `next` field, cast to 39 * _Atomic at the access sites only. The C standard does not guarantee 40 * T * and _Atomic(T *) share representation; on every C11 platform 41 * that matters they do (same size and alignment for pointers). We take 42 * that bet to keep the waiter at one next field and the ST hot path 43 * untouched. 44 * ==================================================================== */ 45 46 #ifdef XCO_MT 47 48 _Thread_local xco_thread_t *xco__thread_current = NULL; 49 50 #define XCO__ANEXT(w) ((_Atomic(xco_waiter_t *) *)&(w)->next) 51 52 void xco_thread_post(xco_thread_t *t, xco_waiter_t *w) { 53 atomic_store_explicit(XCO__ANEXT(w), NULL, memory_order_relaxed); 54 xco_waiter_t *prev = 55 atomic_exchange_explicit(&t->inbox_tail, w, memory_order_acq_rel); 56 atomic_store_explicit(XCO__ANEXT(prev), w, memory_order_release); 57 58 /* Dedup: only the empty -> non-empty edge invokes the wake hook. 59 * seq_cst pairs with the fence in xco_thread_try_park: either this 60 * exchange observes the parking consumer's false (we wake), or the 61 * consumer's re-check observes our push (it doesn't park). */ 62 if (!atomic_exchange_explicit(&t->wakeup_pending, true, 63 memory_order_seq_cst)) { 64 if (t->wakeup) t->wakeup(t); 65 } 66 } 67 68 /* Consumer-only. Standard Vyukov pop, including the two transient 69 * NULL-with-items states: a producer mid-push (tail moved, next link 70 * not yet stored) and the single-element stub re-push dance. Both 71 * resolve on a later call; xco__inbox_nonempty still reports items so 72 * spin sites (run fixpoint, try_park) don't sleep through them. */ 73 static xco_waiter_t *xco__inbox_pop(xco_thread_t *t) { 74 xco_waiter_t *head = t->inbox_head; 75 xco_waiter_t *next = atomic_load_explicit(XCO__ANEXT(head), memory_order_acquire); 76 77 if (head == &t->inbox_stub) { 78 if (!next) return NULL; /* empty */ 79 t->inbox_head = next; /* skip the stub */ 80 head = next; 81 next = atomic_load_explicit(XCO__ANEXT(head), memory_order_acquire); 82 } 83 if (next) { 84 t->inbox_head = next; 85 return head; 86 } 87 88 /* head is the last visible node. If a producer is mid-push behind 89 * it, report empty for now. */ 90 xco_waiter_t *tail = atomic_load_explicit(&t->inbox_tail, memory_order_acquire); 91 if (head != tail) return NULL; /* producer mid-push */ 92 93 /* Single element: re-push the stub so head becomes poppable. */ 94 atomic_store_explicit(XCO__ANEXT(&t->inbox_stub), NULL, memory_order_relaxed); 95 xco_waiter_t *prev = 96 atomic_exchange_explicit(&t->inbox_tail, &t->inbox_stub, memory_order_acq_rel); 97 atomic_store_explicit(XCO__ANEXT(prev), &t->inbox_stub, memory_order_release); 98 99 next = atomic_load_explicit(XCO__ANEXT(head), memory_order_acquire); 100 if (next) { 101 t->inbox_head = next; 102 return head; 103 } 104 return NULL; /* racing producer; resolves later */ 105 } 106 107 /* True if the inbox may hold items (including a producer mid-push). 108 * Fully-drained resting state: cursor at the stub and the producer end 109 * back at the stub. */ 110 static bool xco__inbox_nonempty(xco_thread_t *t) { 111 if (t->inbox_head != &t->inbox_stub) return true; 112 return atomic_load_explicit(&t->inbox_tail, memory_order_acquire) 113 != &t->inbox_stub; 114 } 115 116 /* Fire everything currently poppable. Callers own the TLS install. 117 * xco_waiter_fire (not a direct fire call) so a waiter whose home was 118 * retargeted mid-flight re-routes instead of running here. */ 119 static void xco__inbox_fire_all(xco_thread_t *t) { 120 for (xco_waiter_t *w; (w = xco__inbox_pop(t)) != NULL;) { 121 xco_waiter_fire(w, w->value); 122 } 123 } 124 125 void xco_thread_drain(xco_thread_t *t) { 126 xco_thread_t *prev = xco__thread_current; 127 xco__thread_current = t; 128 xco__inbox_fire_all(t); 129 xco__thread_current = prev; 130 } 131 132 bool xco_thread_try_park(xco_thread_t *t) { 133 atomic_store_explicit(&t->wakeup_pending, false, memory_order_seq_cst); 134 /* Order the flag store before the inbox re-check; pairs with the 135 * producer's seq_cst exchange in xco_thread_post (see there). */ 136 atomic_thread_fence(memory_order_seq_cst); 137 if (xco__inbox_nonempty(t)) { 138 atomic_store_explicit(&t->wakeup_pending, true, memory_order_relaxed); 139 return false; 140 } 141 return true; 142 } 143 144 #endif /* XCO_MT */ 145 146 /* ==================================================================== 147 * Runtime 148 * ==================================================================== */ 149 150 /* xco_rt_init and xco_rt_enqueue are defined inline in xco.h. */ 151 152 static xco_waiter_t *xco_rt_dequeue(xco_runtime_t *rt) { 153 xco_waiter_t *w = rt->head; 154 if (!w) return NULL; 155 rt->head = w->next; 156 if (!rt->head) rt->tail = NULL; 157 /* Hand the waiter back fully detached. Every fire path already clears 158 * prev before firing (and we just walked off the ready-queue), so 159 * w->prev is NULL in practice — making it explicit here means 160 * waker users can re-park without re-init, regardless of which 161 * fire path resumed them. */ 162 w->next = NULL; 163 w->prev = NULL; 164 return w; 165 } 166 167 void xco_rt_run(xco_runtime_t *rt, uint64_t now) { 168 /* The runtime ready queue holds only wakers. Other waiter 169 * shapes (e.g. select_input) fire synchronously inside event 170 * notify paths and never reach here. 171 * 172 * If a timer source is attached, advance it at the top of each 173 * iteration: any timer whose deadline <= now fires, enqueueing its 174 * wakers, which the inner loop then drains. A step may insert 175 * a fresh already-expired timer; the outer loop catches it on the 176 * next pass. Termination: each pass either drains a non-empty 177 * queue or exits, and advance only fires timers it then removes, 178 * so total work is bounded. 179 * 180 * XCO_MT: an attached thread's inbox joins the fixpoint, drained at 181 * the top of each pass with the current-thread TLS installed so 182 * inboxed fires (and their nested fires) route as same-thread. A 183 * remote producer can extend the run; the host's park handshake 184 * (xco_thread_try_park) covers the quiescent -> post race after we 185 * return. */ 186 #ifdef XCO_MT 187 xco_thread_t *prev_t = xco__thread_current; 188 if (rt->thread) xco__thread_current = rt->thread; 189 #endif 190 for (;;) { 191 #ifdef XCO_MT 192 if (rt->thread) xco__inbox_fire_all(rt->thread); 193 #endif 194 if (rt->timers) xco_timers_advance(rt->timers, now); 195 if (!rt->head) { 196 #ifdef XCO_MT 197 /* Items may be transiently unpoppable (producer mid-push); 198 * spin the fixpoint rather than report quiescent. */ 199 if (rt->thread && xco__inbox_nonempty(rt->thread)) continue; 200 #endif 201 break; 202 } 203 for (xco_waiter_t *w; (w = xco_rt_dequeue(rt));) { 204 xco_waker_t *sw = (xco_waker_t *)w; 205 xco_step(sw->mach, sw->resume_value); 206 } 207 } 208 #ifdef XCO_MT 209 xco__thread_current = prev_t; 210 #endif 211 } 212 213 /* ---- Waker ------------------------------------------------------------ */ 214 215 /* Exposed (with leading underscore) so the inline xco_waker_init in 216 * xco.h can install it without dragging the body into the header. */ 217 void xco__waker_fire(xco_waiter_t *w, uintptr_t value) { 218 xco_waker_t *sw = (xco_waker_t *)w; 219 sw->resume_value = value; 220 xco_rt_enqueue(sw->rt, w); 221 } 222 223 /* ==================================================================== 224 * Latch 225 * ==================================================================== */ 226 227 static bool xco_latch_poll(xco_event_t *e, uintptr_t *out, xco_waiter_t *w) { 228 xco_latch_t *l = (xco_latch_t *)e; 229 if (l->set) { 230 if (out) *out = l->value; 231 return true; 232 } 233 if (!w) return false; 234 /* One-waiter invariant: w must arrive clean. Detach paths (xco_latch_set 235 * iterator, xco_latch_unpark, xco_select_event_deinit) all leave waiters in 236 * this state. A trip here means a double-park bug. */ 237 assert(!w->prev && !w->next); 238 w->next = l->waiters; 239 if (l->waiters) l->waiters->prev = w; 240 l->waiters = w; 241 return false; 242 } 243 244 static void xco_latch_unpark(xco_event_t *e, xco_waiter_t *w) { 245 xco_latch_t *l = (xco_latch_t *)e; 246 /* Detect "not on this list" via the invariant maintained by park 247 * and the detach paths: a parked waiter has prev set OR is the 248 * head; a detached one has prev == NULL and is not the head. */ 249 if (!w->prev && l->waiters != w) return; 250 if (w->prev) w->prev->next = w->next; 251 else l->waiters = w->next; 252 if (w->next) w->next->prev = w->prev; 253 w->prev = w->next = NULL; 254 } 255 256 /* Exposed so the inline xco_latch_init in xco.h can reference it. */ 257 const xco_event_vtable_t xco__latch_vt = { 258 .poll = xco_latch_poll, 259 .unpark = xco_latch_unpark, 260 }; 261 262 void xco_latch_set(xco_latch_t *l, uintptr_t value) { 263 if (l->set) return; 264 l->set = true; 265 l->value = value; 266 267 /* Detach the whole waitlist before firing. A waiter's fire callback 268 * might do anything (including unpark a sibling on another event), 269 * but it cannot mutate this list — it's already gone. */ 270 xco_waiter_t *w = l->waiters; 271 l->waiters = NULL; 272 while (w) { 273 xco_waiter_t *next = w->next; /* save before xco_waiter_fire clears */ 274 xco_waiter_fire(w, value); 275 w = next; 276 } 277 } 278 279 /* ==================================================================== 280 * Semaphore 281 * 282 * FIFO doubly-linked waitlist, same shape as the chan_q_* helpers below 283 * but specialized to a xco_semaphore_t (so we don't have to thread the 284 * head/tail pair through chan_q_*). 285 * ==================================================================== */ 286 287 static void xco_sem_q_push(xco_semaphore_t *s, xco_waiter_t *w) { 288 assert(!w->prev && !w->next); 289 w->prev = s->tail; 290 w->next = NULL; 291 if (s->tail) s->tail->next = w; 292 else s->head = w; 293 s->tail = w; 294 } 295 296 static xco_waiter_t *xco_sem_q_pop(xco_semaphore_t *s) { 297 xco_waiter_t *w = s->head; 298 if (!w) return NULL; 299 s->head = w->next; 300 if (s->head) s->head->prev = NULL; 301 else s->tail = NULL; 302 w->prev = w->next = NULL; 303 return w; 304 } 305 306 static void xco_sem_q_remove(xco_semaphore_t *s, xco_waiter_t *w) { 307 if (!w->prev && s->head != w) return; 308 if (w->prev) w->prev->next = w->next; 309 else s->head = w->next; 310 if (w->next) w->next->prev = w->prev; 311 else s->tail = w->prev; 312 w->prev = w->next = NULL; 313 } 314 315 static bool xco_semaphore_poll(xco_event_t *e, uintptr_t *out, xco_waiter_t *w) { 316 xco_semaphore_t *s = (xco_semaphore_t *)e; 317 if (s->permits > 0) { 318 s->permits--; 319 if (out) *out = 1; 320 return true; 321 } 322 if (!w) return false; 323 xco_sem_q_push(s, w); 324 return false; 325 } 326 327 static void xco_semaphore_unpark(xco_event_t *e, xco_waiter_t *w) { 328 xco_semaphore_t *s = (xco_semaphore_t *)e; 329 xco_sem_q_remove(s, w); 330 } 331 332 const xco_event_vtable_t xco__semaphore_acquire_vt = { 333 .poll = xco_semaphore_poll, 334 .unpark = xco_semaphore_unpark, 335 }; 336 337 void xco_semaphore_release(xco_semaphore_t *s, size_t n) { 338 /* Hand a permit directly to each FIFO waiter, then drop any leftover 339 * into the count. Direct handoff prevents a fresh try from jumping 340 * the queue: an arriving acquirer that called try_ would see permits=0 341 * and park behind the existing waiters until everyone ahead has been 342 * served. */ 343 while (n > 0) { 344 xco_waiter_t *w = xco_sem_q_pop(s); 345 if (!w) break; 346 n--; 347 /* Fire value is conventional 1 — "you got a permit". Step-waiter 348 * users ignore the value; select inputs capture it as the input's 349 * value field. */ 350 xco_waiter_fire(w, 1); 351 } 352 s->permits += n; 353 } 354 355 /* ==================================================================== 356 * Select / all-of 357 * ==================================================================== */ 358 359 /* One fire callback serves both modes. A counter `remaining` is decremented 360 * on each fire; done is set when it hits 0. select inits remaining=1 (any 361 * one fire closes); allof inits remaining=n (every input must fire). The 362 * disarm-siblings loop is a no-op for already-fired waiters, so it runs 363 * uniformly: for select it cleans up still-parked losers, for allof it 364 * does nothing (every sibling is already detached by its source). */ 365 static void xco_select_input_fire(xco_waiter_t *w, uintptr_t value) { 366 xco_select_input_t *in = (xco_select_input_t *)w; 367 xco_select_event_t *s = in->parent; 368 369 /* Defensive: guard against any straggler that escaped disarm. */ 370 if (s->done.set) return; 371 /* Capture the input's payload before resuming anyone. Sticky 372 * sources also keep it on themselves; transient sources (channels) 373 * deliver only here, so this is the only durable record. */ 374 in->value = value; 375 if (--s->remaining > 0) return; 376 377 size_t i = (size_t)(in - s->inputs); 378 /* Disarm anyone still parked so their waiters don't dangle on input 379 * waitlists past s's lifetime. Idempotent on already-detached waiters. */ 380 for (size_t j = 0; j < s->n; j++) { 381 if (j != i) xco_event_unpark(s->inputs[j].src, &s->inputs[j].w); 382 } 383 xco_latch_set(&s->done, i); 384 } 385 386 void xco_select_event_init(xco_select_event_t *s, 387 xco_select_input_t *inputs, size_t n, 388 xco_event_t *const *srcs) { 389 xco_latch_init(&s->done); 390 s->inputs = inputs; 391 s->n = n; 392 s->remaining = 1; /* any one fire closes the wait */ 393 394 /* Fast path: an input already ready. Fire and skip parking entirely 395 * so deinit has nothing to disarm. */ 396 for (size_t i = 0; i < n; i++) { 397 uintptr_t v; 398 if (xco_event_poll(srcs[i], &v, NULL)) { 399 inputs[i].value = v; /* captured for inputs[winner].value */ 400 xco_latch_set(&s->done, i); 401 return; 402 } 403 } 404 405 for (size_t i = 0; i < n; i++) { 406 xco_waiter_init(&inputs[i].w, xco_select_input_fire); 407 inputs[i].src = srcs[i]; 408 inputs[i].parent = s; 409 inputs[i].value = 0; 410 (void)xco_event_poll(srcs[i], NULL, &inputs[i].w); 411 } 412 } 413 414 void xco_allof_event_init(xco_select_event_t *s, 415 xco_select_input_t *inputs, size_t n, 416 xco_event_t *const *srcs) { 417 xco_latch_init(&s->done); 418 s->inputs = inputs; 419 s->n = n; 420 s->remaining = n; /* every input must fire to close */ 421 422 if (n == 0) { xco_latch_set(&s->done, 0); return; } 423 424 /* Initialize each input then poll. Fused: an already-ready input is 425 * consumed inline (value captured, remaining--, no parking); the 426 * rest end up parked by the same call. If everyone was inline, fire 427 * done at the end. */ 428 for (size_t i = 0; i < n; i++) { 429 xco_waiter_init(&inputs[i].w, xco_select_input_fire); 430 inputs[i].src = srcs[i]; 431 inputs[i].parent = s; 432 inputs[i].value = 0; 433 434 uintptr_t v; 435 if (xco_event_poll(srcs[i], &v, &inputs[i].w)) { 436 inputs[i].value = v; 437 s->remaining--; 438 } 439 } 440 441 /* All inline-ready: fire done with the last input's index, matching 442 * the "closing index" semantics of the parked path. */ 443 if (s->remaining == 0) xco_latch_set(&s->done, n - 1); 444 } 445 446 void xco_select_event_deinit(xco_select_event_t *s) { 447 /* done.set => the closing fire already disarmed everyone (or the 448 * fast path skipped parking entirely). Otherwise — possible after a 449 * partial allof — some inputs may still be parked; unpark is 450 * idempotent for already-detached waiters. */ 451 if (s->done.set) return; 452 for (size_t i = 0; i < s->n; i++) { 453 xco_event_unpark(s->inputs[i].src, &s->inputs[i].w); 454 } 455 } 456 457 /* ==================================================================== 458 * FIFO waitlist helpers (shared by queue and notify) 459 * 460 * Doubly-linked FIFO push/pop. Same shape as latch's list operations 461 * but with an explicit tail so arrival order is preserved (unlike 462 * latch, where waiter order is irrelevant). The xco_chan_q_* names are 463 * historical — the chan code that originally used them is now folded 464 * into queue (chan = queue at cap=0 + BLOCK). 465 * ==================================================================== */ 466 467 static void xco_chan_q_push(xco_waiter_t **head, xco_waiter_t **tail, xco_waiter_t *w) { 468 assert(!w->prev && !w->next); 469 w->prev = *tail; 470 w->next = NULL; 471 if (*tail) (*tail)->next = w; 472 else *head = w; 473 *tail = w; 474 } 475 476 static xco_waiter_t *xco_chan_q_pop(xco_waiter_t **head, xco_waiter_t **tail) { 477 xco_waiter_t *w = *head; 478 if (!w) return NULL; 479 *head = w->next; 480 if (*head) (*head)->prev = NULL; 481 else *tail = NULL; 482 w->prev = w->next = NULL; 483 return w; 484 } 485 486 static void xco_chan_q_remove(xco_waiter_t **head, xco_waiter_t **tail, xco_waiter_t *w) { 487 /* Same not-on-list test as xco_latch_unpark: a queued waiter has prev 488 * set OR is the head; a detached one has prev == NULL and isn't 489 * the head. */ 490 if (!w->prev && *head != w) return; 491 if (w->prev) w->prev->next = w->next; 492 else *head = w->next; 493 if (w->next) w->next->prev = w->prev; 494 else *tail = w->prev; 495 w->prev = w->next = NULL; 496 } 497 498 499 /* ==================================================================== 500 * Queue 501 * 502 * The FIFO list helpers (xco_chan_q_push/pop/remove) are reused for the 503 * queue's send and recv waitlists — same shape, same invariants. The 504 * ring buffer lives in caller-provided storage; we just track head and 505 * len. cap == 0 leaves the buffer logic dormant: every send either 506 * direct-hands or parks, every recv either takes from a parked sender 507 * or parks — i.e. it degenerates to chan rendezvous. 508 * ==================================================================== */ 509 510 static inline xco_queue_t *xco_queue_of_recv(xco_event_t *e) { 511 return (xco_queue_t *)((char *)e - offsetof(xco_queue_t, recv)); 512 } 513 514 static inline void xco_queue_push_buf(xco_queue_t *q, uintptr_t v) { 515 assert(q->len < q->cap); 516 q->buf[(q->head + q->len) % q->cap] = v; 517 q->len++; 518 } 519 520 static inline uintptr_t xco_queue_pop_buf(xco_queue_t *q) { 521 assert(q->len > 0); 522 uintptr_t v = q->buf[q->head]; 523 q->head = (q->head + 1) % q->cap; 524 q->len--; 525 return v; 526 } 527 528 /* Pop one parked sender's value into the now-free buffer slot, firing 529 * the sender. Caller must have ensured a free slot exists (just popped 530 * from the buffer, or cap > len). Maintains FIFO across the buffer + 531 * sender-waitlist boundary: oldest buffered values come out before any 532 * sender's value (which was queued later). No-op if no sender parked. */ 533 static void xco_queue_drain_one_sender(xco_queue_t *q) { 534 if (!q->send_head) return; 535 xco_waiter_t *w = xco_chan_q_pop(&q->send_head, &q->send_tail); 536 xco_queue_send_waiter_t *qsw = (xco_queue_send_waiter_t *)w; 537 xco_queue_push_buf(q, qsw->value); 538 qsw->delivered = true; 539 /* Fire after pushing so the sender sees its delivery as complete. */ 540 xco_waiter_fire(w, 0); 541 } 542 543 static bool xco_queue_recv_poll(xco_event_t *e, uintptr_t *out, xco_waiter_t *w) { 544 xco_queue_t *q = xco_queue_of_recv(e); 545 if (q->len > 0) { 546 uintptr_t v = xco_queue_pop_buf(q); 547 if (out) *out = v; 548 xco_queue_drain_one_sender(q); 549 return true; 550 } 551 /* Empty buffer. If a sender is parked here it can only mean cap==0 552 * (otherwise the sender would have used the buffer). Hand directly. */ 553 if (q->send_head) { 554 xco_waiter_t *sender = xco_chan_q_pop(&q->send_head, &q->send_tail); 555 xco_queue_send_waiter_t *qsw = (xco_queue_send_waiter_t *)sender; 556 if (out) *out = qsw->value; 557 qsw->delivered = true; 558 xco_waiter_fire(sender, 0); 559 return true; 560 } 561 /* Closed and drained: receivers learn EOF via xco_queue_recv; out is 562 * undefined. */ 563 if (q->closed) { 564 if (out) *out = 0; 565 return true; 566 } 567 if (!w) return false; 568 xco_chan_q_push(&q->recv_head, &q->recv_tail, w); 569 return false; 570 } 571 572 static void xco_queue_recv_unpark(xco_event_t *e, xco_waiter_t *w) { 573 xco_queue_t *q = xco_queue_of_recv(e); 574 xco_chan_q_remove(&q->recv_head, &q->recv_tail, w); 575 } 576 577 const xco_event_vtable_t xco__queue_recv_vt = { 578 .poll = xco_queue_recv_poll, 579 .unpark = xco_queue_recv_unpark, 580 }; 581 582 xco_queue_send_status_t xco_queue_send_poll(xco_queue_t *q, uintptr_t value, 583 xco_queue_send_waiter_t *qsw) { 584 /* Closed is closed regardless of policy: caller learns the truth. */ 585 if (q->closed) return XCO_QSEND_CLOSED; 586 587 /* Direct handoff first: parked receivers always win over the buffer. 588 * This is the rendezvous case and the cap==0 case. */ 589 xco_waiter_t *w = xco_chan_q_pop(&q->recv_head, &q->recv_tail); 590 if (w) { 591 xco_waiter_fire(w, value); 592 return XCO_QSEND_ACCEPTED; 593 } 594 if (q->len < q->cap) { 595 xco_queue_push_buf(q, value); 596 return XCO_QSEND_ACCEPTED; 597 } 598 /* Buffer full and no waiting receiver. */ 599 switch (q->policy) { 600 case XCO_QUEUE_BLOCK: 601 if (!qsw) return XCO_QSEND_BLOCKED; 602 qsw->value = value; 603 xco_chan_q_push(&q->send_head, &q->send_tail, &qsw->sw.base); 604 return XCO_QSEND_BLOCKED; 605 case XCO_QUEUE_DROP_NEWEST: 606 return XCO_QSEND_ACCEPTED; 607 case XCO_QUEUE_DROP_OLDEST: 608 (void)xco_queue_pop_buf(q); 609 xco_queue_push_buf(q, value); 610 return XCO_QSEND_ACCEPTED; 611 } 612 __builtin_unreachable(); 613 } 614 615 void xco_queue_send_unpark(xco_queue_t *q, xco_queue_send_waiter_t *qsw) { 616 xco_chan_q_remove(&q->send_head, &q->send_tail, &qsw->sw.base); 617 } 618 619 xco_recv_status_t xco_queue_recv(xco_queue_t *q, uintptr_t *out) { 620 if (q->len > 0) { 621 uintptr_t v = xco_queue_pop_buf(q); 622 if (out) *out = v; 623 xco_queue_drain_one_sender(q); 624 return XCO_RECV_GOT; 625 } 626 if (q->send_head) { 627 xco_waiter_t *w = xco_chan_q_pop(&q->send_head, &q->send_tail); 628 xco_queue_send_waiter_t *qsw = (xco_queue_send_waiter_t *)w; 629 if (out) *out = qsw->value; 630 qsw->delivered = true; 631 xco_waiter_fire(w, 0); 632 return XCO_RECV_GOT; 633 } 634 if (q->closed) return XCO_RECV_CLOSED; 635 return XCO_RECV_EMPTY; 636 } 637 638 void xco_queue_close(xco_queue_t *q) { 639 if (q->closed) return; 640 q->closed = true; 641 642 /* Drain parked senders with delivered=false. Senders only park 643 * under BLOCK, so this is no-op for DROP_* (their waitlist is 644 * always empty). */ 645 xco_waiter_t *w; 646 while ((w = xco_chan_q_pop(&q->send_head, &q->send_tail)) != NULL) { 647 xco_queue_send_waiter_t *qsw = (xco_queue_send_waiter_t *)w; 648 qsw->delivered = false; 649 xco_waiter_fire(w, 0); 650 } 651 /* Wake parked receivers so they can observe closed via xco_queue_recv. 652 * Receivers may still drain buffered values first — xco_queue_recv's 653 * XCO_RECV_GOT path is hit before the XCO_RECV_CLOSED branch. */ 654 while ((w = xco_chan_q_pop(&q->recv_head, &q->recv_tail)) != NULL) { 655 xco_waiter_fire(w, 0); 656 } 657 } 658 659 /* ---- Queue send op (selectable send) ---------------------------------- */ 660 661 void xco__queue_send_op_fire(xco_waiter_t *w, uintptr_t value) { 662 (void)value; 663 xco_queue_send_op_t *op = (xco_queue_send_op_t *)w; 664 xco_latch_set(&op->done, op->qsw.delivered ? 1 : 0); 665 } 666 667 void xco_queue_send_op_init(xco_queue_send_op_t *op, xco_queue_t *q, uintptr_t value) { 668 xco_queue_send_waiter_init(&op->qsw, NULL, NULL); 669 op->qsw.sw.base.fire = xco__queue_send_op_fire; 670 op->queue = q; 671 xco_latch_init(&op->done); 672 673 switch (xco_queue_send_poll(q, value, &op->qsw)) { 674 case XCO_QSEND_ACCEPTED: 675 op->qsw.delivered = true; 676 xco_latch_set(&op->done, 1); 677 return; 678 case XCO_QSEND_CLOSED: 679 xco_latch_set(&op->done, 0); 680 return; 681 case XCO_QSEND_BLOCKED: 682 /* Only BLOCK + full buffer; parked. */ 683 return; 684 } 685 } 686 687 /* ==================================================================== 688 * Broadcast (slot) 689 * 690 * The waitlist uses the same doubly-linked LIFO shape as latch — there 691 * is no FIFO requirement because publish wakes everyone at once. The 692 * key difference from latch is publish vs set: publish never marks a 693 * sticky bit on the event, so try always returns false (subscribers 694 * wait for the *next* publish), and the waitlist is reusable across 695 * publishes — subscribers re-park to receive subsequent values. 696 * ==================================================================== */ 697 698 static bool xco_broadcast_poll(xco_event_t *e, uintptr_t *out, xco_waiter_t *w) { 699 (void)out; 700 /* Transient: never reports ready; just parks if asked. */ 701 if (!w) return false; 702 xco_broadcast_t *b = (xco_broadcast_t *)e; 703 assert(!w->prev && !w->next); 704 w->next = b->waiters; 705 if (b->waiters) b->waiters->prev = w; 706 b->waiters = w; 707 return false; 708 } 709 710 static void xco_broadcast_unpark(xco_event_t *e, xco_waiter_t *w) { 711 xco_broadcast_t *b = (xco_broadcast_t *)e; 712 if (!w->prev && b->waiters != w) return; 713 if (w->prev) w->prev->next = w->next; 714 else b->waiters = w->next; 715 if (w->next) w->next->prev = w->prev; 716 w->prev = w->next = NULL; 717 } 718 719 const xco_event_vtable_t xco__broadcast_vt = { 720 .poll = xco_broadcast_poll, 721 .unpark = xco_broadcast_unpark, 722 }; 723 724 void xco_broadcast_publish(xco_broadcast_t *b, uintptr_t value) { 725 b->has_value = true; 726 b->value = value; 727 728 /* Detach the waitlist before iterating — same hazard-free pattern as 729 * xco_latch_set. A waiter's fire callback may re-park itself on us (the 730 * common case for a re-arming subscriber); decoupling means that 731 * re-park lands on a fresh waitlist, not on the snapshot we're 732 * walking. */ 733 xco_waiter_t *w = b->waiters; 734 b->waiters = NULL; 735 while (w) { 736 xco_waiter_t *next = w->next; /* save before xco_waiter_fire clears */ 737 xco_waiter_fire(w, value); 738 w = next; 739 } 740 } 741 742 /* ==================================================================== 743 * Notify 744 * 745 * Doubly-linked FIFO waitlist (same shape as the chan/queue waitlists). 746 * xco_notify_one fires the head; xco_notify_all detaches the whole list before 747 * iterating so callbacks can re-park onto a fresh waitlist without 748 * iterator hazards (same pattern as xco_latch_set). xco_event_poll never 749 * reports ready: notify is purely transient. 750 * ==================================================================== */ 751 752 static bool xco_notify_poll(xco_event_t *e, uintptr_t *out, xco_waiter_t *w) { 753 (void)out; 754 if (!w) return false; 755 xco_notify_t *n = (xco_notify_t *)e; 756 xco_chan_q_push(&n->head, &n->tail, w); 757 return false; 758 } 759 760 static void xco_notify_unpark(xco_event_t *e, xco_waiter_t *w) { 761 xco_notify_t *n = (xco_notify_t *)e; 762 xco_chan_q_remove(&n->head, &n->tail, w); 763 } 764 765 const xco_event_vtable_t xco__notify_vt = { 766 .poll = xco_notify_poll, 767 .unpark = xco_notify_unpark, 768 }; 769 770 void xco_notify_one(xco_notify_t *n) { 771 xco_waiter_t *w = xco_chan_q_pop(&n->head, &n->tail); 772 if (!w) return; 773 xco_waiter_fire(w, 0); 774 } 775 776 void xco_notify_all(xco_notify_t *n) { 777 /* Detach before iterating: re-parking inside fire lands on a fresh 778 * (empty) list. Walk the snapshot via saved next pointers. */ 779 xco_waiter_t *w = n->head; 780 n->head = n->tail = NULL; 781 while (w) { 782 xco_waiter_t *next = w->next; 783 xco_waiter_fire(w, 0); 784 w = next; 785 } 786 } 787 788 /* ==================================================================== 789 * Pairing heap 790 * 791 * Standard intrusive pairing heap. Each node carries three link fields: 792 * 793 * child : head of children sibling list (NULL if leaf). 794 * prev : parent if this node is its parent's first child; 795 * previous sibling otherwise; NULL only for a detached tree 796 * root (including h->root). 797 * next : next sibling, or NULL for the last child / a detached root. 798 * 799 * The "prev points to either parent or a sibling" trick lets us splice 800 * a node out in O(1) without an extra parent pointer: parent->child==n 801 * distinguishes "first child" from "non-first sibling." 802 * 803 * meld picks the smaller-deadline root as winner and grafts the loser 804 * as its new first child. Pop-min and remove both rebuild via the 805 * classic two-pass pairwise merge of the resulting children list. 806 * ==================================================================== */ 807 808 /* Merge two detached subtree roots (each with prev=next=NULL). Returns 809 * the merged root (also detached: prev=next=NULL). */ 810 static xco_timer_t *xco_ph_meld(xco_timer_t *a, xco_timer_t *b) { 811 if (!a) return b; 812 if (!b) return a; 813 xco_timer_t *small, *large; 814 if (a->deadline <= b->deadline) { small = a; large = b; } 815 else { small = b; large = a; } 816 /* Graft `large` as the new first child of `small`. */ 817 large->next = small->child; 818 if (small->child) small->child->prev = large; 819 large->prev = small; /* parent link via prev */ 820 small->child = large; 821 small->prev = NULL; 822 small->next = NULL; 823 return small; 824 } 825 826 /* Two-pass pairwise meld of a children sibling list. Detaches each node 827 * before melding so meld inputs satisfy its prev=next=NULL contract. 828 * The output has prev=next=NULL. */ 829 static xco_timer_t *xco_ph_merge_pairs(xco_timer_t *first) { 830 /* Pass 1: walk the sibling list left-to-right, melding consecutive 831 * pairs. Chain results via `next` (ab)use as a temporary list link. */ 832 xco_timer_t *list = NULL; 833 while (first) { 834 xco_timer_t *a = first; 835 xco_timer_t *b = a->next; 836 xco_timer_t *rest = b ? b->next : NULL; 837 a->prev = a->next = NULL; 838 if (b) { b->prev = b->next = NULL; } 839 xco_timer_t *m = xco_ph_meld(a, b); 840 m->next = list; /* prepend to pass-1 list */ 841 list = m; 842 first = rest; 843 } 844 /* Pass 2: meld the pass-1 list into a single root. */ 845 xco_timer_t *acc = NULL; 846 while (list) { 847 xco_timer_t *nxt = list->next; 848 list->prev = NULL; 849 list->next = NULL; 850 acc = xco_ph_meld(acc, list); 851 list = nxt; 852 } 853 return acc; 854 } 855 856 /* Detach n from the tree (must currently be in the heap). Returns the 857 * (possibly new) main-heap root. n is left fully detached: child still 858 * points to its subtree, but prev/next are NULL — caller decides what 859 * to do with that subtree. */ 860 static xco_timer_t *xco_ph_detach(xco_pairing_heap_t *h, xco_timer_t *n) { 861 if (h->root == n) { 862 /* n is the main root; pop it and rebuild from its children. */ 863 xco_timer_t *new_root = xco_ph_merge_pairs(n->child); 864 n->child = NULL; 865 n->prev = NULL; 866 n->next = NULL; 867 return new_root; 868 } 869 /* n has a parent (recorded via prev — either as its parent's first 870 * child or as some sibling's successor). Splice out of the sibling 871 * list. */ 872 if (n->prev->child == n) { 873 /* First child: parent's child link skips us. */ 874 n->prev->child = n->next; 875 } else { 876 /* Mid/last sibling: previous sibling's next skips us. */ 877 n->prev->next = n->next; 878 } 879 if (n->next) n->next->prev = n->prev; 880 n->prev = NULL; 881 n->next = NULL; 882 /* The main-heap root is unchanged structurally; the caller of this 883 * function decides how (or whether) to reintroduce n's subtree. */ 884 return h->root; 885 } 886 887 static void xco_ph_insert(xco_timers_t *ts, xco_timer_t *t) { 888 xco_pairing_heap_t *h = (xco_pairing_heap_t *)ts; 889 /* Singleton tree (prev/next/child already NULL via xco_timer_init). */ 890 h->root = xco_ph_meld(h->root, t); 891 t->in_heap = true; 892 } 893 894 static void xco_ph_cancel(xco_timers_t *ts, xco_timer_t *t) { 895 xco_pairing_heap_t *h = (xco_pairing_heap_t *)ts; 896 if (!t->in_heap) return; 897 h->root = xco_ph_detach(h, t); 898 /* Now meld t's subtree (its children) back into the main heap. */ 899 xco_timer_t *sub = xco_ph_merge_pairs(t->child); 900 t->child = NULL; 901 h->root = xco_ph_meld(h->root, sub); 902 t->in_heap = false; 903 } 904 905 static void xco_ph_advance(xco_timers_t *ts, uint64_t now) { 906 xco_pairing_heap_t *h = (xco_pairing_heap_t *)ts; 907 /* Pop while the min-key timer is due. Each fire may run callbacks 908 * that insert *new* timers (with later deadlines, normally) — those 909 * land back in the heap, and we keep checking the root. */ 910 while (h->root && h->root->deadline <= now) { 911 xco_timer_t *t = h->root; 912 h->root = xco_ph_merge_pairs(t->child); 913 t->child = NULL; 914 t->prev = NULL; 915 t->next = NULL; 916 t->in_heap = false; 917 /* Trigger the latch: drains the waitlist and delivers the 918 * deadline as the fire payload. */ 919 xco_latch_set(&t->done, (uintptr_t)t->deadline); 920 } 921 } 922 923 static uint64_t xco_ph_peek(const xco_timers_t *ts) { 924 const xco_pairing_heap_t *h = (const xco_pairing_heap_t *)ts; 925 return h->root ? h->root->deadline : UINT64_MAX; 926 } 927 928 const xco_timers_vtable_t xco__pairing_heap_vt = { 929 .insert = xco_ph_insert, 930 .cancel = xco_ph_cancel, 931 .advance = xco_ph_advance, 932 .peek = xco_ph_peek, 933 }; 934 935 /* ==================================================================== 936 * Timeout 937 * ==================================================================== */ 938 939 /* Bridge waiter: parked on the timer's latch, fires the cancel when the 940 * timer fires. Two-step indirection so cancel can already have its own 941 * waiters (e.g. a xco_wait_or_cancel select) without the timer's waitlist 942 * caring about cancel internals. */ 943 static void xco__timeout_bridge_fire(xco_waiter_t *w, uintptr_t value) { 944 (void)value; 945 xco_timeout_t *to = (xco_timeout_t *)((char *)w - offsetof(xco_timeout_t, bridge)); 946 xco_cancel_set(&to->cancel); 947 } 948 949 void xco_timeout_init(xco_timeout_t *to, xco_timers_t *ts, uint64_t deadline) { 950 xco_timer_init(&to->timer, ts, deadline); 951 xco_cancel_init(&to->cancel); 952 xco_waiter_init(&to->bridge, xco__timeout_bridge_fire); 953 /* Park the bridge on the timer. If the timer fires, xco_latch_set 954 * detaches the bridge and calls our fire callback inline. The timer 955 * was just inserted into the heap and hasn't been advanced, so poll 956 * always parks here (return value ignored). */ 957 (void)xco_event_poll(xco_timer_event(&to->timer), NULL, &to->bridge); 958 } 959 960 /* ==================================================================== 961 * Ticker 962 * 963 * The ticker's event surface uses the broadcast-style LIFO doubly-linked 964 * waitlist: subscribers are fired all-at-once on each tick, so order 965 * doesn't matter; doubly-linked gives O(1) unpark for cancellation. 966 * ==================================================================== */ 967 968 static bool xco_ticker_poll(xco_event_t *e, uintptr_t *out, xco_waiter_t *w) { 969 (void)out; 970 /* Transient — never reports ready; just parks if asked. */ 971 if (!w) return false; 972 xco_ticker_t *t = (xco_ticker_t *)((char *)e - offsetof(xco_ticker_t, base)); 973 assert(!w->prev && !w->next); 974 w->next = t->waiters; 975 if (t->waiters) t->waiters->prev = w; 976 t->waiters = w; 977 return false; 978 } 979 980 static void xco_ticker_unpark(xco_event_t *e, xco_waiter_t *w) { 981 xco_ticker_t *t = (xco_ticker_t *)((char *)e - offsetof(xco_ticker_t, base)); 982 if (!w->prev && t->waiters != w) return; 983 if (w->prev) w->prev->next = w->next; 984 else t->waiters = w->next; 985 if (w->next) w->next->prev = w->prev; 986 w->prev = w->next = NULL; 987 } 988 989 const xco_event_vtable_t xco__ticker_vt = { 990 .poll = xco_ticker_poll, 991 .unpark = xco_ticker_unpark, 992 }; 993 994 /* Bridge waiter: parks on the underlying timer's latch. On fire, compute 995 * the next deadline (skip-ahead-safe), reinstall the timer, re-park the 996 * bridge on the new timer, then fire every parked subscriber with the 997 * just-fired deadline. The waitlist is detached before iteration so 998 * subscribers can re-park inside their fire callbacks. */ 999 static void xco__ticker_bridge_fire(xco_waiter_t *w, uintptr_t value) { 1000 xco_ticker_t *t = (xco_ticker_t *)((char *)w - offsetof(xco_ticker_t, bridge)); 1001 uint64_t fired = (uint64_t)value; 1002 uint64_t next = fired + t->period; 1003 /* Skip-ahead: in the rare overflow case (period = 0 or wraparound), 1004 * step forward enough to keep next > fired. */ 1005 if (next <= fired) { 1006 next += ((fired - next) / t->period + 1) * t->period; 1007 } 1008 /* Reinstall the timer for the next tick. The latch's storage is 1009 * reused — xco_timer_init runs xco_latch_init on it. */ 1010 xco_timer_init(&t->timer, t->src, next); 1011 /* Bridge waiter is fully detached (xco_waiter_fire just cleared its 1012 * links); park it on the freshly-armed timer (which has not yet been 1013 * advanced, so poll parks). */ 1014 (void)xco_event_poll(xco_timer_event(&t->timer), NULL, &t->bridge); 1015 1016 /* Fire the subscribers. Detach the waitlist first so re-park inside 1017 * fire lands on the now-empty list. */ 1018 xco_waiter_t *waiters = t->waiters; 1019 t->waiters = NULL; 1020 while (waiters) { 1021 xco_waiter_t *nxt = waiters->next; 1022 xco_waiter_fire(waiters, (uintptr_t)fired); 1023 waiters = nxt; 1024 } 1025 } 1026 1027 void xco_ticker_init(xco_ticker_t *t, xco_timers_t *ts, 1028 uint64_t period, uint64_t first_deadline) { 1029 /* period must be positive — the skip-ahead computation in the bridge 1030 * divides by period, and a zero-period ticker would loop forever 1031 * inside xco_ph_advance. */ 1032 assert(period > 0); 1033 t->base.vt = &xco__ticker_vt; 1034 t->src = ts; 1035 t->period = period; 1036 t->waiters = NULL; 1037 1038 xco_timer_init(&t->timer, ts, first_deadline); 1039 1040 xco_waiter_init(&t->bridge, xco__ticker_bridge_fire); 1041 (void)xco_event_poll(xco_timer_event(&t->timer), NULL, &t->bridge); 1042 } 1043 1044 void xco_ticker_deinit(xco_ticker_t *t) { 1045 /* Pull the bridge off the timer (no-op if already fired) and cancel 1046 * the timer. Subscribers' waiters are the caller's storage; nothing 1047 * to free here. */ 1048 xco_event_unpark(xco_timer_event(&t->timer), &t->bridge); 1049 xco_timer_deinit(&t->timer); 1050 } 1051 1052 /* ==================================================================== 1053 * Task group 1054 * 1055 * Each attach contributes one to the countdown and parks a bridge waiter 1056 * on the task's done event. Bridge fire splices the slot out of the 1057 * group's list and decrements the countdown — the join event fires when 1058 * the last attached task finishes. 1059 * 1060 * Cancellation is fan-out: walk the list, set each task's cancel, then 1061 * set the group-level cancel. Bodies cooperate by composing their work 1062 * with xco_task_cancel(self); the group-level cancel is for non-task 1063 * waiters that want to react to "the group has been told to stop." 1064 * ==================================================================== */ 1065 1066 static void xco__task_group_detach_slot(xco_task_group_t *g, xco_group_attach_t *slot) { 1067 /* Doubly-linked, head/tail tracked; same shape as other waitlists. */ 1068 if (slot->prev) slot->prev->next = slot->next; 1069 else g->head = slot->next; 1070 if (slot->next) slot->next->prev = slot->prev; 1071 else g->tail = slot->prev; 1072 slot->prev = slot->next = NULL; 1073 } 1074 1075 static void xco__task_group_bridge_fire(xco_waiter_t *w, uintptr_t value) { 1076 (void)value; 1077 xco_group_attach_t *slot = (xco_group_attach_t *)((char *)w - offsetof(xco_group_attach_t, bridge)); 1078 xco_task_group_t *g = slot->group; 1079 xco__task_group_detach_slot(g, slot); 1080 xco_countdown_done(&g->pending); 1081 } 1082 1083 void xco_task_group_init(xco_task_group_t *g) { 1084 /* Don't go through xco_countdown_init(0) — that fires the latch 1085 * immediately, which would make the very first attach's 1086 * xco_countdown_add UB. The group's join must remain not-fired until at 1087 * least one attached task has finished, so we open with 1088 * remaining=0 and an unset latch. The first attach lifts remaining 1089 * to 1, and matching countdown_dones bring it back to 0, firing 1090 * the latch. */ 1091 xco_latch_init(&g->pending.done); 1092 g->pending.remaining = 0; 1093 xco_cancel_init(&g->cancel); 1094 g->head = g->tail = NULL; 1095 } 1096 1097 void xco_task_group_attach(xco_task_group_t *g, xco_task_t *t, xco_group_attach_t *slot) { 1098 xco_countdown_add(&g->pending, 1); 1099 1100 slot->task = t; 1101 slot->group = g; 1102 1103 /* Append to the group's list (FIFO; ordering doesn't affect cancel 1104 * fan-out semantics, but consistent with other waitlists in the 1105 * codebase). */ 1106 slot->prev = g->tail; 1107 slot->next = NULL; 1108 if (g->tail) g->tail->next = slot; 1109 else g->head = slot; 1110 g->tail = slot; 1111 1112 xco_waiter_init(&slot->bridge, xco__task_group_bridge_fire); 1113 1114 /* Park on the task's done event. Re-attaching a finished task is UB 1115 * per the contract; the latch's clean-waiter assert inside poll 1116 * would catch a double-park. */ 1117 (void)xco_event_poll(xco_task_done_event(t), NULL, &slot->bridge); 1118 } 1119 1120 void xco_task_group_cancel(xco_task_group_t *g) { 1121 /* Fan-out cancel: signal each attached task. Walk the snapshot 1122 * (cancel doesn't detach the slot — only task done does — so the 1123 * list is stable across iteration). */ 1124 for (xco_group_attach_t *s = g->head; s; s = s->next) { 1125 xco_cancel_set(&s->task->cancel); 1126 } 1127 /* Group-level cancel for anyone awaiting "the group as a whole." */ 1128 xco_cancel_set(&g->cancel); 1129 } 1130 1131 /* ==================================================================== 1132 * xco — coroutines 1133 * ==================================================================== */ 1134 1135 typedef struct xco_impl { 1136 xco_mach_t base; /* must be first; aliases xco_coro_t.base */ 1137 _Alignas(XCO__CTX_ALIGN) unsigned char ctx_buf[XCO__CTX_SIZE]; 1138 xco_platform_ctx_t *resumer_ctx; /* where suspend/return goes */ 1139 xco_fn fn; 1140 } xco_impl_t; 1141 1142 _Static_assert(sizeof(xco_impl_t) <= sizeof(xco_coro_t), "xco_coro_t too small"); 1143 _Static_assert(_Alignof(xco_impl_t) <= _Alignof(xco_coro_t), "xco_coro_t under-aligned"); 1144 1145 static inline xco_impl_t *xco_impl_of(xco_coro_t *c) { return (xco_impl_t *)c; } 1146 static inline xco_platform_ctx_t *xco_ctx_of(xco_impl_t *ci) { 1147 return (xco_platform_ctx_t *)ci->ctx_buf; 1148 } 1149 1150 /* Per-thread state. */ 1151 static _Thread_local xco_impl_t *xco_coro_current = NULL; 1152 static _Thread_local _Alignas(XCO__CTX_ALIGN) 1153 unsigned char xco_t_main_ctx_buf[XCO__CTX_SIZE]; 1154 static inline xco_platform_ctx_t *xco_main_ctx(void) { 1155 return (xco_platform_ctx_t *)xco_t_main_ctx_buf; 1156 } 1157 1158 /* Trampoline: runs on the coroutine's own stack, invoked by the 1159 * platform layer on the first switch into a fresh context. The 1160 * argument is the value passed to that first xco_step. The coroutine 1161 * identifies itself via xco_coro_current, set by the resumer just before 1162 * switching. */ 1163 static void xco_trampoline(uintptr_t arg) { 1164 xco_impl_t *self = xco_coro_current; 1165 uintptr_t ret = self->fn(arg); 1166 1167 self->base.status = XCO_STEP_DEAD; 1168 (void)xco_platform_switch(xco_ctx_of(self), self->resumer_ctx, ret); 1169 __builtin_unreachable(); 1170 } 1171 1172 /* xco_step_fn entry point wired into base.step at init time. All callers 1173 * — generic xco_step consumers and xco-aware code alike — route through 1174 * here. */ 1175 static xco_step_result_t xco_co_step(xco_mach_t *s, uintptr_t value) { 1176 xco_impl_t *next = (xco_impl_t *)s; 1177 assert(next->base.status == XCO_STEP_INIT || next->base.status == XCO_STEP_SUSPENDED); 1178 1179 xco_impl_t *prev = xco_coro_current; 1180 next->resumer_ctx = prev ? xco_ctx_of(prev) : xco_main_ctx(); 1181 next->base.status = XCO_STEP_RUNNING; 1182 xco_coro_current = next; 1183 1184 uintptr_t back = xco_platform_switch(next->resumer_ctx, 1185 xco_ctx_of(next), value); 1186 1187 /* Coroutine has either suspended or returned; status is already 1188 * set correctly by xco_suspend or by the xco_trampoline. */ 1189 xco_coro_current = prev; 1190 return (xco_step_result_t){ .value = back, .status = next->base.status }; 1191 } 1192 1193 void xco_init(xco_coro_t *c, xco_fn fn, 1194 void *stack_base, size_t stack_len) { 1195 xco_impl_t *ci = xco_impl_of(c); 1196 ci->base.step = xco_co_step; 1197 ci->base.status = XCO_STEP_INIT; 1198 ci->fn = fn; 1199 ci->resumer_ctx = NULL; 1200 xco_platform_init(xco_ctx_of(ci), stack_base, stack_len, xco_trampoline); 1201 } 1202 1203 uintptr_t xco_suspend(uintptr_t value) { 1204 xco_impl_t *self = xco_coro_current; 1205 assert(self != NULL); 1206 self->base.status = XCO_STEP_SUSPENDED; 1207 return xco_platform_switch(xco_ctx_of(self), self->resumer_ctx, value); 1208 } 1209 1210 xco_coro_t *xco_self(void) { 1211 return (xco_coro_t *)xco_coro_current; 1212 } 1213 1214 /* ---- xco-backed task -------------------------------------------------- */ 1215 1216 /* The xco_trampoline: runs as the coroutine's xco_fn. Recovers the owning 1217 * xco_cotask_t via container_of on the embedded co (xco_self() returns 1218 * the running xco), dispatches the user fn, and surfaces the return 1219 * value through xco_task_done — so a joiner waiting on xco_task_done_event 1220 * wakes with the body's return without the body needing to know about 1221 * the task surface at all. */ 1222 static uintptr_t xco_cotask_trampoline(uintptr_t arg) { 1223 xco_coro_t *self = xco_self(); 1224 xco_cotask_t *xt = (xco_cotask_t *)((char *)self - offsetof(xco_cotask_t, co)); 1225 uintptr_t r = xt->fn(&xt->task, arg); 1226 xco_task_done(&xt->task, r); 1227 return r; 1228 } 1229 1230 void xco_cotask_init(xco_cotask_t *xt, xco_cotask_fn fn, 1231 void *stack_base, size_t stack_len) { 1232 xco_task_init(&xt->task, &xt->co.base); 1233 xt->fn = fn; 1234 xco_init(&xt->co, xco_cotask_trampoline, stack_base, stack_len); 1235 } 1236 1237 /* ==================================================================== 1238 * xco_op — generic effect/IO layer 1239 * 1240 * The runtime owns a doubly-linked pending-ops list (op_head/op_tail). 1241 * Submit appends; cancel-while-PENDING splices in O(1); take detaches 1242 * the whole list in O(1) and lets the host iterate via the next pointers 1243 * (which are then the host's to reuse for in-flight tracking). 1244 * 1245 * Resolution always goes through xco_latch_set on op->done; awaiters 1246 * compose with the standard event API. Status is delivered as the latch 1247 * payload. 1248 * ==================================================================== */ 1249 1250 void xco_op_submit(xco_runtime_t *rt, xco_op_t *op) { 1251 xco_latch_init(&op->done); 1252 op->cancel_requested = false; 1253 op->rt = rt; 1254 op->epoch = rt->op_epoch; /* matches → PENDING */ 1255 op->next = NULL; 1256 op->prev = rt->op_tail; 1257 if (rt->op_tail) rt->op_tail->next = op; 1258 else rt->op_head = op; 1259 rt->op_tail = op; 1260 } 1261 1262 void xco_op_cancel(xco_op_t *op) { 1263 if (op->done.set) return; /* RESOLVED — no-op */ 1264 if (xco_op_is_pending(op)) { 1265 /* PENDING: splice from rt's pending list and resolve as cancelled. */ 1266 xco_runtime_t *rt = op->rt; 1267 if (op->prev) op->prev->next = op->next; 1268 else rt->op_head = op->next; 1269 if (op->next) op->next->prev = op->prev; 1270 else rt->op_tail = op->prev; 1271 op->prev = op->next = NULL; 1272 xco_latch_set(&op->done, (uintptr_t)XCO_OP_CANCELLED); 1273 return; 1274 } 1275 /* IN_FLIGHT: advisory. Host sees cancel_requested and decides whether 1276 * to honor it; xco_op_complete is still the final word either way. */ 1277 op->cancel_requested = true; 1278 } 1279 1280 void xco_op_complete(xco_op_t *op, xco_op_status_t status) { 1281 /* xco_latch_set is idempotent; second/late completion is a no-op. 1282 * Per the contract, host calls this only after take, so op is not 1283 * on rt's pending list. */ 1284 xco_latch_set(&op->done, (uintptr_t)status); 1285 } 1286 1287 xco_op_t *xco_rt_take_ops(xco_runtime_t *rt, xco_op_t **tail_out) { 1288 xco_op_t *head = rt->op_head; 1289 if (tail_out) *tail_out = rt->op_tail; 1290 rt->op_head = rt->op_tail = NULL; 1291 /* Bump the epoch: every op currently on the (now-detached) batch had 1292 * its epoch field set to the old value at submit, so they all flip to 1293 * IN_FLIGHT in O(1). Submits after this point land in the new 1294 * generation. */ 1295 rt->op_epoch++; 1296 return head; 1297 }