sx/library/modules/std/sched.sx

// Stream B1 (fibers) B1.5a — the M:1 cooperative fiber scheduler core.
//
// A `Scheduler` drives any number of `Fiber`s, each running a stackful
// `body: Closure() -> void` on its own guarded `mmap` stack (the §8.1.1 guard
// page turns a stack overflow into an immediate fault instead of silent
// neighbor corruption). Fibers cooperate: a running fiber hands control back to
// the scheduler loop via `yield_now` (re-enqueued, round-robin) or
// `suspend_self` (parked off-queue until an external `wake`). When a body
// returns, the fiber reaches `.done`, its stack is `munmap`'d and its heap
// `Fiber` freed.
//
// Built on the proven primitives from examples/concurrency/1807-1809:
//   - `swap_context` (aarch64 `abi(.naked)`, 13-slot save area: x19..x28, fp,
//     lr, sp) saves the callee-saved registers + SP into `*from` and loads them
//     from `*to`, then `ret`s onto `to`'s stack.
//   - the `_fib_tramp` global-asm first-entry trampoline: x19 holds the
//     bootstrapped `*Fiber`; it moves it to x0 and `bl`s the exported generic
//     dispatch `fib_dispatch`, which calls the body then switches back to the
//     scheduler.
//   - guarded `mmap` stacks: `[GUARD | usable]`, low GUARD page `mprotect`'d
//     PROT_NONE, 16-aligned top returned as the bootstrapped SP.
//
// aarch64-macOS-pinned: the `swap_context` asm + the 13-slot save area are
// per-arch; the `mmap` flag constants (MAP_ANON = 0x1000) and the 16 KB guard
// page are Apple-specific. Runs end-to-end on a matching host, ir-only on a
// mismatch.
#import "modules/std.sx";
kqb :: #import "modules/std/net/kqueue.sx";

// --- libc mmap stack primitives -------------------------------------------

mmap     :: (addr: *void, len: i64, prot: i32, flags: i32, fd: i32, off: i64) -> *void extern libc "mmap";
mprotect :: (addr: *void, len: i64, prot: i32) -> i32 extern libc "mprotect";
munmap   :: (addr: *void, len: i64) -> i32 extern libc "munmap";
abort    :: () -> noreturn extern libc "abort";

PROT_NONE :: 0;
PROT_RW   :: 3;          // PROT_READ | PROT_WRITE
MAP_AP    :: 0x1002;     // macOS MAP_PRIVATE (0x2) | MAP_ANON (0x1000)
GUARD     :: 16384;      // one 16 KB page (aarch64-macOS)
STACK     :: 131072;     // 128 KB usable per fiber

// Max fd events drained per kqueue wait (B1.4c). Sized for the M:1 model's
// small fiber counts; a wait that fills it just drains the rest on the next
// loop iteration (the woken fibers run, the queue re-drains, the still-pending
// waiters block again).
MAXEV     :: 16;

// --- core types ------------------------------------------------------------

// Saved context: x19..x28 (10), x29/fp, x30/lr, sp — 13 u64 slots.
FiberCtx :: struct { regs: [13]u64; }

FiberState :: enum { ready; running; suspended; done; }

Fiber :: struct {
    ctx: FiberCtx;
    body: Closure() -> void;
    state: FiberState;
    sched: *Scheduler;
    stack_region: *void;   // mmap base — for munmap on reap
    stack_len: i64;        // GUARD + STACK, for munmap
    id: i64;
    next: *Fiber;          // intrusive FIFO ready-queue link
}

// A pending virtual-time timer: wake `fiber` once the virtual clock reaches
// `deadline_ms`. Stored in `Scheduler.timers` (a `List`) in insertion order, so
// a linear min-scan that takes the FIRST entry at the minimum deadline gives a
// stable FIFO tiebreak for equal deadlines.
Timer :: struct {
    deadline_ms: i64;
    fiber: *Fiber;
}

// B1.4c: a fiber parked on REAL fd readiness. Unlike a `Timer` (virtual
// time), an `IoWaiter` blocks the whole scheduler on `kevent` until the
// kernel reports `fd` readable, then wakes `fiber`. Stored in
// `Scheduler.io_waiters`; the registration is one-shot (EV_ONESHOT), so the
// kernel auto-removes it after firing — we only have to drop the waiter
// record. `cancel_io_waiter_for` evicts a stale record (mirror of
// `cancel_timer_for`) so a reaped fiber's waiter can never be woken.
IoWaiter :: struct {
    fd: i32;
    fiber: *Fiber;
}

Scheduler :: struct {
    sched_ctx: FiberCtx;        // the scheduler loop's own saved context
    current: *Fiber;            // running fiber; null while in the scheduler loop
    ready_head: *Fiber;
    ready_tail: *Fiber;
    own_allocator: Allocator;   // captured at init — fibers outlive their spawn scope
    next_id: i64;
    n_spawned: i64;
    n_suspended: i64;           // fibers parked off-queue (suspend_self minus wake)

    // --- B1.4b: deterministic virtual-time timer scheduling ----------------
    clock_ms: i64;              // the VIRTUAL clock (ms). Starts 0; advances ONLY
                                // when the ready queue drains and the earliest
                                // pending timer fires. No real wall clock is ever
                                // read — wake ORDER + timestamps are reproducible.
    timers: List(Timer);        // pending sleep timers, in insertion order. Grown
                                // through `own_allocator` (long-lived-container
                                // rule: a timer outlives the `sleep` call's scope).

    // --- B1.4c: real fd-readiness blocking via kqueue ----------------------
    kq: i32;                    // the kqueue fd. LAZY: -1 until the first
                                // `block_on_fd` opens it, so a pure-compute /
                                // virtual-timer scheduler never opens a kqueue
                                // fd (no leak for the common case). Once opened it
                                // lives for the scheduler's lifetime; there is no
                                // deinit yet, so it leaks one fd at program exit
                                // (bounded, harmless — same class as the spawn
                                // env / go Task leaks documented above).
    io_waiters: List(IoWaiter); // fibers parked on fd readiness, grown through
                                // `own_allocator` (long-lived-container rule: a
                                // waiter outlives the `block_on_fd` call's scope).

    // Construct a scheduler BY VALUE (allocator value-return convention).
    // Captures the current `context.allocator` into `own_allocator` — fibers and
    // their heap `Fiber` structs outlive their spawn scope, so all internal
    // allocation must go through this captured (long-lived) allocator, not
    // whatever transient one happens to be current at a later call.
    init :: () -> Scheduler {
        s : Scheduler = ---;
        s.current = null;
        s.ready_head = null;
        s.ready_tail = null;
        s.own_allocator = context.allocator;
        s.next_id = 0;
        s.n_spawned = 0;
        s.n_suspended = 0;
        s.clock_ms = 0;
        s.timers = .{};
        s.kq = -1;              // lazy: opened by the first block_on_fd
        s.io_waiters = .{};
        return s;
    }

    // Spawn a fiber running `body`. Heap-allocates the `Fiber` and a guarded
    // stack, bootstraps the saved context (x19 = *Fiber, fp = 0, lr =
    // trampoline, sp = stack top), enqueues it ready (FIFO), returns the
    // `*Fiber`.
    // KNOWN LIMITATION (env leak): `body` is a fat `{fn_ptr, env}` whose env is
    // heap-allocated at the closure-literal site. The reap path frees the Fiber
    // struct + unmaps the stack, but sx exposes no way to free a closure's env
    // (the scheduler can't name the env pointer), so ONE env per spawned fiber
    // leaks until program exit. Bounded by the spawn count; under the default
    // GPA (which frees at exit) it is invisible, but a long-running scheduler
    // under an arena/tracking allocator accumulates one env per fiber. Freeing
    // it needs a language affordance for closure-env ownership — deferred.
    spawn :: (self: *Scheduler, body: Closure() -> void) -> *Fiber {
        raw := self.own_allocator.alloc_bytes(size_of(Fiber));
        if raw == null {
            print("sched: out of memory allocating a Fiber\n");
            abort();
        }
        f : *Fiber = xx raw;
        f.body = body;
        f.sched = self;
        f.id = self.next_id;
        f.next = null;
        self.next_id = self.next_id + 1;
        self.n_spawned = self.n_spawned + 1;

        top := boot_stack(f, STACK);
        f.ctx.regs[0]  = xx f;           // x19 = self
        f.ctx.regs[10] = 0;              // fp
        f.ctx.regs[11] = xx fib_tramp;   // lr → trampoline
        f.ctx.regs[12] = top;            // sp

        f.state = .ready;
        enqueue(self, f);
        return f;
    }

    // The running fiber yields cooperatively: mark ready, switch back to the
    // scheduler. The run loop re-enqueues it (round-robin). MUST be called from
    // inside a fiber (there must be a running fiber to yield).
    yield_now :: (self: *Scheduler) {
        cur := self.current;
        if cur == null {
            print("sched: yield_now() called outside a fiber (no running fiber)\n");
            abort();
        }
        cur.state = .ready;
        swap_context(@cur.ctx, @self.sched_ctx);
    }

    // The running fiber parks itself: mark suspended, switch back to the
    // scheduler. The run loop does NOT re-enqueue a suspended fiber — an
    // external `wake` must re-add it. (Used by FiberIo to park on a blocking
    // op until completion.) MUST be called from inside a fiber — a null
    // `current` (called from the bare scheduler/main context) would deref null;
    // bail loudly instead of segfaulting.
    suspend_self :: (self: *Scheduler) {
        cur := self.current;
        if cur == null {
            print("sched: suspend_self() called outside a fiber (no running fiber)\n");
            abort();
        }
        cur.state = .suspended;
        self.n_suspended = self.n_suspended + 1;
        swap_context(@cur.ctx, @self.sched_ctx);
    }

    // Re-ready a parked (suspended) fiber and enqueue it. Called from outside
    // the fiber (e.g. an I/O completion or another fiber) to wake it.
    //
    // GUARDED on `.suspended`: enqueue links `f` into the FIFO, so waking a
    // fiber that is ALREADY queued (`.ready`) or running (`.running`) would
    // re-link a node already in the list — nulling its `next` mid-list and
    // cycling `ready_tail` back onto it, corrupting the queue (a spurious /
    // double wake, or waking a yielded-not-parked fiber, would segfault). Only
    // a genuinely parked fiber may be re-enqueued; any other wake is a no-op.
    wake :: (self: *Scheduler, f: *Fiber) {
        if f.state != .suspended { return; }
        // Evict any pending sleep timer for `f`. EVERY path that re-readies a
        // suspended fiber funnels through `wake` (a manual/Task wake, or the
        // timer-fire in `run` — which already removed the fired timer, so this
        // is a harmless re-scan there). Without this, a fiber that armed a
        // `sleep` timer but was woken EARLY by another path would run to
        // completion and be reaped (stack munmap'd + Fiber freed) while its
        // Timer still held a dangling `*Fiber` — a later fire would dereference
        // freed memory (use-after-free). One timer per fiber max in the M:1
        // model, so a single eviction suffices; it also prevents a stale timer
        // from spuriously re-waking a since-re-slept fiber.
        cancel_timer_for(self, f);
        // Same UAF reasoning for fd waiters: every path that re-readies a
        // suspended fiber funnels through `wake`. If a fiber armed `block_on_fd`
        // but was woken by another path (a manual wake, a Task completion), its
        // `IoWaiter` would otherwise survive pointing at a fiber that runs to
        // completion and is reaped (stack munmap'd + Fiber freed). A later
        // kqueue drain matching that stale record would `wake` freed memory.
        // Evict it here. NOTE: we do NOT EV_DELETE the kqueue registration — it
        // is EV_ONESHOT, so a never-fired registration simply lingers in the
        // kernel queue until the fd is readable, at which point the drain finds
        // no matching waiter and ignores it (see `run`). The fd is the example's
        // to close; closing it auto-removes any pending registration.
        cancel_io_waiter_for(self, f);
        self.n_suspended = self.n_suspended - 1;
        f.state = .ready;
        enqueue(self, f);
    }

    // Read the VIRTUAL clock — the simulated millisecond time. Advances only as
    // timers fire (in `run`), never from a real wall clock, so two runs of the
    // same fiber program observe identical timestamps. A fiber that just woke
    // from `sleep(ms)` sees `now_ms()` equal to its deadline.
    now_ms :: (self: *Scheduler) -> i64 {
        return self.clock_ms;
    }

    // Sleep the running fiber for `ms` simulated milliseconds: arm a timer at
    // `clock_ms + ms`, then park off-queue. The scheduler advances the virtual
    // clock to this deadline and wakes the fiber once the ready queue has fully
    // drained AND no earlier timer is pending (deadline order, FIFO tiebreak).
    // MUST be called from inside a fiber (there must be a `current` to park);
    // a null `current` bails loudly, mirroring `suspend_self`.
    //
    // Virtual time only moves forward: `ms >= 0` makes the deadline
    // `>= clock_ms`, so a fired timer never rewinds the clock.
    sleep :: (self: *Scheduler, ms: i64) {
        cur := self.current;
        if cur == null {
            print("sched: sleep() called outside a fiber (no running fiber)\n");
            abort();
        }
        // The virtual clock is MONOTONIC — it only advances as timers fire. A
        // negative duration would arm a deadline in the past, rewinding the
        // clock when it fired and breaking every ordering contract. Reject it
        // loudly rather than silently corrupting time. (`sleep(0)` is allowed: a
        // same-tick yield to the timer wheel.)
        if ms < 0 {
            print("sched: sleep({}) — negative duration would rewind the virtual clock\n", ms);
            abort();
        }
        t : Timer = .{ deadline_ms = self.clock_ms + ms, fiber = cur };
        // Long-lived-container rule: a timer outlives this `sleep` call's scope
        // (it survives in `self.timers` until the scheduler fires it), so grow
        // through the captured `own_allocator`, never the transient current one.
        self.timers.append(t, self.own_allocator);
        self.suspend_self();   // parks `cur` off-queue; the timer fire re-wakes it
    }

    // --- B1.4c: block the running fiber until `fd` is readable --------------
    //
    // Register `fd` for EVFILT_READ with the scheduler's kqueue (lazily
    // opening it on first use), record an `IoWaiter`, then park the fiber
    // off-queue. The run loop blocks on `kevent` once nothing else is runnable
    // and wakes this fiber when the kernel reports `fd` ready (EV_ONESHOT — the
    // kernel auto-removes the registration after it fires, so the run loop only
    // has to drop the waiter record + `wake` the fiber).
    //
    // `want_read` is the readiness direction; only read-readiness is wired for
    // now (a write-readiness EVFILT_WRITE path would mirror this exactly). A
    // false `want_read` would be a write-wait — not yet implemented, so bail
    // loudly rather than silently arming a read filter (silent-wrong-arm rule).
    //
    // MUST be called from inside a fiber (there must be a `current` to park); a
    // null `current` bails loudly, mirroring `suspend_self` / `sleep`.
    block_on_fd :: (self: *Scheduler, fd: i32, want_read: bool) {
        cur := self.current;
        if cur == null {
            print("sched: block_on_fd() called outside a fiber (no running fiber)\n");
            abort();
        }
        if !want_read {
            print("sched: block_on_fd(want_read=false) — write-readiness not implemented\n");
            abort();
        }
        // ONE waiter per fd (enforced). macOS `EV_ADD` for an existing
        // (ident, filter) REPLACES the registration rather than stacking, so a
        // second fiber blocking on the same fd would leave only one live
        // registration: when the fd fires, the kernel delivers a single event,
        // one waiter wakes, and the other is stranded in `io_waiters` with no
        // registration — the next `kq_wait` then blocks forever. The M:1 model
        // (and `wake_io_waiter_for_fd`, which wakes the first match) assumes a
        // single waiter per fd; enforce it loudly instead of silently hanging.
        j := 0;
        while j < self.io_waiters.len {
            if self.io_waiters.items[j].fd == fd {
                print("sched: block_on_fd: fd {} already has a waiter (one waiter per fd in the M:1 model)\n", fd);
                abort();
            }
            j = j + 1;
        }
        // Lazily open the kqueue fd the first time fd-blocking is used.
        if self.kq < 0 {
            self.kq = kqb.kqueue();
            if self.kq < 0 {
                print("sched: kqueue() failed to open the event queue\n");
                abort();
            }
        }
        // Arm a one-shot read-readiness registration for `fd`. udata is unused
        // (we match the waiter by fd in the drain), so pass 0.
        chg := kqb.kev_change(fd, kqb.EVFILT_READ, kqb.EV_ADD | kqb.EV_ENABLE | kqb.EV_ONESHOT, 0);
        if !kqb.kq_apply(self.kq, chg) {
            print("sched: kevent() failed to register fd {} for read readiness\n", fd);
            abort();
        }
        // Record the waiter BEFORE parking — the run loop matches the fired
        // event's ident back to this record. Long-lived-container rule: the
        // waiter outlives this call's scope (it survives in `self.io_waiters`
        // until the kqueue drain wakes it), so grow through `own_allocator`.
        w : IoWaiter = .{ fd = fd, fiber = cur };
        self.io_waiters.append(w, self.own_allocator);
        self.suspend_self();   // parks `cur` off-queue; the kqueue drain re-wakes it
    }

    // The scheduler loop. Drives ready fibers to quiescence, then advances the
    // virtual clock by firing the earliest pending timer (which re-readies its
    // sleeper), and repeats — until both the ready queue and the timer set are
    // empty. Within the inner drain each iteration: dequeue the next fiber,
    // switch into it, and — on its switch back — reap it if done (munmap stack,
    // free the Fiber), re-enqueue it if it yielded, or leave it parked if it
    // suspended.
    run :: (self: *Scheduler) {
        while true {
            while self.ready_head != null {
                f := dequeue(self);
                self.current = f;
                f.state = .running;
                swap_context(@self.sched_ctx, @f.ctx);   // returns here when f yields / suspends / finishes
                self.current = null;
                if f.state == .done {
                    // We've switched OFF f's stack already (the final swap landed
                    // here), so the stack is free to unmap. Free the Fiber struct
                    // AFTER munmap.
                    munmap(f.stack_region, f.stack_len);
                    self.own_allocator.dealloc_bytes(xx f);
                } else if f.state == .ready {
                    enqueue(self, f);
                }
                // .suspended: leave it parked (not in any queue; `wake` re-adds it).
            }
            // Ready queue drained. Decide what advances the world next.
            //
            // Mode 1 — VIRTUAL TIME: fire the earliest pending timer (advancing
            // the virtual clock to it), re-readying its sleeper. Timers take
            // precedence over fd-blocking: a program uses `sleep` OR fds, not
            // both at once. (Documented limitation: virtual-time timers and real
            // kqueue timeouts are NOT unified — if both a timer and an io-waiter
            // are pending we always fire the timer first and never block on
            // kqueue while a timer is outstanding. A program that genuinely
            // needs "fd-or-real-timeout" wants a kqueue EVFILT_TIMER, future
            // work.)
            idx := earliest_timer(self);
            if idx >= 0 {
                t := self.timers.items[idx];
                remove_timer(self, idx);
                self.clock_ms = t.deadline_ms;   // advance VIRTUAL time forward
                self.wake(t.fiber);              // re-enqueue the sleeper → drain again
                continue;
            }
            // Mode 2 — REAL fd readiness: nothing is runnable and no timer is
            // pending, but fibers are parked on fds. BLOCK on kqueue until the
            // kernel reports at least one fd ready, then wake every waiter whose
            // fd fired. (null timeout via -1 → wait forever.)
            if self.io_waiters.len > 0 {
                evbuf : [MAXEV]kqb.Kevent = ---;
                n := kqb.kq_wait(self.kq, @evbuf[0], MAXEV, -1);
                if n < 0 {
                    print("sched: kevent() wait failed while blocking on fd readiness\n");
                    abort();
                }
                // For each fired event, find the io-waiter whose fd matches its
                // ident, evict it, and wake its fiber. EV_ONESHOT already removed
                // the kernel registration, so we only drop the waiter record.
                i := 0;
                while i < n {
                    ready_fd : i32 = xx evbuf[i].ident;
                    wake_io_waiter_for_fd(self, ready_fd);
                    i = i + 1;
                }
                continue;
            }
            // Nothing runnable, no timer, no fd waiter → done.
            break;
        }
        // The ready queue, the timer set, AND the io-waiter set are all empty. If
        // a fiber is STILL parked, nothing will ever wake it (a `suspend_self`
        // without an armed timer or fd registration, never externally woken) —
        // its stack + struct are leaked and the program believes it finished.
        // That is a genuine deadlock; surface it loudly. (Timer sleepers and fd
        // waiters are balanced: each arming path increments `n_suspended` via
        // `suspend_self`, and its wake decrements it — so once every timer has
        // fired and every io-waiter has been woken, `n_suspended` counts only
        // these true orphans.)
        //
        // SCOPE — fd waiters are NOT covered by this check, BY DESIGN, not as an
        // oversight. While `io_waiters.len > 0` the loop above blocks in
        // `kq_wait(-1)` and never reaches here. A fiber blocked on an fd that the
        // OS never reports ready blocks FOREVER — which is the correct semantics
        // of an event loop (a server idling on a socket is indistinguishable from
        // one whose peer never sends; the scheduler cannot know an fd will never
        // become ready, so it must keep waiting). That is a caller-side logic
        // issue (blocking on input that never arrives), not a scheduler deadlock
        // to abort on. This check covers only pure `suspend_self` parks with no
        // pending wake source at all.
        if self.n_suspended != 0 {
            print("sched: deadlock — {} fiber(s) suspended with an empty run queue\n", self.n_suspended);
            abort();
        }
    }
}

// --- the context switch (naked) + first-entry trampoline -------------------

// x0 = from, x1 = to (read straight from the ABI registers — a naked fn has no
// frame, so its params are never spilled). SP-in ≠ SP-out by design.
swap_context :: (from: *FiberCtx, to: *FiberCtx) abi(.naked) {
    asm volatile {
        #string ASM
        stp x19, x20, [x0, #0]
        stp x21, x22, [x0, #16]
        stp x23, x24, [x0, #32]
        stp x25, x26, [x0, #48]
        stp x27, x28, [x0, #64]
        stp x29, x30, [x0, #80]
        mov x9, sp
        str x9, [x0, #96]
        ldp x19, x20, [x1, #0]
        ldp x21, x22, [x1, #16]
        ldp x23, x24, [x1, #32]
        ldp x25, x26, [x1, #48]
        ldp x27, x28, [x1, #64]
        ldp x29, x30, [x1, #80]
        ldr x9, [x1, #96]
        mov sp, x9
        ret
ASM
    };
}

// First-entry trampoline: a fiber's bootstrapped LR points here. x19 holds the
// `*Fiber` (preset in the saved context); move it to x0 and call the generic
// dispatch.
asm {
    #string T
.global _fib_tramp
_fib_tramp:
    mov x0, x19
    bl _fib_dispatch
    brk #0
T,
};
fib_tramp :: () extern;

// The ONE place that runs a fiber body. Reached only from `_fib_tramp` on first
// entry, on the fiber's own fresh stack. Runs the body, marks the fiber done,
// and switches back to the scheduler — never returns past the final switch.
fib_dispatch :: (self: *Fiber) export "fib_dispatch" {
    self.body();
    self.state = .done;
    swap_context(@self.ctx, @self.sched.sched_ctx);
}

// --- guarded stack bootstrap ----------------------------------------------

// mmap a [guard | usable-stack] region, mprotect the low guard page PROT_NONE.
// Stores the region base + len on the fiber (for munmap on reap) and returns
// the 16-aligned stack top (the bootstrapped SP).
boot_stack :: (f: *Fiber, size: i64) -> u64 {
    total := GUARD + size;
    region : *void = mmap(null, total, PROT_RW, MAP_AP, -1, 0);
    // mmap signals failure with MAP_FAILED = (void*)-1 (NOT null). Handing a
    // wild SP to the switch would `ret` onto garbage — bail loudly instead.
    if (xx region) == (xx (0 - 1)) {
        print("sched: mmap failed for a {}-byte fiber stack\n", total);
        abort();
    }
    f.stack_region = region;
    f.stack_len = total;
    // Guard-arm: turn the low page unwritable so overflow faults at the
    // boundary. The guard is mandatory (§8.1.1); a stack handed out without it
    // would silently corrupt a neighbor on overflow, so a failed mprotect is
    // fatal, not ignorable.
    if mprotect(region, GUARD, PROT_NONE) != 0 {
        print("sched: mprotect(PROT_NONE) failed to arm the stack guard page\n");
        abort();
    }
    usable : u64 = (xx region) + GUARD;
    top : u64 = usable + size;
    return top - (top % 16);   // 16-byte aligned stack top (AAPCS)
}

// --- intrusive FIFO ready-queue -------------------------------------------

enqueue :: (self: *Scheduler, f: *Fiber) {
    f.next = null;
    if self.ready_tail == null {
        self.ready_head = f;
        self.ready_tail = f;
    } else {
        self.ready_tail.next = f;
        self.ready_tail = f;
    }
}

dequeue :: (self: *Scheduler) -> *Fiber {
    f := self.ready_head;
    if f == null { return null; }
    self.ready_head = f.next;
    if self.ready_head == null { self.ready_tail = null; }
    f.next = null;
    return f;
}

// --- virtual-time timer set (linear min-scan, FIFO tiebreak) ---------------
//
// The timer set is a plain `List(Timer)` kept in INSERTION order. Fiber counts
// are tiny, so a linear scan for the minimum deadline is ideal — no heap to
// maintain — and "first entry at the minimum" naturally gives FIFO ordering for
// equal deadlines (the earlier-inserted timer is visited first, so it wins the
// tie). Removal shifts the tail down by one to preserve that insertion order for
// the remaining entries.

// Index of the earliest-deadline pending timer, or -1 if none. On a deadline
// tie the lowest index (earliest inserted) wins → deterministic FIFO wake order.
earliest_timer :: (self: *Scheduler) -> i64 {
    if self.timers.len == 0 { return -1; }
    best := 0;
    i := 1;
    while i < self.timers.len {
        // Strict `<` so equal deadlines do NOT displace the earlier (lower)
        // index — that is the FIFO tiebreak.
        if self.timers.items[i].deadline_ms < self.timers.items[best].deadline_ms {
            best = i;
        }
        i = i + 1;
    }
    return best;
}

// Remove the timer at `idx`, shifting every later entry down one slot so the
// remaining timers keep their insertion order (preserving the FIFO tiebreak).
remove_timer :: (self: *Scheduler, idx: i64) {
    i := idx;
    while i < self.timers.len - 1 {
        self.timers.items[i] = self.timers.items[i + 1];
        i = i + 1;
    }
    self.timers.len = self.timers.len - 1;
}

// Remove a pending sleep timer referencing fiber `f`, if any. A fiber has at
// most one pending timer in the M:1 model (it can only `sleep` once before
// suspending), so the first match is the only one. No-op if `f` has none.
cancel_timer_for :: (self: *Scheduler, f: *Fiber) {
    i := 0;
    while i < self.timers.len {
        if self.timers.items[i].fiber == f {
            remove_timer(self, i);
            return;
        }
        i = i + 1;
    }
}

// --- B1.4c: fd-waiter set (linear scan, fd-keyed) --------------------------
//
// Like the timer set, a plain `List(IoWaiter)` scanned linearly — fiber counts
// are tiny. Removal shifts the tail down one slot.

// Remove the io-waiter at `idx`, shifting later entries down one slot.
remove_io_waiter :: (self: *Scheduler, idx: i64) {
    i := idx;
    while i < self.io_waiters.len - 1 {
        self.io_waiters.items[i] = self.io_waiters.items[i + 1];
        i = i + 1;
    }
    self.io_waiters.len = self.io_waiters.len - 1;
}

// Remove a pending fd-waiter referencing fiber `f`, if any. A fiber has at most
// one pending io-waiter in the M:1 model (it can only `block_on_fd` once before
// suspending), so the first match is the only one. No-op if `f` has none. Used
// by `wake` to evict a waiter when the fiber is re-readied by another path.
cancel_io_waiter_for :: (self: *Scheduler, f: *Fiber) {
    i := 0;
    while i < self.io_waiters.len {
        if self.io_waiters.items[i].fiber == f {
            remove_io_waiter(self, i);
            return;
        }
        i = i + 1;
    }
}

// A fired kqueue event for `fd`: find the waiter registered on it, evict the
// record, and wake its fiber. No-op if no waiter matches (a stale one-shot
// registration whose fiber was already woken another way — see `wake`). Only
// the FIRST match is woken: one waiter per fd in this model (a single fiber
// blocks on a given read fd at a time).
wake_io_waiter_for_fd :: (self: *Scheduler, fd: i32) {
    i := 0;
    while i < self.io_waiters.len {
        if self.io_waiters.items[i].fd == fd {
            wf := self.io_waiters.items[i].fiber;
            remove_io_waiter(self, i);
            self.wake(wf);   // re-enqueues the parked fiber (also calls
                             // cancel_io_waiter_for, now a harmless no-op —
                             // the record is already removed)
            return;
        }
        i = i + 1;
    }
}

// The public API lives as methods on `Scheduler` (above): `init`, `spawn`,
// `yield_now`, `suspend_self`, `wake`, `run`, `now_ms`, `sleep`.

// --- B1.4a: truly-suspending fiber-task async (`go` / `wait` / `cancel`) ----
//
// An async-task layer on top of the M:1 scheduler: `s.go(work)` runs `work` as
// a REAL fiber, and `t.wait()` SUSPENDS the caller fiber until the task's fiber
// completes — genuine interleaving, in contrast with io.sx's `context.io.async`
// (which runs the worker inline to completion before returning). Distinct from
// io.sx's `Future` by design: `Task` is defined here so the two modules stay
// decoupled (no cross-import; sched.sx must keep importing only `std.sx`, since
// a different import path re-emits the module's global `_fib_tramp` asm and
// duplicates the symbol).
//
// THE NULLARY-THUNK RATIONALE. `work` is a NULLARY thunk `Closure() -> $R`, not
// a worker-plus-`..args` pair like io.sx's `async`. A variadic pack is
// comptime-only and segfaults if captured into a deferred closure that crosses
// the fiber boundary (issue 0156 Part 2). So instead of forwarding inputs as a
// pack, the user captures any inputs in the lambda AT THE CALL SITE (where
// they're live): `s.go(() -> i64 => compute(a, b))`. Nothing variadic ever
// crosses into the fiber — the thunk is a plain `{fn_ptr, env}` fat closure.
//
// KNOWN LIMITATION (heap-Task leak): `go` heap-allocates the `Task` (it outlives
// the call — the fiber fills `value`/`state` later, after `go` has returned), but
// B1.4a never frees it. Like the closure-env leak documented on `spawn` above,
// this is bounded by the `go` count and invisible under the default GPA (frees
// at exit); a long-running scheduler under an arena/tracking allocator
// accumulates one `Task` per `go`. Freeing it safely needs join-point ownership
// tracking — deferred.
//
// WAKE-AFTER-COMPLETE ORDERING (both orderings are correct):
//   - worker finishes BEFORE `wait`: the worker set `t.state = .ready` and saw
//     `t.waiter == null`, so it issued no wake. `wait` sees `.ready` (not
//     `.pending`), does NOT park, and returns `t.value` — no lost wakeup.
//   - `wait` runs BEFORE the worker finishes: `wait` registers itself as
//     `t.waiter` and parks via `suspend_self`. When the worker finishes it sees
//     a non-null `t.waiter` and `wake`s it; `wait` resumes and returns the value.

TaskState :: enum { pending; ready; canceled; }

// The `!` channel for `wait`. Defined LOCALLY (not reusing io.sx's `IoErr`):
// `IoErr` is reachable here only as a re-export alias through std.sx, and the
// failable-type detection behind `raise` does not see through that alias to the
// underlying `error` set — so `raise error.Canceled` against `(.., !IoErr)`
// here is rejected as "not a failable function". A local `error` decl is
// recognized directly. (Same `.Canceled` contract as io.sx model (a).)
TaskErr :: error { Canceled }

Task :: struct ($R: Type) {
    value: R;
    state: TaskState = .pending;
    waiter: *void = null;        // the single parked awaiter (opaque *Fiber); M:1 → at most one
    sched: *Scheduler;           // owning scheduler (for park/wake in `wait`)
    canceled: i64;               // cooperative cancel flag (M:1: no preemption → no atomics)
}

// Spawn `work` as a fiber; return a heap `*Task` that completes when the fiber
// finishes. Mirrors `spawn`'s alloc + null-check + abort.
go :: ufcs (self: *Scheduler, work: Closure() -> $R) -> *Task($R) {
    raw := self.own_allocator.alloc_bytes(size_of(Task($R)));
    if raw == null {
        print("sched: out of memory allocating a Task\n");
        abort();
    }
    t : *Task($R) = xx raw;
    t.state = .pending;
    t.waiter = null;
    t.sched = self;
    t.canceled = 0;
    self.spawn(() => {
        // Cooperative cancel: skip the work entirely if cancel already landed
        // before this fiber was scheduled (saves the compute + side effects). A
        // cancel that lands DURING `work()` still lets it finish (no preemption
        // in the M:1 model) — cancel suppresses DELIVERY, never an in-flight run.
        if t.canceled == 0 {
            t.value = work();
            t.state = .ready;
        }
        // Wake the awaiter only if one already parked (else `wait` will not park).
        if t.waiter != null { self.wake(xx t.waiter); }
    });
    return t;
}

// Suspend the caller until the task completes; return its value (or raise on
// cancel). MUST be called from inside a fiber (so there is a `self.current` to
// park) — typically from a fiber spawned via `s.spawn(...)`.
wait :: ufcs (t: *Task($R)) -> ($R, !TaskErr) {
    if t.canceled != 0 { raise error.Canceled; }
    if t.state == .pending {
        // ONE waiter per task (enforced). A `Task` holds a single `waiter` slot;
        // a second concurrent `wait` on the same pending task would OVERWRITE the
        // first, and completion would wake only the second — the first fiber
        // would stay suspended forever (silent deadlock). The M:1 model is
        // single-await per task; enforce it loudly (mirrors `block_on_fd`'s
        // one-waiter-per-fd guard). A multi-waiter task would need a waiter list.
        if t.waiter != null {
            print("sched: wait() — task already has a waiter (one awaiter per task in the M:1 model)\n");
            abort();
        }
        t.waiter = xx t.sched.current;   // register self as the waiter
        t.sched.suspend_self();          // park until the task's fiber wakes us
    }
    if t.canceled != 0 or t.state == .canceled { raise error.Canceled; }
    return t.value;
}

// Request cancellation — rides the `!` channel (model (a), like io.sx 1806). M:1
// cooperative: the worker fiber may already have run; cancel still makes a
// subsequent (or in-flight) `wait` raise `.Canceled`.
cancel :: ufcs (t: *Task($R)) {
    t.canceled = 1;
    t.state = .canceled;
}