feat: async/await colorblind over the fiber Io (Phase 2 of Io unification)
`context.io.async(worker)` / `await` now run over the `Io` PROTOCOL, so the
same code interleaves under the fiber scheduler or runs inline under the
blocking `CBlockingIo` — one async stack, reached purely through `context.io`.
- Protocol: `suspend_raw(park: *ParkToken)` (was by-value). A suspending impl
records the parked execution context into `park.handle` before parking, so a
cross-context `ready(park)` knows whom to resume; `Scheduler.suspend_raw`
writes `self.current`, `CBlockingIo` ignores it.
- io.sx async layer rewritten colorblind: `async` submits the worker through
`io.spawn_raw` (inline under blocking, a fiber under the scheduler) and returns
a HEAP `*Future($R)` the worker fills later; `await` suspends via `suspend_raw`
until ready, then returns/raises. The generic worker is bridged to spawn_raw's
raw `(*void)->void` entry via a monomorphic `ThunkBox` (a heap-boxed nullary
completion closure) — all genericity lives in the closure env. Workers are
nullary (inputs captured at the call site) because a variadic pack can't cross
the fiber boundary. `CBlockingIo.spawn_raw` now runs the worker inline.
- Migrated 1805/1806 to the nullary `*Future` form; retrofit 1822/1823 to the
`push .{ … }` partial-context literal (inherits allocator/data).
- The async machinery adds a few prelude types, shifting the type-name table —
40 `.ir` snapshots regenerated (no behavior change; only `.exit`/`.stdout`/
`.stderr` would signal that, and none changed).
Locked by examples/concurrency/1824 — two async tasks under the fiber Io, the
completion log proving deferral (1 2 then 10 20 then 123). Suite 829/0,
byte-identical aarch64-macOS host + aarch64-linux container.
This commit is contained in:
@@ -49,13 +49,19 @@ impl Io for CBlockingIo {
|
||||
// is here for the protocol shape the scheduler [B1.3] will use; the
|
||||
// blocking impl never routes through it, so it is a no-op handle.
|
||||
spawn_raw :: (self: *CBlockingIo, entry: *void, arg: *void, opts: SpawnOpts) -> *void {
|
||||
// The blocking model has no scheduler: run the worker thunk INLINE to
|
||||
// completion right here, so the `async` free-fn's Future is born `.ready`.
|
||||
// (A suspending impl — the fiber scheduler — instead defers `entry` onto a
|
||||
// fiber.) Same `(*void)->void` erased-thunk contract `spawn_raw` mandates.
|
||||
entry_fn : (*void) -> void = xx entry;
|
||||
entry_fn(arg);
|
||||
return null;
|
||||
}
|
||||
// Blocking never suspends — a suspend at the bottom of the M:1 stack
|
||||
// would deadlock. No-op (returns success). The `!` is part of the
|
||||
// protocol contract (a suspending impl raises `.Canceled` out here),
|
||||
// so the conforming blocking impl keeps it even though it never raises.
|
||||
suspend_raw :: (self: *CBlockingIo, park: ParkToken) -> ! {
|
||||
suspend_raw :: (self: *CBlockingIo, park: *ParkToken) -> ! {
|
||||
return;
|
||||
}
|
||||
ready :: (self: *CBlockingIo, park: ParkToken) {}
|
||||
@@ -89,31 +95,74 @@ Future :: struct ($R: Type) {
|
||||
}
|
||||
|
||||
// --- The async ergonomic layer (generic free-fns over the protocol) ---
|
||||
|
||||
// `async(io, worker, ..args)` — submit `worker(..args)`. Blocking: runs
|
||||
// the worker to completion inline, Future born `.ready`. The worker is a
|
||||
// `Closure(..$args) -> $R` (a lambda whose params are annotated at the
|
||||
// call site); `..$args` forwards the call-site arguments to it.
|
||||
//
|
||||
// NOTE on construction shape: the Future is built with `= ---` + per-field
|
||||
// assignment, NOT a `return Future.{...}` struct-literal. A struct-literal
|
||||
// in `return` position trips a generic-instantiation gap for the `Atomic`
|
||||
// field; the `= ---` (uninit) + field-assign form is the verified idiom.
|
||||
async :: ufcs (io: Io, worker: Closure(..$args) -> $R, ..$args) -> Future($R) {
|
||||
f : Future($R) = ---;
|
||||
f.value = worker(..args);
|
||||
f.state = .ready;
|
||||
// COLORBLIND over the `Io` impl: `async` always submits the worker through
|
||||
// `io.spawn_raw`, so the SAME code runs the worker inline under `CBlockingIo`
|
||||
// (Future born `.ready`) or as a real fiber under the scheduler (Future born
|
||||
// `.pending`, completed later — `await` suspends until then). The only protocol-
|
||||
// level value `spawn_raw` accepts is a raw `(*void)->void` entry + a `*void`
|
||||
// arg, so the generic worker is bridged via a MONOMORPHIC boxed-closure thunk
|
||||
// (`sx_run_boxed_closure`): all the generic-ness lives in the closure's env, and
|
||||
// the thunk is one fixed `Closure()->void` invoker — no per-instantiation entry.
|
||||
|
||||
// The one fixed entry `spawn_raw` ever calls: cast the arg back to the heap-boxed
|
||||
// completion closure and run it. Monomorphic (over `Closure()->void`), so a
|
||||
// single top-level symbol serves every `async($R)` instantiation.
|
||||
// The heap box the bridge carries: a struct holding the nullary completion
|
||||
// closure. A struct field is the one position a `Closure() -> void` type parses
|
||||
// in (a bare alias / `size_of(Closure()->void)` trips the parser), and it gives
|
||||
// the bridge a concrete `*ThunkBox` to `size_of`/cast/call through.
|
||||
ThunkBox :: struct { run: Closure() -> void; }
|
||||
|
||||
sx_run_boxed_closure :: (arg: *void) {
|
||||
b : *ThunkBox = xx arg;
|
||||
b.run();
|
||||
}
|
||||
|
||||
// `async(io, worker)` — submit a NULLARY `worker: Closure() -> $R` and get a
|
||||
// `*Future($R)` handle. The worker must be nullary because under the fiber impl
|
||||
// the body crosses a fiber boundary, and a captured variadic pack segfaults there
|
||||
// (issue 0156 Part 2) — so any inputs are captured at the CALL SITE in the lambda
|
||||
// (`context.io.async(() -> i64 => compute(a, b))`), exactly like `sched.go`.
|
||||
//
|
||||
// The Future is HEAP-allocated (not returned by value): under the fiber impl the
|
||||
// worker fills it AFTER `async` returns, so the awaiter and the worker must share
|
||||
// one stable object. Like `sched.go`'s Task, it currently leaks (bounded by the
|
||||
// async count; invisible under the default GPA). Freeing it needs join-point
|
||||
// ownership — deferred.
|
||||
async :: ufcs (io: Io, worker: Closure() -> $R) -> *Future($R) {
|
||||
raw := context.allocator.alloc_bytes(size_of(Future($R)));
|
||||
f : *Future($R) = xx raw;
|
||||
f.state = .pending;
|
||||
f.park = .{ handle = null };
|
||||
f.task = null;
|
||||
f.canceled = Atomic(bool).init(false);
|
||||
// The completion closure: run the worker, publish the result, wake any parked
|
||||
// awaiter. Heap-boxed so it survives until the worker actually runs (deferred
|
||||
// under the fiber impl). It captures `f` + `worker`; nothing variadic crosses.
|
||||
braw := context.allocator.alloc_bytes(size_of(ThunkBox));
|
||||
b : *ThunkBox = xx braw;
|
||||
b.run = () => {
|
||||
f.value = worker();
|
||||
f.state = .ready;
|
||||
context.io.ready(f.park); // no-op if no awaiter parked yet
|
||||
};
|
||||
f.task = io.spawn_raw(xx sx_run_boxed_closure, xx b, .{});
|
||||
return f;
|
||||
}
|
||||
|
||||
// (A nullary worker needs no separate entry: the variadic `async` above binds
|
||||
// `..$args` to the empty pack, so `context.io.async(() -> $R => …)` calls
|
||||
// `worker()` and returns `Future($R)`. Locked by examples/1805.)
|
||||
|
||||
// `await(f)` — value-carrying failable. `.ready` → the result; `.failed`
|
||||
// / `.canceled` → raise the stored / cancellation error.
|
||||
// `await(f)` — value-carrying failable. Suspends the caller until `f` completes
|
||||
// (no-op under the blocking impl, where it is already `.ready`), then `.ready` →
|
||||
// the result; `.failed`/`.canceled` → raise. Under the fiber impl the caller is a
|
||||
// fiber; `suspend_raw` records it into `f.park` so the worker's `ready(f.park)`
|
||||
// resumes it. Re-checks state after the wake (the worker set `.ready` before
|
||||
// waking). A worker that finished BEFORE `await` leaves `.ready`, so no park, no
|
||||
// lost wakeup.
|
||||
await :: ufcs (f: *Future($R)) -> $R !IoErr {
|
||||
if f.canceled.load(.acquire) { raise error.Canceled; }
|
||||
if f.state == .pending {
|
||||
context.io.suspend_raw(@f.park) catch {}; // Phase 3 propagates Canceled
|
||||
}
|
||||
if f.canceled.load(.acquire) { raise error.Canceled; }
|
||||
if f.state == .canceled { raise error.Canceled; }
|
||||
if f.state == .failed { raise error.Failed; }
|
||||
|
||||
Reference in New Issue
Block a user