xco

xco.h (54688B)

---

 /*
  * xco.h — minimal C11 concurrency library.
  *
  * Four layers in this header, bottom-up:
  *
  *   xco_mach_t    Generic resumable function. A value that can be
  *                 driven forward one step at a time; each step takes a
  *                 uintptr_t in, returns one out, and reports whether
  *                 the function suspended or finished. Substrate shared
  *                 by stack-switching coroutines (xco) and hand-coded
  *                 state machines.
  *
  *   waiter / event / runtime
  *                 Pollable event substrate (poll / unpark) with a
  *                 single-threaded FIFO ready queue. Concrete events:
  *                 latch, countdown, notify, semaphore/mutex, select,
  *                 allof, channel, queue, broadcast, cancel, timer,
  *                 pairing-heap timer source, timeout, ticker. All
  *                 storage caller-provided, no allocation, no atomics.
  *
  *   xco_task_t    Lifecycle handle for a running xco_step. Bundles a
  *                 done latch (fires with the step's return value) and
  *                 a cancel latch (the cooperative wind-down signal).
  *                 xco_task_group_t fans these in/out across a dynamic
  *                 set of tasks. Storage caller-provided; the step
  *                 itself lives wherever the caller put it.
  *
  *   xco_coro_t    Stack-switching coroutine. xco_coro_t embeds xco_mach_t as
  *                 its first member, so a coroutine is one concrete
  *                 kind of resumable function: generic code holding an
  *                 xco_mach_t* works on coroutines and hand-coded state
  *                 machines uniformly. Values pass between caller and
  *                 coroutine through a single uintptr_t channel; pack
  *                 richer data behind a pointer.
  *                 Asymmetric: xco_suspend always returns to the most
  *                 recent resumer. Resumes nest like function calls.
  *                 Thread portability: the platform switch saves only
  *                 callee-saved regs (no TLS register, no signal mask),
  *                 so a fully-suspended coroutine could in principle be
  *                 resumed on another thread. In practice don't: the
  *                 runtime/event/waiter substrate is single-threaded
  *                 (no atomics), so a coroutine parked on any event is
  *                 tethered to that runtime's thread; and user code
  *                 silently rebinds errno / _Thread_local / thread-affine
  *                 OS handles to whichever thread resumed it.
  *
  *   xco_cotask_t  xco specialization of xco_task_t. The xco_trampoline calls
  *                 fn(&xt->task, arg) and then xco_task_done with the
  *                 return value, so xco_wait_or_cancel-style teardown works
  *                 without the user wiring anything.
  *
  * One-waiter invariant. An xco_step, while suspended, is parked on at most
  * one event. Multi-wait is composed in the event graph: build a
  * select_event (or any future combinator — all-of, timeout, ...) and
  * park on that. The xco_step never sees more than one event directly.
  * This is what lets a single xco_waker_t live inline in the xco_step
  * and a single next/prev pair serve both event waitlists and the
  * runtime ready queue (the two list memberships are disjoint in time).
  */
 
 #ifndef XCO_H
 #define XCO_H
 
 #include <assert.h>
 #include <stdbool.h>
 #include <stddef.h>
 #include <stdint.h>
 
 /* Provides XCO_SIZE, XCO_ALIGN, XCO_STACK_ALIGN, XCO__CTX_SIZE,
  * XCO__CTX_ALIGN; resolved by the build to the platform-specific
  * copy via the include path (-Iplatform/$(PLATFORM)). */
 #include "xco_platform.h"
 
 /* ====================================================================
  * xco_step — generic resumable function interface.
  *
  * The first-member convention: embed xco_mach_t as the first field of
  * your concrete type so a pointer to your type can be passed wherever
  * an xco_mach_t * is expected.
  *
  *   typedef struct {
  *       xco_mach_t base;
  *       int     phase;
  *       ...
  *   } parser_t;
  *
  *   static xco_step_result_t parser_step(xco_mach_t *s, uintptr_t v) {
  *       parser_t *p = (parser_t *)s;
  *       switch (p->phase) {
  *       case 0: p->phase = 1; return (xco_step_result_t){v + 1, XCO_STEP_SUSPENDED};
  *       case 1:               return (xco_step_result_t){v * 2, XCO_STEP_DEAD};
  *       }
  *       __builtin_unreachable();
  *   }
  *
  *   parser_t p = { .base = {.step = parser_step, .status = XCO_STEP_INIT} };
  *   xco_step_result_t r = xco_step(&p.base, 0);
  * ==================================================================== */
 
 typedef enum {
     XCO_STEP_INIT,        /* created, never stepped */
     XCO_STEP_RUNNING,     /* inside step(), or in an active resume chain */
     XCO_STEP_SUSPENDED,   /* yielded; resumable */
     XCO_STEP_DEAD,        /* function returned */
 } xco_mach_status_t;
 
 typedef struct {
     uintptr_t      value;
     xco_mach_status_t status;
 } xco_step_result_t;
 
 typedef struct xco_mach xco_mach_t;
 typedef xco_step_result_t (*xco_step_fn)(xco_mach_t *s, uintptr_t value);
 
 struct xco_mach {
     xco_step_fn       step;
     xco_mach_status_t status;   /* cached; xco_step() syncs from each result */
 };
 
 /* Drive one step. The wrapper updates s->status from the returned
  * result so step implementations only need to populate the result. */
 static inline xco_step_result_t xco_step(xco_mach_t *s, uintptr_t value) {
     xco_step_result_t r = s->step(s, value);
     s->status = r.status;
     return r;
 }
 
 static inline xco_mach_status_t xco_mach_status(const xco_mach_t *s) {
     return s->status;
 }
 
 /* ====================================================================
  * Event substrate.
  *
  * Fused try-or-park: xco_event_poll(e, &out, w) returns true if the
  * event is ready (writes the value to *out, w is NOT parked) and false
  * otherwise (parks w on the waitlist iff w is non-NULL). The two
  * degenerate forms:
  *
  *   xco_event_poll(e, &v,  NULL)   — pure try (peek; never parks).
  *   xco_event_poll(e, NULL, w)     — park-if-not-ready (out discarded).
  *
  * Fusing closes the try/park TOCTOU window an MT impl would otherwise
  * have to internally re-check, and lets each event arm a waiter
  * atomically with its readiness check.
  *
  * Standard usage from a state machine:
  *
  *   uintptr_t v;
  *   if (xco_event_poll(e, &v, &my_waiter.base)) { ... use v ... }
  *   else                                        { return SUSPENDED; }
  *
  * Standard usage from an xco coroutine wrapper:
  *
  *   uintptr_t v;
  *   xco_waker_t sw;
  *   xco_waker_init(&sw, rt, &xco_self()->base);
  *   if (!xco_event_poll(e, &v, &sw.base)) {
  *       xco_suspend(0);
  *       (void)xco_event_poll(e, &v, NULL);  // sticky: now ready
  *   }
  * ==================================================================== */
 
 /* ---- Waiter ------------------------------------------------------------ */
 
 typedef struct xco_waiter xco_waiter_t;
 struct xco_waiter {
     /* Doubly-linked while parked on an event waitlist, so unpark is
      * O(1). Reused as the singly-linked next pointer while on the
      * runtime ready queue (FIFO, no removal from middle); prev is
      * undefined in that state and reset on the next park. */
     xco_waiter_t *next;
     xco_waiter_t *prev;
     /* Fire callback. value is the event's payload at fire time — sticky
      * events also store it on themselves, transient events (channels,
      * one-shot signals) deliver only here. Waiters that don't care
      * about the value just ignore the parameter.
      *
      * Invoke via xco_waiter_fire (below), not directly: the helper enforces
      * the "fire receives a fully detached waiter" contract that makes it
      * safe to re-park inside the callback. */
     void (*fire)(xco_waiter_t *w, uintptr_t value);
 };
 
 /* Canonical way to invoke a waiter's fire callback. Hands the callback a
  * fully detached waiter so the callback (or whatever the resumed step
  * does) can re-park on a fresh waitlist without colliding with stale
  * link state. Detachers that lead into fire (queue pops, latch drains,
  * etc.) don't need to clear prev/next themselves. */
 static inline void xco_waiter_fire(xco_waiter_t *w, uintptr_t value) {
     w->prev = NULL;
     w->next = NULL;
     w->fire(w, value);
 }
 
 /* ---- Event ------------------------------------------------------------ */
 
 typedef struct xco_event xco_event_t;
 
 typedef struct {
     /* Fused try + park. If the event is ready, write its value to *out
      * (when out != NULL), do NOT park w, and return true. Otherwise, if
      * w != NULL park it on the waitlist and return false; if w == NULL
      * just return false without parking. *out is left untouched on the
      * not-ready path. */
     bool (*poll)(xco_event_t *e, uintptr_t *out, xco_waiter_t *w);
     /* Remove w from the waitlist. Idempotent: no-op if not parked. */
     void (*unpark)(xco_event_t *e, xco_waiter_t *w);
 } xco_event_vtable_t;
 
 struct xco_event { const xco_event_vtable_t *vt; };
 
 static inline bool xco_event_poll(xco_event_t *e, uintptr_t *out, xco_waiter_t *w) {
     return e->vt->poll(e, out, w);
 }
 static inline void xco_event_unpark(xco_event_t *e, xco_waiter_t *w) { e->vt->unpark(e, w); }
 
 /* ---- Runtime ---------------------------------------------------------- */
 
 /* Forward-declared: the optional timer source attached to the runtime.
  * Defined in the timer section below. */
 typedef struct xco_timers xco_timers_t;
 
 typedef struct xco_op xco_op_t;
 
 typedef struct xco_runtime {
     xco_waiter_t  *head, *tail;
     xco_timers_t *timers;        /* optional; advanced inside xco_rt_run */
     /* Pending ops list (xco_op layer). The runtime never inspects ops; it
      * only owns this list-head pair so the host can pull a batch via
      * xco_rt_take_ops. op_epoch is bumped each take and recorded on each
      * submit; PENDING vs IN_FLIGHT is a generation match (see xco_op). */
     xco_op_t     *op_head, *op_tail;
     uint64_t      op_epoch;
 } xco_runtime_t;
 
 static inline void xco_rt_init(xco_runtime_t *rt) {
     rt->head     = rt->tail = NULL;
     rt->timers   = NULL;
     rt->op_head  = rt->op_tail = NULL;
     rt->op_epoch = 0;
 }
 
 /* Attach (or detach with NULL) a timer source. While attached, xco_rt_run
  * advances it each pass with the now value the caller supplied; firing
  * timers may enqueue more waiters, which the same xco_rt_run call then drains. */
 static inline void xco_rt_attach_timers(xco_runtime_t *rt, xco_timers_t *ts) {
     rt->timers = ts;
 }
 
 /* Append w to the ready queue. Used by xco__waker_fire and by anyone
  * else that wants a waiter resumed by the scheduler. */
 static inline void xco_rt_enqueue(xco_runtime_t *rt, xco_waiter_t *w) {
     w->next = NULL;
     if (rt->tail) rt->tail->next = w;
     else          rt->head       = w;
     rt->tail = w;
 }
 
 /* Drain the ready queue, resuming each waker's xco_step, until empty.
  * Steps may re-arm on events (and thus leave the queue) or enqueue
  * other steps; xco_rt_run keeps going until quiescent. now is forwarded to
  * any attached timer source's advance(); pass 0 (or anything) when no
  * source is attached. The library never reads a clock — now is always
  * caller-supplied. */
 void xco_rt_run(xco_runtime_t *rt, uint64_t now);
 
 /* The canonical bridge between events and the scheduler. When fired,
  * stashes the value and enqueues itself onto rt; xco_rt_run pops it and
  * calls xco_step(mach, value), so the resumed step receives the event's
  * payload directly without a re-try.
  *
  * Init once, re-park freely. The runtime hands the waiter back fully
  * detached (next/prev cleared) before invoking the resumed step, so a
  * subscriber that wants to wait on the next event can call xco_event_poll
  * directly — no re-init needed unless rt or step changes. */
 typedef struct {
     xco_waiter_t    base;
     xco_runtime_t *rt;
     xco_mach_t   *mach;
     uintptr_t  resume_value;     /* set by fire, consumed by xco_rt_run */
 } xco_waker_t;
 
 /* Defined in xco.c; declared here so xco_waker_init can install it. */
 void xco__waker_fire(xco_waiter_t *w, uintptr_t value);
 
 static inline void xco_waker_init(xco_waker_t *sw, xco_runtime_t *rt, xco_mach_t *m) {
     sw->base.next     = NULL;
     sw->base.prev     = NULL;
     sw->base.fire     = xco__waker_fire;
     sw->rt            = rt;
     sw->mach          = m;
     sw->resume_value  = 0;
 }
 
 /* ---- Latch ------------------------------------------------------------ */
 
 /* One-shot sticky event. set() flips the bit, stores the payload, and
  * fires every waiter. Subsequent set() calls are ignored. To re-arm,
  * reinitialize a fresh latch. */
 typedef struct {
     xco_event_t   base;
     bool      set;
     uintptr_t value;
     xco_waiter_t  *waiters;
 } xco_latch_t;
 
 /* Defined in xco.c; referenced by xco_latch_init. */
 extern const xco_event_vtable_t xco__latch_vt;
 
 static inline void xco_latch_init(xco_latch_t *l) {
     l->base.vt = &xco__latch_vt;
     l->set     = false;
     l->value   = 0;
     l->waiters = NULL;
 }
 
 void xco_latch_set(xco_latch_t *l, uintptr_t value);
 
 /* ---- Countdown -------------------------------------------------------- */
 
 /* One-shot fan-in counter. Fires its embedded latch (payload 0) when
  * remaining hits 0. xco_countdown_add(n) is legal while remaining > 0;
  * xco_countdown_done decrements; both are UB once the latch has fired.
  *
  * Compose with the standard event API via xco_countdown_event(). */
 typedef struct xco_countdown {
     xco_latch_t done;
     size_t  remaining;
 } xco_countdown_t;
 
 static inline void xco_countdown_init(xco_countdown_t *c, size_t n) {
     xco_latch_init(&c->done);
     c->remaining = n;
     if (n == 0) xco_latch_set(&c->done, 0);
 }
 
 static inline void xco_countdown_add(xco_countdown_t *c, size_t n) {
     assert(!c->done.set);
     c->remaining += n;
 }
 
 static inline void xco_countdown_done(xco_countdown_t *c) {
     assert(c->remaining > 0);
     if (--c->remaining == 0) xco_latch_set(&c->done, 0);
 }
 
 static inline xco_event_t *xco_countdown_event(xco_countdown_t *c) { return &c->done.base; }
 static inline bool     xco_countdown_fired(const xco_countdown_t *c) { return c->done.set; }
 
 /* ---- Notify (wake-one / wake-all) ------------------------------------- */
 
 /* Transient signal with no sticky state. xco_notify_one fires (and detaches)
  * the head of a FIFO waitlist; xco_notify_all fires every parked waiter. Both
  * are no-ops when the waitlist is empty. Subscribers must re-park to see
  * subsequent notifications.
  *
  * xco_event_poll never reports ready: there is no "ready now" state — a
  * subscriber waits for the *next* notify. */
 typedef struct xco_notify {
     xco_event_t  base;
     xco_waiter_t *head, *tail;
 } xco_notify_t;
 
 extern const xco_event_vtable_t xco__notify_vt;
 
 static inline void xco_notify_init(xco_notify_t *n) {
     n->base.vt = &xco__notify_vt;
     n->head    = n->tail = NULL;
 }
 
 static inline xco_event_t *xco_notify_event(xco_notify_t *n) { return &n->base; }
 
 void xco_notify_one(xco_notify_t *n);
 void xco_notify_all(xco_notify_t *n);
 
 /* ---- Semaphore -------------------------------------------------------- */
 
 /* Counting semaphore. acquire is exposed as xco_event_t (composable with
  * select / xco_wait_or_cancel): xco_event_poll succeeds and decrements when
  * permits > 0; otherwise the waiter parks FIFO. xco_semaphore_release(n)
  * hands one permit to each of up to n waiting waiters (each is fired,
  * which the receiver treats as "you got a permit") before adding any
  * leftover to the count.
  *
  * One permit per acquire. Bulk acquire isn't expressible in xco_event_t's
  * shape; if you need it, call sequentially. For binary use (mutex-style
  * critical section across awaits) init with permits = 1.
  *
  * Fairness: FIFO at the waitlist. A racing inline xco_event_poll by a fresh
  * caller can jump ahead of parked waiters when permits are released
  * back to count rather than directly handed off — release prefers
  * parked waiters first to avoid that. */
 typedef struct xco_semaphore {
     xco_event_t  acquire;
     size_t   permits;
     xco_waiter_t *head, *tail;
 } xco_semaphore_t;
 
 extern const xco_event_vtable_t xco__semaphore_acquire_vt;
 
 static inline void xco_semaphore_init(xco_semaphore_t *s, size_t initial) {
     s->acquire.vt = &xco__semaphore_acquire_vt;
     s->permits    = initial;
     s->head       = s->tail = NULL;
 }
 
 static inline xco_event_t *xco_semaphore_event(xco_semaphore_t *s) { return &s->acquire; }
 
 void xco_semaphore_release(xco_semaphore_t *s, size_t n);
 
 /* ---- Mutex ------------------------------------------------------------ */
 
 /* Binary semaphore wrapper for vocabulary at call sites. xco_mutex_init is
  * xco_semaphore_init(s, 1); the xco_event_t fires once per release; xco_mutex_release
  * hands the permit to the next waiter (or returns it to the count). */
 typedef xco_semaphore_t xco_mutex_t;
 
 static inline void     xco_mutex_init   (xco_mutex_t *m) { xco_semaphore_init(m, 1); }
 static inline xco_event_t *xco_mutex_event  (xco_mutex_t *m) { return xco_semaphore_event(m); }
 static inline void     xco_mutex_release(xco_mutex_t *m) { xco_semaphore_release(m, 1); }
 
 /* ---- Select / all-of -------------------------------------------------- */
 
 /* Wait over N input events. Two semantics share the same storage shape,
  * so a caller can switch between them by changing only the init call:
  *
  *   xco_select_event_init   fires when ANY input fires  (any-of)
  *   xco_allof_event_init    fires when ALL inputs fire  (all-of)
  *
  * In both cases done's payload is the index of the input whose firing
  * closed the wait — the winner for select, the last-to-fire for allof —
  * and inputs[i].value carries each fired input's payload (works
  * uniformly for sticky and transient sources, where re-trying the input
  * would either succeed or fail).
  *
  * Composes: a select_event is itself an event. */
 
 typedef struct xco_select_event xco_select_event_t;
 
 /* Per-input arming record. Caller-allocated as an array of n alongside
  * the select_event. After fire, .value holds whatever the input
  * delivered; other fields are internal. */
 typedef struct {
     xco_waiter_t         w;
     xco_event_t        *src;
     xco_select_event_t *parent;
     uintptr_t       value;       /* captured at fire time */
 } xco_select_input_t;
 
 struct xco_select_event {
     xco_latch_t         done;       /* fires with the closing input's index */
     xco_select_input_t *inputs;
     size_t          n;
     size_t          remaining;  /* counts down; done fires at 0
                                  * (select: starts at 1, allof: at n) */
 };
 
 /* Initialize as a select (any-of). inputs[] is caller-provided storage
  * for n nodes; srcs[] is the array of n input event pointers (read
  * once during init). If any input is already ready, the select fires
  * immediately and no waiters are parked. Use &s->done.base as the
  * resulting xco_event_t. */
 void xco_select_event_init(xco_select_event_t *s,
                        xco_select_input_t *inputs, size_t n,
                        xco_event_t *const *srcs);
 
 /* Initialize as an allof (all-of). Inputs already ready at init are
  * consumed inline (no parking, value captured); if every input is
  * ready, done fires immediately. n == 0 fires done with payload 0. */
 void xco_allof_event_init(xco_select_event_t *s,
                       xco_select_input_t *inputs, size_t n,
                       xco_event_t *const *srcs);
 
 /* Disarm any inputs still parked. Safe to call after fire (no-op) and
  * after partial completion (allof). Required before s leaves scope if
  * it has not yet fired. */
 void xco_select_event_deinit(xco_select_event_t *s);
 
 
 /* ---- Queue ------------------------------------------------------------ */
 
 /* Bounded FIFO of uintptr_t. Caller provides the ring buffer storage.
  * Recv side is exposed as xco_event_t (composable with select). Send side
  * is a typed API (carries a value), shaped after the event-poll fusion:
  * fused try + park, NULL-waiter degenerates to pure-try.
  *
  * Three full-buffer policies, fixed at init:
  *   XCO_QUEUE_BLOCK         senders park until a receiver makes room.
  *   XCO_QUEUE_DROP_NEWEST   xco_queue_send_poll silently discards the new value.
  *   XCO_QUEUE_DROP_OLDEST   xco_queue_send_poll evicts the head and pushes new tail.
  *
  * Senders never park under DROP_* policies — passing a non-NULL qsw is
  * only meaningful under XCO_QUEUE_BLOCK. xco_queue_send_unpark is
  * idempotent (cancellation-safe).
  *
  * Direct-handoff: xco_queue_send_poll first checks for a parked receiver and
  * delivers inline if present (payload bypasses the buffer), regardless
  * of policy.
  *
  * Rendezvous matrix (BLOCK + cap=0; the xco_chan_* aliases at the bottom
  * of this section name this configuration explicitly):
  *   send + parked recv    fire recv with value, sender continues inline.
  *   send + no recv        sender parks (xco_queue_send_poll); peer pulls later.
  *   recv + parked sender  read sender's value, fire sender (delivery
  *                         confirmation), receiver continues inline.
  *   recv + no sender      receiver parks (xco_event_poll on recv); peer
  *                         delivers later.
  *
  * FIFO order on both waitlists.
  *
  * Close: optional EOF semantics. After xco_queue_close, xco_queue_send_poll
  * returns XCO_QSEND_CLOSED regardless of policy (no delivery), parked
  * senders are drained with delivered=false, and parked receivers are
  * woken so they can observe XCO_RECV_CLOSED via xco_queue_recv. The recv
  * event is "ready" iff a value is available OR the queue is closed —
  * receivers must call xco_queue_recv to disambiguate value vs EOF. */
 
 /* Result of a typed receive on a queue (or chan, which is just a queue). */
 typedef enum {
     XCO_RECV_GOT,        /* *out holds the delivered value */
     XCO_RECV_EMPTY,      /* nothing available right now; caller may park */
     XCO_RECV_CLOSED,     /* peer closed and no values remain */
 } xco_recv_status_t;
 
 typedef enum {
     XCO_QUEUE_BLOCK,
     XCO_QUEUE_DROP_NEWEST,
     XCO_QUEUE_DROP_OLDEST,
 } xco_queue_policy_t;
 
 typedef struct xco_queue {
     xco_event_t        recv;
     uintptr_t     *buf;
     size_t         cap, head, len;
     xco_queue_policy_t policy;
     xco_waiter_t       *send_head, *send_tail;
     xco_waiter_t       *recv_head, *recv_tail;
     bool           closed;
 } xco_queue_t;
 
 extern const xco_event_vtable_t xco__queue_recv_vt;
 
 static inline void xco_queue_init(xco_queue_t *q, uintptr_t *buf, size_t cap,
                               xco_queue_policy_t policy) {
     q->recv.vt    = &xco__queue_recv_vt;
     q->buf        = buf;
     q->cap        = cap;
     q->head       = 0;
     q->len        = 0;
     q->policy     = policy;
     q->send_head  = q->send_tail = NULL;
     q->recv_head  = q->recv_tail = NULL;
     q->closed     = false;
 }
 
 static inline xco_event_t *xco_queue_recv_event(xco_queue_t *q) { return &q->recv; }
 
 /* Sender-side waiter for XCO_QUEUE_BLOCK. Same shape as xco_chan_send_waiter_t:
  * a waker plus a value slot the receiver / close-drain reads back on the
  * park path. `delivered` is set by the closing side: true on a normal
  * handoff, false on a close drain. */
 typedef struct {
     xco_waker_t  sw;
     uintptr_t    value;       /* set by xco_queue_send_poll on the park path */
     bool         delivered;
 } xco_queue_send_waiter_t;
 
 static inline void xco_queue_send_waiter_init(xco_queue_send_waiter_t *qsw,
                                               xco_runtime_t *rt, xco_mach_t *m) {
     xco_waker_init(&qsw->sw, rt, m);
     qsw->value     = 0;
     qsw->delivered = false;
 }
 
 /* Result of xco_queue_send_poll. */
 typedef enum {
     XCO_QSEND_ACCEPTED,   /* delivered to a parked receiver, buffered, or
                              accepted-by-policy (silently dropped under
                              DROP_NEWEST, evicted-and-pushed under DROP_OLDEST) */
     XCO_QSEND_BLOCKED,    /* BLOCK + full; parked iff qsw != NULL */
     XCO_QSEND_CLOSED,     /* queue closed; never parks. Returned regardless
                              of policy — closed is closed. */
 } xco_queue_send_status_t;
 
 /* Fused try + park for a sender. Direct-delivers to a parked receiver if
  * one is waiting (returns XCO_QSEND_ACCEPTED, qsw not parked). Otherwise:
  *
  *   XCO_QUEUE_BLOCK + room       buffered → XCO_QSEND_ACCEPTED.
  *   XCO_QUEUE_BLOCK + full       XCO_QSEND_BLOCKED; if qsw != NULL the
  *                                sender's value is stashed in qsw->value
  *                                and qsw is parked.
  *   XCO_QUEUE_DROP_NEWEST + full silently drops → XCO_QSEND_ACCEPTED.
  *   XCO_QUEUE_DROP_OLDEST + full evicts head, pushes new tail → ACCEPTED.
  *   closed (any policy)          XCO_QSEND_CLOSED; never parks.
  *
  * The two degenerate forms mirror xco_event_poll:
  *   xco_queue_send_poll(q, v, NULL)   — pure try (peek; never parks).
  *   xco_queue_send_poll(q, v, qsw)    — fused try-or-park (BLOCK only). */
 xco_queue_send_status_t xco_queue_send_poll(xco_queue_t *q, uintptr_t value,
                                             xco_queue_send_waiter_t *qsw);
 
 void xco_queue_send_unpark(xco_queue_t *q, xco_queue_send_waiter_t *qsw);
 
 /* Typed receive. Disambiguates value vs EOF where xco_event_poll cannot:
  * returns XCO_RECV_GOT (value popped from the buffer or directly from a
  * parked sender), XCO_RECV_CLOSED (closed and drained), or
  * XCO_RECV_EMPTY (caller may park). */
 xco_recv_status_t xco_queue_recv(xco_queue_t *q, uintptr_t *out);
 
 /* Close the queue. Idempotent. Drains parked senders (delivered=false)
  * and wakes parked receivers. After close, xco_queue_send_poll returns
  * XCO_QSEND_CLOSED regardless of policy. */
 void xco_queue_close(xco_queue_t *q);
 static inline bool xco_queue_is_closed(const xco_queue_t *q) { return q->closed; }
 
 /* Selectable send op. A per-call object that holds the value, parks on
  * the queue (only meaningful under XCO_QUEUE_BLOCK), and exposes
  * &op->done.base as the event that fires when the send resolves.
  *
  * The op embeds a xco_queue_send_waiter_t (so the queue's send list stays
  * uniform — receivers read .value at the same offset for both direct
  * and op senders) but rewires its fire callback: instead of resuming an
  * xco_step, fire sets op->done. Polymorphism via the function pointer.
  *
  * The done latch's payload is 1 on XCO_QSEND_ACCEPTED (handed to a
  * receiver, buffered, or accepted under DROP_*) and 0 on
  * XCO_QSEND_CLOSED / close-drain.
  *
  * Under DROP_* policies the send always resolves inline at init (the
  * poll returns ACCEPTED and op->done fires immediately).
  *
  * Lifecycle: init → wait on &op->done.base → deinit. Always deinit;
  * it's a no-op after resolution and unparks the queue-side waiter if not. */
 typedef struct {
     xco_queue_send_waiter_t qsw;        /* parked on queue; fire overridden */
     xco_queue_t           *queue;
     xco_latch_t            done;
 } xco_queue_send_op_t;
 
 void xco__queue_send_op_fire(xco_waiter_t *w, uintptr_t value);
 
 void xco_queue_send_op_init(xco_queue_send_op_t *op, xco_queue_t *q, uintptr_t value);
 static inline void xco_queue_send_op_deinit(xco_queue_send_op_t *op) {
     if (op->done.set) return;
     xco_queue_send_unpark(op->queue, &op->qsw);
 }
 
 /* ---- Channel (alias) -------------------------------------------------- */
 
 /* Unbuffered rendezvous channel: a queue at cap=0 with XCO_QUEUE_BLOCK
  * policy. Senders and receivers wait on each other; whichever arrives
  * first parks until its peer shows up. The pending value lives in the
  * sender's xco_chan_send_waiter_t for the duration of any wait — no
  * per-channel buffer storage.
  *
  * The xco_chan_* names are thin aliases over the queue API: a chan IS a
  * queue. They exist so call sites can name "rendezvous" explicitly
  * rather than spelling out cap=0+BLOCK. The queue's recv event, send
  * poll, close, recv, and selectable send op all carry over unchanged.
  *
  * Storage: an xco_chan_t carries the queue's buffer/cap/policy fields
  * even though they're inert at cap=0 (~40 bytes of overhead vs a
  * dedicated rendezvous struct). Below the noise floor for typical use. */
 
 typedef xco_queue_t              xco_chan_t;
 typedef xco_queue_send_waiter_t  xco_chan_send_waiter_t;
 typedef xco_queue_send_status_t  xco_chan_send_status_t;
 typedef xco_queue_send_op_t      xco_chan_send_op_t;
 
 /* Status aliases. Same enumerators, vocabulary at call sites: a chan
  * "delivers" rather than "accepts" — but the underlying constants are
  * the queue's. */
 #define XCO_SEND_DELIVERED  XCO_QSEND_ACCEPTED
 #define XCO_SEND_BLOCKED    XCO_QSEND_BLOCKED
 #define XCO_SEND_CLOSED     XCO_QSEND_CLOSED
 
 static inline void xco_chan_init(xco_chan_t *c) {
     xco_queue_init(c, NULL, 0, XCO_QUEUE_BLOCK);
 }
 static inline xco_event_t *xco_chan_recv_event(xco_chan_t *c) {
     return xco_queue_recv_event(c);
 }
 static inline void xco_chan_send_waiter_init(xco_chan_send_waiter_t *csw,
                                              xco_runtime_t *rt, xco_mach_t *m) {
     xco_queue_send_waiter_init(csw, rt, m);
 }
 static inline xco_chan_send_status_t xco_chan_send_poll(xco_chan_t *c, uintptr_t value,
                                                         xco_chan_send_waiter_t *csw) {
     return xco_queue_send_poll(c, value, csw);
 }
 static inline void xco_chan_send_unpark(xco_chan_t *c, xco_chan_send_waiter_t *csw) {
     xco_queue_send_unpark(c, csw);
 }
 static inline xco_recv_status_t xco_chan_recv(xco_chan_t *c, uintptr_t *out) {
     return xco_queue_recv(c, out);
 }
 static inline void xco_chan_close(xco_chan_t *c) { xco_queue_close(c); }
 static inline bool xco_chan_is_closed(const xco_chan_t *c) {
     return xco_queue_is_closed(c);
 }
 static inline void xco_chan_send_op_init(xco_chan_send_op_t *op, xco_chan_t *c, uintptr_t value) {
     xco_queue_send_op_init(op, c, value);
 }
 static inline void xco_chan_send_op_deinit(xco_chan_send_op_t *op) {
     xco_queue_send_op_deinit(op);
 }
 
 /* ---- Broadcast (slot) ------------------------------------------------- */
 
 /* Re-armable signal carrying a "latest value" slot. Subscribers park on
  * the event; xco_broadcast_publish stores the new value, fires every parked
  * subscriber with that value, and clears the waitlist — subscribers must
  * re-park to see further publishes. Subscribers that aren't parked at
  * publish time miss that publish but will see the next one. This is the
  * coalescing "watch a slot" semantics, not lossless fan-out.
  *
  * xco_event_poll never reports ready: there is no "ready now" state — a
  * subscriber waits for the *next* publish. To read the latest published
  * value at any time, use xco_broadcast_value (valid once xco_broadcast_has_value
  * returns true).
  *
  * For lossless multi-consumer delivery, give each subscriber its own
  * queue and have the producer write to all of them. */
 
 typedef struct xco_broadcast {
     xco_event_t   base;
     bool      has_value;
     uintptr_t value;
     xco_waiter_t  *waiters;
 } xco_broadcast_t;
 
 extern const xco_event_vtable_t xco__broadcast_vt;
 
 static inline void xco_broadcast_init(xco_broadcast_t *b) {
     b->base.vt   = &xco__broadcast_vt;
     b->has_value = false;
     b->value     = 0;
     b->waiters   = NULL;
 }
 
 static inline xco_event_t  *xco_broadcast_event    (xco_broadcast_t *b)       { return &b->base; }
 static inline bool      xco_broadcast_has_value(const xco_broadcast_t *b) { return b->has_value; }
 static inline uintptr_t xco_broadcast_value    (const xco_broadcast_t *b) { return b->value; }
 
 void xco_broadcast_publish(xco_broadcast_t *b, uintptr_t value);
 
 /* ---- Cancellation ----------------------------------------------------- */
 
 /* A cancellation token is a sticky latch — these aliases exist for
  * vocabulary at call sites. xco_cancel_set fires every parked waiter; the
  * idempotency of xco_latch_set means racing cancellers are fine.
  *
  * Pair a xco_cancel_t with any blocking event via xco_wait_or_cancel to get
  * "await X, or be cancelled."
  *
  * Discipline: cancellation notifies; it never drops in-flight values.
  * A cancelled await returns control to its caller, which is responsible
  * for draining whatever it owns — deinit a pending chan_send_op so its
  * value goes back to the sender, deinit a select_event so input waiters
  * detach, drive a cancellable coroutine to XCO_STEP_DEAD before freeing
  * its stack. The xco layer does no unwinding; the coroutine cooperates. */
 
 typedef xco_latch_t xco_cancel_t;
 
 static inline void     xco_cancel_init(xco_cancel_t *c)         { xco_latch_init(c); }
 static inline void     xco_cancel_set(xco_cancel_t *c)          { xco_latch_set(c, 0); }
 static inline bool     xco_cancel_is_set(const xco_cancel_t *c) { return c->set; }
 static inline xco_event_t *xco_cancel_event(xco_cancel_t *c)        { return &c->base; }
 
 /* Outcome indices for xco_wait_or_cancel — match the inputs[] order so the
  * value the resumer receives from the latched select reads as one of
  * these directly. */
 enum {
     XCO_WAIT_OK        = 0,   /* ev fired; inputs[0].value holds its payload */
     XCO_WAIT_CANCELLED = 1,   /* cancel fired before ev */
 };
 
 /* Build a select over (ev, cancel) using caller-provided storage. If
  * either is already ready at init the select fast-paths and never parks
  * anyone (ev is checked first, so an event that has already resolved
  * wins over a concurrent cancel). Treat &out->done.base as the event
  * to wait on. Always pair with xco_select_event_deinit before storage
  * leaves scope (no-op once fired). */
 static inline void xco_wait_or_cancel(xco_select_event_t *out,
                                   xco_select_input_t inputs[2],
                                   xco_event_t *ev, xco_cancel_t *c) {
     xco_event_t *srcs[2] = {ev, xco_cancel_event(c)};
     xco_select_event_init(out, inputs, 2, srcs);
 }
 
 /* ---- Timers ----------------------------------------------------------- */
 
 /* A timer is a sticky event keyed on a u64 deadline. It fires (exactly
  * once) when the attached timer source is advanced past that deadline.
  * The library never reads a clock; the caller provides `now` to
  * xco_timers_advance (or via xco_rt_run).
  *
  * Storage is pluggable through the timers vtable so callers can swap a
  * pairing heap (in-tree, O(log n) amortized everywhere including cancel)
  * for a wheel or other structure without touching the timer/timeout
  * surface. The timer struct holds the heap link fields inline; the source
  * impl interprets them.
  *
  * Lifecycle:
  *   xco_timer_init(t, ts, deadline)  // inserts into ts
  *   ... wait on xco_timer_event(t), or compose into select/xco_wait_or_cancel ...
  *   xco_timer_deinit(t)              // removes from ts if not yet fired
  *
  * Fire payload is the deadline. Re-arming = reinit a fresh timer. */
 
 typedef struct xco_timer xco_timer_t;
 
 typedef struct {
     /* Insert t into the source. t must be initialized but not yet
      * inserted; insert sets t's heap link fields. */
     void (*insert) (xco_timers_t *ts, xco_timer_t *t);
     /* Remove t from the source if currently inserted. Caller must
      * ensure t was inserted into this same source. */
     void (*cancel) (xco_timers_t *ts, xco_timer_t *t);
     /* Fire every timer whose deadline <= now, in deadline order, popping
      * each from the source. Each fire drains the timer's waiter list. */
     void (*advance)(xco_timers_t *ts, uint64_t now);
     /* Return the earliest queued deadline, or UINT64_MAX if no timer
      * is queued. */
     uint64_t (*peek)(const xco_timers_t *ts);
 } xco_timers_vtable_t;
 
 struct xco_timers {
     const xco_timers_vtable_t *vt;
     uint64_t now;       /* most recent advance() input; monotonic */
 };
 
 static inline void     xco_timers_insert (xco_timers_t *ts, xco_timer_t *t)   { ts->vt->insert (ts, t); }
 static inline void     xco_timers_cancel (xco_timers_t *ts, xco_timer_t *t)   { ts->vt->cancel (ts, t); }
 static inline void     xco_timers_advance(xco_timers_t *ts, uint64_t now) {
     assert(now >= ts->now);
     ts->now = now;
     ts->vt->advance(ts, now);
 }
 static inline uint64_t xco_timers_peek   (const xco_timers_t *ts)             { return ts->vt->peek(ts); }
 static inline uint64_t xco_timers_now    (const xco_timers_t *ts)             { return ts->now; }
 
 /* Concrete timer. Embeds a latch so try/park/unpark and the fire-all
  * waitlist handling come for free; the timer source manipulates only
  * the heap link fields and triggers the latch on fire. The latch's
  * payload after fire is the timer's deadline. */
 struct xco_timer {
     xco_latch_t   done;            /* fires once, payload = deadline */
     uint64_t  deadline;
     xco_timers_t *src;             /* source this timer is registered with */
     bool      in_heap;         /* true between insert and fire/cancel */
     /* Pairing-heap link fields; opaque to anyone but the source impl.
      * prev is parent if first child, else previous sibling, else NULL. */
     xco_timer_t  *child, *prev, *next;
 };
 
 static inline xco_event_t *xco_timer_event(xco_timer_t *t)        { return &t->done.base; }
 static inline bool     xco_timer_fired(const xco_timer_t *t)  { return t->done.set; }
 
 static inline void xco_timer_init(xco_timer_t *t, xco_timers_t *ts, uint64_t deadline) {
     xco_latch_init(&t->done);
     t->deadline = deadline;
     t->src      = ts;
     t->in_heap  = false;
     t->child    = NULL;
     t->prev     = NULL;
     t->next     = NULL;
     xco_timers_insert(ts, t);
 }
 
 static inline void xco_timer_deinit(xco_timer_t *t) {
     if (t->in_heap) xco_timers_cancel(t->src, t);
 }
 
 /* In-tree timer source: intrusive pairing heap. O(1) amortized insert
  * and meld; O(log n) amortized advance and cancel. No per-source
  * allocation — the heap is just a root pointer; nodes live in the
  * caller's xco_timer_t's. */
 typedef struct {
     xco_timers_t  base;
     xco_timer_t  *root;
 } xco_pairing_heap_t;
 
 extern const xco_timers_vtable_t xco__pairing_heap_vt;
 
 static inline void xco_pairing_heap_init(xco_pairing_heap_t *h) {
     h->base.vt  = &xco__pairing_heap_vt;
     h->base.now = 0;
     h->root     = NULL;
 }
 
 /* ---- Timeout ---------------------------------------------------------- */
 
 /* Bundle: a timer that fires a xco_cancel_t on expiration. The natural
  * pairing for "await ev, or be cancelled by deadline":
  *
  *   xco_timeout_t to;
  *   xco_timeout_init(&to, ts, now + budget);
  *   xco_select_event_t sel; xco_select_input_t inputs[2];
  *   xco_wait_or_cancel(&sel, inputs, ev, &to.cancel);
  *   ... wait on &sel.done.base ...
  *   xco_select_event_deinit(&sel);
  *   xco_timeout_deinit(&to);   // safe whether the timer fired or not
  *
  * The bridge waiter is parked on the timer; when it fires it sets the
  * cancel. Bridge fire is idempotent vs xco_cancel_set (a sticky latch). */
 typedef struct xco_timeout {
     xco_timer_t  timer;
     xco_cancel_t cancel;
     xco_waiter_t  bridge;
 } xco_timeout_t;
 
 void xco_timeout_init(xco_timeout_t *to, xco_timers_t *ts, uint64_t deadline);
 static inline void xco_timeout_deinit(xco_timeout_t *to) {
     xco_event_unpark(xco_timer_event(&to->timer), &to->bridge);
     xco_timer_deinit(&to->timer);
 }
 
 /* ---- Ticker ----------------------------------------------------------- */
 
 /* Re-armable transient signal driven by a timer source. Each time the
  * underlying timer fires, the ticker computes the next deadline (period
  * past the just-fired one, with skip-ahead for catch-up after overflow),
  * reinstalls the timer, and fires every parked subscriber with the
  * just-fired deadline as the payload. Subscribers that aren't parked at
  * a fire miss it (transient — same coalescing semantics as broadcast).
  *
  *   xco_ticker_init(&t, ts, period, first_deadline);
  *   ... wait on xco_ticker_event(&t), re-park to see further ticks ...
  *   xco_ticker_deinit(&t);   // cancels the in-flight timer
  *
  * xco_event_poll never reports ready; subscribers wait for the *next* tick. */
 typedef struct xco_ticker {
     xco_timer_t   timer;
     xco_timers_t *src;
     uint64_t  period;
     xco_event_t   base;
     xco_waiter_t  *waiters;
     xco_waiter_t   bridge;       /* internal: parks on xco_timer_event */
 } xco_ticker_t;
 
 extern const xco_event_vtable_t xco__ticker_vt;
 
 void     xco_ticker_init  (xco_ticker_t *t, xco_timers_t *ts,
                        uint64_t period, uint64_t first_deadline);
 void     xco_ticker_deinit(xco_ticker_t *t);
 static inline xco_event_t *xco_ticker_event(xco_ticker_t *t) { return &t->base; }
 
 /* ---- Task ------------------------------------------------------------- */
 
 /* Lifecycle handle for a running xco_step. Bundles a done latch (fires when
  * the xco_step returns, payload = its return value) with a cancel latch
  * (the canonical signal to ask the xco_step to wind down). The xco_step itself
  * is caller-allocated; the task holds a pointer to it.
  *
  * Who fires done:
  *   - Hand-coded state machine: call xco_task_done(t, ret) in the same arm
  *     that returns XCO_STEP_DEAD.
  *   - xco-backed task (see xco_cotask_t below): the xco_trampoline calls
  *     xco_task_done automatically with the coroutine's return value.
  *
  * Cooperation: cancellation only notifies — the xco_step is responsible for
  * draining what it owns and reaching XCO_STEP_DEAD. The task's cancel is a
  * normal xco_cancel_t, so the xco_step typically composes xco_wait_or_cancel against
  * it on every blocking await.
  *
  * Joining: callers wait on xco_task_done_event with the standard event API
  * (try / park, or compose into select / xco_wait_or_cancel). On fire the
  * latch's payload is the xco_step's return value. */
 
 typedef struct xco_task {
     xco_mach_t  *mach;
     xco_latch_t   done;
     xco_cancel_t  cancel;
 } xco_task_t;
 
 static inline void xco_task_init(xco_task_t *t, xco_mach_t *mach) {
     t->mach = mach;
     xco_latch_init(&t->done);
     xco_cancel_init(&t->cancel);
 }
 
 /* Mark the task complete with its return value. Idempotent (xco_latch_set is). */
 static inline void xco_task_done(xco_task_t *t, uintptr_t value) {
     xco_latch_set(&t->done, value);
 }
 
 static inline xco_event_t  *xco_task_done_event  (xco_task_t *t)       { return &t->done.base; }
 static inline xco_cancel_t *xco_task_cancel      (xco_task_t *t)       { return &t->cancel; }
 static inline bool      xco_task_finished    (const xco_task_t *t) { return t->done.set; }
 static inline bool      xco_task_is_cancelled(const xco_task_t *t) { return xco_cancel_is_set(&t->cancel); }
 static inline xco_mach_t  *xco_task_mach        (xco_task_t *t)       { return t->mach; }
 
 /* ---- Task group ------------------------------------------------------- */
 
 /* Fan-in join + fan-out cancel for a dynamic set of tasks. Caller
  * provides storage for each per-attachment record (xco_group_attach_t), so
  * the group itself does no allocation.
  *
  * xco_task_group_attach(g, t, slot):
  *   xco_countdown_add(g->pending, 1); slot's bridge waiter parks on
  *   xco_task_done_event(t); slot is appended to g's list. When the task's
  *   done fires, the bridge fires: it splices the slot out of g's list
  *   and calls xco_countdown_done(&g->pending). Re-attaching a finished
  *   task is UB.
  *
  * xco_task_group_cancel(g): walks the attachment list and xco_cancel_set's
  * each &slot->task->cancel, then xco_cancel_set's g->cancel. Bodies that
  * compose xco_wait_or_cancel against xco_task_cancel(t) wind down cooperatively;
  * meanwhile, anything waiting on g->cancel observes the group-level
  * signal directly.
  *
  * xco_task_group_join_event(g): fires when every attached task has reached
  * xco_task_done. Compose with select / xco_wait_or_cancel like any event. */
 
 typedef struct xco_group_attach xco_group_attach_t;
 
 typedef struct xco_task_group {
     xco_countdown_t     pending;
     xco_cancel_t        cancel;
     xco_group_attach_t *head, *tail;
 } xco_task_group_t;
 
 struct xco_group_attach {
     xco_waiter_t         bridge;     /* parked on xco_task_done_event(task) */
     xco_task_t         *task;
     xco_task_group_t   *group;
     xco_group_attach_t *next, *prev;
 };
 
 void      xco_task_group_init   (xco_task_group_t *g);
 void      xco_task_group_attach (xco_task_group_t *g, xco_task_t *t,
                              xco_group_attach_t *slot);
 void      xco_task_group_cancel (xco_task_group_t *g);
 static inline xco_event_t  *xco_task_group_join_event   (xco_task_group_t *g) {
     return xco_countdown_event(&g->pending);
 }
 static inline xco_cancel_t *xco_task_group_cancel_handle(xco_task_group_t *g) {
     return &g->cancel;
 }
 
 /* ====================================================================
  * xco — stack-switching coroutines.
  *
  * Teardown: this layer does not unwind a suspended coroutine's stack.
  * Drive a coroutine to return (e.g. by passing a cancel sentinel it
  * is expected to handle) before freeing its stack memory.
  * ==================================================================== */
 
 /* Coroutine entry point. The argument is the value supplied to the
  * first xco_step on this xco. The return value is delivered to the resumer
  * as the final xco_step_result, with status XCO_STEP_DEAD. */
 typedef uintptr_t (*xco_fn)(uintptr_t arg);
 
 /* Coroutine control block. Allocate anywhere — on a stack, in a
  * struct, on the heap. xco_mach_t base is first so xco_coro_t * can be passed
  * directly to xco_step() or any generic xco_mach_t * consumer. The trailing
  * priv storage holds the saved register context and bookkeeping;
  * its contents are private to the implementation. */
 typedef struct xco_coro {
     xco_mach_t base;
     _Alignas(XCO_ALIGN) unsigned char priv[XCO_SIZE];
 } xco_coro_t;
 
 /* Initialize *c to run fn on [stack_base, stack_base + stack_len).
  * stack_base must be XCO_STACK_ALIGN-aligned; the runtime picks the
  * starting SP based on the architecture's stack growth direction.
  * Status after init is XCO_STEP_INIT. */
 void xco_init(xco_coro_t *c, xco_fn fn,
               void *stack_base, size_t stack_len);
 
 /* Suspend the currently running coroutine, returning value to its
  * resumer. Returns the value passed by the next resume. Undefined if
  * called outside a coroutine. */
 uintptr_t xco_suspend(uintptr_t value);
 
 /* The currently running coroutine, or NULL if the caller is not in
  * one. The runtime maintains this for xco_suspend. */
 xco_coro_t *xco_self(void);
 
 /* Resuming a coroutine is just driving its xco_step: callers use
  * xco_step(&c->base, v) directly. Reading status without resuming is
  * xco_mach_status(&c->base). The xco layer adds no separate vocabulary
  * for these — that's the unification with hand-coded state machines.
  * Resuming a coroutine that is not XCO_STEP_INIT or XCO_STEP_SUSPENDED is
  * undefined. */
 
 /* Convenience: init then first-step in one call. */
 static inline xco_step_result_t xco_spawn(xco_coro_t *c, xco_fn fn,
                                        void *stack_base, size_t stack_len,
                                        uintptr_t arg) {
     xco_init(c, fn, stack_base, stack_len);
     return xco_step(&c->base, arg);
 }
 
 /* Cooperative yield. Enqueues self on rt's ready queue and suspends;
  * the next xco_rt_run pass resumes us. Useful for fairness when a coroutine
  * wants to give other ready work a turn between long-running steps.
  * Must be called from inside a coroutine driven by rt. */
 static inline void xco_yield(xco_runtime_t *rt) {
     xco_coro_t *self = xco_self();
     assert(self != NULL);
     xco_waker_t sw;
     xco_waker_init(&sw, rt, &self->base);
     xco_rt_enqueue(rt, &sw.base);
     xco_suspend(0);
 }
 
 /* Await an event from inside a coroutine. The standard poll-suspend
  * dance, in one call. Returns the event's value (delivered by fire on
  * the slow path, by the inline poll on the fast path). Must be called
  * from inside a coroutine driven by rt. */
 static inline uintptr_t xco_await(xco_runtime_t *rt, xco_event_t *e) {
     xco_coro_t *self = xco_self();
     assert(self != NULL);
     uintptr_t v;
     xco_waker_t sw;
     xco_waker_init(&sw, rt, &self->base);
     if (xco_event_poll(e, &v, &sw.base)) return v;
     xco_suspend(0);
     (void)xco_event_poll(e, &v, NULL);
     return v;
 }
 
 /* Await ev or be cancelled by c. Returns true if ev fired (its payload
  * is written to *out, which may be NULL); false if cancelled. The
  * internal select_event is always cleaned up before return.
  *
  * The canonical shape for cooperative work inside a task body — pair
  * with xco_task_cancel(self) on every blocking await. */
 static inline bool xco_await_or_cancel(xco_runtime_t *rt, xco_event_t *ev,
                                        xco_cancel_t *c, uintptr_t *out) {
     xco_select_event_t sel;
     xco_select_input_t inputs[2];
     xco_wait_or_cancel(&sel, inputs, ev, c);
     uintptr_t winner = xco_await(rt, &sel.done.base);
     bool ok = (winner == XCO_WAIT_OK);
     if (ok && out) *out = inputs[XCO_WAIT_OK].value;
     xco_select_event_deinit(&sel);
     return ok;
 }
 
 /* ---- xco-backed task ------------------------------------------------- */
 
 /* xco specialization of xco_task_t. Bundles the user-visible task handle
  * with the xco that runs it; the xco_trampoline calls fn(&xt->task, arg)
  * and then xco_task_done with its return value, so xco_wait_or_cancel-style
  * teardown works without the user wiring anything.
  *
  * The xco_trampoline recovers xt from xco_self() at first entry (container_of
  * on the embedded co), so the first-resume uintptr_t is preserved as
  * fn's arg under normal xco semantics. Subsequent resumes pass values
  * to the coroutine in the usual way.
  *
  * Storage shape: caller allocates xco_cotask_t and a stack. The xco_task_t
  * inside is the handle to wait/cancel on; cancel via xco_cancel_set on
  * &xt->task.cancel and the body is expected to observe it (typically
  * by composing xco_wait_or_cancel against xco_task_cancel(&xt->task)). */
 
 typedef uintptr_t (*xco_cotask_fn)(xco_task_t *t, uintptr_t arg);
 
 typedef struct xco_cotask {
     xco_task_t      task;
     xco_coro_t       co;
     xco_cotask_fn fn;
 } xco_cotask_t;
 
 /* Initialize xt to run fn on the given stack. After this the embedded
  * xco is XCO_STEP_INIT; drive it with xco_step(&xt->co.base, v) or use
  * xco_cotask_spawn for the init-and-first-step convenience. */
 void xco_cotask_init(xco_cotask_t *xt, xco_cotask_fn fn,
                    void *stack_base, size_t stack_len);
 
 /* Convenience: init then first-step in one call. arg is delivered as
  * fn's argument. */
 static inline xco_step_result_t xco_cotask_spawn(xco_cotask_t *xt, xco_cotask_fn fn,
                                             void *stack_base, size_t stack_len,
                                             uintptr_t arg) {
     xco_cotask_init(xt, fn, stack_base, stack_len);
     return xco_step(&xt->co.base, arg);
 }
 
 /* ====================================================================
  * xco_op — generic effect/IO layer.
  *
  * Coroutines submit ops describing work to do; the runtime accumulates
  * them on a pending list; after xco_rt_run quiesces, the host pulls the
  * batch via xco_rt_take_ops, executes them however it likes (io_uring,
  * threads, mocks, replay), and injects completions back via
  * xco_op_complete. The xco library itself reads no clocks and makes no
  * syscalls — that property extends to IO via this layer.
  *
  * An op is just an event with a side-channel describing what to do.
  * Embed xco_op_t as the first member of a kind-specific payload struct;
  * the host pattern-matches on `kind` (an open tag space — the runtime
  * never inspects it).
  *
  * State machine:
  *   PENDING   op->epoch == op->rt->op_epoch, done unset   (on rt's list)
  *   IN_FLIGHT op->epoch != op->rt->op_epoch, done unset   (host has taken)
  *   RESOLVED  done.set                                     (terminal)
  *
  * The epoch counter on the runtime is bumped each xco_rt_take_ops, so
  * "still on the pending list" is a generation match — submit + take are
  * both O(1) and the runtime never walks the batch.
  *
  * Threading: single-threaded, same as the rest of xco. submit, cancel,
  * and complete must all be called on the runtime's thread.
  * ==================================================================== */
 
 typedef enum {
     XCO_OP_PENDING,        /* not used as a fire payload, but conceptual */
     XCO_OP_COMPLETED,      /* host finished; result lives in embedder fields */
     XCO_OP_CANCELLED,      /* resolved without a real result */
 } xco_op_status_t;
 
 struct xco_op {
     xco_latch_t      done;              /* fires once, payload = status */
     xco_op_t        *next, *prev;       /* intrusive on rt's pending list */
     xco_runtime_t   *rt;                /* set on submit; stays set after take */
     uint64_t         epoch;             /* matches rt->op_epoch iff PENDING */
     uint32_t         kind;              /* open tag space; host pattern-matches */
     bool             cancel_requested;  /* advisory to host post-take */
 };
 
 /* True iff op is still on rt's current pending batch (i.e. PENDING).
  * False for IN_FLIGHT or RESOLVED. O(1). */
 static inline bool xco_op_is_pending(const xco_op_t *op) {
     return op->rt && op->epoch == op->rt->op_epoch && !op->done.set;
 }
 
 /* Submit op to rt's pending list. Initializes done, clears
  * cancel_requested, and sets op->rt = rt. The caller pre-populates op->kind
  * and any embedder fields (fd, buf, len, ...) before submit. After submit
  * the awaiter typically waits on &op->done.base — composes with select,
  * xco_wait_or_cancel, etc. */
 void xco_op_submit  (xco_runtime_t *rt, xco_op_t *op);
 
 /* Request cancellation. Behavior depends on state:
  *   PENDING   splice from rt list, fire done(CANCELLED).
  *   IN_FLIGHT set cancel_requested=true (advisory; host decides).
  *   RESOLVED  no-op.
  * Idempotent. */
 void xco_op_cancel  (xco_op_t *op);
 
 /* Host's final word for an op it took via xco_rt_take_ops. Fires done
  * with the given status. Idempotent (the latch is). */
 void xco_op_complete(xco_op_t *op, xco_op_status_t status);
 
 /* Detach the runtime's pending ops list and return its head. Bumps the
  * runtime's op_epoch, transitioning the whole batch to IN_FLIGHT in O(1)
  * (no walk — each op's stored epoch now no longer matches). The host
  * owns the returned list — the next/prev fields are the host's to reuse
  * however it wants for in-flight tracking. Returns NULL if the list is
  * empty.
  *
  * If tail_out is non-NULL, *tail_out receives the list's tail (or NULL
  * for an empty list) — handy for splicing the batch onto a host-side
  * in-flight list in O(1) without walking. */
 xco_op_t *xco_rt_take_ops(xco_runtime_t *rt, xco_op_t **tail_out);
 
 #endif /* XCO_H */