xco

xco.c (44826B)

---

 /*
  * xco.c — implementation for xco.h.
  *
  * Two parts:
  *   - Event substrate (runtime, latch, semaphore, select/allof, queue
  *     (chan is a typedef alias for the cap=0+BLOCK case), broadcast,
  *     notify, timer, pairing heap, timeout, ticker, task group). Each
  *     event type is a small struct with a static vtable.
  *     Waitlists are intrusive linked lists; fire-all detaches the whole
  *     list before iterating so callbacks can do anything, including
  *     re-park or unpark sibling waiters, without iterator hazards.
  *
  *   - Stack-switching coroutines (xco). The platform layer
  *     (arch/<name>/xco_arch.c) supplies the register save/restore
  *     primitive (xco_platform_switch) and the initial-context setup
  *     (xco_platform_init). Everything else — the state machine, the
  *     resume chain, the xco_trampoline wrapping user fn entry/exit, and
  *     the xco_self() TLS pointer — lives here.
  *
  * This translation unit is architecture-neutral. The platform context
  * type is forward-declared (in xco_platform.h) and only ever referred
  * to by pointer, so its actual size and layout never cross into this
  * file. We reserve raw space for it inside xco_impl_t using
  * XCO__CTX_SIZE / XCO__CTX_ALIGN from the arch's xco_arch.h, and cast
  * to xco_platform_ctx_t * when calling into the platform layer.
  */
 
 #include "xco.h"
 #include "xco_platform_internal.h"
 
 #include <assert.h>
 #include <stddef.h>
 #include <stdint.h>
 
 /* ====================================================================
  * Runtime
  * ==================================================================== */
 
 /* xco_rt_init and xco_rt_enqueue are defined inline in xco.h. */
 
 static xco_waiter_t *xco_rt_dequeue(xco_runtime_t *rt) {
     xco_waiter_t *w = rt->head;
     if (!w) return NULL;
     rt->head = w->next;
     if (!rt->head) rt->tail = NULL;
     /* Hand the waiter back fully detached. Every fire path already clears
      * prev before firing (and we just walked off the ready-queue), so
      * w->prev is NULL in practice — making it explicit here means
      * waker users can re-park without re-init, regardless of which
      * fire path resumed them. */
     w->next = NULL;
     w->prev = NULL;
     return w;
 }
 
 void xco_rt_run(xco_runtime_t *rt, uint64_t now) {
     /* The runtime ready queue holds only wakers. Other waiter
      * shapes (e.g. select_input) fire synchronously inside event
      * notify paths and never reach here.
      *
      * If a timer source is attached, advance it at the top of each
      * iteration: any timer whose deadline <= now fires, enqueueing its
      * wakers, which the inner loop then drains. A step may insert
      * a fresh already-expired timer; the outer loop catches it on the
      * next pass. Termination: each pass either drains a non-empty
      * queue or exits, and advance only fires timers it then removes,
      * so total work is bounded. */
     for (;;) {
         if (rt->timers) xco_timers_advance(rt->timers, now);
         if (!rt->head) return;
         for (xco_waiter_t *w; (w = xco_rt_dequeue(rt));) {
             xco_waker_t *sw = (xco_waker_t *)w;
             xco_step(sw->mach, sw->resume_value);
         }
     }
 }
 
 /* ---- Waker ------------------------------------------------------------ */
 
 /* Exposed (with leading underscore) so the inline xco_waker_init in
  * xco.h can install it without dragging the body into the header. */
 void xco__waker_fire(xco_waiter_t *w, uintptr_t value) {
     xco_waker_t *sw = (xco_waker_t *)w;
     sw->resume_value = value;
     xco_rt_enqueue(sw->rt, w);
 }
 
 /* ====================================================================
  * Latch
  * ==================================================================== */
 
 static bool xco_latch_poll(xco_event_t *e, uintptr_t *out, xco_waiter_t *w) {
     xco_latch_t *l = (xco_latch_t *)e;
     if (l->set) {
         if (out) *out = l->value;
         return true;
     }
     if (!w) return false;
     /* One-waiter invariant: w must arrive clean. Detach paths (xco_latch_set
      * iterator, xco_latch_unpark, xco_select_event_deinit) all leave waiters in
      * this state. A trip here means a double-park bug. */
     assert(!w->prev && !w->next);
     w->next = l->waiters;
     if (l->waiters) l->waiters->prev = w;
     l->waiters = w;
     return false;
 }
 
 static void xco_latch_unpark(xco_event_t *e, xco_waiter_t *w) {
     xco_latch_t *l = (xco_latch_t *)e;
     /* Detect "not on this list" via the invariant maintained by park
      * and the detach paths: a parked waiter has prev set OR is the
      * head; a detached one has prev == NULL and is not the head. */
     if (!w->prev && l->waiters != w) return;
     if (w->prev) w->prev->next = w->next;
     else         l->waiters    = w->next;
     if (w->next) w->next->prev = w->prev;
     w->prev = w->next = NULL;
 }
 
 /* Exposed so the inline xco_latch_init in xco.h can reference it. */
 const xco_event_vtable_t xco__latch_vt = {
     .poll   = xco_latch_poll,
     .unpark = xco_latch_unpark,
 };
 
 void xco_latch_set(xco_latch_t *l, uintptr_t value) {
     if (l->set) return;
     l->set   = true;
     l->value = value;
 
     /* Detach the whole waitlist before firing. A waiter's fire callback
      * might do anything (including unpark a sibling on another event),
      * but it cannot mutate this list — it's already gone. */
     xco_waiter_t *w = l->waiters;
     l->waiters = NULL;
     while (w) {
         xco_waiter_t *next = w->next;       /* save before xco_waiter_fire clears */
         xco_waiter_fire(w, value);
         w = next;
     }
 }
 
 /* ====================================================================
  * Semaphore
  *
  * FIFO doubly-linked waitlist, same shape as the chan_q_* helpers below
  * but specialized to a xco_semaphore_t (so we don't have to thread the
  * head/tail pair through chan_q_*).
  * ==================================================================== */
 
 static void xco_sem_q_push(xco_semaphore_t *s, xco_waiter_t *w) {
     assert(!w->prev && !w->next);
     w->prev = s->tail;
     w->next = NULL;
     if (s->tail) s->tail->next = w;
     else         s->head       = w;
     s->tail = w;
 }
 
 static xco_waiter_t *xco_sem_q_pop(xco_semaphore_t *s) {
     xco_waiter_t *w = s->head;
     if (!w) return NULL;
     s->head = w->next;
     if (s->head) s->head->prev = NULL;
     else         s->tail       = NULL;
     w->prev = w->next = NULL;
     return w;
 }
 
 static void xco_sem_q_remove(xco_semaphore_t *s, xco_waiter_t *w) {
     if (!w->prev && s->head != w) return;
     if (w->prev) w->prev->next = w->next;
     else         s->head       = w->next;
     if (w->next) w->next->prev = w->prev;
     else         s->tail       = w->prev;
     w->prev = w->next = NULL;
 }
 
 static bool xco_semaphore_poll(xco_event_t *e, uintptr_t *out, xco_waiter_t *w) {
     xco_semaphore_t *s = (xco_semaphore_t *)e;
     if (s->permits > 0) {
         s->permits--;
         if (out) *out = 1;
         return true;
     }
     if (!w) return false;
     xco_sem_q_push(s, w);
     return false;
 }
 
 static void xco_semaphore_unpark(xco_event_t *e, xco_waiter_t *w) {
     xco_semaphore_t *s = (xco_semaphore_t *)e;
     xco_sem_q_remove(s, w);
 }
 
 const xco_event_vtable_t xco__semaphore_acquire_vt = {
     .poll   = xco_semaphore_poll,
     .unpark = xco_semaphore_unpark,
 };
 
 void xco_semaphore_release(xco_semaphore_t *s, size_t n) {
     /* Hand a permit directly to each FIFO waiter, then drop any leftover
      * into the count. Direct handoff prevents a fresh try from jumping
      * the queue: an arriving acquirer that called try_ would see permits=0
      * and park behind the existing waiters until everyone ahead has been
      * served. */
     while (n > 0) {
         xco_waiter_t *w = xco_sem_q_pop(s);
         if (!w) break;
         n--;
         /* Fire value is conventional 1 — "you got a permit". Step-waiter
          * users ignore the value; select inputs capture it as the input's
          * value field. */
         xco_waiter_fire(w, 1);
     }
     s->permits += n;
 }
 
 /* ====================================================================
  * Select / all-of
  * ==================================================================== */
 
 /* One fire callback serves both modes. A counter `remaining` is decremented
  * on each fire; done is set when it hits 0. select inits remaining=1 (any
  * one fire closes); allof inits remaining=n (every input must fire). The
  * disarm-siblings loop is a no-op for already-fired waiters, so it runs
  * uniformly: for select it cleans up still-parked losers, for allof it
  * does nothing (every sibling is already detached by its source). */
 static void xco_select_input_fire(xco_waiter_t *w, uintptr_t value) {
     xco_select_input_t *in = (xco_select_input_t *)w;
     xco_select_event_t *s  = in->parent;
 
     /* Defensive: guard against any straggler that escaped disarm. */
     if (s->done.set) return;
     /* Capture the input's payload before resuming anyone. Sticky
      * sources also keep it on themselves; transient sources (channels)
      * deliver only here, so this is the only durable record. */
     in->value = value;
     if (--s->remaining > 0) return;
 
     size_t i = (size_t)(in - s->inputs);
     /* Disarm anyone still parked so their waiters don't dangle on input
      * waitlists past s's lifetime. Idempotent on already-detached waiters. */
     for (size_t j = 0; j < s->n; j++) {
         if (j != i) xco_event_unpark(s->inputs[j].src, &s->inputs[j].w);
     }
     xco_latch_set(&s->done, i);
 }
 
 void xco_select_event_init(xco_select_event_t *s,
                        xco_select_input_t *inputs, size_t n,
                        xco_event_t *const *srcs) {
     xco_latch_init(&s->done);
     s->inputs    = inputs;
     s->n         = n;
     s->remaining = 1;       /* any one fire closes the wait */
 
     /* Fast path: an input already ready. Fire and skip parking entirely
      * so deinit has nothing to disarm. */
     for (size_t i = 0; i < n; i++) {
         uintptr_t v;
         if (xco_event_poll(srcs[i], &v, NULL)) {
             inputs[i].value = v;       /* captured for inputs[winner].value */
             xco_latch_set(&s->done, i);
             return;
         }
     }
 
     for (size_t i = 0; i < n; i++) {
         inputs[i].w.next = NULL;
         inputs[i].w.prev = NULL;
         inputs[i].w.fire = xco_select_input_fire;
         inputs[i].src    = srcs[i];
         inputs[i].parent = s;
         inputs[i].value  = 0;
         (void)xco_event_poll(srcs[i], NULL, &inputs[i].w);
     }
 }
 
 void xco_allof_event_init(xco_select_event_t *s,
                       xco_select_input_t *inputs, size_t n,
                       xco_event_t *const *srcs) {
     xco_latch_init(&s->done);
     s->inputs    = inputs;
     s->n         = n;
     s->remaining = n;       /* every input must fire to close */
 
     if (n == 0) { xco_latch_set(&s->done, 0); return; }
 
     /* Initialize each input then poll. Fused: an already-ready input is
      * consumed inline (value captured, remaining--, no parking); the
      * rest end up parked by the same call. If everyone was inline, fire
      * done at the end. */
     for (size_t i = 0; i < n; i++) {
         inputs[i].w.next = NULL;
         inputs[i].w.prev = NULL;
         inputs[i].w.fire = xco_select_input_fire;
         inputs[i].src    = srcs[i];
         inputs[i].parent = s;
         inputs[i].value  = 0;
 
         uintptr_t v;
         if (xco_event_poll(srcs[i], &v, &inputs[i].w)) {
             inputs[i].value = v;
             s->remaining--;
         }
     }
 
     /* All inline-ready: fire done with the last input's index, matching
      * the "closing index" semantics of the parked path. */
     if (s->remaining == 0) xco_latch_set(&s->done, n - 1);
 }
 
 void xco_select_event_deinit(xco_select_event_t *s) {
     /* done.set => the closing fire already disarmed everyone (or the
      * fast path skipped parking entirely). Otherwise — possible after a
      * partial allof — some inputs may still be parked; unpark is
      * idempotent for already-detached waiters. */
     if (s->done.set) return;
     for (size_t i = 0; i < s->n; i++) {
         xco_event_unpark(s->inputs[i].src, &s->inputs[i].w);
     }
 }
 
 /* ====================================================================
  * FIFO waitlist helpers (shared by queue and notify)
  *
  * Doubly-linked FIFO push/pop. Same shape as latch's list operations
  * but with an explicit tail so arrival order is preserved (unlike
  * latch, where waiter order is irrelevant). The xco_chan_q_* names are
  * historical — the chan code that originally used them is now folded
  * into queue (chan = queue at cap=0 + BLOCK).
  * ==================================================================== */
 
 static void xco_chan_q_push(xco_waiter_t **head, xco_waiter_t **tail, xco_waiter_t *w) {
     assert(!w->prev && !w->next);
     w->prev = *tail;
     w->next = NULL;
     if (*tail) (*tail)->next = w;
     else       *head         = w;
     *tail = w;
 }
 
 static xco_waiter_t *xco_chan_q_pop(xco_waiter_t **head, xco_waiter_t **tail) {
     xco_waiter_t *w = *head;
     if (!w) return NULL;
     *head = w->next;
     if (*head) (*head)->prev = NULL;
     else       *tail         = NULL;
     w->prev = w->next = NULL;
     return w;
 }
 
 static void xco_chan_q_remove(xco_waiter_t **head, xco_waiter_t **tail, xco_waiter_t *w) {
     /* Same not-on-list test as xco_latch_unpark: a queued waiter has prev
      * set OR is the head; a detached one has prev == NULL and isn't
      * the head. */
     if (!w->prev && *head != w) return;
     if (w->prev) w->prev->next = w->next;
     else         *head         = w->next;
     if (w->next) w->next->prev = w->prev;
     else         *tail         = w->prev;
     w->prev = w->next = NULL;
 }
 
 
 /* ====================================================================
  * Queue
  *
  * The FIFO list helpers (xco_chan_q_push/pop/remove) are reused for the
  * queue's send and recv waitlists — same shape, same invariants. The
  * ring buffer lives in caller-provided storage; we just track head and
  * len. cap == 0 leaves the buffer logic dormant: every send either
  * direct-hands or parks, every recv either takes from a parked sender
  * or parks — i.e. it degenerates to chan rendezvous.
  * ==================================================================== */
 
 static inline xco_queue_t *xco_queue_of_recv(xco_event_t *e) {
     return (xco_queue_t *)((char *)e - offsetof(xco_queue_t, recv));
 }
 
 static inline void xco_queue_push_buf(xco_queue_t *q, uintptr_t v) {
     assert(q->len < q->cap);
     q->buf[(q->head + q->len) % q->cap] = v;
     q->len++;
 }
 
 static inline uintptr_t xco_queue_pop_buf(xco_queue_t *q) {
     assert(q->len > 0);
     uintptr_t v = q->buf[q->head];
     q->head = (q->head + 1) % q->cap;
     q->len--;
     return v;
 }
 
 /* Pop one parked sender's value into the now-free buffer slot, firing
  * the sender. Caller must have ensured a free slot exists (just popped
  * from the buffer, or cap > len). Maintains FIFO across the buffer +
  * sender-waitlist boundary: oldest buffered values come out before any
  * sender's value (which was queued later). No-op if no sender parked. */
 static void xco_queue_drain_one_sender(xco_queue_t *q) {
     if (!q->send_head) return;
     xco_waiter_t *w = xco_chan_q_pop(&q->send_head, &q->send_tail);
     xco_queue_send_waiter_t *qsw = (xco_queue_send_waiter_t *)w;
     xco_queue_push_buf(q, qsw->value);
     qsw->delivered = true;
     /* Fire after pushing so the sender sees its delivery as complete. */
     xco_waiter_fire(w, 0);
 }
 
 static bool xco_queue_recv_poll(xco_event_t *e, uintptr_t *out, xco_waiter_t *w) {
     xco_queue_t *q = xco_queue_of_recv(e);
     if (q->len > 0) {
         uintptr_t v = xco_queue_pop_buf(q);
         if (out) *out = v;
         xco_queue_drain_one_sender(q);
         return true;
     }
     /* Empty buffer. If a sender is parked here it can only mean cap==0
      * (otherwise the sender would have used the buffer). Hand directly. */
     if (q->send_head) {
         xco_waiter_t *sender = xco_chan_q_pop(&q->send_head, &q->send_tail);
         xco_queue_send_waiter_t *qsw = (xco_queue_send_waiter_t *)sender;
         if (out) *out = qsw->value;
         qsw->delivered = true;
         xco_waiter_fire(sender, 0);
         return true;
     }
     /* Closed and drained: receivers learn EOF via xco_queue_recv; out is
      * undefined. */
     if (q->closed) {
         if (out) *out = 0;
         return true;
     }
     if (!w) return false;
     xco_chan_q_push(&q->recv_head, &q->recv_tail, w);
     return false;
 }
 
 static void xco_queue_recv_unpark(xco_event_t *e, xco_waiter_t *w) {
     xco_queue_t *q = xco_queue_of_recv(e);
     xco_chan_q_remove(&q->recv_head, &q->recv_tail, w);
 }
 
 const xco_event_vtable_t xco__queue_recv_vt = {
     .poll   = xco_queue_recv_poll,
     .unpark = xco_queue_recv_unpark,
 };
 
 xco_queue_send_status_t xco_queue_send_poll(xco_queue_t *q, uintptr_t value,
                                             xco_queue_send_waiter_t *qsw) {
     /* Closed is closed regardless of policy: caller learns the truth. */
     if (q->closed) return XCO_QSEND_CLOSED;
 
     /* Direct handoff first: parked receivers always win over the buffer.
      * This is the rendezvous case and the cap==0 case. */
     xco_waiter_t *w = xco_chan_q_pop(&q->recv_head, &q->recv_tail);
     if (w) {
         xco_waiter_fire(w, value);
         return XCO_QSEND_ACCEPTED;
     }
     if (q->len < q->cap) {
         xco_queue_push_buf(q, value);
         return XCO_QSEND_ACCEPTED;
     }
     /* Buffer full and no waiting receiver. */
     switch (q->policy) {
     case XCO_QUEUE_BLOCK:
         if (!qsw) return XCO_QSEND_BLOCKED;
         qsw->value = value;
         xco_chan_q_push(&q->send_head, &q->send_tail, &qsw->sw.base);
         return XCO_QSEND_BLOCKED;
     case XCO_QUEUE_DROP_NEWEST:
         return XCO_QSEND_ACCEPTED;
     case XCO_QUEUE_DROP_OLDEST:
         (void)xco_queue_pop_buf(q);
         xco_queue_push_buf(q, value);
         return XCO_QSEND_ACCEPTED;
     }
     __builtin_unreachable();
 }
 
 void xco_queue_send_unpark(xco_queue_t *q, xco_queue_send_waiter_t *qsw) {
     xco_chan_q_remove(&q->send_head, &q->send_tail, &qsw->sw.base);
 }
 
 xco_recv_status_t xco_queue_recv(xco_queue_t *q, uintptr_t *out) {
     if (q->len > 0) {
         uintptr_t v = xco_queue_pop_buf(q);
         if (out) *out = v;
         xco_queue_drain_one_sender(q);
         return XCO_RECV_GOT;
     }
     if (q->send_head) {
         xco_waiter_t *w = xco_chan_q_pop(&q->send_head, &q->send_tail);
         xco_queue_send_waiter_t *qsw = (xco_queue_send_waiter_t *)w;
         if (out) *out = qsw->value;
         qsw->delivered = true;
         xco_waiter_fire(w, 0);
         return XCO_RECV_GOT;
     }
     if (q->closed) return XCO_RECV_CLOSED;
     return XCO_RECV_EMPTY;
 }
 
 void xco_queue_close(xco_queue_t *q) {
     if (q->closed) return;
     q->closed = true;
 
     /* Drain parked senders with delivered=false. Senders only park
      * under BLOCK, so this is no-op for DROP_* (their waitlist is
      * always empty). */
     xco_waiter_t *w;
     while ((w = xco_chan_q_pop(&q->send_head, &q->send_tail)) != NULL) {
         xco_queue_send_waiter_t *qsw = (xco_queue_send_waiter_t *)w;
         qsw->delivered = false;
         xco_waiter_fire(w, 0);
     }
     /* Wake parked receivers so they can observe closed via xco_queue_recv.
      * Receivers may still drain buffered values first — xco_queue_recv's
      * XCO_RECV_GOT path is hit before the XCO_RECV_CLOSED branch. */
     while ((w = xco_chan_q_pop(&q->recv_head, &q->recv_tail)) != NULL) {
         xco_waiter_fire(w, 0);
     }
 }
 
 /* ---- Queue send op (selectable send) ---------------------------------- */
 
 void xco__queue_send_op_fire(xco_waiter_t *w, uintptr_t value) {
     (void)value;
     xco_queue_send_op_t *op = (xco_queue_send_op_t *)w;
     xco_latch_set(&op->done, op->qsw.delivered ? 1 : 0);
 }
 
 void xco_queue_send_op_init(xco_queue_send_op_t *op, xco_queue_t *q, uintptr_t value) {
     xco_queue_send_waiter_init(&op->qsw, NULL, NULL);
     op->qsw.sw.base.fire = xco__queue_send_op_fire;
     op->queue = q;
     xco_latch_init(&op->done);
 
     switch (xco_queue_send_poll(q, value, &op->qsw)) {
     case XCO_QSEND_ACCEPTED:
         op->qsw.delivered = true;
         xco_latch_set(&op->done, 1);
         return;
     case XCO_QSEND_CLOSED:
         xco_latch_set(&op->done, 0);
         return;
     case XCO_QSEND_BLOCKED:
         /* Only BLOCK + full buffer; parked. */
         return;
     }
 }
 
 /* ====================================================================
  * Broadcast (slot)
  *
  * The waitlist uses the same doubly-linked LIFO shape as latch — there
  * is no FIFO requirement because publish wakes everyone at once. The
  * key difference from latch is publish vs set: publish never marks a
  * sticky bit on the event, so try always returns false (subscribers
  * wait for the *next* publish), and the waitlist is reusable across
  * publishes — subscribers re-park to receive subsequent values.
  * ==================================================================== */
 
 static bool xco_broadcast_poll(xco_event_t *e, uintptr_t *out, xco_waiter_t *w) {
     (void)out;
     /* Transient: never reports ready; just parks if asked. */
     if (!w) return false;
     xco_broadcast_t *b = (xco_broadcast_t *)e;
     assert(!w->prev && !w->next);
     w->next = b->waiters;
     if (b->waiters) b->waiters->prev = w;
     b->waiters = w;
     return false;
 }
 
 static void xco_broadcast_unpark(xco_event_t *e, xco_waiter_t *w) {
     xco_broadcast_t *b = (xco_broadcast_t *)e;
     if (!w->prev && b->waiters != w) return;
     if (w->prev) w->prev->next = w->next;
     else         b->waiters    = w->next;
     if (w->next) w->next->prev = w->prev;
     w->prev = w->next = NULL;
 }
 
 const xco_event_vtable_t xco__broadcast_vt = {
     .poll   = xco_broadcast_poll,
     .unpark = xco_broadcast_unpark,
 };
 
 void xco_broadcast_publish(xco_broadcast_t *b, uintptr_t value) {
     b->has_value = true;
     b->value     = value;
 
     /* Detach the waitlist before iterating — same hazard-free pattern as
      * xco_latch_set. A waiter's fire callback may re-park itself on us (the
      * common case for a re-arming subscriber); decoupling means that
      * re-park lands on a fresh waitlist, not on the snapshot we're
      * walking. */
     xco_waiter_t *w = b->waiters;
     b->waiters = NULL;
     while (w) {
         xco_waiter_t *next = w->next;       /* save before xco_waiter_fire clears */
         xco_waiter_fire(w, value);
         w = next;
     }
 }
 
 /* ====================================================================
  * Notify
  *
  * Doubly-linked FIFO waitlist (same shape as the chan/queue waitlists).
  * xco_notify_one fires the head; xco_notify_all detaches the whole list before
  * iterating so callbacks can re-park onto a fresh waitlist without
  * iterator hazards (same pattern as xco_latch_set). xco_event_poll never
  * reports ready: notify is purely transient.
  * ==================================================================== */
 
 static bool xco_notify_poll(xco_event_t *e, uintptr_t *out, xco_waiter_t *w) {
     (void)out;
     if (!w) return false;
     xco_notify_t *n = (xco_notify_t *)e;
     xco_chan_q_push(&n->head, &n->tail, w);
     return false;
 }
 
 static void xco_notify_unpark(xco_event_t *e, xco_waiter_t *w) {
     xco_notify_t *n = (xco_notify_t *)e;
     xco_chan_q_remove(&n->head, &n->tail, w);
 }
 
 const xco_event_vtable_t xco__notify_vt = {
     .poll   = xco_notify_poll,
     .unpark = xco_notify_unpark,
 };
 
 void xco_notify_one(xco_notify_t *n) {
     xco_waiter_t *w = xco_chan_q_pop(&n->head, &n->tail);
     if (!w) return;
     xco_waiter_fire(w, 0);
 }
 
 void xco_notify_all(xco_notify_t *n) {
     /* Detach before iterating: re-parking inside fire lands on a fresh
      * (empty) list. Walk the snapshot via saved next pointers. */
     xco_waiter_t *w = n->head;
     n->head = n->tail = NULL;
     while (w) {
         xco_waiter_t *next = w->next;
         xco_waiter_fire(w, 0);
         w = next;
     }
 }
 
 /* ====================================================================
  * Pairing heap
  *
  * Standard intrusive pairing heap. Each node carries three link fields:
  *
  *   child : head of children sibling list (NULL if leaf).
  *   prev  : parent if this node is its parent's first child;
  *           previous sibling otherwise; NULL only for a detached tree
  *           root (including h->root).
  *   next  : next sibling, or NULL for the last child / a detached root.
  *
  * The "prev points to either parent or a sibling" trick lets us splice
  * a node out in O(1) without an extra parent pointer: parent->child==n
  * distinguishes "first child" from "non-first sibling."
  *
  * meld picks the smaller-deadline root as winner and grafts the loser
  * as its new first child. Pop-min and remove both rebuild via the
  * classic two-pass pairwise merge of the resulting children list.
  * ==================================================================== */
 
 /* Merge two detached subtree roots (each with prev=next=NULL). Returns
  * the merged root (also detached: prev=next=NULL). */
 static xco_timer_t *xco_ph_meld(xco_timer_t *a, xco_timer_t *b) {
     if (!a) return b;
     if (!b) return a;
     xco_timer_t *small, *large;
     if (a->deadline <= b->deadline) { small = a; large = b; }
     else                            { small = b; large = a; }
     /* Graft `large` as the new first child of `small`. */
     large->next = small->child;
     if (small->child) small->child->prev = large;
     large->prev  = small;       /* parent link via prev */
     small->child = large;
     small->prev  = NULL;
     small->next  = NULL;
     return small;
 }
 
 /* Two-pass pairwise meld of a children sibling list. Detaches each node
  * before melding so meld inputs satisfy its prev=next=NULL contract.
  * The output has prev=next=NULL. */
 static xco_timer_t *xco_ph_merge_pairs(xco_timer_t *first) {
     /* Pass 1: walk the sibling list left-to-right, melding consecutive
      * pairs. Chain results via `next` (ab)use as a temporary list link. */
     xco_timer_t *list = NULL;
     while (first) {
         xco_timer_t *a = first;
         xco_timer_t *b = a->next;
         xco_timer_t *rest = b ? b->next : NULL;
         a->prev = a->next = NULL;
         if (b) { b->prev = b->next = NULL; }
         xco_timer_t *m = xco_ph_meld(a, b);
         m->next = list;          /* prepend to pass-1 list */
         list = m;
         first = rest;
     }
     /* Pass 2: meld the pass-1 list into a single root. */
     xco_timer_t *acc = NULL;
     while (list) {
         xco_timer_t *nxt = list->next;
         list->prev = NULL;
         list->next = NULL;
         acc = xco_ph_meld(acc, list);
         list = nxt;
     }
     return acc;
 }
 
 /* Detach n from the tree (must currently be in the heap). Returns the
  * (possibly new) main-heap root. n is left fully detached: child still
  * points to its subtree, but prev/next are NULL — caller decides what
  * to do with that subtree. */
 static xco_timer_t *xco_ph_detach(xco_pairing_heap_t *h, xco_timer_t *n) {
     if (h->root == n) {
         /* n is the main root; pop it and rebuild from its children. */
         xco_timer_t *new_root = xco_ph_merge_pairs(n->child);
         n->child = NULL;
         n->prev  = NULL;
         n->next  = NULL;
         return new_root;
     }
     /* n has a parent (recorded via prev — either as its parent's first
      * child or as some sibling's successor). Splice out of the sibling
      * list. */
     if (n->prev->child == n) {
         /* First child: parent's child link skips us. */
         n->prev->child = n->next;
     } else {
         /* Mid/last sibling: previous sibling's next skips us. */
         n->prev->next = n->next;
     }
     if (n->next) n->next->prev = n->prev;
     n->prev = NULL;
     n->next = NULL;
     /* The main-heap root is unchanged structurally; the caller of this
      * function decides how (or whether) to reintroduce n's subtree. */
     return h->root;
 }
 
 static void xco_ph_insert(xco_timers_t *ts, xco_timer_t *t) {
     xco_pairing_heap_t *h = (xco_pairing_heap_t *)ts;
     /* Singleton tree (prev/next/child already NULL via xco_timer_init). */
     h->root = xco_ph_meld(h->root, t);
     t->in_heap = true;
 }
 
 static void xco_ph_cancel(xco_timers_t *ts, xco_timer_t *t) {
     xco_pairing_heap_t *h = (xco_pairing_heap_t *)ts;
     if (!t->in_heap) return;
     h->root = xco_ph_detach(h, t);
     /* Now meld t's subtree (its children) back into the main heap. */
     xco_timer_t *sub = xco_ph_merge_pairs(t->child);
     t->child = NULL;
     h->root = xco_ph_meld(h->root, sub);
     t->in_heap = false;
 }
 
 static void xco_ph_advance(xco_timers_t *ts, uint64_t now) {
     xco_pairing_heap_t *h = (xco_pairing_heap_t *)ts;
     /* Pop while the min-key timer is due. Each fire may run callbacks
      * that insert *new* timers (with later deadlines, normally) — those
      * land back in the heap, and we keep checking the root. */
     while (h->root && h->root->deadline <= now) {
         xco_timer_t *t = h->root;
         h->root = xco_ph_merge_pairs(t->child);
         t->child   = NULL;
         t->prev    = NULL;
         t->next    = NULL;
         t->in_heap = false;
         /* Trigger the latch: drains the waitlist and delivers the
          * deadline as the fire payload. */
         xco_latch_set(&t->done, (uintptr_t)t->deadline);
     }
 }
 
 static uint64_t xco_ph_peek(const xco_timers_t *ts) {
     const xco_pairing_heap_t *h = (const xco_pairing_heap_t *)ts;
     return h->root ? h->root->deadline : UINT64_MAX;
 }
 
 const xco_timers_vtable_t xco__pairing_heap_vt = {
     .insert  = xco_ph_insert,
     .cancel  = xco_ph_cancel,
     .advance = xco_ph_advance,
     .peek    = xco_ph_peek,
 };
 
 /* ====================================================================
  * Timeout
  * ==================================================================== */
 
 /* Bridge waiter: parked on the timer's latch, fires the cancel when the
  * timer fires. Two-step indirection so cancel can already have its own
  * waiters (e.g. a xco_wait_or_cancel select) without the timer's waitlist
  * caring about cancel internals. */
 static void xco__timeout_bridge_fire(xco_waiter_t *w, uintptr_t value) {
     (void)value;
     xco_timeout_t *to = (xco_timeout_t *)((char *)w - offsetof(xco_timeout_t, bridge));
     xco_cancel_set(&to->cancel);
 }
 
 void xco_timeout_init(xco_timeout_t *to, xco_timers_t *ts, uint64_t deadline) {
     xco_timer_init(&to->timer, ts, deadline);
     xco_cancel_init(&to->cancel);
     to->bridge.next = NULL;
     to->bridge.prev = NULL;
     to->bridge.fire = xco__timeout_bridge_fire;
     /* Park the bridge on the timer. If the timer fires, xco_latch_set
      * detaches the bridge and calls our fire callback inline. The timer
      * was just inserted into the heap and hasn't been advanced, so poll
      * always parks here (return value ignored). */
     (void)xco_event_poll(xco_timer_event(&to->timer), NULL, &to->bridge);
 }
 
 /* ====================================================================
  * Ticker
  *
  * The ticker's event surface uses the broadcast-style LIFO doubly-linked
  * waitlist: subscribers are fired all-at-once on each tick, so order
  * doesn't matter; doubly-linked gives O(1) unpark for cancellation.
  * ==================================================================== */
 
 static bool xco_ticker_poll(xco_event_t *e, uintptr_t *out, xco_waiter_t *w) {
     (void)out;
     /* Transient — never reports ready; just parks if asked. */
     if (!w) return false;
     xco_ticker_t *t = (xco_ticker_t *)((char *)e - offsetof(xco_ticker_t, base));
     assert(!w->prev && !w->next);
     w->next = t->waiters;
     if (t->waiters) t->waiters->prev = w;
     t->waiters = w;
     return false;
 }
 
 static void xco_ticker_unpark(xco_event_t *e, xco_waiter_t *w) {
     xco_ticker_t *t = (xco_ticker_t *)((char *)e - offsetof(xco_ticker_t, base));
     if (!w->prev && t->waiters != w) return;
     if (w->prev) w->prev->next = w->next;
     else         t->waiters    = w->next;
     if (w->next) w->next->prev = w->prev;
     w->prev = w->next = NULL;
 }
 
 const xco_event_vtable_t xco__ticker_vt = {
     .poll   = xco_ticker_poll,
     .unpark = xco_ticker_unpark,
 };
 
 /* Bridge waiter: parks on the underlying timer's latch. On fire, compute
  * the next deadline (skip-ahead-safe), reinstall the timer, re-park the
  * bridge on the new timer, then fire every parked subscriber with the
  * just-fired deadline. The waitlist is detached before iteration so
  * subscribers can re-park inside their fire callbacks. */
 static void xco__ticker_bridge_fire(xco_waiter_t *w, uintptr_t value) {
     xco_ticker_t *t = (xco_ticker_t *)((char *)w - offsetof(xco_ticker_t, bridge));
     uint64_t  fired  = (uint64_t)value;
     uint64_t  next   = fired + t->period;
     /* Skip-ahead: in the rare overflow case (period = 0 or wraparound),
      * step forward enough to keep next > fired. */
     if (next <= fired) {
         next += ((fired - next) / t->period + 1) * t->period;
     }
     /* Reinstall the timer for the next tick. The latch's storage is
      * reused — xco_timer_init runs xco_latch_init on it. */
     xco_timer_init(&t->timer, t->src, next);
     /* Bridge waiter is fully detached (xco_waiter_fire just cleared its
      * links); park it on the freshly-armed timer (which has not yet been
      * advanced, so poll parks). */
     (void)xco_event_poll(xco_timer_event(&t->timer), NULL, &t->bridge);
 
     /* Fire the subscribers. Detach the waitlist first so re-park inside
      * fire lands on the now-empty list. */
     xco_waiter_t *waiters = t->waiters;
     t->waiters = NULL;
     while (waiters) {
         xco_waiter_t *nxt = waiters->next;
         xco_waiter_fire(waiters, (uintptr_t)fired);
         waiters = nxt;
     }
 }
 
 void xco_ticker_init(xco_ticker_t *t, xco_timers_t *ts,
                  uint64_t period, uint64_t first_deadline) {
     /* period must be positive — the skip-ahead computation in the bridge
      * divides by period, and a zero-period ticker would loop forever
      * inside xco_ph_advance. */
     assert(period > 0);
     t->base.vt   = &xco__ticker_vt;
     t->src       = ts;
     t->period    = period;
     t->waiters   = NULL;
 
     xco_timer_init(&t->timer, ts, first_deadline);
 
     t->bridge.next = NULL;
     t->bridge.prev = NULL;
     t->bridge.fire = xco__ticker_bridge_fire;
     (void)xco_event_poll(xco_timer_event(&t->timer), NULL, &t->bridge);
 }
 
 void xco_ticker_deinit(xco_ticker_t *t) {
     /* Pull the bridge off the timer (no-op if already fired) and cancel
      * the timer. Subscribers' waiters are the caller's storage; nothing
      * to free here. */
     xco_event_unpark(xco_timer_event(&t->timer), &t->bridge);
     xco_timer_deinit(&t->timer);
 }
 
 /* ====================================================================
  * Task group
  *
  * Each attach contributes one to the countdown and parks a bridge waiter
  * on the task's done event. Bridge fire splices the slot out of the
  * group's list and decrements the countdown — the join event fires when
  * the last attached task finishes.
  *
  * Cancellation is fan-out: walk the list, set each task's cancel, then
  * set the group-level cancel. Bodies cooperate by composing their work
  * with xco_task_cancel(self); the group-level cancel is for non-task
  * waiters that want to react to "the group has been told to stop."
  * ==================================================================== */
 
 static void xco__task_group_detach_slot(xco_task_group_t *g, xco_group_attach_t *slot) {
     /* Doubly-linked, head/tail tracked; same shape as other waitlists. */
     if (slot->prev) slot->prev->next = slot->next;
     else            g->head          = slot->next;
     if (slot->next) slot->next->prev = slot->prev;
     else            g->tail          = slot->prev;
     slot->prev = slot->next = NULL;
 }
 
 static void xco__task_group_bridge_fire(xco_waiter_t *w, uintptr_t value) {
     (void)value;
     xco_group_attach_t *slot = (xco_group_attach_t *)((char *)w - offsetof(xco_group_attach_t, bridge));
     xco_task_group_t   *g    = slot->group;
     xco__task_group_detach_slot(g, slot);
     xco_countdown_done(&g->pending);
 }
 
 void xco_task_group_init(xco_task_group_t *g) {
     /* Don't go through xco_countdown_init(0) — that fires the latch
      * immediately, which would make the very first attach's
      * xco_countdown_add UB. The group's join must remain not-fired until at
      * least one attached task has finished, so we open with
      * remaining=0 and an unset latch. The first attach lifts remaining
      * to 1, and matching countdown_dones bring it back to 0, firing
      * the latch. */
     xco_latch_init(&g->pending.done);
     g->pending.remaining = 0;
     xco_cancel_init(&g->cancel);
     g->head = g->tail = NULL;
 }
 
 void xco_task_group_attach(xco_task_group_t *g, xco_task_t *t, xco_group_attach_t *slot) {
     xco_countdown_add(&g->pending, 1);
 
     slot->task  = t;
     slot->group = g;
 
     /* Append to the group's list (FIFO; ordering doesn't affect cancel
      * fan-out semantics, but consistent with other waitlists in the
      * codebase). */
     slot->prev = g->tail;
     slot->next = NULL;
     if (g->tail) g->tail->next = slot;
     else         g->head       = slot;
     g->tail = slot;
 
     slot->bridge.next = NULL;
     slot->bridge.prev = NULL;
     slot->bridge.fire = xco__task_group_bridge_fire;
 
     /* Park on the task's done event. Re-attaching a finished task is UB
      * per the contract; the latch's clean-waiter assert inside poll
      * would catch a double-park. */
     (void)xco_event_poll(xco_task_done_event(t), NULL, &slot->bridge);
 }
 
 void xco_task_group_cancel(xco_task_group_t *g) {
     /* Fan-out cancel: signal each attached task. Walk the snapshot
      * (cancel doesn't detach the slot — only task done does — so the
      * list is stable across iteration). */
     for (xco_group_attach_t *s = g->head; s; s = s->next) {
         xco_cancel_set(&s->task->cancel);
     }
     /* Group-level cancel for anyone awaiting "the group as a whole." */
     xco_cancel_set(&g->cancel);
 }
 
 /* ====================================================================
  * xco — coroutines
  * ==================================================================== */
 
 typedef struct xco_impl {
     xco_mach_t             base;        /* must be first; aliases xco_coro_t.base */
     _Alignas(XCO__CTX_ALIGN) unsigned char ctx_buf[XCO__CTX_SIZE];
     xco_platform_ctx_t *resumer_ctx; /* where suspend/return goes */
     xco_fn              fn;
 } xco_impl_t;
 
 _Static_assert(sizeof(xco_impl_t)   <= sizeof(xco_coro_t),   "xco_coro_t too small");
 _Static_assert(_Alignof(xco_impl_t) <= _Alignof(xco_coro_t), "xco_coro_t under-aligned");
 
 static inline xco_impl_t         *xco_impl_of(xco_coro_t *c)         { return (xco_impl_t *)c; }
 static inline xco_platform_ctx_t *xco_ctx_of(xco_impl_t *ci) {
     return (xco_platform_ctx_t *)ci->ctx_buf;
 }
 
 /* Per-thread state. */
 static _Thread_local xco_impl_t *xco_coro_current = NULL;
 static _Thread_local _Alignas(XCO__CTX_ALIGN)
        unsigned char xco_t_main_ctx_buf[XCO__CTX_SIZE];
 static inline xco_platform_ctx_t *xco_main_ctx(void) {
     return (xco_platform_ctx_t *)xco_t_main_ctx_buf;
 }
 
 /* Trampoline: runs on the coroutine's own stack, invoked by the
  * platform layer on the first switch into a fresh context. The
  * argument is the value passed to that first xco_step. The coroutine
  * identifies itself via xco_coro_current, set by the resumer just before
  * switching. */
 static void xco_trampoline(uintptr_t arg) {
     xco_impl_t *self = xco_coro_current;
     uintptr_t   ret  = self->fn(arg);
 
     self->base.status = XCO_STEP_DEAD;
     (void)xco_platform_switch(xco_ctx_of(self), self->resumer_ctx, ret);
     __builtin_unreachable();
 }
 
 /* xco_step_fn entry point wired into base.step at init time. All callers
  * — generic xco_step consumers and xco-aware code alike — route through
  * here. */
 static xco_step_result_t xco_co_step(xco_mach_t *s, uintptr_t value) {
     xco_impl_t *next = (xco_impl_t *)s;
     assert(next->base.status == XCO_STEP_INIT || next->base.status == XCO_STEP_SUSPENDED);
 
     xco_impl_t *prev  = xco_coro_current;
     next->resumer_ctx = prev ? xco_ctx_of(prev) : xco_main_ctx();
     next->base.status = XCO_STEP_RUNNING;
     xco_coro_current         = next;
 
     uintptr_t back = xco_platform_switch(next->resumer_ctx,
                                          xco_ctx_of(next), value);
 
     /* Coroutine has either suspended or returned; status is already
      * set correctly by xco_suspend or by the xco_trampoline. */
     xco_coro_current = prev;
     return (xco_step_result_t){ .value = back, .status = next->base.status };
 }
 
 void xco_init(xco_coro_t *c, xco_fn fn,
               void *stack_base, size_t stack_len) {
     xco_impl_t *ci  = xco_impl_of(c);
     ci->base.step   = xco_co_step;
     ci->base.status = XCO_STEP_INIT;
     ci->fn          = fn;
     ci->resumer_ctx = NULL;
     xco_platform_init(xco_ctx_of(ci), stack_base, stack_len, xco_trampoline);
 }
 
 uintptr_t xco_suspend(uintptr_t value) {
     xco_impl_t *self = xco_coro_current;
     assert(self != NULL);
     self->base.status = XCO_STEP_SUSPENDED;
     return xco_platform_switch(xco_ctx_of(self), self->resumer_ctx, value);
 }
 
 xco_coro_t *xco_self(void) {
     return (xco_coro_t *)xco_coro_current;
 }
 
 /* ---- xco-backed task -------------------------------------------------- */
 
 /* The xco_trampoline: runs as the coroutine's xco_fn. Recovers the owning
  * xco_cotask_t via container_of on the embedded co (xco_self() returns
  * the running xco), dispatches the user fn, and surfaces the return
  * value through xco_task_done — so a joiner waiting on xco_task_done_event
  * wakes with the body's return without the body needing to know about
  * the task surface at all. */
 static uintptr_t xco_cotask_trampoline(uintptr_t arg) {
     xco_coro_t      *self = xco_self();
     xco_cotask_t *xt   = (xco_cotask_t *)((char *)self - offsetof(xco_cotask_t, co));
     uintptr_t   r    = xt->fn(&xt->task, arg);
     xco_task_done(&xt->task, r);
     return r;
 }
 
 void xco_cotask_init(xco_cotask_t *xt, xco_cotask_fn fn,
                    void *stack_base, size_t stack_len) {
     xco_task_init(&xt->task, &xt->co.base);
     xt->fn = fn;
     xco_init(&xt->co, xco_cotask_trampoline, stack_base, stack_len);
 }
 
 /* ====================================================================
  * xco_op — generic effect/IO layer
  *
  * The runtime owns a doubly-linked pending-ops list (op_head/op_tail).
  * Submit appends; cancel-while-PENDING splices in O(1); take detaches
  * the whole list in O(1) and lets the host iterate via the next pointers
  * (which are then the host's to reuse for in-flight tracking).
  *
  * Resolution always goes through xco_latch_set on op->done; awaiters
  * compose with the standard event API. Status is delivered as the latch
  * payload.
  * ==================================================================== */
 
 void xco_op_submit(xco_runtime_t *rt, xco_op_t *op) {
     xco_latch_init(&op->done);
     op->cancel_requested = false;
     op->rt    = rt;
     op->epoch = rt->op_epoch;       /* matches → PENDING */
     op->next  = NULL;
     op->prev  = rt->op_tail;
     if (rt->op_tail) rt->op_tail->next = op;
     else             rt->op_head       = op;
     rt->op_tail = op;
 }
 
 void xco_op_cancel(xco_op_t *op) {
     if (op->done.set) return;            /* RESOLVED — no-op */
     if (xco_op_is_pending(op)) {
         /* PENDING: splice from rt's pending list and resolve as cancelled. */
         xco_runtime_t *rt = op->rt;
         if (op->prev) op->prev->next = op->next;
         else          rt->op_head    = op->next;
         if (op->next) op->next->prev = op->prev;
         else          rt->op_tail    = op->prev;
         op->prev = op->next = NULL;
         xco_latch_set(&op->done, (uintptr_t)XCO_OP_CANCELLED);
         return;
     }
     /* IN_FLIGHT: advisory. Host sees cancel_requested and decides whether
      * to honor it; xco_op_complete is still the final word either way. */
     op->cancel_requested = true;
 }
 
 void xco_op_complete(xco_op_t *op, xco_op_status_t status) {
     /* xco_latch_set is idempotent; second/late completion is a no-op.
      * Per the contract, host calls this only after take, so op is not
      * on rt's pending list. */
     xco_latch_set(&op->done, (uintptr_t)status);
 }
 
 xco_op_t *xco_rt_take_ops(xco_runtime_t *rt, xco_op_t **tail_out) {
     xco_op_t *head = rt->op_head;
     if (tail_out) *tail_out = rt->op_tail;
     rt->op_head = rt->op_tail = NULL;
     /* Bump the epoch: every op currently on the (now-detached) batch had
      * its epoch field set to the old value at submit, so they all flip to
      * IN_FLIGHT in O(1). Submits after this point land in the new
      * generation. */
     rt->op_epoch++;
     return head;
 }