commit ddf9d2429466b379f60ba9d5435fdc3d971445c2
parent b452af6ba0dc945a4cce2a32a57dcaeefaa9eeaf
Author: Ryan Sepassi <rsepassi@gmail.com>
Date: Thu, 23 Jul 2026 06:52:49 -0700
Add XCO_MT: per-thread MPSC inboxes behind a freestanding thread abstraction
xco_thread_t = Vyukov MPSC inbox + host wakeup hook + park handshake;
thread creation, blocking, and waking stay host bindings. Waiters gain
a home thread and xco_waiter_fire becomes the routing seam: off-home,
the waiter itself is posted as the message and fires on its home
thread's drain. Threads without runtimes (workers) are first-class —
their job queue is the inbox. Event mutators, including op completion,
remain owner-thread-only; the op layer is unchanged in both modes.
Gated behind XCO_MT (struct layouts change): make builds libxco.a and
mt/libxco_mt.a, and the ST test suite runs against both to prove
no-threads-attached equivalence. tests/test_mt.c covers inbox FIFO,
wake-edge dedup, fire routing and mid-flight retarget, the deferred
waker path, a pthread-worker two-leg job round-trip, and a
multi-producer stress with per-producer FIFO checks; clean under TSan
and ASan/UBSan. Design doc: XCO_MT.md.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01DmQCcgqbYfXmnGBz2GnBN5
Diffstat:
| M | Makefile | | | 26 | +++++++++++++++++++++++--- |
| M | README.md | | | 13 | +++++++++++++ |
| A | XCO_MT.md | | | 482 | +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ |
| A | tests/test_mt.c | | | 500 | +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ |
| M | xco.c | | | 159 | ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++--------- |
| M | xco.h | | | 183 | ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++--- |
6 files changed, 1336 insertions(+), 27 deletions(-)
diff --git a/Makefile b/Makefile
@@ -21,24 +21,44 @@ SRCS := xco.c $(PLATFORMDIR)/xco_platform.c
OBJS := $(SRCS:%.c=$(BUILD)/%.o)
LIB := $(BUILD)/libxco.a
+# XCO_MT changes struct layouts, so the MT build is a separate archive;
+# every object and test in the mt tree compiles with -DXCO_MT.
+MT_OBJS := $(SRCS:%.c=$(BUILD)/mt/%.o)
+MT_LIB := $(BUILD)/mt/libxco_mt.a
+
TEST_SRCS := tests/test_xco.c tests/test_event.c tests/test_op.c
TEST_BINS := $(TEST_SRCS:tests/%.c=$(BUILD)/%)
-all: $(LIB)
+# The ST suite also runs against the MT build (an MT library with no
+# threads attached must behave identically), plus the MT-only suite.
+MT_TEST_BINS := $(TEST_SRCS:tests/%.c=$(BUILD)/mt/%) $(BUILD)/mt/test_mt
+
+all: $(LIB) $(MT_LIB)
$(LIB): $(OBJS)
$(AR) rcs $@ $^
+$(MT_LIB): $(MT_OBJS)
+ $(AR) rcs $@ $^
+
+$(BUILD)/mt/%.o: %.c
+ @mkdir -p $(dir $@)
+ $(CC) -DXCO_MT $(CPPFLAGS) $(CFLAGS) -c -o $@ $<
+
$(BUILD)/%.o: %.c
@mkdir -p $(dir $@)
$(CC) $(CPPFLAGS) $(CFLAGS) -c -o $@ $<
+$(BUILD)/mt/test_%: tests/test_%.c $(MT_LIB)
+ @mkdir -p $(dir $@)
+ $(CC) -DXCO_MT -I. $(CPPFLAGS) $(CFLAGS) -o $@ $< $(MT_LIB) -lpthread
+
$(BUILD)/test_%: tests/test_%.c $(LIB)
@mkdir -p $(dir $@)
$(CC) -I. $(CPPFLAGS) $(CFLAGS) -o $@ $< $(LIB)
-test: $(TEST_BINS)
- @for t in $(TEST_BINS); do echo "==> $$t"; $$t || exit 1; done
+test: $(TEST_BINS) $(MT_TEST_BINS)
+ @for t in $(TEST_BINS) $(MT_TEST_BINS); do echo "==> $$t"; $$t || exit 1; done
clean:
rm -rf $(BUILD)
diff --git a/README.md b/README.md
@@ -2,6 +2,8 @@
A minimal C11 concurrency library. No allocation, no atomics, no
hidden threads, no clocks. Single-threaded. Caller-provided storage.
+An opt-in `XCO_MT` build extends it to N pinned threads (see below);
+everything in the previous sentences stays true of the default build.
## Layers
@@ -32,6 +34,17 @@ Bottom-up; each layer is a thin abstraction over the one below.
completions back via `xco_op_complete`. The library makes no
syscalls — that property extends to IO via this layer.
+- **xco_thread (XCO_MT, opt-in)** — freestanding thread abstraction:
+ a Vyukov MPSC inbox, a host wakeup fn pointer, and a park handshake.
+ Waiters gain a *home* thread and `xco_waiter_fire` becomes a routing
+ seam: off-home, the waiter itself is posted as the message and fires
+ on its home thread's drain. Threads without runtimes (workers) are
+ first-class — their job queue is the inbox. Thread creation,
+ blocking, and waking stay host bindings (condvar, eventfd, `WFE`);
+ xco adds only `<stdatomic.h>`. Design and contracts: `XCO_MT.md`.
+ Changes struct layouts, so a whole program must agree on the flag;
+ the build produces `build/libxco.a` and `build/mt/libxco_mt.a`.
+
## Properties
- **No allocation.** Every event, task, queue buffer, and coroutine
diff --git a/XCO_MT.md b/XCO_MT.md
@@ -0,0 +1,482 @@
+# XCO_MT — multi-threaded extension
+
+**Status: implemented** — gated behind `XCO_MT` in xco.h/xco.c, tested
+in `tests/test_mt.c` (inbox mechanics, fire routing, deferred waker,
+pthread-worker job round-trip, MPSC stress; the ST suite also runs
+against the MT build to prove no-threads-attached equivalence). The
+build produces both `libxco.a` and `libxco_mt.a`; the flag changes
+struct layouts, so a whole program must agree on it.
+
+Extends xco from one runtime to N pinned threads communicating through
+per-thread MPSC inboxes. Single-threaded fast paths are unchanged; the
+extension is gated behind `XCO_MT`.
+
+The unit of the extension is `xco_thread_t` — a freestanding thread
+abstraction, not a runtime attachment. It holds the atomic inbox, a host
+wakeup hook, and the dedup flag, and nothing else. A runtime *may* have
+one attached (that's how cross-thread waiter fires find it), but a
+thread without any runtime — a worker executing jobs — is equally
+first-class. Everything OS-specific lives behind two host-supplied
+points: the `wakeup` fn pointer (how to rouse this thread) and the
+host's own blocking call (how this thread sleeps). After thread
+creation, the pthread implementation — or eventfd, or kqueue user
+events, or WFE/SEV on bare metal — is entirely behind the abstraction.
+xco itself gains a dependency on `<stdatomic.h>` and nothing more: no
+pthread, no syscalls, no TLS beyond one `_Thread_local` pointer pair.
+
+## Model
+
+- One `xco_thread_t` per participating thread. All non-atomic
+ runtime/event/timer state is touched only by its owning thread.
+- A runtime pinned to a thread attaches that thread's `xco_thread_t`.
+ Without one, the runtime is single-threaded — today's behavior.
+- **The waiter is the message.** It already carries a `fire` fn pointer;
+ we add a value slot and a `home` pointer. No allocation, no boxing.
+- Each waiter has a *home* thread (where its `fire` must run). Each
+ event has an *owner* runtime (whose thread mutates its waitlist). The
+ two are independent — that's how cross-thread waits work.
+- A job for a worker is just a waiter whose `home` is the worker's
+ thread: posting work and routing completions are the same primitive.
+
+## The thread abstraction
+
+```c
+typedef struct xco_thread xco_thread_t;
+struct xco_thread {
+ /* Vyukov MPSC inbox. */
+ xco_waiter_t inbox_stub; /* sentinel */
+ _Atomic(xco_waiter_t *) inbox_tail; /* producers xchg */
+ xco_waiter_t *inbox_head; /* consumer-only */
+ /* Wake dedup. */
+ _Atomic bool wakeup_pending;
+ /* Host hook: rouse this thread. Called from arbitrary threads
+ * (and, being lock-free upstream, safe from signal handlers if the
+ * host's implementation is). */
+ void (*wakeup)(xco_thread_t *);
+ void *wakeup_ud;
+ /* The runtime pinned to this thread, or NULL for a bare worker.
+ * Set by xco_rt_attach_thread. */
+ xco_runtime_t *rt;
+};
+
+void xco_thread_init(xco_thread_t *t,
+ void (*wakeup)(xco_thread_t *), void *ud);
+
+/* Producer side, any thread: push w onto t's inbox and wake t on the
+ * empty -> non-empty edge. The primitive under cross-thread fires; also
+ * called directly to hand a job to a worker. */
+void xco_thread_post(xco_thread_t *t, xco_waiter_t *w);
+
+/* Consumer side, owner thread only: pop and run every inboxed waiter's
+ * fire. Installs/restores the current-thread TLS around the fires so
+ * nested fires and event mutations route correctly. */
+void xco_thread_drain(xco_thread_t *t);
+
+/* Consumer side, owner thread only: the lost-wake handshake. Clears
+ * wakeup_pending, then re-checks the inbox. True = inbox empty, safe to
+ * block on the host primitive; false = drain again first. */
+bool xco_thread_try_park(xco_thread_t *t);
+
+/* Attach to a runtime (same shape as xco_rt_attach_timers); sets both
+ * rt->thread and t->rt. */
+static inline void xco_rt_attach_thread(xco_runtime_t *rt, xco_thread_t *t);
+```
+
+A general `xco_waiter_init(w, fire)` helper (both builds) initializes a
+detached waiter and, under `XCO_MT`, clears `home`/`value` — every
+library init path routes through it, and hand-built waiters should too,
+setting `home` explicitly afterward when routing is wanted.
+
+What binds it to an actual OS thread is host code only:
+
+| Thread | `wakeup` impl | blocking call |
+|---|---|---|
+| event-loop thread | write eventfd / kevent `EVFILT_USER` / self-pipe | `epoll_wait` / `kevent` |
+| pthread worker | `pthread_cond_signal` | `pthread_cond_wait` |
+| bare-metal / ISR target | `SEV` (or nothing) | `WFE` |
+
+xco never sees any of these; it calls `t->wakeup(t)` and returns.
+
+## The fire seam
+
+`xco_waiter_fire` becomes the routing point:
+
+```c
+static inline void xco_waiter_fire(xco_waiter_t *w, uintptr_t v) {
+ w->prev = NULL; w->next = NULL;
+ if (!w->home || w->home == xco__thread_current) {
+ w->fire(w, v); /* fast path, unchanged */
+ return;
+ }
+ w->value = v;
+ xco_thread_post(w->home, w); /* Vyukov + dedup wake */
+}
+```
+
+`xco__thread_current` is a thread-local set for the duration of
+`xco_rt_run` and `xco_thread_drain` (save-and-restore, so nesting is
+fine). NULL when neither is active. One TLS read on the fast path;
+compiles to nothing when `XCO_MT` is off.
+
+Setting TLS inside the library (rather than asking the host to do it) is
+forced by the inbox drain: drain's `fire` callbacks would otherwise
+re-route into the same inbox they were just popped from. Same-thread
+fires *between* drains are rare in practice — host IO callbacks running
+on the owner thread typically call event mutators (`xco_latch_set`,
+etc.) directly rather than going through the fire path.
+
+## Required additions
+
+### Waiter
+
+```c
+struct xco_waiter {
+ xco_waiter_t *next, *prev;
+ void (*fire)(xco_waiter_t *, uintptr_t);
+ /* MT (XCO_MT only): */
+ xco_thread_t *home; /* NULL = fire-anywhere */
+ uintptr_t value; /* set by remote firer; also waker's resume value */
+};
+```
+
+`value` subsumes `xco_waker_t.resume_value` — the same "stash payload
+between fire and resume" slot, generalized.
+
+### Runtime
+
+The runtime gains a single `xco_thread_t *thread` field; everything else
+lives in the thread struct.
+
+### Waker after the change
+
+The waker keeps its `rt` and `resume_value` fields in both builds; only
+`home` is new (set from `rt->thread` at init, NULL when rt is NULL or
+has no thread attached). Fire uses `sw->rt` directly — it always runs
+on the home thread by construction, where touching rt's ready queue is
+legal — so no TLS-runtime recovery is needed, and a bare-thread drain
+(a worker with no runtime) can fire arbitrary waiters without any
+runtime existing on that thread. The base's `value` slot is inbox
+transit only; `resume_value` remains the fire→step handoff.
+
+```c
+static inline void xco_waker_init(xco_waker_t *sw, xco_runtime_t *rt, xco_mach_t *m) {
+ xco_waiter_init(&sw->base, xco__waker_fire);
+#ifdef XCO_MT
+ sw->base.home = rt ? rt->thread : NULL;
+#endif
+ sw->rt = rt;
+ sw->mach = m;
+ sw->resume_value = 0;
+}
+
+void xco__waker_fire(xco_waiter_t *w, uintptr_t v) {
+ /* Always on home thread by construction (xco_waiter_fire routed us). */
+ xco_waker_t *sw = (xco_waker_t *)w;
+ sw->resume_value = v;
+ xco_rt_enqueue(sw->rt, w);
+}
+```
+
+## Inbox mechanics — reusing `next`
+
+The waiter's list memberships (event waitlist / ready queue / inbox)
+remain disjoint in time, so `next` does triple duty. The inbox path
+casts to `_Atomic` at the access site:
+
+```c
+void xco_thread_post(xco_thread_t *t, xco_waiter_t *w) {
+ _Atomic(xco_waiter_t *) *w_next = (_Atomic(xco_waiter_t *) *)&w->next;
+ atomic_store_explicit(w_next, NULL, memory_order_relaxed);
+
+ xco_waiter_t *prev = atomic_exchange_explicit(
+ &t->inbox_tail, w, memory_order_acq_rel);
+
+ _Atomic(xco_waiter_t *) *prev_next = (_Atomic(xco_waiter_t *) *)&prev->next;
+ atomic_store_explicit(prev_next, w, memory_order_release);
+
+ /* Dedup: only the empty -> non-empty transition triggers the wake.
+ * seq_cst pairs with the fence in xco_thread_try_park. */
+ if (!atomic_exchange_explicit(&t->wakeup_pending, true, memory_order_seq_cst)) {
+ if (t->wakeup) t->wakeup(t);
+ }
+}
+```
+
+Caveat on the cast. The C standard does not guarantee `T *` and
+`_Atomic(T *)` share representation; in practice on every C11 platform
+that matters they do (same size and alignment for pointers). We take
+that bet to keep the waiter at one `next` field and the ST hot path
+untouched.
+
+Pop (consumer-only, owner thread):
+
+```c
+xco_waiter_t *xco__inbox_pop(xco_thread_t *t) {
+ xco_waiter_t *head = t->inbox_head;
+ _Atomic(xco_waiter_t *) *head_next = (_Atomic(xco_waiter_t *) *)&head->next;
+ xco_waiter_t *next = atomic_load_explicit(head_next, memory_order_acquire);
+ if (head == &t->inbox_stub) {
+ if (!next) return NULL;
+ t->inbox_head = next;
+ head = next;
+ head_next = (_Atomic(xco_waiter_t *) *)&head->next;
+ next = atomic_load_explicit(head_next, memory_order_acquire);
+ }
+ if (next) { t->inbox_head = next; return head; }
+ /* Single-element: re-link stub at tail. Standard Vyukov dance. */
+ /* ... */
+}
+```
+
+Safe to re-park inside `fire` because `next` is rewritten on each
+membership transition.
+
+## Parking protocol
+
+`wakeup_pending` is set on post (atomic exchange returns the old value;
+only the false→true edge calls `wakeup`). The consumer clears it just
+before blocking, then re-checks the inbox to close the lost-wake window
+— that pair is `xco_thread_try_park`:
+
+```c
+bool xco_thread_try_park(xco_thread_t *t) {
+ atomic_store_explicit(&t->wakeup_pending, false, memory_order_seq_cst);
+ /* Order the flag store before the inbox re-check; pairs with the
+ * producer's seq_cst flag exchange in xco_thread_post. Either the
+ * producer sees our false and wakes, or we see its push and don't
+ * park. */
+ atomic_thread_fence(memory_order_seq_cst);
+ if (xco__inbox_nonempty(t)) {
+ atomic_store_explicit(&t->wakeup_pending, true, memory_order_relaxed);
+ return false; /* drain again, don't block */
+ }
+ return true; /* safe to block on the host primitive */
+}
+```
+
+Net cost: one wake per quiescent → busy transition, regardless of how
+many producers pile in.
+
+## Workers: threads without runtimes
+
+A worker owns an `xco_thread_t` and nothing else — no runtime, no
+events. Its job queue *is* the inbox; a job is a waiter with
+`home = &worker->t` whose `fire` executes the work on the worker
+thread. Completion is the same move in reverse: fire a waiter whose
+home is the submitter's thread.
+
+```c
+/* Worker loop — the only pthread code is the host's blocking call and
+ * the condvar behind t.wakeup: */
+for (;;) {
+ xco_thread_drain(&self->t); /* runs job fires, TLS installed */
+ if (xco_thread_try_park(&self->t))
+ host_cond_wait(self); /* woken by t.wakeup's signal */
+}
+```
+
+The inbox is MPSC, so each worker has its own and the producer picks
+one (round-robin or affinity — a host policy). There is no shared
+queue and no stealing; pinned means pinned, for jobs as for coroutines.
+
+## Affinity rules
+
+- Mutators of an event (`xco_latch_set`, `xco_notify_*`,
+ `xco_semaphore_release`, `xco_queue_send_poll`, `xco_queue_close`,
+ `xco_broadcast_publish`, `xco_op_cancel`, `xco_op_complete`, `xco_timer_*`)
+ must run on the event's owner thread. Cross-thread invocation goes
+ through a *trampoline waiter* (next section).
+- A waiter parked on event E may have any `home`. On fire, it's detached
+ locally on E's owner thread, then routed home.
+- Waiters never migrate `home` while parked or inboxed.
+
+## Trampoline waiters
+
+A trampoline is a waiter whose `fire` performs the local mutation you
+wanted to do from a foreign thread. The waiter itself is the message;
+the mutation runs on the destination thread.
+
+```c
+/* Long-lived trampoline owned by a producer thread, used to publish to a
+ * broadcast that lives on rt B. */
+typedef struct {
+ xco_waiter_t base;
+ xco_broadcast_t *bcast;
+ uintptr_t value;
+} bcast_publish_trampoline_t;
+
+static void bcast_publish_fire(xco_waiter_t *w, uintptr_t _v) {
+ bcast_publish_trampoline_t *t = (bcast_publish_trampoline_t *)w;
+ xco_broadcast_publish(t->bcast, t->value); /* runs on B */
+}
+
+/* Init once: */
+bcast_publish_trampoline_t tr = {
+ .base = { .fire = bcast_publish_fire, .home = rt_B->thread },
+ .bcast = b,
+};
+
+/* Per cross-thread publish (any thread): */
+tr.value = the_value;
+xco_waiter_fire(&tr.base, 0); /* routes; B runs publish on drain */
+```
+
+Reuse rule: a trampoline can't be re-fired while still on an inbox.
+Either allocate one per pending message, or coalesce on the producer
+side (e.g., "latest-value" semantics line up naturally with broadcast).
+For one-shot operations the trampoline is typically embedded in the same
+struct as the operation it completes (see xco_op below).
+
+## Op round-trip: complete on the home thread
+
+`xco_op_complete` and `xco_op_cancel` stay owner-thread-only — no MT
+branch, no hidden routing, no extra fields on `xco_op_t`. A worker that
+finishes an op does not complete it; it posts the op back to its home
+thread, and completion runs there. The canonical shape embeds one job
+waiter in the op struct and rides it both ways — out to the worker as
+the job, back home as the completion notice:
+
+```c
+typedef struct {
+ xco_op_t base; /* done latch lives here */
+ xco_waiter_t job; /* leg 1: worker inbox; leg 2: home inbox */
+ xco_thread_t *home; /* where complete must run */
+ int fd; void *buf; size_t len; /* request */
+ ssize_t result; int err; /* reply, filled by worker */
+} read_op_t;
+
+/* Leg 2 — home thread (via drain): */
+static void read_op_completed(xco_waiter_t *w, uintptr_t _v) {
+ read_op_t *op = container_of(w, read_op_t, job);
+ xco_op_complete(&op->base, XCO_OP_COMPLETED); /* owner thread: legal */
+}
+
+/* Leg 1 — worker thread (via drain): */
+static void read_op_execute(xco_waiter_t *w, uintptr_t _v) {
+ read_op_t *op = container_of(w, read_op_t, job);
+ op->result = host_read(op->fd, op->buf, op->len, &op->err);
+ w->fire = read_op_completed;
+ w->home = op->home;
+ xco_waiter_fire(w, 0); /* routes home; completion runs there */
+}
+
+/* Submitter, home thread, after xco_rt_take_ops: */
+op->job.fire = read_op_execute;
+op->job.home = &worker->t;
+xco_waiter_fire(&op->job, 0); /* routes to the worker */
+```
+
+Both legs are ordinary `xco_waiter_fire` routing; the reuse rule holds
+because the waiter's inbox memberships are disjoint in time. Foreign-
+thread cancel is the same move: post a trampoline home and call
+`xco_op_cancel` there (mid-flight, the advisory `cancel_requested` flag
+is the only thing an executor reads off-home).
+
+## Atomics and freestanding inventory
+
+Atomic:
+- `xco_thread_t.inbox_tail`
+- `xco_thread_t.wakeup_pending`
+- `xco_waiter.next` *at the inbox post/pop access sites only*
+
+Not atomic: every other field. Event waitlists, ready queue, timer heap,
+op pending list, latch state, semaphore permits, queue buffers — all
+stay plain, governed by the owner-thread rule.
+
+Freestanding: `XCO_MT` adds `<stdatomic.h>` and one `_Thread_local`
+pointer pair. It adds no pthread usage, no allocation, no syscalls, and
+no knowledge of how any thread was created, blocks, or wakes — those
+exist only in host code, behind `wakeup` and the host's blocking call.
+
+## Host main loop
+
+`xco_rt_run` absorbs both the inbox drain and the TLS install/restore,
+so a runtime thread's loop is just:
+
+```c
+for (;;) {
+ xco_rt_run(rt, host_now()); /* drains inbox + ready + timers */
+ if (no_io_pending(rt) && xco_thread_try_park(rt->thread))
+ host_block_on_io(rt->thread); /* eventfd / kevent / self-pipe */
+}
+```
+
+Internally:
+
+```c
+void xco_rt_run(xco_runtime_t *rt, uint64_t now) {
+ xco_thread_t *prev_t = xco__current_thread();
+ xco_runtime_t *prev_rt = xco__current_rt();
+ xco__current_thread_set(rt->thread);
+ xco__current_rt_set(rt);
+
+ for (;;) {
+ if (rt->thread) xco__drain_inbox(rt->thread); /* TLS already set */
+ drain_ready_queue(rt);
+ if (rt->timers) xco_timers_advance(rt->timers, now);
+ if (quiescent(rt)) break; /* fixpoint over all three */
+ }
+
+ xco__current_rt_set(prev_rt);
+ xco__current_thread_set(prev_t);
+}
+```
+
+## Canonical example: interrupt handler
+
+A signal handler (or hardware ISR analog) resumes a coroutine waiting on
+a hardware event. Setup runs on the runtime's thread; the handler holds
+only the waiter pointer. On bare metal the same shape works with
+`wakeup` = `SEV` (or nothing) and the idle loop blocking in `WFE`.
+
+```c
+typedef struct {
+ xco_waiter_t base;
+ xco_coro_t *co;
+} irq_waiter_t;
+
+static void irq_resume(xco_waiter_t *w, uintptr_t v) {
+ irq_waiter_t *iw = (irq_waiter_t *)w;
+ xco_step(&iw->co->base, v); /* on home thread */
+}
+
+/* On rt's thread: */
+irq_waiter_t iw = {
+ .base = { .fire = irq_resume, .home = rt->thread },
+ .co = xco_self(),
+};
+atomic_store(&g_irq_pending, &iw.base); /* publish to ISR */
+xco_suspend(0);
+
+/* In the signal handler — async-signal-safe: */
+void on_sigio(int sig, siginfo_t *si, void *ctx) {
+ xco_waiter_t *w = atomic_exchange(&g_irq_pending, NULL);
+ if (w) xco_waiter_fire(w, (uintptr_t)si->si_value.sival_int);
+}
+```
+
+The handler never allocates, never locks, never touches event internals.
+Net cost: one atomic exchange (inbox tail), one release store
+(prev->next), and — only on the empty-inbox edge — one wakeup call. The
+coroutine resumes on its own thread with the ISR's value.
+
+## What stays out of scope
+
+- Thread creation, joining, blocking, waking. All host bindings; xco
+ defines only the mailbox and the handshake.
+- Work stealing / coroutine or job migration. Pinned means pinned.
+- Bounded inbox / backpressure. The natural bound is parked-op count;
+ unbounded growth is a host-design problem, not an xco one.
+- Lock-free events. Every primitive remains owner-thread-only;
+ cross-thread use goes through trampoline waiters. This keeps
+ primitives tiny and the ST build identical to today.
+
+## Build
+
+`XCO_MT` toggles the new waiter fields, the `xco_thread_t` API, and the
+TLS install/restore and inbox drain in `xco_rt_run`/`xco_thread_drain`.
+The op layer is untouched in both modes — completion is always an
+owner-thread call. Without `XCO_MT`, xco compiles to exactly today's
+library — no atomics, no TLS, no extra bytes per waiter, no `home` field
+on the base, and `xco_rt_run` reduces to just the ready-queue + timers
+fixpoint it is today.
diff --git a/tests/test_mt.c b/tests/test_mt.c
@@ -0,0 +1,500 @@
+/*
+ * test_mt.c — exercises the XCO_MT extension (compile with -DXCO_MT).
+ *
+ * Layered like the extension itself:
+ * - Inbox mechanics on one OS thread: post/drain FIFO, wake-edge
+ * dedup, try_park handshake. xco_thread_t is freestanding, so two
+ * logical threads on one OS thread exercise all routing paths
+ * deterministically.
+ * - The fire seam: home routing, mid-flight retarget re-route, the
+ * deferred waker path (event set outside xco_rt_run routes the
+ * waker through the runtime's own inbox).
+ * - Real pthreads: the two-leg job round-trip from XCO_MT.md's "Op
+ * round-trip" section — coroutines await ops, a worker executes
+ * leg 1 on its thread, completion runs on the runtime's thread —
+ * and a multi-producer MPSC stress with per-producer FIFO checks.
+ *
+ * Host bindings here are pthread mutex/condvar pairs; xco itself never
+ * sees them — only the wakeup fn pointer and the try_park handshake.
+ */
+
+#include "xco.h"
+
+#include <assert.h>
+#include <pthread.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+
+#define CONTAINER_OF(ptr, type, member) \
+ ((type *)((char *)(ptr) - offsetof(type, member)))
+
+#define STACK_BYTES (64 * 1024)
+
+/* ---- Host binding: a thread that parks on a condvar ---------------- */
+
+typedef struct {
+ xco_thread_t t; /* first member: wake hook casts back */
+ pthread_mutex_t mu;
+ pthread_cond_t cv;
+ int wakes; /* wake-hook invocations (hook-side) */
+} host_thread_t;
+
+static void host_wake(xco_thread_t *t) {
+ host_thread_t *h = (host_thread_t *)t;
+ pthread_mutex_lock(&h->mu);
+ h->wakes++;
+ pthread_cond_signal(&h->cv);
+ pthread_mutex_unlock(&h->mu);
+}
+
+static void host_thread_init(host_thread_t *h) {
+ xco_thread_init(&h->t, host_wake, NULL);
+ pthread_mutex_init(&h->mu, NULL);
+ pthread_cond_init(&h->cv, NULL);
+ h->wakes = 0;
+}
+
+/* Park until the inbox has work. The mutex serializes the wake hook
+ * against the park handshake, closing the signal-before-wait window. */
+static void host_block(host_thread_t *h) {
+ pthread_mutex_lock(&h->mu);
+ while (xco_thread_try_park(&h->t))
+ pthread_cond_wait(&h->cv, &h->mu);
+ pthread_mutex_unlock(&h->mu);
+}
+
+/* ---- Recording waiter ---------------------------------------------- */
+
+typedef struct {
+ xco_waiter_t w;
+ int fired;
+ uintptr_t value;
+ xco_thread_t *seen_thread; /* xco__thread_current at fire time */
+ int order; /* global fire sequence, from *order_ctr */
+ int *order_ctr;
+} rec_waiter_t;
+
+static void rec_fire(xco_waiter_t *w, uintptr_t value) {
+ rec_waiter_t *r = CONTAINER_OF(w, rec_waiter_t, w);
+ r->fired++;
+ r->value = value;
+ r->seen_thread = xco__thread_current;
+ if (r->order_ctr) r->order = (*r->order_ctr)++;
+}
+
+static void rec_init(rec_waiter_t *r, xco_thread_t *home, int *order_ctr) {
+ xco_waiter_init(&r->w, rec_fire);
+ r->w.home = home;
+ r->fired = 0;
+ r->value = 0;
+ r->seen_thread = NULL;
+ r->order = -1;
+ r->order_ctr = order_ctr;
+}
+
+/* ---- Post/drain FIFO, single OS thread ------------------------------ */
+
+static void test_post_drain_fifo(void) {
+ host_thread_t h;
+ host_thread_init(&h);
+
+ int order = 0;
+ rec_waiter_t r[5];
+ for (int i = 0; i < 5; i++) {
+ rec_init(&r[i], &h.t, &order);
+ r[i].w.value = (uintptr_t)(100 + i); /* as if stashed by a router */
+ xco_thread_post(&h.t, &r[i].w);
+ }
+ for (int i = 0; i < 5; i++) assert(!r[i].fired);
+
+ xco_thread_drain(&h.t);
+ for (int i = 0; i < 5; i++) {
+ assert(r[i].fired == 1);
+ assert(r[i].value == (uintptr_t)(100 + i));
+ assert(r[i].order == i); /* FIFO */
+ assert(r[i].seen_thread == &h.t); /* TLS installed by drain */
+ }
+ assert(xco__thread_current == NULL); /* restored */
+ assert(xco_thread_try_park(&h.t)); /* drained: parkable */
+}
+
+/* ---- Wake-edge dedup + try_park handshake --------------------------- */
+
+static void test_wake_dedup(void) {
+ host_thread_t h;
+ host_thread_init(&h);
+
+ rec_waiter_t r[3];
+ for (int i = 0; i < 3; i++) rec_init(&r[i], &h.t, NULL);
+
+ /* Only the empty -> non-empty edge wakes. */
+ xco_thread_post(&h.t, &r[0].w);
+ xco_thread_post(&h.t, &r[1].w);
+ assert(h.wakes == 1);
+
+ /* Non-empty: try_park refuses and re-arms the pending flag, so the
+ * still-queued items don't re-wake either. */
+ assert(!xco_thread_try_park(&h.t));
+ xco_thread_post(&h.t, &r[2].w);
+ assert(h.wakes == 1);
+
+ xco_thread_drain(&h.t);
+ assert(r[0].fired && r[1].fired && r[2].fired);
+
+ /* Parked again: the next post is a fresh edge. */
+ assert(xco_thread_try_park(&h.t));
+ rec_init(&r[0], &h.t, NULL);
+ xco_thread_post(&h.t, &r[0].w);
+ assert(h.wakes == 2);
+ xco_thread_drain(&h.t);
+ assert(r[0].fired);
+}
+
+/* ---- Fire seam: routing + mid-flight retarget ----------------------- */
+
+static void test_fire_routing(void) {
+ host_thread_t a, b;
+ host_thread_init(&a);
+ host_thread_init(&b);
+
+ /* home == NULL: fires inline on the calling thread. */
+ rec_waiter_t inl;
+ rec_init(&inl, NULL, NULL);
+ xco_waiter_fire(&inl.w, 7);
+ assert(inl.fired == 1 && inl.value == 7 && inl.seen_thread == NULL);
+
+ /* home == a, fired from outside any drain: becomes the message. */
+ rec_waiter_t ra;
+ rec_init(&ra, &a.t, NULL);
+ xco_waiter_fire(&ra.w, 42);
+ assert(!ra.fired && a.wakes == 1);
+ xco_thread_drain(&a.t);
+ assert(ra.fired == 1 && ra.value == 42 && ra.seen_thread == &a.t);
+
+ /* Mid-flight retarget: posted to a, but home says b — a's drain
+ * re-routes instead of firing. */
+ rec_waiter_t rb;
+ rec_init(&rb, &b.t, NULL);
+ xco_thread_post(&a.t, &rb.w);
+ rb.w.value = 9; /* payload rides the inbox */
+ xco_thread_drain(&a.t);
+ assert(!rb.fired && b.wakes == 1);
+ xco_thread_drain(&b.t);
+ assert(rb.fired == 1 && rb.value == 9 && rb.seen_thread == &b.t);
+
+ /* Same-home fire during drain runs inline (no self re-post). */
+ assert(xco_thread_try_park(&a.t));
+}
+
+/* ---- Deferred waker: event set outside xco_rt_run ------------------- */
+
+typedef struct {
+ xco_runtime_t *rt;
+ xco_latch_t *latch;
+} await_ctx_t;
+
+static uintptr_t await_latch_fn(uintptr_t arg) {
+ await_ctx_t *c = (await_ctx_t *)arg;
+ return xco_await(c->rt, &c->latch->base);
+}
+
+static void test_deferred_waker(void) {
+ host_thread_t h;
+ host_thread_init(&h);
+ xco_runtime_t rt;
+ xco_rt_init(&rt);
+ xco_rt_attach_thread(&rt, &h.t);
+ assert(h.t.rt == &rt);
+
+ xco_latch_t latch;
+ xco_latch_init(&latch);
+
+ static unsigned char stack[STACK_BYTES] __attribute__((aligned(XCO_STACK_ALIGN)));
+ xco_coro_t co;
+ await_ctx_t ctx = { .rt = &rt, .latch = &latch };
+ xco_spawn(&co, await_latch_fn, stack, sizeof stack, (uintptr_t)&ctx);
+ assert(xco_mach_status(&co.base) == XCO_STEP_SUSPENDED);
+
+ /* Set the latch on the owner thread but outside xco_rt_run: TLS is
+ * NULL, the waker's home is h.t, so it rides the runtime's own
+ * inbox rather than enqueueing inline. */
+ xco_latch_set(&latch, 1234);
+ assert(xco_mach_status(&co.base) == XCO_STEP_SUSPENDED);
+ assert(h.wakes == 1);
+
+ xco_rt_run(&rt, 0); /* drains inbox -> enqueue -> step */
+ assert(xco_mach_status(&co.base) == XCO_STEP_DEAD);
+
+ /* Inside xco_rt_run the same fire is inline: re-run with a second
+ * co whose latch is set by the first co's body — covered implicitly
+ * by the op round-trip test below; here just confirm quiescence. */
+ assert(xco_thread_try_park(&h.t));
+}
+
+/* ---- Two-leg job round-trip over real pthreads ---------------------- */
+
+/* Worker: a bare xco_thread_t (no runtime) on its own pthread. Jobs are
+ * waiters with home = the worker; a poison job flips `stop`. */
+typedef struct {
+ host_thread_t ht;
+ pthread_t pt;
+ int stop; /* set by the poison job, on this thread */
+ pthread_t tid; /* filled at thread start */
+} worker_t;
+
+static void *worker_main(void *arg) {
+ worker_t *wk = (worker_t *)arg;
+ wk->tid = pthread_self();
+ for (;;) {
+ xco_thread_drain(&wk->ht.t);
+ if (wk->stop) break;
+ host_block(&wk->ht);
+ }
+ return NULL;
+}
+
+typedef struct {
+ xco_waiter_t w;
+ worker_t *wk;
+} poison_t;
+
+static void poison_fire(xco_waiter_t *w, uintptr_t v) {
+ (void)v;
+ CONTAINER_OF(w, poison_t, w)->wk->stop = 1;
+}
+
+enum { OP_DOUBLE = 1 };
+
+typedef struct {
+ xco_op_t base;
+ xco_waiter_t job; /* leg 1: worker inbox; leg 2: home inbox */
+ xco_thread_t *home; /* where complete must run */
+ uintptr_t input;
+ uintptr_t result; /* filled by the worker */
+ pthread_t exec_tid; /* thread leg 1 ran on */
+ pthread_t complete_tid; /* thread leg 2 ran on */
+} double_op_t;
+
+/* Leg 2 — home thread (via drain): completion is an owner-thread call. */
+static void double_op_completed(xco_waiter_t *w, uintptr_t v) {
+ (void)v;
+ double_op_t *op = CONTAINER_OF(w, double_op_t, job);
+ op->complete_tid = pthread_self();
+ xco_op_complete(&op->base, XCO_OP_COMPLETED);
+}
+
+/* Leg 1 — worker thread (via drain): execute, then ride the same waiter
+ * home. */
+static void double_op_execute(xco_waiter_t *w, uintptr_t v) {
+ (void)v;
+ double_op_t *op = CONTAINER_OF(w, double_op_t, job);
+ op->result = op->input * 2;
+ op->exec_tid = pthread_self();
+ w->fire = double_op_completed;
+ w->home = op->home;
+ xco_waiter_fire(w, 0); /* routes home; completion runs there */
+}
+
+enum { N_COS = 8, OPS_PER_CO = 4, N_WORKERS = 2 };
+
+typedef struct {
+ xco_runtime_t *rt;
+ xco_thread_t *home;
+ double_op_t ops[OPS_PER_CO];
+ uintptr_t seed;
+} round_trip_ctx_t;
+
+static uintptr_t round_trip_fn(xco_task_t *task, uintptr_t arg) {
+ (void)task;
+ round_trip_ctx_t *c = (round_trip_ctx_t *)arg;
+ uintptr_t sum = 0;
+ for (int i = 0; i < OPS_PER_CO; i++) {
+ double_op_t *op = &c->ops[i];
+ op->base.kind = OP_DOUBLE;
+ op->home = c->home;
+ op->input = c->seed + (uintptr_t)i;
+ op->result = 0;
+ xco_op_submit(c->rt, &op->base);
+ uintptr_t st = xco_await(c->rt, &op->base.done.base);
+ assert(st == XCO_OP_COMPLETED);
+ assert(op->result == op->input * 2);
+ sum += op->result;
+ }
+ return sum;
+}
+
+static void test_job_round_trip(void) {
+ host_thread_t hm; /* the "main"/runtime thread */
+ host_thread_init(&hm);
+ xco_runtime_t rt;
+ xco_rt_init(&rt);
+ xco_rt_attach_thread(&rt, &hm.t);
+
+ worker_t workers[N_WORKERS];
+ for (int i = 0; i < N_WORKERS; i++) {
+ host_thread_init(&workers[i].ht);
+ workers[i].stop = 0;
+ int rc = pthread_create(&workers[i].pt, NULL, worker_main, &workers[i]);
+ assert(rc == 0);
+ }
+
+ static round_trip_ctx_t ctxs[N_COS];
+ static xco_cotask_t tasks[N_COS];
+ static unsigned char stacks[N_COS][STACK_BYTES]
+ __attribute__((aligned(XCO_STACK_ALIGN)));
+ uintptr_t expect = 0;
+ for (int i = 0; i < N_COS; i++) {
+ ctxs[i].rt = &rt;
+ ctxs[i].home = &hm.t;
+ ctxs[i].seed = (uintptr_t)(i * 100 + 1);
+ for (int k = 0; k < OPS_PER_CO; k++)
+ expect += (ctxs[i].seed + (uintptr_t)k) * 2;
+ xco_cotask_spawn(&tasks[i], round_trip_fn,
+ stacks[i], STACK_BYTES, (uintptr_t)&ctxs[i]);
+ }
+
+ /* Host main loop: run to quiescence, dispatch the batch round-robin,
+ * park until completions ride home. */
+ int rr = 0;
+ for (;;) {
+ xco_rt_run(&rt, 0);
+
+ int finished = 0;
+ for (int i = 0; i < N_COS; i++)
+ if (xco_task_finished(&tasks[i].task)) finished++;
+ if (finished == N_COS) break;
+
+ xco_op_t *batch = xco_rt_take_ops(&rt, NULL);
+ while (batch) {
+ xco_op_t *next = batch->next; /* posting clobbers links */
+ double_op_t *op = (double_op_t *)batch;
+ assert(op->base.kind == OP_DOUBLE);
+ op->job.fire = double_op_execute;
+ op->job.home = &workers[rr].ht.t;
+ rr = (rr + 1) % N_WORKERS;
+ xco_waiter_fire(&op->job, 0); /* routes to the worker */
+ batch = next;
+ }
+
+ host_block(&hm);
+ }
+
+ uintptr_t total = 0;
+ for (int i = 0; i < N_COS; i++) total += tasks[i].task.done.value;
+ assert(total == expect);
+
+ /* Every leg ran on the right thread. */
+ pthread_t self = pthread_self();
+ for (int i = 0; i < N_COS; i++) {
+ for (int k = 0; k < OPS_PER_CO; k++) {
+ double_op_t *op = &ctxs[i].ops[k];
+ assert(pthread_equal(op->complete_tid, self));
+ int on_worker = 0;
+ for (int wkr = 0; wkr < N_WORKERS; wkr++)
+ if (pthread_equal(op->exec_tid, workers[wkr].tid)) on_worker = 1;
+ assert(on_worker);
+ }
+ }
+
+ /* Shut the workers down via poison jobs. */
+ poison_t poison[N_WORKERS];
+ for (int i = 0; i < N_WORKERS; i++) {
+ xco_waiter_init(&poison[i].w, poison_fire);
+ poison[i].wk = &workers[i];
+ xco_thread_post(&workers[i].ht.t, &poison[i].w);
+ pthread_join(workers[i].pt, NULL);
+ }
+}
+
+/* ---- MPSC stress: N producers, per-producer FIFO -------------------- */
+
+enum { N_PRODUCERS = 4, ITEMS_PER_PRODUCER = 10000 };
+
+typedef struct {
+ xco_waiter_t w;
+ int producer;
+ int seq;
+} stress_item_t;
+
+static struct {
+ host_thread_t consumer;
+ int received;
+ int next_seq[N_PRODUCERS];
+} g_stress;
+
+static void stress_fire(xco_waiter_t *w, uintptr_t v) {
+ (void)v;
+ stress_item_t *it = CONTAINER_OF(w, stress_item_t, w);
+ assert(xco__thread_current == &g_stress.consumer.t);
+ assert(g_stress.next_seq[it->producer] == it->seq); /* per-producer FIFO */
+ g_stress.next_seq[it->producer]++;
+ g_stress.received++;
+}
+
+typedef struct {
+ stress_item_t *items;
+ int producer;
+ pthread_t pt;
+} producer_t;
+
+static void *producer_main(void *arg) {
+ producer_t *p = (producer_t *)arg;
+ for (int i = 0; i < ITEMS_PER_PRODUCER; i++) {
+ stress_item_t *it = &p->items[i];
+ xco_waiter_init(&it->w, stress_fire);
+ it->w.home = &g_stress.consumer.t;
+ it->producer = p->producer;
+ it->seq = i;
+ xco_waiter_fire(&it->w, 0); /* off-home: becomes the message */
+ }
+ return NULL;
+}
+
+static void test_mpsc_stress(void) {
+ host_thread_init(&g_stress.consumer);
+ g_stress.received = 0;
+ memset(g_stress.next_seq, 0, sizeof g_stress.next_seq);
+
+ producer_t producers[N_PRODUCERS];
+ for (int i = 0; i < N_PRODUCERS; i++) {
+ producers[i].items = malloc(sizeof(stress_item_t) * ITEMS_PER_PRODUCER);
+ assert(producers[i].items);
+ producers[i].producer = i;
+ int rc = pthread_create(&producers[i].pt, NULL, producer_main, &producers[i]);
+ assert(rc == 0);
+ }
+
+ const int total = N_PRODUCERS * ITEMS_PER_PRODUCER;
+ while (g_stress.received < total) {
+ xco_thread_drain(&g_stress.consumer.t);
+ if (g_stress.received < total) host_block(&g_stress.consumer);
+ }
+ assert(g_stress.received == total);
+ for (int i = 0; i < N_PRODUCERS; i++)
+ assert(g_stress.next_seq[i] == ITEMS_PER_PRODUCER);
+
+ for (int i = 0; i < N_PRODUCERS; i++) {
+ pthread_join(producers[i].pt, NULL);
+ free(producers[i].items);
+ }
+}
+
+/* -------------------------------------------------------------------- */
+
+int main(void) {
+ test_post_drain_fifo();
+ printf("test_post_drain_fifo OK\n");
+ test_wake_dedup();
+ printf("test_wake_dedup OK\n");
+ test_fire_routing();
+ printf("test_fire_routing OK\n");
+ test_deferred_waker();
+ printf("test_deferred_waker OK\n");
+ test_job_round_trip();
+ printf("test_job_round_trip OK\n");
+ test_mpsc_stress();
+ printf("test_mpsc_stress OK\n");
+ printf("test_mt: all OK\n");
+ return 0;
+}
diff --git a/xco.c b/xco.c
@@ -33,6 +33,117 @@
#include <stdint.h>
/* ====================================================================
+ * Thread (XCO_MT)
+ *
+ * The Vyukov MPSC inbox rides the waiter's `next` field, cast to
+ * _Atomic at the access sites only. The C standard does not guarantee
+ * T * and _Atomic(T *) share representation; on every C11 platform
+ * that matters they do (same size and alignment for pointers). We take
+ * that bet to keep the waiter at one next field and the ST hot path
+ * untouched.
+ * ==================================================================== */
+
+#ifdef XCO_MT
+
+_Thread_local xco_thread_t *xco__thread_current = NULL;
+
+#define XCO__ANEXT(w) ((_Atomic(xco_waiter_t *) *)&(w)->next)
+
+void xco_thread_post(xco_thread_t *t, xco_waiter_t *w) {
+ atomic_store_explicit(XCO__ANEXT(w), NULL, memory_order_relaxed);
+ xco_waiter_t *prev =
+ atomic_exchange_explicit(&t->inbox_tail, w, memory_order_acq_rel);
+ atomic_store_explicit(XCO__ANEXT(prev), w, memory_order_release);
+
+ /* Dedup: only the empty -> non-empty edge invokes the wake hook.
+ * seq_cst pairs with the fence in xco_thread_try_park: either this
+ * exchange observes the parking consumer's false (we wake), or the
+ * consumer's re-check observes our push (it doesn't park). */
+ if (!atomic_exchange_explicit(&t->wakeup_pending, true,
+ memory_order_seq_cst)) {
+ if (t->wakeup) t->wakeup(t);
+ }
+}
+
+/* Consumer-only. Standard Vyukov pop, including the two transient
+ * NULL-with-items states: a producer mid-push (tail moved, next link
+ * not yet stored) and the single-element stub re-push dance. Both
+ * resolve on a later call; xco__inbox_nonempty still reports items so
+ * spin sites (run fixpoint, try_park) don't sleep through them. */
+static xco_waiter_t *xco__inbox_pop(xco_thread_t *t) {
+ xco_waiter_t *head = t->inbox_head;
+ xco_waiter_t *next = atomic_load_explicit(XCO__ANEXT(head), memory_order_acquire);
+
+ if (head == &t->inbox_stub) {
+ if (!next) return NULL; /* empty */
+ t->inbox_head = next; /* skip the stub */
+ head = next;
+ next = atomic_load_explicit(XCO__ANEXT(head), memory_order_acquire);
+ }
+ if (next) {
+ t->inbox_head = next;
+ return head;
+ }
+
+ /* head is the last visible node. If a producer is mid-push behind
+ * it, report empty for now. */
+ xco_waiter_t *tail = atomic_load_explicit(&t->inbox_tail, memory_order_acquire);
+ if (head != tail) return NULL; /* producer mid-push */
+
+ /* Single element: re-push the stub so head becomes poppable. */
+ atomic_store_explicit(XCO__ANEXT(&t->inbox_stub), NULL, memory_order_relaxed);
+ xco_waiter_t *prev =
+ atomic_exchange_explicit(&t->inbox_tail, &t->inbox_stub, memory_order_acq_rel);
+ atomic_store_explicit(XCO__ANEXT(prev), &t->inbox_stub, memory_order_release);
+
+ next = atomic_load_explicit(XCO__ANEXT(head), memory_order_acquire);
+ if (next) {
+ t->inbox_head = next;
+ return head;
+ }
+ return NULL; /* racing producer; resolves later */
+}
+
+/* True if the inbox may hold items (including a producer mid-push).
+ * Fully-drained resting state: cursor at the stub and the producer end
+ * back at the stub. */
+static bool xco__inbox_nonempty(xco_thread_t *t) {
+ if (t->inbox_head != &t->inbox_stub) return true;
+ return atomic_load_explicit(&t->inbox_tail, memory_order_acquire)
+ != &t->inbox_stub;
+}
+
+/* Fire everything currently poppable. Callers own the TLS install.
+ * xco_waiter_fire (not a direct fire call) so a waiter whose home was
+ * retargeted mid-flight re-routes instead of running here. */
+static void xco__inbox_fire_all(xco_thread_t *t) {
+ for (xco_waiter_t *w; (w = xco__inbox_pop(t)) != NULL;) {
+ xco_waiter_fire(w, w->value);
+ }
+}
+
+void xco_thread_drain(xco_thread_t *t) {
+ xco_thread_t *prev = xco__thread_current;
+ xco__thread_current = t;
+ xco__inbox_fire_all(t);
+ xco__thread_current = prev;
+}
+
+bool xco_thread_try_park(xco_thread_t *t) {
+ atomic_store_explicit(&t->wakeup_pending, false, memory_order_seq_cst);
+ /* Order the flag store before the inbox re-check; pairs with the
+ * producer's seq_cst exchange in xco_thread_post (see there). */
+ atomic_thread_fence(memory_order_seq_cst);
+ if (xco__inbox_nonempty(t)) {
+ atomic_store_explicit(&t->wakeup_pending, true, memory_order_relaxed);
+ return false;
+ }
+ return true;
+}
+
+#endif /* XCO_MT */
+
+/* ====================================================================
* Runtime
* ==================================================================== */
@@ -64,15 +175,39 @@ void xco_rt_run(xco_runtime_t *rt, uint64_t now) {
* a fresh already-expired timer; the outer loop catches it on the
* next pass. Termination: each pass either drains a non-empty
* queue or exits, and advance only fires timers it then removes,
- * so total work is bounded. */
+ * so total work is bounded.
+ *
+ * XCO_MT: an attached thread's inbox joins the fixpoint, drained at
+ * the top of each pass with the current-thread TLS installed so
+ * inboxed fires (and their nested fires) route as same-thread. A
+ * remote producer can extend the run; the host's park handshake
+ * (xco_thread_try_park) covers the quiescent -> post race after we
+ * return. */
+#ifdef XCO_MT
+ xco_thread_t *prev_t = xco__thread_current;
+ if (rt->thread) xco__thread_current = rt->thread;
+#endif
for (;;) {
+#ifdef XCO_MT
+ if (rt->thread) xco__inbox_fire_all(rt->thread);
+#endif
if (rt->timers) xco_timers_advance(rt->timers, now);
- if (!rt->head) return;
+ if (!rt->head) {
+#ifdef XCO_MT
+ /* Items may be transiently unpoppable (producer mid-push);
+ * spin the fixpoint rather than report quiescent. */
+ if (rt->thread && xco__inbox_nonempty(rt->thread)) continue;
+#endif
+ break;
+ }
for (xco_waiter_t *w; (w = xco_rt_dequeue(rt));) {
xco_waker_t *sw = (xco_waker_t *)w;
xco_step(sw->mach, sw->resume_value);
}
}
+#ifdef XCO_MT
+ xco__thread_current = prev_t;
+#endif
}
/* ---- Waker ------------------------------------------------------------ */
@@ -268,9 +403,7 @@ void xco_select_event_init(xco_select_event_t *s,
}
for (size_t i = 0; i < n; i++) {
- inputs[i].w.next = NULL;
- inputs[i].w.prev = NULL;
- inputs[i].w.fire = xco_select_input_fire;
+ xco_waiter_init(&inputs[i].w, xco_select_input_fire);
inputs[i].src = srcs[i];
inputs[i].parent = s;
inputs[i].value = 0;
@@ -293,9 +426,7 @@ void xco_allof_event_init(xco_select_event_t *s,
* rest end up parked by the same call. If everyone was inline, fire
* done at the end. */
for (size_t i = 0; i < n; i++) {
- inputs[i].w.next = NULL;
- inputs[i].w.prev = NULL;
- inputs[i].w.fire = xco_select_input_fire;
+ xco_waiter_init(&inputs[i].w, xco_select_input_fire);
inputs[i].src = srcs[i];
inputs[i].parent = s;
inputs[i].value = 0;
@@ -818,9 +949,7 @@ static void xco__timeout_bridge_fire(xco_waiter_t *w, uintptr_t value) {
void xco_timeout_init(xco_timeout_t *to, xco_timers_t *ts, uint64_t deadline) {
xco_timer_init(&to->timer, ts, deadline);
xco_cancel_init(&to->cancel);
- to->bridge.next = NULL;
- to->bridge.prev = NULL;
- to->bridge.fire = xco__timeout_bridge_fire;
+ xco_waiter_init(&to->bridge, xco__timeout_bridge_fire);
/* Park the bridge on the timer. If the timer fires, xco_latch_set
* detaches the bridge and calls our fire callback inline. The timer
* was just inserted into the heap and hasn't been advanced, so poll
@@ -908,9 +1037,7 @@ void xco_ticker_init(xco_ticker_t *t, xco_timers_t *ts,
xco_timer_init(&t->timer, ts, first_deadline);
- t->bridge.next = NULL;
- t->bridge.prev = NULL;
- t->bridge.fire = xco__ticker_bridge_fire;
+ xco_waiter_init(&t->bridge, xco__ticker_bridge_fire);
(void)xco_event_poll(xco_timer_event(&t->timer), NULL, &t->bridge);
}
@@ -982,9 +1109,7 @@ void xco_task_group_attach(xco_task_group_t *g, xco_task_t *t, xco_group_attach_
else g->head = slot;
g->tail = slot;
- slot->bridge.next = NULL;
- slot->bridge.prev = NULL;
- slot->bridge.fire = xco__task_group_bridge_fire;
+ xco_waiter_init(&slot->bridge, xco__task_group_bridge_fire);
/* Park on the task's done event. Re-attaching a finished task is UB
* per the contract; the latch's clean-waiter assert inside poll
diff --git a/xco.h b/xco.h
@@ -56,6 +56,19 @@
* 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).
+ *
+ * XCO_MT (compile-time opt-in, see XCO_MT.md). Extends xco to N pinned
+ * threads communicating through per-thread MPSC inboxes. The unit is
+ * xco_thread_t — a freestanding thread abstraction (atomic inbox + host
+ * wakeup hook + park handshake); thread creation, blocking, and waking
+ * are host bindings, never xco's. The waiter gains a `home` thread and
+ * xco_waiter_fire becomes the routing seam: fire on the home thread (or
+ * home == NULL) runs inline, anywhere else the waiter itself is posted
+ * to home's inbox as the message. Event mutators remain owner-thread-
+ * only; all struct layouts change, so the whole program must agree on
+ * the flag. Without XCO_MT this library compiles to exactly the
+ * single-threaded description above — no atomics, no TLS, no extra
+ * waiter fields.
*/
#ifndef XCO_H
@@ -66,6 +79,10 @@
#include <stddef.h>
#include <stdint.h>
+#ifdef XCO_MT
+#include <stdatomic.h>
+#endif
+
/* 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)). */
@@ -163,12 +180,21 @@ static inline xco_mach_status_t xco_mach_status(const xco_mach_t *s) {
/* ---- Waiter ------------------------------------------------------------ */
+/* Forward-declared here so the waiter and its fire seam can reference
+ * it; the full definition (and the rest of the MT surface) follows the
+ * runtime section below. */
+typedef struct xco_thread xco_thread_t;
+
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. */
+ * undefined in that state and reset on the next park. Under XCO_MT,
+ * next is additionally the intrusive inbox link (accessed through
+ * _Atomic casts at the inbox push/pop sites only); the three list
+ * memberships — waitlist, ready queue, inbox — are disjoint in
+ * time. */
xco_waiter_t *next;
xco_waiter_t *prev;
/* Fire callback. value is the event's payload at fire time — sticky
@@ -180,16 +206,60 @@ struct xco_waiter {
* 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);
+#ifdef XCO_MT
+ /* The thread fire must run on; NULL = fire-anywhere (inline on the
+ * calling thread). Never migrates while parked or inboxed. */
+ xco_thread_t *home;
+ /* Payload stash while the waiter rides an inbox: set by the remote
+ * firer, handed to fire by the drain. */
+ uintptr_t value;
+#endif
};
+/* Initialize a waiter: detached links, the given fire callback, and (under
+ * XCO_MT) home = NULL / value = 0. Library init paths all route through
+ * this; callers building waiters by hand should too, then set `home`
+ * explicitly if the waiter must fire on a particular thread. */
+static inline void xco_waiter_init(xco_waiter_t *w,
+ void (*fire)(xco_waiter_t *, uintptr_t)) {
+ w->next = NULL;
+ w->prev = NULL;
+ w->fire = fire;
+#ifdef XCO_MT
+ w->home = NULL;
+ w->value = 0;
+#endif
+}
+
+#ifdef XCO_MT
+/* Producer side of the inbox; defined with the rest of the thread
+ * abstraction below, declared here for the fire seam. */
+void xco_thread_post(xco_thread_t *t, xco_waiter_t *w);
+/* The thread whose xco_rt_run / xco_thread_drain is currently active on
+ * this OS thread, or NULL. Maintained by those two entry points
+ * (save-and-restore, so nesting is fine). */
+extern _Thread_local xco_thread_t *xco__thread_current;
+#endif
+
/* 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. */
+ * etc.) don't need to clear prev/next themselves.
+ *
+ * Under XCO_MT this is the routing seam: a waiter whose home is another
+ * thread is not fired here — it becomes the message, posted to home's
+ * inbox; the home thread's drain fires it with the stashed value. */
static inline void xco_waiter_fire(xco_waiter_t *w, uintptr_t value) {
w->prev = NULL;
w->next = NULL;
+#ifdef XCO_MT
+ if (w->home && w->home != xco__thread_current) {
+ w->value = value;
+ xco_thread_post(w->home, w);
+ return;
+ }
+#endif
w->fire(w, value);
}
@@ -232,6 +302,9 @@ typedef struct xco_runtime {
* submit; PENDING vs IN_FLIGHT is a generation match (see xco_op). */
xco_op_t *op_head, *op_tail;
uint64_t op_epoch;
+#ifdef XCO_MT
+ xco_thread_t *thread; /* optional; drained inside xco_rt_run */
+#endif
} xco_runtime_t;
static inline void xco_rt_init(xco_runtime_t *rt) {
@@ -239,6 +312,9 @@ static inline void xco_rt_init(xco_runtime_t *rt) {
rt->timers = NULL;
rt->op_head = rt->op_tail = NULL;
rt->op_epoch = 0;
+#ifdef XCO_MT
+ rt->thread = NULL;
+#endif
}
/* Attach (or detach with NULL) a timer source. While attached, xco_rt_run
@@ -262,7 +338,9 @@ static inline void xco_rt_enqueue(xco_runtime_t *rt, xco_waiter_t *w) {
* 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. */
+ * caller-supplied. Under XCO_MT, an attached thread's inbox joins the
+ * fixpoint (drained with the current-thread TLS installed), so
+ * quiescent means ready queue, due timers, and inbox are all empty. */
void xco_rt_run(xco_runtime_t *rt, uint64_t now);
/* The canonical bridge between events and the scheduler. When fired,
@@ -273,7 +351,11 @@ void xco_rt_run(xco_runtime_t *rt, uint64_t now);
* 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. */
+ * directly — no re-init needed unless rt or step changes.
+ *
+ * Under XCO_MT the waker's home is rt's attached thread, so a fire from
+ * any other thread routes the waker through rt's inbox and the enqueue
+ * runs on rt's own thread — the ready queue stays plain. */
typedef struct {
xco_waiter_t base;
xco_runtime_t *rt;
@@ -285,14 +367,101 @@ typedef struct {
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;
+ xco_waiter_init(&sw->base, xco__waker_fire);
+#ifdef XCO_MT
+ sw->base.home = rt ? rt->thread : NULL;
+#endif
sw->rt = rt;
sw->mach = m;
sw->resume_value = 0;
}
+#ifdef XCO_MT
+/* ---- Thread (XCO_MT) --------------------------------------------------- */
+
+/* Freestanding thread abstraction: a Vyukov MPSC inbox, a host wakeup
+ * hook, and the wake-dedup flag. Nothing else — thread creation,
+ * joining, blocking, and waking are host bindings. What ties an
+ * xco_thread_t to an OS thread is only the host's discipline: one
+ * thread owns it (calls drain/try_park), everyone else only posts.
+ *
+ * A runtime pinned to a thread attaches that thread's xco_thread_t
+ * (xco_rt_attach_thread); cross-thread waiter fires find the runtime's
+ * inbox through it. A thread without a runtime — a worker draining
+ * posted jobs — is equally first-class: its job queue IS the inbox.
+ *
+ * The consumer loop for a bare worker:
+ *
+ * for (;;) {
+ * xco_thread_drain(&self->t);
+ * lock(mu);
+ * while (xco_thread_try_park(&self->t)) cond_wait(cv, mu);
+ * unlock(mu);
+ * }
+ *
+ * with t.wakeup = lock(mu); signal(cv); unlock(mu). An event-loop
+ * thread blocks in epoll/kevent instead, with t.wakeup writing an
+ * eventfd / triggering EVFILT_USER. */
+struct xco_thread {
+ /* Vyukov MPSC inbox. Producers xchg inbox_tail; the owner thread
+ * follows inbox_head. Links ride the waiter's next field through
+ * _Atomic casts at the push/pop sites. */
+ xco_waiter_t inbox_stub; /* sentinel */
+ _Atomic(xco_waiter_t *) inbox_tail;
+ xco_waiter_t *inbox_head; /* consumer-only */
+ /* Wake dedup: only the empty -> non-empty post edge invokes wakeup. */
+ _Atomic bool wakeup_pending;
+ /* Host hook: rouse this thread. Called from arbitrary threads; must
+ * be safe to invoke concurrently and redundantly. NULL = no hook
+ * (a thread that never blocks, or polls). */
+ void (*wakeup)(xco_thread_t *t);
+ void *wakeup_ud;
+ /* The runtime pinned to this thread, or NULL for a bare worker.
+ * Set by xco_rt_attach_thread. */
+ xco_runtime_t *rt;
+};
+
+static inline void xco_thread_init(xco_thread_t *t,
+ void (*wakeup)(xco_thread_t *), void *ud) {
+ xco_waiter_init(&t->inbox_stub, NULL);
+ t->inbox_head = &t->inbox_stub;
+ atomic_store_explicit(&t->inbox_tail, &t->inbox_stub, memory_order_relaxed);
+ atomic_store_explicit(&t->wakeup_pending, false, memory_order_relaxed);
+ t->wakeup = wakeup;
+ t->wakeup_ud = ud;
+ t->rt = NULL;
+}
+
+/* Attach t to rt (same shape as xco_rt_attach_timers): xco_rt_run will
+ * drain t's inbox in its fixpoint, wakers created against rt route home
+ * to t, and xco_thread_drain on a bare t=NULL detach is a no-op. */
+static inline void xco_rt_attach_thread(xco_runtime_t *rt, xco_thread_t *t) {
+ rt->thread = t;
+ if (t) t->rt = rt;
+}
+
+/* xco_thread_post — producer side, any thread (declared above with the
+ * fire seam): push w onto t's inbox and invoke t->wakeup on the
+ * empty -> non-empty edge. w must be detached (not parked, queued, or
+ * already inboxed). This is the primitive under cross-thread fires; call
+ * it directly to hand a job waiter to a worker. Lock-free; safe from a
+ * signal handler if t->wakeup is. */
+
+/* Consumer side, owner thread only: pop and fire every inboxed waiter
+ * (installing/restoring the current-thread TLS around the fires, so
+ * nested fires and re-routes behave). Waiters whose home was retargeted
+ * mid-flight are re-routed, not fired here. */
+void xco_thread_drain(xco_thread_t *t);
+
+/* Consumer side, owner thread only: the lost-wake handshake. Clears
+ * wakeup_pending, then re-checks the inbox. True = inbox empty, safe to
+ * block on the host primitive; false = more arrived, drain again first.
+ * Call with the same mutual exclusion against t->wakeup that the host's
+ * blocking primitive requires (see the worker-loop sketch above). */
+bool xco_thread_try_park(xco_thread_t *t);
+
+#endif /* XCO_MT */
+
/* ---- Latch ------------------------------------------------------------ */
/* One-shot sticky event. set() flips the bit, stores the payload, and
