xco.h (62152B)
1 /* 2 * xco.h — minimal C11 concurrency library. 3 * 4 * Four layers in this header, bottom-up: 5 * 6 * xco_mach_t Generic resumable function. A value that can be 7 * driven forward one step at a time; each step takes a 8 * uintptr_t in, returns one out, and reports whether 9 * the function suspended or finished. Substrate shared 10 * by stack-switching coroutines (xco) and hand-coded 11 * state machines. 12 * 13 * waiter / event / runtime 14 * Pollable event substrate (poll / unpark) with a 15 * single-threaded FIFO ready queue. Concrete events: 16 * latch, countdown, notify, semaphore/mutex, select, 17 * allof, channel, queue, broadcast, cancel, timer, 18 * pairing-heap timer source, timeout, ticker. All 19 * storage caller-provided, no allocation, no atomics. 20 * 21 * xco_task_t Lifecycle handle for a running xco_step. Bundles a 22 * done latch (fires with the step's return value) and 23 * a cancel latch (the cooperative wind-down signal). 24 * xco_task_group_t fans these in/out across a dynamic 25 * set of tasks. Storage caller-provided; the step 26 * itself lives wherever the caller put it. 27 * 28 * xco_coro_t Stack-switching coroutine. xco_coro_t embeds xco_mach_t as 29 * its first member, so a coroutine is one concrete 30 * kind of resumable function: generic code holding an 31 * xco_mach_t* works on coroutines and hand-coded state 32 * machines uniformly. Values pass between caller and 33 * coroutine through a single uintptr_t channel; pack 34 * richer data behind a pointer. 35 * Asymmetric: xco_suspend always returns to the most 36 * recent resumer. Resumes nest like function calls. 37 * Thread portability: the platform switch saves only 38 * callee-saved regs (no TLS register, no signal mask), 39 * so a fully-suspended coroutine could in principle be 40 * resumed on another thread. In practice don't: the 41 * runtime/event/waiter substrate is single-threaded 42 * (no atomics), so a coroutine parked on any event is 43 * tethered to that runtime's thread; and user code 44 * silently rebinds errno / _Thread_local / thread-affine 45 * OS handles to whichever thread resumed it. 46 * 47 * xco_cotask_t xco specialization of xco_task_t. The xco_trampoline calls 48 * fn(&xt->task, arg) and then xco_task_done with the 49 * return value, so xco_wait_or_cancel-style teardown works 50 * without the user wiring anything. 51 * 52 * One-waiter invariant. An xco_step, while suspended, is parked on at most 53 * one event. Multi-wait is composed in the event graph: build a 54 * select_event (or any future combinator — all-of, timeout, ...) and 55 * park on that. The xco_step never sees more than one event directly. 56 * This is what lets a single xco_waker_t live inline in the xco_step 57 * and a single next/prev pair serve both event waitlists and the 58 * runtime ready queue (the two list memberships are disjoint in time). 59 * 60 * XCO_MT (compile-time opt-in, see XCO_MT.md). Extends xco to N pinned 61 * threads communicating through per-thread MPSC inboxes. The unit is 62 * xco_thread_t — a freestanding thread abstraction (atomic inbox + host 63 * wakeup hook + park handshake); thread creation, blocking, and waking 64 * are host bindings, never xco's. The waiter gains a `home` thread and 65 * xco_waiter_fire becomes the routing seam: fire on the home thread (or 66 * home == NULL) runs inline, anywhere else the waiter itself is posted 67 * to home's inbox as the message. Event mutators remain owner-thread- 68 * only; all struct layouts change, so the whole program must agree on 69 * the flag. Without XCO_MT this library compiles to exactly the 70 * single-threaded description above — no atomics, no TLS, no extra 71 * waiter fields. 72 */ 73 74 #ifndef XCO_H 75 #define XCO_H 76 77 #include <assert.h> 78 #include <stdbool.h> 79 #include <stddef.h> 80 #include <stdint.h> 81 82 #ifdef XCO_MT 83 #include <stdatomic.h> 84 #endif 85 86 /* Provides XCO_SIZE, XCO_ALIGN, XCO_STACK_ALIGN, XCO__CTX_SIZE, 87 * XCO__CTX_ALIGN; resolved by the build to the platform-specific 88 * copy via the include path (-Iplatform/$(PLATFORM)). */ 89 #include "xco_platform.h" 90 91 /* ==================================================================== 92 * xco_step — generic resumable function interface. 93 * 94 * The first-member convention: embed xco_mach_t as the first field of 95 * your concrete type so a pointer to your type can be passed wherever 96 * an xco_mach_t * is expected. 97 * 98 * typedef struct { 99 * xco_mach_t base; 100 * int phase; 101 * ... 102 * } parser_t; 103 * 104 * static xco_step_result_t parser_step(xco_mach_t *s, uintptr_t v) { 105 * parser_t *p = (parser_t *)s; 106 * switch (p->phase) { 107 * case 0: p->phase = 1; return (xco_step_result_t){v + 1, XCO_STEP_SUSPENDED}; 108 * case 1: return (xco_step_result_t){v * 2, XCO_STEP_DEAD}; 109 * } 110 * __builtin_unreachable(); 111 * } 112 * 113 * parser_t p = { .base = {.step = parser_step, .status = XCO_STEP_INIT} }; 114 * xco_step_result_t r = xco_step(&p.base, 0); 115 * ==================================================================== */ 116 117 typedef enum { 118 XCO_STEP_INIT, /* created, never stepped */ 119 XCO_STEP_RUNNING, /* inside step(), or in an active resume chain */ 120 XCO_STEP_SUSPENDED, /* yielded; resumable */ 121 XCO_STEP_DEAD, /* function returned */ 122 } xco_mach_status_t; 123 124 typedef struct { 125 uintptr_t value; 126 xco_mach_status_t status; 127 } xco_step_result_t; 128 129 typedef struct xco_mach xco_mach_t; 130 typedef xco_step_result_t (*xco_step_fn)(xco_mach_t *s, uintptr_t value); 131 132 struct xco_mach { 133 xco_step_fn step; 134 xco_mach_status_t status; /* cached; xco_step() syncs from each result */ 135 }; 136 137 /* Drive one step. The wrapper updates s->status from the returned 138 * result so step implementations only need to populate the result. */ 139 static inline xco_step_result_t xco_step(xco_mach_t *s, uintptr_t value) { 140 xco_step_result_t r = s->step(s, value); 141 s->status = r.status; 142 return r; 143 } 144 145 static inline xco_mach_status_t xco_mach_status(const xco_mach_t *s) { 146 return s->status; 147 } 148 149 /* ==================================================================== 150 * Event substrate. 151 * 152 * Fused try-or-park: xco_event_poll(e, &out, w) returns true if the 153 * event is ready (writes the value to *out, w is NOT parked) and false 154 * otherwise (parks w on the waitlist iff w is non-NULL). The two 155 * degenerate forms: 156 * 157 * xco_event_poll(e, &v, NULL) — pure try (peek; never parks). 158 * xco_event_poll(e, NULL, w) — park-if-not-ready (out discarded). 159 * 160 * Fusing closes the try/park TOCTOU window an MT impl would otherwise 161 * have to internally re-check, and lets each event arm a waiter 162 * atomically with its readiness check. 163 * 164 * Standard usage from a state machine: 165 * 166 * uintptr_t v; 167 * if (xco_event_poll(e, &v, &my_waiter.base)) { ... use v ... } 168 * else { return SUSPENDED; } 169 * 170 * Standard usage from an xco coroutine wrapper: 171 * 172 * uintptr_t v; 173 * xco_waker_t sw; 174 * xco_waker_init(&sw, rt, &xco_self()->base); 175 * if (!xco_event_poll(e, &v, &sw.base)) { 176 * xco_suspend(0); 177 * (void)xco_event_poll(e, &v, NULL); // sticky: now ready 178 * } 179 * ==================================================================== */ 180 181 /* ---- Waiter ------------------------------------------------------------ */ 182 183 /* Forward-declared here so the waiter and its fire seam can reference 184 * it; the full definition (and the rest of the MT surface) follows the 185 * runtime section below. */ 186 typedef struct xco_thread xco_thread_t; 187 188 typedef struct xco_waiter xco_waiter_t; 189 struct xco_waiter { 190 /* Doubly-linked while parked on an event waitlist, so unpark is 191 * O(1). Reused as the singly-linked next pointer while on the 192 * runtime ready queue (FIFO, no removal from middle); prev is 193 * undefined in that state and reset on the next park. Under XCO_MT, 194 * next is additionally the intrusive inbox link (accessed through 195 * _Atomic casts at the inbox push/pop sites only); the three list 196 * memberships — waitlist, ready queue, inbox — are disjoint in 197 * time. */ 198 xco_waiter_t *next; 199 xco_waiter_t *prev; 200 /* Fire callback. value is the event's payload at fire time — sticky 201 * events also store it on themselves, transient events (channels, 202 * one-shot signals) deliver only here. Waiters that don't care 203 * about the value just ignore the parameter. 204 * 205 * Invoke via xco_waiter_fire (below), not directly: the helper enforces 206 * the "fire receives a fully detached waiter" contract that makes it 207 * safe to re-park inside the callback. */ 208 void (*fire)(xco_waiter_t *w, uintptr_t value); 209 #ifdef XCO_MT 210 /* The thread fire must run on; NULL = fire-anywhere (inline on the 211 * calling thread). Never migrates while parked or inboxed. */ 212 xco_thread_t *home; 213 /* Payload stash while the waiter rides an inbox: set by the remote 214 * firer, handed to fire by the drain. */ 215 uintptr_t value; 216 #endif 217 }; 218 219 /* Initialize a waiter: detached links, the given fire callback, and (under 220 * XCO_MT) home = NULL / value = 0. Library init paths all route through 221 * this; callers building waiters by hand should too, then set `home` 222 * explicitly if the waiter must fire on a particular thread. */ 223 static inline void xco_waiter_init(xco_waiter_t *w, 224 void (*fire)(xco_waiter_t *, uintptr_t)) { 225 w->next = NULL; 226 w->prev = NULL; 227 w->fire = fire; 228 #ifdef XCO_MT 229 w->home = NULL; 230 w->value = 0; 231 #endif 232 } 233 234 #ifdef XCO_MT 235 /* Producer side of the inbox; defined with the rest of the thread 236 * abstraction below, declared here for the fire seam. */ 237 void xco_thread_post(xco_thread_t *t, xco_waiter_t *w); 238 /* The thread whose xco_rt_run / xco_thread_drain is currently active on 239 * this OS thread, or NULL. Maintained by those two entry points 240 * (save-and-restore, so nesting is fine). */ 241 extern _Thread_local xco_thread_t *xco__thread_current; 242 #endif 243 244 /* Canonical way to invoke a waiter's fire callback. Hands the callback a 245 * fully detached waiter so the callback (or whatever the resumed step 246 * does) can re-park on a fresh waitlist without colliding with stale 247 * link state. Detachers that lead into fire (queue pops, latch drains, 248 * etc.) don't need to clear prev/next themselves. 249 * 250 * Under XCO_MT this is the routing seam: a waiter whose home is another 251 * thread is not fired here — it becomes the message, posted to home's 252 * inbox; the home thread's drain fires it with the stashed value. */ 253 static inline void xco_waiter_fire(xco_waiter_t *w, uintptr_t value) { 254 w->prev = NULL; 255 w->next = NULL; 256 #ifdef XCO_MT 257 if (w->home && w->home != xco__thread_current) { 258 w->value = value; 259 xco_thread_post(w->home, w); 260 return; 261 } 262 #endif 263 w->fire(w, value); 264 } 265 266 /* ---- Event ------------------------------------------------------------ */ 267 268 typedef struct xco_event xco_event_t; 269 270 typedef struct { 271 /* Fused try + park. If the event is ready, write its value to *out 272 * (when out != NULL), do NOT park w, and return true. Otherwise, if 273 * w != NULL park it on the waitlist and return false; if w == NULL 274 * just return false without parking. *out is left untouched on the 275 * not-ready path. */ 276 bool (*poll)(xco_event_t *e, uintptr_t *out, xco_waiter_t *w); 277 /* Remove w from the waitlist. Idempotent: no-op if not parked. */ 278 void (*unpark)(xco_event_t *e, xco_waiter_t *w); 279 } xco_event_vtable_t; 280 281 struct xco_event { const xco_event_vtable_t *vt; }; 282 283 static inline bool xco_event_poll(xco_event_t *e, uintptr_t *out, xco_waiter_t *w) { 284 return e->vt->poll(e, out, w); 285 } 286 static inline void xco_event_unpark(xco_event_t *e, xco_waiter_t *w) { e->vt->unpark(e, w); } 287 288 /* ---- Runtime ---------------------------------------------------------- */ 289 290 /* Forward-declared: the optional timer source attached to the runtime. 291 * Defined in the timer section below. */ 292 typedef struct xco_timers xco_timers_t; 293 294 typedef struct xco_op xco_op_t; 295 296 typedef struct xco_runtime { 297 xco_waiter_t *head, *tail; 298 xco_timers_t *timers; /* optional; advanced inside xco_rt_run */ 299 /* Pending ops list (xco_op layer). The runtime never inspects ops; it 300 * only owns this list-head pair so the host can pull a batch via 301 * xco_rt_take_ops. op_epoch is bumped each take and recorded on each 302 * submit; PENDING vs IN_FLIGHT is a generation match (see xco_op). */ 303 xco_op_t *op_head, *op_tail; 304 uint64_t op_epoch; 305 #ifdef XCO_MT 306 xco_thread_t *thread; /* optional; drained inside xco_rt_run */ 307 #endif 308 } xco_runtime_t; 309 310 static inline void xco_rt_init(xco_runtime_t *rt) { 311 rt->head = rt->tail = NULL; 312 rt->timers = NULL; 313 rt->op_head = rt->op_tail = NULL; 314 rt->op_epoch = 0; 315 #ifdef XCO_MT 316 rt->thread = NULL; 317 #endif 318 } 319 320 /* Attach (or detach with NULL) a timer source. While attached, xco_rt_run 321 * advances it each pass with the now value the caller supplied; firing 322 * timers may enqueue more waiters, which the same xco_rt_run call then drains. */ 323 static inline void xco_rt_attach_timers(xco_runtime_t *rt, xco_timers_t *ts) { 324 rt->timers = ts; 325 } 326 327 /* Append w to the ready queue. Used by xco__waker_fire and by anyone 328 * else that wants a waiter resumed by the scheduler. */ 329 static inline void xco_rt_enqueue(xco_runtime_t *rt, xco_waiter_t *w) { 330 w->next = NULL; 331 if (rt->tail) rt->tail->next = w; 332 else rt->head = w; 333 rt->tail = w; 334 } 335 336 /* Drain the ready queue, resuming each waker's xco_step, until empty. 337 * Steps may re-arm on events (and thus leave the queue) or enqueue 338 * other steps; xco_rt_run keeps going until quiescent. now is forwarded to 339 * any attached timer source's advance(); pass 0 (or anything) when no 340 * source is attached. The library never reads a clock — now is always 341 * caller-supplied. Under XCO_MT, an attached thread's inbox joins the 342 * fixpoint (drained with the current-thread TLS installed), so 343 * quiescent means ready queue, due timers, and inbox are all empty. */ 344 void xco_rt_run(xco_runtime_t *rt, uint64_t now); 345 346 /* The canonical bridge between events and the scheduler. When fired, 347 * stashes the value and enqueues itself onto rt; xco_rt_run pops it and 348 * calls xco_step(mach, value), so the resumed step receives the event's 349 * payload directly without a re-try. 350 * 351 * Init once, re-park freely. The runtime hands the waiter back fully 352 * detached (next/prev cleared) before invoking the resumed step, so a 353 * subscriber that wants to wait on the next event can call xco_event_poll 354 * directly — no re-init needed unless rt or step changes. 355 * 356 * Under XCO_MT the waker's home is rt's attached thread, so a fire from 357 * any other thread routes the waker through rt's inbox and the enqueue 358 * runs on rt's own thread — the ready queue stays plain. */ 359 typedef struct { 360 xco_waiter_t base; 361 xco_runtime_t *rt; 362 xco_mach_t *mach; 363 uintptr_t resume_value; /* set by fire, consumed by xco_rt_run */ 364 } xco_waker_t; 365 366 /* Defined in xco.c; declared here so xco_waker_init can install it. */ 367 void xco__waker_fire(xco_waiter_t *w, uintptr_t value); 368 369 static inline void xco_waker_init(xco_waker_t *sw, xco_runtime_t *rt, xco_mach_t *m) { 370 xco_waiter_init(&sw->base, xco__waker_fire); 371 #ifdef XCO_MT 372 sw->base.home = rt ? rt->thread : NULL; 373 #endif 374 sw->rt = rt; 375 sw->mach = m; 376 sw->resume_value = 0; 377 } 378 379 #ifdef XCO_MT 380 /* ---- Thread (XCO_MT) --------------------------------------------------- */ 381 382 /* Freestanding thread abstraction: a Vyukov MPSC inbox, a host wakeup 383 * hook, and the wake-dedup flag. Nothing else — thread creation, 384 * joining, blocking, and waking are host bindings. What ties an 385 * xco_thread_t to an OS thread is only the host's discipline: one 386 * thread owns it (calls drain/try_park), everyone else only posts. 387 * 388 * A runtime pinned to a thread attaches that thread's xco_thread_t 389 * (xco_rt_attach_thread); cross-thread waiter fires find the runtime's 390 * inbox through it. A thread without a runtime — a worker draining 391 * posted jobs — is equally first-class: its job queue IS the inbox. 392 * 393 * The consumer loop for a bare worker: 394 * 395 * for (;;) { 396 * xco_thread_drain(&self->t); 397 * lock(mu); 398 * while (xco_thread_try_park(&self->t)) cond_wait(cv, mu); 399 * unlock(mu); 400 * } 401 * 402 * with t.wakeup = lock(mu); signal(cv); unlock(mu). An event-loop 403 * thread blocks in epoll/kevent instead, with t.wakeup writing an 404 * eventfd / triggering EVFILT_USER. */ 405 struct xco_thread { 406 /* Vyukov MPSC inbox. Producers xchg inbox_tail; the owner thread 407 * follows inbox_head. Links ride the waiter's next field through 408 * _Atomic casts at the push/pop sites. */ 409 xco_waiter_t inbox_stub; /* sentinel */ 410 _Atomic(xco_waiter_t *) inbox_tail; 411 xco_waiter_t *inbox_head; /* consumer-only */ 412 /* Wake dedup: only the empty -> non-empty post edge invokes wakeup. */ 413 _Atomic bool wakeup_pending; 414 /* Host hook: rouse this thread. Called from arbitrary threads; must 415 * be safe to invoke concurrently and redundantly. NULL = no hook 416 * (a thread that never blocks, or polls). */ 417 void (*wakeup)(xco_thread_t *t); 418 void *wakeup_ud; 419 /* The runtime pinned to this thread, or NULL for a bare worker. 420 * Set by xco_rt_attach_thread. */ 421 xco_runtime_t *rt; 422 }; 423 424 static inline void xco_thread_init(xco_thread_t *t, 425 void (*wakeup)(xco_thread_t *), void *ud) { 426 xco_waiter_init(&t->inbox_stub, NULL); 427 t->inbox_head = &t->inbox_stub; 428 atomic_store_explicit(&t->inbox_tail, &t->inbox_stub, memory_order_relaxed); 429 atomic_store_explicit(&t->wakeup_pending, false, memory_order_relaxed); 430 t->wakeup = wakeup; 431 t->wakeup_ud = ud; 432 t->rt = NULL; 433 } 434 435 /* Attach t to rt (same shape as xco_rt_attach_timers): xco_rt_run will 436 * drain t's inbox in its fixpoint, wakers created against rt route home 437 * to t, and xco_thread_drain on a bare t=NULL detach is a no-op. */ 438 static inline void xco_rt_attach_thread(xco_runtime_t *rt, xco_thread_t *t) { 439 rt->thread = t; 440 if (t) t->rt = rt; 441 } 442 443 /* xco_thread_post — producer side, any thread (declared above with the 444 * fire seam): push w onto t's inbox and invoke t->wakeup on the 445 * empty -> non-empty edge. w must be detached (not parked, queued, or 446 * already inboxed). This is the primitive under cross-thread fires; call 447 * it directly to hand a job waiter to a worker. Lock-free; safe from a 448 * signal handler if t->wakeup is. */ 449 450 /* Consumer side, owner thread only: pop and fire every inboxed waiter 451 * (installing/restoring the current-thread TLS around the fires, so 452 * nested fires and re-routes behave). Waiters whose home was retargeted 453 * mid-flight are re-routed, not fired here. */ 454 void xco_thread_drain(xco_thread_t *t); 455 456 /* Consumer side, owner thread only: the lost-wake handshake. Clears 457 * wakeup_pending, then re-checks the inbox. True = inbox empty, safe to 458 * block on the host primitive; false = more arrived, drain again first. 459 * Call with the same mutual exclusion against t->wakeup that the host's 460 * blocking primitive requires (see the worker-loop sketch above). */ 461 bool xco_thread_try_park(xco_thread_t *t); 462 463 #endif /* XCO_MT */ 464 465 /* ---- Latch ------------------------------------------------------------ */ 466 467 /* One-shot sticky event. set() flips the bit, stores the payload, and 468 * fires every waiter. Subsequent set() calls are ignored. To re-arm, 469 * reinitialize a fresh latch. */ 470 typedef struct { 471 xco_event_t base; 472 bool set; 473 uintptr_t value; 474 xco_waiter_t *waiters; 475 } xco_latch_t; 476 477 /* Defined in xco.c; referenced by xco_latch_init. */ 478 extern const xco_event_vtable_t xco__latch_vt; 479 480 static inline void xco_latch_init(xco_latch_t *l) { 481 l->base.vt = &xco__latch_vt; 482 l->set = false; 483 l->value = 0; 484 l->waiters = NULL; 485 } 486 487 void xco_latch_set(xco_latch_t *l, uintptr_t value); 488 489 /* ---- Countdown -------------------------------------------------------- */ 490 491 /* One-shot fan-in counter. Fires its embedded latch (payload 0) when 492 * remaining hits 0. xco_countdown_add(n) is legal while remaining > 0; 493 * xco_countdown_done decrements; both are UB once the latch has fired. 494 * 495 * Compose with the standard event API via xco_countdown_event(). */ 496 typedef struct xco_countdown { 497 xco_latch_t done; 498 size_t remaining; 499 } xco_countdown_t; 500 501 static inline void xco_countdown_init(xco_countdown_t *c, size_t n) { 502 xco_latch_init(&c->done); 503 c->remaining = n; 504 if (n == 0) xco_latch_set(&c->done, 0); 505 } 506 507 static inline void xco_countdown_add(xco_countdown_t *c, size_t n) { 508 assert(!c->done.set); 509 c->remaining += n; 510 } 511 512 static inline void xco_countdown_done(xco_countdown_t *c) { 513 assert(c->remaining > 0); 514 if (--c->remaining == 0) xco_latch_set(&c->done, 0); 515 } 516 517 static inline xco_event_t *xco_countdown_event(xco_countdown_t *c) { return &c->done.base; } 518 static inline bool xco_countdown_fired(const xco_countdown_t *c) { return c->done.set; } 519 520 /* ---- Notify (wake-one / wake-all) ------------------------------------- */ 521 522 /* Transient signal with no sticky state. xco_notify_one fires (and detaches) 523 * the head of a FIFO waitlist; xco_notify_all fires every parked waiter. Both 524 * are no-ops when the waitlist is empty. Subscribers must re-park to see 525 * subsequent notifications. 526 * 527 * xco_event_poll never reports ready: there is no "ready now" state — a 528 * subscriber waits for the *next* notify. */ 529 typedef struct xco_notify { 530 xco_event_t base; 531 xco_waiter_t *head, *tail; 532 } xco_notify_t; 533 534 extern const xco_event_vtable_t xco__notify_vt; 535 536 static inline void xco_notify_init(xco_notify_t *n) { 537 n->base.vt = &xco__notify_vt; 538 n->head = n->tail = NULL; 539 } 540 541 static inline xco_event_t *xco_notify_event(xco_notify_t *n) { return &n->base; } 542 543 void xco_notify_one(xco_notify_t *n); 544 void xco_notify_all(xco_notify_t *n); 545 546 /* ---- Semaphore -------------------------------------------------------- */ 547 548 /* Counting semaphore. acquire is exposed as xco_event_t (composable with 549 * select / xco_wait_or_cancel): xco_event_poll succeeds and decrements when 550 * permits > 0; otherwise the waiter parks FIFO. xco_semaphore_release(n) 551 * hands one permit to each of up to n waiting waiters (each is fired, 552 * which the receiver treats as "you got a permit") before adding any 553 * leftover to the count. 554 * 555 * One permit per acquire. Bulk acquire isn't expressible in xco_event_t's 556 * shape; if you need it, call sequentially. For binary use (mutex-style 557 * critical section across awaits) init with permits = 1. 558 * 559 * Fairness: FIFO at the waitlist. A racing inline xco_event_poll by a fresh 560 * caller can jump ahead of parked waiters when permits are released 561 * back to count rather than directly handed off — release prefers 562 * parked waiters first to avoid that. */ 563 typedef struct xco_semaphore { 564 xco_event_t acquire; 565 size_t permits; 566 xco_waiter_t *head, *tail; 567 } xco_semaphore_t; 568 569 extern const xco_event_vtable_t xco__semaphore_acquire_vt; 570 571 static inline void xco_semaphore_init(xco_semaphore_t *s, size_t initial) { 572 s->acquire.vt = &xco__semaphore_acquire_vt; 573 s->permits = initial; 574 s->head = s->tail = NULL; 575 } 576 577 static inline xco_event_t *xco_semaphore_event(xco_semaphore_t *s) { return &s->acquire; } 578 579 void xco_semaphore_release(xco_semaphore_t *s, size_t n); 580 581 /* ---- Mutex ------------------------------------------------------------ */ 582 583 /* Binary semaphore wrapper for vocabulary at call sites. xco_mutex_init is 584 * xco_semaphore_init(s, 1); the xco_event_t fires once per release; xco_mutex_release 585 * hands the permit to the next waiter (or returns it to the count). */ 586 typedef xco_semaphore_t xco_mutex_t; 587 588 static inline void xco_mutex_init (xco_mutex_t *m) { xco_semaphore_init(m, 1); } 589 static inline xco_event_t *xco_mutex_event (xco_mutex_t *m) { return xco_semaphore_event(m); } 590 static inline void xco_mutex_release(xco_mutex_t *m) { xco_semaphore_release(m, 1); } 591 592 /* ---- Select / all-of -------------------------------------------------- */ 593 594 /* Wait over N input events. Two semantics share the same storage shape, 595 * so a caller can switch between them by changing only the init call: 596 * 597 * xco_select_event_init fires when ANY input fires (any-of) 598 * xco_allof_event_init fires when ALL inputs fire (all-of) 599 * 600 * In both cases done's payload is the index of the input whose firing 601 * closed the wait — the winner for select, the last-to-fire for allof — 602 * and inputs[i].value carries each fired input's payload (works 603 * uniformly for sticky and transient sources, where re-trying the input 604 * would either succeed or fail). 605 * 606 * Composes: a select_event is itself an event. */ 607 608 typedef struct xco_select_event xco_select_event_t; 609 610 /* Per-input arming record. Caller-allocated as an array of n alongside 611 * the select_event. After fire, .value holds whatever the input 612 * delivered; other fields are internal. */ 613 typedef struct { 614 xco_waiter_t w; 615 xco_event_t *src; 616 xco_select_event_t *parent; 617 uintptr_t value; /* captured at fire time */ 618 } xco_select_input_t; 619 620 struct xco_select_event { 621 xco_latch_t done; /* fires with the closing input's index */ 622 xco_select_input_t *inputs; 623 size_t n; 624 size_t remaining; /* counts down; done fires at 0 625 * (select: starts at 1, allof: at n) */ 626 }; 627 628 /* Initialize as a select (any-of). inputs[] is caller-provided storage 629 * for n nodes; srcs[] is the array of n input event pointers (read 630 * once during init). If any input is already ready, the select fires 631 * immediately and no waiters are parked. Use &s->done.base as the 632 * resulting xco_event_t. */ 633 void xco_select_event_init(xco_select_event_t *s, 634 xco_select_input_t *inputs, size_t n, 635 xco_event_t *const *srcs); 636 637 /* Initialize as an allof (all-of). Inputs already ready at init are 638 * consumed inline (no parking, value captured); if every input is 639 * ready, done fires immediately. n == 0 fires done with payload 0. */ 640 void xco_allof_event_init(xco_select_event_t *s, 641 xco_select_input_t *inputs, size_t n, 642 xco_event_t *const *srcs); 643 644 /* Disarm any inputs still parked. Safe to call after fire (no-op) and 645 * after partial completion (allof). Required before s leaves scope if 646 * it has not yet fired. */ 647 void xco_select_event_deinit(xco_select_event_t *s); 648 649 650 /* ---- Queue ------------------------------------------------------------ */ 651 652 /* Bounded FIFO of uintptr_t. Caller provides the ring buffer storage. 653 * Recv side is exposed as xco_event_t (composable with select). Send side 654 * is a typed API (carries a value), shaped after the event-poll fusion: 655 * fused try + park, NULL-waiter degenerates to pure-try. 656 * 657 * Three full-buffer policies, fixed at init: 658 * XCO_QUEUE_BLOCK senders park until a receiver makes room. 659 * XCO_QUEUE_DROP_NEWEST xco_queue_send_poll silently discards the new value. 660 * XCO_QUEUE_DROP_OLDEST xco_queue_send_poll evicts the head and pushes new tail. 661 * 662 * Senders never park under DROP_* policies — passing a non-NULL qsw is 663 * only meaningful under XCO_QUEUE_BLOCK. xco_queue_send_unpark is 664 * idempotent (cancellation-safe). 665 * 666 * Direct-handoff: xco_queue_send_poll first checks for a parked receiver and 667 * delivers inline if present (payload bypasses the buffer), regardless 668 * of policy. 669 * 670 * Rendezvous matrix (BLOCK + cap=0; the xco_chan_* aliases at the bottom 671 * of this section name this configuration explicitly): 672 * send + parked recv fire recv with value, sender continues inline. 673 * send + no recv sender parks (xco_queue_send_poll); peer pulls later. 674 * recv + parked sender read sender's value, fire sender (delivery 675 * confirmation), receiver continues inline. 676 * recv + no sender receiver parks (xco_event_poll on recv); peer 677 * delivers later. 678 * 679 * FIFO order on both waitlists. 680 * 681 * Close: optional EOF semantics. After xco_queue_close, xco_queue_send_poll 682 * returns XCO_QSEND_CLOSED regardless of policy (no delivery), parked 683 * senders are drained with delivered=false, and parked receivers are 684 * woken so they can observe XCO_RECV_CLOSED via xco_queue_recv. The recv 685 * event is "ready" iff a value is available OR the queue is closed — 686 * receivers must call xco_queue_recv to disambiguate value vs EOF. */ 687 688 /* Result of a typed receive on a queue (or chan, which is just a queue). */ 689 typedef enum { 690 XCO_RECV_GOT, /* *out holds the delivered value */ 691 XCO_RECV_EMPTY, /* nothing available right now; caller may park */ 692 XCO_RECV_CLOSED, /* peer closed and no values remain */ 693 } xco_recv_status_t; 694 695 typedef enum { 696 XCO_QUEUE_BLOCK, 697 XCO_QUEUE_DROP_NEWEST, 698 XCO_QUEUE_DROP_OLDEST, 699 } xco_queue_policy_t; 700 701 typedef struct xco_queue { 702 xco_event_t recv; 703 uintptr_t *buf; 704 size_t cap, head, len; 705 xco_queue_policy_t policy; 706 xco_waiter_t *send_head, *send_tail; 707 xco_waiter_t *recv_head, *recv_tail; 708 bool closed; 709 } xco_queue_t; 710 711 extern const xco_event_vtable_t xco__queue_recv_vt; 712 713 static inline void xco_queue_init(xco_queue_t *q, uintptr_t *buf, size_t cap, 714 xco_queue_policy_t policy) { 715 q->recv.vt = &xco__queue_recv_vt; 716 q->buf = buf; 717 q->cap = cap; 718 q->head = 0; 719 q->len = 0; 720 q->policy = policy; 721 q->send_head = q->send_tail = NULL; 722 q->recv_head = q->recv_tail = NULL; 723 q->closed = false; 724 } 725 726 static inline xco_event_t *xco_queue_recv_event(xco_queue_t *q) { return &q->recv; } 727 728 /* Sender-side waiter for XCO_QUEUE_BLOCK. Same shape as xco_chan_send_waiter_t: 729 * a waker plus a value slot the receiver / close-drain reads back on the 730 * park path. `delivered` is set by the closing side: true on a normal 731 * handoff, false on a close drain. */ 732 typedef struct { 733 xco_waker_t sw; 734 uintptr_t value; /* set by xco_queue_send_poll on the park path */ 735 bool delivered; 736 } xco_queue_send_waiter_t; 737 738 static inline void xco_queue_send_waiter_init(xco_queue_send_waiter_t *qsw, 739 xco_runtime_t *rt, xco_mach_t *m) { 740 xco_waker_init(&qsw->sw, rt, m); 741 qsw->value = 0; 742 qsw->delivered = false; 743 } 744 745 /* Result of xco_queue_send_poll. */ 746 typedef enum { 747 XCO_QSEND_ACCEPTED, /* delivered to a parked receiver, buffered, or 748 accepted-by-policy (silently dropped under 749 DROP_NEWEST, evicted-and-pushed under DROP_OLDEST) */ 750 XCO_QSEND_BLOCKED, /* BLOCK + full; parked iff qsw != NULL */ 751 XCO_QSEND_CLOSED, /* queue closed; never parks. Returned regardless 752 of policy — closed is closed. */ 753 } xco_queue_send_status_t; 754 755 /* Fused try + park for a sender. Direct-delivers to a parked receiver if 756 * one is waiting (returns XCO_QSEND_ACCEPTED, qsw not parked). Otherwise: 757 * 758 * XCO_QUEUE_BLOCK + room buffered → XCO_QSEND_ACCEPTED. 759 * XCO_QUEUE_BLOCK + full XCO_QSEND_BLOCKED; if qsw != NULL the 760 * sender's value is stashed in qsw->value 761 * and qsw is parked. 762 * XCO_QUEUE_DROP_NEWEST + full silently drops → XCO_QSEND_ACCEPTED. 763 * XCO_QUEUE_DROP_OLDEST + full evicts head, pushes new tail → ACCEPTED. 764 * closed (any policy) XCO_QSEND_CLOSED; never parks. 765 * 766 * The two degenerate forms mirror xco_event_poll: 767 * xco_queue_send_poll(q, v, NULL) — pure try (peek; never parks). 768 * xco_queue_send_poll(q, v, qsw) — fused try-or-park (BLOCK only). */ 769 xco_queue_send_status_t xco_queue_send_poll(xco_queue_t *q, uintptr_t value, 770 xco_queue_send_waiter_t *qsw); 771 772 void xco_queue_send_unpark(xco_queue_t *q, xco_queue_send_waiter_t *qsw); 773 774 /* Typed receive. Disambiguates value vs EOF where xco_event_poll cannot: 775 * returns XCO_RECV_GOT (value popped from the buffer or directly from a 776 * parked sender), XCO_RECV_CLOSED (closed and drained), or 777 * XCO_RECV_EMPTY (caller may park). */ 778 xco_recv_status_t xco_queue_recv(xco_queue_t *q, uintptr_t *out); 779 780 /* Close the queue. Idempotent. Drains parked senders (delivered=false) 781 * and wakes parked receivers. After close, xco_queue_send_poll returns 782 * XCO_QSEND_CLOSED regardless of policy. */ 783 void xco_queue_close(xco_queue_t *q); 784 static inline bool xco_queue_is_closed(const xco_queue_t *q) { return q->closed; } 785 786 /* Selectable send op. A per-call object that holds the value, parks on 787 * the queue (only meaningful under XCO_QUEUE_BLOCK), and exposes 788 * &op->done.base as the event that fires when the send resolves. 789 * 790 * The op embeds a xco_queue_send_waiter_t (so the queue's send list stays 791 * uniform — receivers read .value at the same offset for both direct 792 * and op senders) but rewires its fire callback: instead of resuming an 793 * xco_step, fire sets op->done. Polymorphism via the function pointer. 794 * 795 * The done latch's payload is 1 on XCO_QSEND_ACCEPTED (handed to a 796 * receiver, buffered, or accepted under DROP_*) and 0 on 797 * XCO_QSEND_CLOSED / close-drain. 798 * 799 * Under DROP_* policies the send always resolves inline at init (the 800 * poll returns ACCEPTED and op->done fires immediately). 801 * 802 * Lifecycle: init → wait on &op->done.base → deinit. Always deinit; 803 * it's a no-op after resolution and unparks the queue-side waiter if not. */ 804 typedef struct { 805 xco_queue_send_waiter_t qsw; /* parked on queue; fire overridden */ 806 xco_queue_t *queue; 807 xco_latch_t done; 808 } xco_queue_send_op_t; 809 810 void xco__queue_send_op_fire(xco_waiter_t *w, uintptr_t value); 811 812 void xco_queue_send_op_init(xco_queue_send_op_t *op, xco_queue_t *q, uintptr_t value); 813 static inline void xco_queue_send_op_deinit(xco_queue_send_op_t *op) { 814 if (op->done.set) return; 815 xco_queue_send_unpark(op->queue, &op->qsw); 816 } 817 818 /* ---- Channel (alias) -------------------------------------------------- */ 819 820 /* Unbuffered rendezvous channel: a queue at cap=0 with XCO_QUEUE_BLOCK 821 * policy. Senders and receivers wait on each other; whichever arrives 822 * first parks until its peer shows up. The pending value lives in the 823 * sender's xco_chan_send_waiter_t for the duration of any wait — no 824 * per-channel buffer storage. 825 * 826 * The xco_chan_* names are thin aliases over the queue API: a chan IS a 827 * queue. They exist so call sites can name "rendezvous" explicitly 828 * rather than spelling out cap=0+BLOCK. The queue's recv event, send 829 * poll, close, recv, and selectable send op all carry over unchanged. 830 * 831 * Storage: an xco_chan_t carries the queue's buffer/cap/policy fields 832 * even though they're inert at cap=0 (~40 bytes of overhead vs a 833 * dedicated rendezvous struct). Below the noise floor for typical use. */ 834 835 typedef xco_queue_t xco_chan_t; 836 typedef xco_queue_send_waiter_t xco_chan_send_waiter_t; 837 typedef xco_queue_send_status_t xco_chan_send_status_t; 838 typedef xco_queue_send_op_t xco_chan_send_op_t; 839 840 /* Status aliases. Same enumerators, vocabulary at call sites: a chan 841 * "delivers" rather than "accepts" — but the underlying constants are 842 * the queue's. */ 843 #define XCO_SEND_DELIVERED XCO_QSEND_ACCEPTED 844 #define XCO_SEND_BLOCKED XCO_QSEND_BLOCKED 845 #define XCO_SEND_CLOSED XCO_QSEND_CLOSED 846 847 static inline void xco_chan_init(xco_chan_t *c) { 848 xco_queue_init(c, NULL, 0, XCO_QUEUE_BLOCK); 849 } 850 static inline xco_event_t *xco_chan_recv_event(xco_chan_t *c) { 851 return xco_queue_recv_event(c); 852 } 853 static inline void xco_chan_send_waiter_init(xco_chan_send_waiter_t *csw, 854 xco_runtime_t *rt, xco_mach_t *m) { 855 xco_queue_send_waiter_init(csw, rt, m); 856 } 857 static inline xco_chan_send_status_t xco_chan_send_poll(xco_chan_t *c, uintptr_t value, 858 xco_chan_send_waiter_t *csw) { 859 return xco_queue_send_poll(c, value, csw); 860 } 861 static inline void xco_chan_send_unpark(xco_chan_t *c, xco_chan_send_waiter_t *csw) { 862 xco_queue_send_unpark(c, csw); 863 } 864 static inline xco_recv_status_t xco_chan_recv(xco_chan_t *c, uintptr_t *out) { 865 return xco_queue_recv(c, out); 866 } 867 static inline void xco_chan_close(xco_chan_t *c) { xco_queue_close(c); } 868 static inline bool xco_chan_is_closed(const xco_chan_t *c) { 869 return xco_queue_is_closed(c); 870 } 871 static inline void xco_chan_send_op_init(xco_chan_send_op_t *op, xco_chan_t *c, uintptr_t value) { 872 xco_queue_send_op_init(op, c, value); 873 } 874 static inline void xco_chan_send_op_deinit(xco_chan_send_op_t *op) { 875 xco_queue_send_op_deinit(op); 876 } 877 878 /* ---- Broadcast (slot) ------------------------------------------------- */ 879 880 /* Re-armable signal carrying a "latest value" slot. Subscribers park on 881 * the event; xco_broadcast_publish stores the new value, fires every parked 882 * subscriber with that value, and clears the waitlist — subscribers must 883 * re-park to see further publishes. Subscribers that aren't parked at 884 * publish time miss that publish but will see the next one. This is the 885 * coalescing "watch a slot" semantics, not lossless fan-out. 886 * 887 * xco_event_poll never reports ready: there is no "ready now" state — a 888 * subscriber waits for the *next* publish. To read the latest published 889 * value at any time, use xco_broadcast_value (valid once xco_broadcast_has_value 890 * returns true). 891 * 892 * For lossless multi-consumer delivery, give each subscriber its own 893 * queue and have the producer write to all of them. */ 894 895 typedef struct xco_broadcast { 896 xco_event_t base; 897 bool has_value; 898 uintptr_t value; 899 xco_waiter_t *waiters; 900 } xco_broadcast_t; 901 902 extern const xco_event_vtable_t xco__broadcast_vt; 903 904 static inline void xco_broadcast_init(xco_broadcast_t *b) { 905 b->base.vt = &xco__broadcast_vt; 906 b->has_value = false; 907 b->value = 0; 908 b->waiters = NULL; 909 } 910 911 static inline xco_event_t *xco_broadcast_event (xco_broadcast_t *b) { return &b->base; } 912 static inline bool xco_broadcast_has_value(const xco_broadcast_t *b) { return b->has_value; } 913 static inline uintptr_t xco_broadcast_value (const xco_broadcast_t *b) { return b->value; } 914 915 void xco_broadcast_publish(xco_broadcast_t *b, uintptr_t value); 916 917 /* ---- Cancellation ----------------------------------------------------- */ 918 919 /* A cancellation token is a sticky latch — these aliases exist for 920 * vocabulary at call sites. xco_cancel_set fires every parked waiter; the 921 * idempotency of xco_latch_set means racing cancellers are fine. 922 * 923 * Pair a xco_cancel_t with any blocking event via xco_wait_or_cancel to get 924 * "await X, or be cancelled." 925 * 926 * Discipline: cancellation notifies; it never drops in-flight values. 927 * A cancelled await returns control to its caller, which is responsible 928 * for draining whatever it owns — deinit a pending chan_send_op so its 929 * value goes back to the sender, deinit a select_event so input waiters 930 * detach, drive a cancellable coroutine to XCO_STEP_DEAD before freeing 931 * its stack. The xco layer does no unwinding; the coroutine cooperates. */ 932 933 typedef xco_latch_t xco_cancel_t; 934 935 static inline void xco_cancel_init(xco_cancel_t *c) { xco_latch_init(c); } 936 static inline void xco_cancel_set(xco_cancel_t *c) { xco_latch_set(c, 0); } 937 static inline bool xco_cancel_is_set(const xco_cancel_t *c) { return c->set; } 938 static inline xco_event_t *xco_cancel_event(xco_cancel_t *c) { return &c->base; } 939 940 /* Outcome indices for xco_wait_or_cancel — match the inputs[] order so the 941 * value the resumer receives from the latched select reads as one of 942 * these directly. */ 943 enum { 944 XCO_WAIT_OK = 0, /* ev fired; inputs[0].value holds its payload */ 945 XCO_WAIT_CANCELLED = 1, /* cancel fired before ev */ 946 }; 947 948 /* Build a select over (ev, cancel) using caller-provided storage. If 949 * either is already ready at init the select fast-paths and never parks 950 * anyone (ev is checked first, so an event that has already resolved 951 * wins over a concurrent cancel). Treat &out->done.base as the event 952 * to wait on. Always pair with xco_select_event_deinit before storage 953 * leaves scope (no-op once fired). */ 954 static inline void xco_wait_or_cancel(xco_select_event_t *out, 955 xco_select_input_t inputs[2], 956 xco_event_t *ev, xco_cancel_t *c) { 957 xco_event_t *srcs[2] = {ev, xco_cancel_event(c)}; 958 xco_select_event_init(out, inputs, 2, srcs); 959 } 960 961 /* ---- Timers ----------------------------------------------------------- */ 962 963 /* A timer is a sticky event keyed on a u64 deadline. It fires (exactly 964 * once) when the attached timer source is advanced past that deadline. 965 * The library never reads a clock; the caller provides `now` to 966 * xco_timers_advance (or via xco_rt_run). 967 * 968 * Storage is pluggable through the timers vtable so callers can swap a 969 * pairing heap (in-tree, O(log n) amortized everywhere including cancel) 970 * for a wheel or other structure without touching the timer/timeout 971 * surface. The timer struct holds the heap link fields inline; the source 972 * impl interprets them. 973 * 974 * Lifecycle: 975 * xco_timer_init(t, ts, deadline) // inserts into ts 976 * ... wait on xco_timer_event(t), or compose into select/xco_wait_or_cancel ... 977 * xco_timer_deinit(t) // removes from ts if not yet fired 978 * 979 * Fire payload is the deadline. Re-arming = reinit a fresh timer. */ 980 981 typedef struct xco_timer xco_timer_t; 982 983 typedef struct { 984 /* Insert t into the source. t must be initialized but not yet 985 * inserted; insert sets t's heap link fields. */ 986 void (*insert) (xco_timers_t *ts, xco_timer_t *t); 987 /* Remove t from the source if currently inserted. Caller must 988 * ensure t was inserted into this same source. */ 989 void (*cancel) (xco_timers_t *ts, xco_timer_t *t); 990 /* Fire every timer whose deadline <= now, in deadline order, popping 991 * each from the source. Each fire drains the timer's waiter list. */ 992 void (*advance)(xco_timers_t *ts, uint64_t now); 993 /* Return the earliest queued deadline, or UINT64_MAX if no timer 994 * is queued. */ 995 uint64_t (*peek)(const xco_timers_t *ts); 996 } xco_timers_vtable_t; 997 998 struct xco_timers { 999 const xco_timers_vtable_t *vt; 1000 uint64_t now; /* most recent advance() input; monotonic */ 1001 }; 1002 1003 static inline void xco_timers_insert (xco_timers_t *ts, xco_timer_t *t) { ts->vt->insert (ts, t); } 1004 static inline void xco_timers_cancel (xco_timers_t *ts, xco_timer_t *t) { ts->vt->cancel (ts, t); } 1005 static inline void xco_timers_advance(xco_timers_t *ts, uint64_t now) { 1006 assert(now >= ts->now); 1007 ts->now = now; 1008 ts->vt->advance(ts, now); 1009 } 1010 static inline uint64_t xco_timers_peek (const xco_timers_t *ts) { return ts->vt->peek(ts); } 1011 static inline uint64_t xco_timers_now (const xco_timers_t *ts) { return ts->now; } 1012 1013 /* Concrete timer. Embeds a latch so try/park/unpark and the fire-all 1014 * waitlist handling come for free; the timer source manipulates only 1015 * the heap link fields and triggers the latch on fire. The latch's 1016 * payload after fire is the timer's deadline. */ 1017 struct xco_timer { 1018 xco_latch_t done; /* fires once, payload = deadline */ 1019 uint64_t deadline; 1020 xco_timers_t *src; /* source this timer is registered with */ 1021 bool in_heap; /* true between insert and fire/cancel */ 1022 /* Pairing-heap link fields; opaque to anyone but the source impl. 1023 * prev is parent if first child, else previous sibling, else NULL. */ 1024 xco_timer_t *child, *prev, *next; 1025 }; 1026 1027 static inline xco_event_t *xco_timer_event(xco_timer_t *t) { return &t->done.base; } 1028 static inline bool xco_timer_fired(const xco_timer_t *t) { return t->done.set; } 1029 1030 static inline void xco_timer_init(xco_timer_t *t, xco_timers_t *ts, uint64_t deadline) { 1031 xco_latch_init(&t->done); 1032 t->deadline = deadline; 1033 t->src = ts; 1034 t->in_heap = false; 1035 t->child = NULL; 1036 t->prev = NULL; 1037 t->next = NULL; 1038 xco_timers_insert(ts, t); 1039 } 1040 1041 static inline void xco_timer_deinit(xco_timer_t *t) { 1042 if (t->in_heap) xco_timers_cancel(t->src, t); 1043 } 1044 1045 /* In-tree timer source: intrusive pairing heap. O(1) amortized insert 1046 * and meld; O(log n) amortized advance and cancel. No per-source 1047 * allocation — the heap is just a root pointer; nodes live in the 1048 * caller's xco_timer_t's. */ 1049 typedef struct { 1050 xco_timers_t base; 1051 xco_timer_t *root; 1052 } xco_pairing_heap_t; 1053 1054 extern const xco_timers_vtable_t xco__pairing_heap_vt; 1055 1056 static inline void xco_pairing_heap_init(xco_pairing_heap_t *h) { 1057 h->base.vt = &xco__pairing_heap_vt; 1058 h->base.now = 0; 1059 h->root = NULL; 1060 } 1061 1062 /* ---- Timeout ---------------------------------------------------------- */ 1063 1064 /* Bundle: a timer that fires a xco_cancel_t on expiration. The natural 1065 * pairing for "await ev, or be cancelled by deadline": 1066 * 1067 * xco_timeout_t to; 1068 * xco_timeout_init(&to, ts, now + budget); 1069 * xco_select_event_t sel; xco_select_input_t inputs[2]; 1070 * xco_wait_or_cancel(&sel, inputs, ev, &to.cancel); 1071 * ... wait on &sel.done.base ... 1072 * xco_select_event_deinit(&sel); 1073 * xco_timeout_deinit(&to); // safe whether the timer fired or not 1074 * 1075 * The bridge waiter is parked on the timer; when it fires it sets the 1076 * cancel. Bridge fire is idempotent vs xco_cancel_set (a sticky latch). */ 1077 typedef struct xco_timeout { 1078 xco_timer_t timer; 1079 xco_cancel_t cancel; 1080 xco_waiter_t bridge; 1081 } xco_timeout_t; 1082 1083 void xco_timeout_init(xco_timeout_t *to, xco_timers_t *ts, uint64_t deadline); 1084 static inline void xco_timeout_deinit(xco_timeout_t *to) { 1085 xco_event_unpark(xco_timer_event(&to->timer), &to->bridge); 1086 xco_timer_deinit(&to->timer); 1087 } 1088 1089 /* ---- Ticker ----------------------------------------------------------- */ 1090 1091 /* Re-armable transient signal driven by a timer source. Each time the 1092 * underlying timer fires, the ticker computes the next deadline (period 1093 * past the just-fired one, with skip-ahead for catch-up after overflow), 1094 * reinstalls the timer, and fires every parked subscriber with the 1095 * just-fired deadline as the payload. Subscribers that aren't parked at 1096 * a fire miss it (transient — same coalescing semantics as broadcast). 1097 * 1098 * xco_ticker_init(&t, ts, period, first_deadline); 1099 * ... wait on xco_ticker_event(&t), re-park to see further ticks ... 1100 * xco_ticker_deinit(&t); // cancels the in-flight timer 1101 * 1102 * xco_event_poll never reports ready; subscribers wait for the *next* tick. */ 1103 typedef struct xco_ticker { 1104 xco_timer_t timer; 1105 xco_timers_t *src; 1106 uint64_t period; 1107 xco_event_t base; 1108 xco_waiter_t *waiters; 1109 xco_waiter_t bridge; /* internal: parks on xco_timer_event */ 1110 } xco_ticker_t; 1111 1112 extern const xco_event_vtable_t xco__ticker_vt; 1113 1114 void xco_ticker_init (xco_ticker_t *t, xco_timers_t *ts, 1115 uint64_t period, uint64_t first_deadline); 1116 void xco_ticker_deinit(xco_ticker_t *t); 1117 static inline xco_event_t *xco_ticker_event(xco_ticker_t *t) { return &t->base; } 1118 1119 /* ---- Task ------------------------------------------------------------- */ 1120 1121 /* Lifecycle handle for a running xco_step. Bundles a done latch (fires when 1122 * the xco_step returns, payload = its return value) with a cancel latch 1123 * (the canonical signal to ask the xco_step to wind down). The xco_step itself 1124 * is caller-allocated; the task holds a pointer to it. 1125 * 1126 * Who fires done: 1127 * - Hand-coded state machine: call xco_task_done(t, ret) in the same arm 1128 * that returns XCO_STEP_DEAD. 1129 * - xco-backed task (see xco_cotask_t below): the xco_trampoline calls 1130 * xco_task_done automatically with the coroutine's return value. 1131 * 1132 * Cooperation: cancellation only notifies — the xco_step is responsible for 1133 * draining what it owns and reaching XCO_STEP_DEAD. The task's cancel is a 1134 * normal xco_cancel_t, so the xco_step typically composes xco_wait_or_cancel against 1135 * it on every blocking await. 1136 * 1137 * Joining: callers wait on xco_task_done_event with the standard event API 1138 * (try / park, or compose into select / xco_wait_or_cancel). On fire the 1139 * latch's payload is the xco_step's return value. */ 1140 1141 typedef struct xco_task { 1142 xco_mach_t *mach; 1143 xco_latch_t done; 1144 xco_cancel_t cancel; 1145 } xco_task_t; 1146 1147 static inline void xco_task_init(xco_task_t *t, xco_mach_t *mach) { 1148 t->mach = mach; 1149 xco_latch_init(&t->done); 1150 xco_cancel_init(&t->cancel); 1151 } 1152 1153 /* Mark the task complete with its return value. Idempotent (xco_latch_set is). */ 1154 static inline void xco_task_done(xco_task_t *t, uintptr_t value) { 1155 xco_latch_set(&t->done, value); 1156 } 1157 1158 static inline xco_event_t *xco_task_done_event (xco_task_t *t) { return &t->done.base; } 1159 static inline xco_cancel_t *xco_task_cancel (xco_task_t *t) { return &t->cancel; } 1160 static inline bool xco_task_finished (const xco_task_t *t) { return t->done.set; } 1161 static inline bool xco_task_is_cancelled(const xco_task_t *t) { return xco_cancel_is_set(&t->cancel); } 1162 static inline xco_mach_t *xco_task_mach (xco_task_t *t) { return t->mach; } 1163 1164 /* ---- Task group ------------------------------------------------------- */ 1165 1166 /* Fan-in join + fan-out cancel for a dynamic set of tasks. Caller 1167 * provides storage for each per-attachment record (xco_group_attach_t), so 1168 * the group itself does no allocation. 1169 * 1170 * xco_task_group_attach(g, t, slot): 1171 * xco_countdown_add(g->pending, 1); slot's bridge waiter parks on 1172 * xco_task_done_event(t); slot is appended to g's list. When the task's 1173 * done fires, the bridge fires: it splices the slot out of g's list 1174 * and calls xco_countdown_done(&g->pending). Re-attaching a finished 1175 * task is UB. 1176 * 1177 * xco_task_group_cancel(g): walks the attachment list and xco_cancel_set's 1178 * each &slot->task->cancel, then xco_cancel_set's g->cancel. Bodies that 1179 * compose xco_wait_or_cancel against xco_task_cancel(t) wind down cooperatively; 1180 * meanwhile, anything waiting on g->cancel observes the group-level 1181 * signal directly. 1182 * 1183 * xco_task_group_join_event(g): fires when every attached task has reached 1184 * xco_task_done. Compose with select / xco_wait_or_cancel like any event. */ 1185 1186 typedef struct xco_group_attach xco_group_attach_t; 1187 1188 typedef struct xco_task_group { 1189 xco_countdown_t pending; 1190 xco_cancel_t cancel; 1191 xco_group_attach_t *head, *tail; 1192 } xco_task_group_t; 1193 1194 struct xco_group_attach { 1195 xco_waiter_t bridge; /* parked on xco_task_done_event(task) */ 1196 xco_task_t *task; 1197 xco_task_group_t *group; 1198 xco_group_attach_t *next, *prev; 1199 }; 1200 1201 void xco_task_group_init (xco_task_group_t *g); 1202 void xco_task_group_attach (xco_task_group_t *g, xco_task_t *t, 1203 xco_group_attach_t *slot); 1204 void xco_task_group_cancel (xco_task_group_t *g); 1205 static inline xco_event_t *xco_task_group_join_event (xco_task_group_t *g) { 1206 return xco_countdown_event(&g->pending); 1207 } 1208 static inline xco_cancel_t *xco_task_group_cancel_handle(xco_task_group_t *g) { 1209 return &g->cancel; 1210 } 1211 1212 /* ==================================================================== 1213 * xco — stack-switching coroutines. 1214 * 1215 * Teardown: this layer does not unwind a suspended coroutine's stack. 1216 * Drive a coroutine to return (e.g. by passing a cancel sentinel it 1217 * is expected to handle) before freeing its stack memory. 1218 * ==================================================================== */ 1219 1220 /* Coroutine entry point. The argument is the value supplied to the 1221 * first xco_step on this xco. The return value is delivered to the resumer 1222 * as the final xco_step_result, with status XCO_STEP_DEAD. */ 1223 typedef uintptr_t (*xco_fn)(uintptr_t arg); 1224 1225 /* Coroutine control block. Allocate anywhere — on a stack, in a 1226 * struct, on the heap. xco_mach_t base is first so xco_coro_t * can be passed 1227 * directly to xco_step() or any generic xco_mach_t * consumer. The trailing 1228 * priv storage holds the saved register context and bookkeeping; 1229 * its contents are private to the implementation. */ 1230 typedef struct xco_coro { 1231 xco_mach_t base; 1232 _Alignas(XCO_ALIGN) unsigned char priv[XCO_SIZE]; 1233 } xco_coro_t; 1234 1235 /* Initialize *c to run fn on [stack_base, stack_base + stack_len). 1236 * stack_base must be XCO_STACK_ALIGN-aligned; the runtime picks the 1237 * starting SP based on the architecture's stack growth direction. 1238 * Status after init is XCO_STEP_INIT. */ 1239 void xco_init(xco_coro_t *c, xco_fn fn, 1240 void *stack_base, size_t stack_len); 1241 1242 /* Suspend the currently running coroutine, returning value to its 1243 * resumer. Returns the value passed by the next resume. Undefined if 1244 * called outside a coroutine. */ 1245 uintptr_t xco_suspend(uintptr_t value); 1246 1247 /* The currently running coroutine, or NULL if the caller is not in 1248 * one. The runtime maintains this for xco_suspend. */ 1249 xco_coro_t *xco_self(void); 1250 1251 /* Resuming a coroutine is just driving its xco_step: callers use 1252 * xco_step(&c->base, v) directly. Reading status without resuming is 1253 * xco_mach_status(&c->base). The xco layer adds no separate vocabulary 1254 * for these — that's the unification with hand-coded state machines. 1255 * Resuming a coroutine that is not XCO_STEP_INIT or XCO_STEP_SUSPENDED is 1256 * undefined. */ 1257 1258 /* Convenience: init then first-step in one call. */ 1259 static inline xco_step_result_t xco_spawn(xco_coro_t *c, xco_fn fn, 1260 void *stack_base, size_t stack_len, 1261 uintptr_t arg) { 1262 xco_init(c, fn, stack_base, stack_len); 1263 return xco_step(&c->base, arg); 1264 } 1265 1266 /* Cooperative yield. Enqueues self on rt's ready queue and suspends; 1267 * the next xco_rt_run pass resumes us. Useful for fairness when a coroutine 1268 * wants to give other ready work a turn between long-running steps. 1269 * Must be called from inside a coroutine driven by rt. */ 1270 static inline void xco_yield(xco_runtime_t *rt) { 1271 xco_coro_t *self = xco_self(); 1272 assert(self != NULL); 1273 xco_waker_t sw; 1274 xco_waker_init(&sw, rt, &self->base); 1275 xco_rt_enqueue(rt, &sw.base); 1276 xco_suspend(0); 1277 } 1278 1279 /* Await an event from inside a coroutine. The standard poll-suspend 1280 * dance, in one call. Returns the event's value (delivered by fire on 1281 * the slow path, by the inline poll on the fast path). Must be called 1282 * from inside a coroutine driven by rt. */ 1283 static inline uintptr_t xco_await(xco_runtime_t *rt, xco_event_t *e) { 1284 xco_coro_t *self = xco_self(); 1285 assert(self != NULL); 1286 uintptr_t v; 1287 xco_waker_t sw; 1288 xco_waker_init(&sw, rt, &self->base); 1289 if (xco_event_poll(e, &v, &sw.base)) return v; 1290 xco_suspend(0); 1291 (void)xco_event_poll(e, &v, NULL); 1292 return v; 1293 } 1294 1295 /* Await ev or be cancelled by c. Returns true if ev fired (its payload 1296 * is written to *out, which may be NULL); false if cancelled. The 1297 * internal select_event is always cleaned up before return. 1298 * 1299 * The canonical shape for cooperative work inside a task body — pair 1300 * with xco_task_cancel(self) on every blocking await. */ 1301 static inline bool xco_await_or_cancel(xco_runtime_t *rt, xco_event_t *ev, 1302 xco_cancel_t *c, uintptr_t *out) { 1303 xco_select_event_t sel; 1304 xco_select_input_t inputs[2]; 1305 xco_wait_or_cancel(&sel, inputs, ev, c); 1306 uintptr_t winner = xco_await(rt, &sel.done.base); 1307 bool ok = (winner == XCO_WAIT_OK); 1308 if (ok && out) *out = inputs[XCO_WAIT_OK].value; 1309 xco_select_event_deinit(&sel); 1310 return ok; 1311 } 1312 1313 /* ---- xco-backed task ------------------------------------------------- */ 1314 1315 /* xco specialization of xco_task_t. Bundles the user-visible task handle 1316 * with the xco that runs it; the xco_trampoline calls fn(&xt->task, arg) 1317 * and then xco_task_done with its return value, so xco_wait_or_cancel-style 1318 * teardown works without the user wiring anything. 1319 * 1320 * The xco_trampoline recovers xt from xco_self() at first entry (container_of 1321 * on the embedded co), so the first-resume uintptr_t is preserved as 1322 * fn's arg under normal xco semantics. Subsequent resumes pass values 1323 * to the coroutine in the usual way. 1324 * 1325 * Storage shape: caller allocates xco_cotask_t and a stack. The xco_task_t 1326 * inside is the handle to wait/cancel on; cancel via xco_cancel_set on 1327 * &xt->task.cancel and the body is expected to observe it (typically 1328 * by composing xco_wait_or_cancel against xco_task_cancel(&xt->task)). */ 1329 1330 typedef uintptr_t (*xco_cotask_fn)(xco_task_t *t, uintptr_t arg); 1331 1332 typedef struct xco_cotask { 1333 xco_task_t task; 1334 xco_coro_t co; 1335 xco_cotask_fn fn; 1336 } xco_cotask_t; 1337 1338 /* Initialize xt to run fn on the given stack. After this the embedded 1339 * xco is XCO_STEP_INIT; drive it with xco_step(&xt->co.base, v) or use 1340 * xco_cotask_spawn for the init-and-first-step convenience. */ 1341 void xco_cotask_init(xco_cotask_t *xt, xco_cotask_fn fn, 1342 void *stack_base, size_t stack_len); 1343 1344 /* Convenience: init then first-step in one call. arg is delivered as 1345 * fn's argument. */ 1346 static inline xco_step_result_t xco_cotask_spawn(xco_cotask_t *xt, xco_cotask_fn fn, 1347 void *stack_base, size_t stack_len, 1348 uintptr_t arg) { 1349 xco_cotask_init(xt, fn, stack_base, stack_len); 1350 return xco_step(&xt->co.base, arg); 1351 } 1352 1353 /* ==================================================================== 1354 * xco_op — generic effect/IO layer. 1355 * 1356 * Coroutines submit ops describing work to do; the runtime accumulates 1357 * them on a pending list; after xco_rt_run quiesces, the host pulls the 1358 * batch via xco_rt_take_ops, executes them however it likes (io_uring, 1359 * threads, mocks, replay), and injects completions back via 1360 * xco_op_complete. The xco library itself reads no clocks and makes no 1361 * syscalls — that property extends to IO via this layer. 1362 * 1363 * An op is just an event with a side-channel describing what to do. 1364 * Embed xco_op_t as the first member of a kind-specific payload struct; 1365 * the host pattern-matches on `kind` (an open tag space — the runtime 1366 * never inspects it). 1367 * 1368 * State machine: 1369 * PENDING op->epoch == op->rt->op_epoch, done unset (on rt's list) 1370 * IN_FLIGHT op->epoch != op->rt->op_epoch, done unset (host has taken) 1371 * RESOLVED done.set (terminal) 1372 * 1373 * The epoch counter on the runtime is bumped each xco_rt_take_ops, so 1374 * "still on the pending list" is a generation match — submit + take are 1375 * both O(1) and the runtime never walks the batch. 1376 * 1377 * Threading: single-threaded, same as the rest of xco. submit, cancel, 1378 * and complete must all be called on the runtime's thread. 1379 * ==================================================================== */ 1380 1381 typedef enum { 1382 XCO_OP_PENDING, /* not used as a fire payload, but conceptual */ 1383 XCO_OP_COMPLETED, /* host finished; result lives in embedder fields */ 1384 XCO_OP_CANCELLED, /* resolved without a real result */ 1385 } xco_op_status_t; 1386 1387 struct xco_op { 1388 xco_latch_t done; /* fires once, payload = status */ 1389 xco_op_t *next, *prev; /* intrusive on rt's pending list */ 1390 xco_runtime_t *rt; /* set on submit; stays set after take */ 1391 uint64_t epoch; /* matches rt->op_epoch iff PENDING */ 1392 uint32_t kind; /* open tag space; host pattern-matches */ 1393 bool cancel_requested; /* advisory to host post-take */ 1394 }; 1395 1396 /* True iff op is still on rt's current pending batch (i.e. PENDING). 1397 * False for IN_FLIGHT or RESOLVED. O(1). */ 1398 static inline bool xco_op_is_pending(const xco_op_t *op) { 1399 return op->rt && op->epoch == op->rt->op_epoch && !op->done.set; 1400 } 1401 1402 /* Submit op to rt's pending list. Initializes done, clears 1403 * cancel_requested, and sets op->rt = rt. The caller pre-populates op->kind 1404 * and any embedder fields (fd, buf, len, ...) before submit. After submit 1405 * the awaiter typically waits on &op->done.base — composes with select, 1406 * xco_wait_or_cancel, etc. */ 1407 void xco_op_submit (xco_runtime_t *rt, xco_op_t *op); 1408 1409 /* Request cancellation. Behavior depends on state: 1410 * PENDING splice from rt list, fire done(CANCELLED). 1411 * IN_FLIGHT set cancel_requested=true (advisory; host decides). 1412 * RESOLVED no-op. 1413 * Idempotent. */ 1414 void xco_op_cancel (xco_op_t *op); 1415 1416 /* Host's final word for an op it took via xco_rt_take_ops. Fires done 1417 * with the given status. Idempotent (the latch is). */ 1418 void xco_op_complete(xco_op_t *op, xco_op_status_t status); 1419 1420 /* Detach the runtime's pending ops list and return its head. Bumps the 1421 * runtime's op_epoch, transitioning the whole batch to IN_FLIGHT in O(1) 1422 * (no walk — each op's stored epoch now no longer matches). The host 1423 * owns the returned list — the next/prev fields are the host's to reuse 1424 * however it wants for in-flight tracking. Returns NULL if the list is 1425 * empty. 1426 * 1427 * If tail_out is non-NULL, *tail_out receives the list's tail (or NULL 1428 * for an empty list) — handy for splicing the batch onto a host-side 1429 * in-flight list in O(1) without walking. */ 1430 xco_op_t *xco_rt_take_ops(xco_runtime_t *rt, xco_op_t **tail_out); 1431 1432 #endif /* XCO_H */