danos/library/runtime/thread.zig

//! `runtime.Thread` — threads for danos, shaped like Zig's `std.Thread` but built on
//! danos's private thread ABI (docs/threading.md). Several tasks share one address
//! space; `spawn` starts one, the kernel delivers the closure pointer in the new
//! thread's rdi, a plain Zig trampoline runs the user function and calls `thread_exit`,
//! and `join` blocks on the thread's exit notification. See docs/threading.md for why
//! this mirrors `std.Thread`'s API rather than being the literal type.
//!
//! The closure (the function's captured args) lives at the **top of the thread's own
//! stack**, not the heap — each thread's stack is private, so there is no shared-heap
//! concurrency in the spawn/join machinery (the runtime heap is not yet thread-safe).
//! A binary must be built multi-threaded (`addThreadedUserBinary`) before it may spawn.

const std = @import("std");
const abi = @import("abi");
const sc = @import("system-call.zig");
const system = @import("system.zig");
const ipc = @import("ipc.zig");

/// A thread stack, if the caller does not override it. 64 KiB of mmap'd, zeroed pages.
pub const default_stack_size: usize = 64 * 1024;

pub const Thread = struct {
    /// The kernel task id of the spawned thread.
    tid: u32,
    /// The endpoint the kernel notifies when this thread ends — what `join` blocks on.
    exit_endpoint: ipc.Handle,
    /// The mmap'd stack, reclaimed by `join` (or at process exit after `detach`).
    stack_base: usize,
    stack_size: usize,

    pub const Id = u32;

    pub const SpawnConfig = struct {
        /// Bytes of stack, rounded up to whole pages by the kernel's mmap.
        stack_size: usize = default_stack_size,
    };

    pub const SpawnError = error{
        /// The kernel refused the thread, the stack mmap failed, or no endpoint was free.
        SystemResources,
    };

    /// Start `function(args...)` on a new thread sharing this address space. Mirrors
    /// `std.Thread.spawn`. The thread's return value is discarded (as in `std.Thread`);
    /// return data through shared state.
    pub fn spawn(config: SpawnConfig, comptime function: anytype, args: anytype) SpawnError!Thread {
        const Args = @TypeOf(args);
        const Closure = struct {
            args: Args,
            /// Entered directly by the kernel with `self` in rdi (C ABI). Runs the user
            /// function, then ends the thread — never returns.
            fn entry(self_addr: usize) callconv(.c) noreturn {
                const self: *@This() = @ptrFromInt(self_addr);
                @call(.auto, function, self.args);
                exitThread();
            }
        };

        // The endpoint the kernel posts this thread's exit notification to.
        const endpoint = ipc.createIpcEndpoint() orelse return error.SystemResources;

        const base = system.mmap(config.stack_size, system.PROT_READ | system.PROT_WRITE);
        if (system.mmapFailed(base)) return error.SystemResources;

        // Lay the closure at the very top of the thread's own stack, then start the
        // thread's rsp just below it (16-aligned minus 8, the alignment a `call` leaves
        // for a C-ABI entry) so the growing stack never overwrites the args.
        var closure_addr = (base + config.stack_size) - @sizeOf(Closure);
        closure_addr &= ~@as(usize, @alignOf(Closure) - 1); // align the closure down
        const closure: *Closure = @ptrFromInt(closure_addr);
        closure.* = .{ .args = args };

        var stack_top = closure_addr & ~@as(usize, 15); // 16-align below the closure
        stack_top -= 8; // ...then rsp % 16 == 8 at the C entry

        const tid = threadSpawn(@intFromPtr(&Closure.entry), stack_top, closure_addr, endpoint);
        if (threadSpawnFailed(tid)) {
            _ = system.munmap(base, config.stack_size);
            return error.SystemResources;
        }
        return .{ .tid = @intCast(tid), .exit_endpoint = endpoint, .stack_base = base, .stack_size = config.stack_size };
    }

    /// Block until this thread finishes, then reclaim its stack. Mirrors
    /// `std.Thread.join`. The exit endpoint is private to this thread, so the first
    /// child-exit notification on it is this thread's.
    pub fn join(self: Thread) void {
        var receive: [0]u8 = undefined;
        while (true) {
            const got = ipc.replyWait(self.exit_endpoint, &.{}, &receive, null);
            if (got.isChildExit() and got.childProcessId() == self.tid) break;
        }
        _ = system.munmap(self.stack_base, self.stack_size);
    }

    /// Relinquish the right to join: never wait for or reclaim this thread. Its stack is
    /// reclaimed at process exit (docs/threading-plan.md M3 — kernel-reaper stack reclaim
    /// for detached threads is a later refinement). Mirrors `std.Thread.detach`.
    pub fn detach(self: Thread) void {
        _ = self;
    }

    /// The calling thread's id (its kernel task id). Mirrors `std.Thread.getCurrentId`.
    pub fn getCurrentId() Id {
        return @intCast(sc.systemCall0(.thread_self));
    }

    /// The dense 0-based index of the core the calling thread is running on. A danos
    /// extension beyond `std.Thread`, used to observe genuine cross-core parallelism.
    pub fn currentCore() Id {
        return @intCast(sc.systemCall0(.current_core));
    }

    /// `std.Thread.Futex`-shaped block/wake on a `u32` atomic — the primitive the
    /// blocking `Mutex`/`Condition`/`Semaphore` are built on. Waiters park in the
    /// kernel (no busy-wait), so an idle core still halts (docs/halting.md).
    pub const Futex = struct {
        /// Block while `ptr.* == expect`. Returns when woken by `wake`, or promptly if
        /// the value already differs (safe against spurious returns, as in std): the
        /// caller re-checks its condition in a loop.
        pub fn wait(ptr: *const std.atomic.Value(u32), expect: u32) void {
            _ = futexWait(@intFromPtr(ptr), expect, 0);
        }

        /// As `wait`, but returns `error.Timeout` if `timeout_ns` elapses first.
        pub fn timedWait(ptr: *const std.atomic.Value(u32), expect: u32, timeout_ns: u64) error{Timeout}!void {
            if (futexWait(@intFromPtr(ptr), expect, timeout_ns) == abi.futex_timed_out) return error.Timeout;
        }

        /// Wake up to `max_waiters` threads blocked on `ptr`.
        pub fn wake(ptr: *const std.atomic.Value(u32), max_waiters: u32) void {
            _ = futexWake(@intFromPtr(ptr), max_waiters);
        }
    };

    /// A mutual-exclusion lock, `std.Thread.Mutex`-shaped. The classic three-state
    /// futex mutex (unlocked / locked / contended): the fast path is a single CAS, and
    /// only a contended lock ever enters the kernel.
    pub const Mutex = struct {
        state: std.atomic.Value(u32) = std.atomic.Value(u32).init(unlocked),

        const unlocked: u32 = 0;
        const locked: u32 = 1;
        const contended: u32 = 2;

        /// Try to take the lock without blocking; returns whether it was acquired.
        pub fn tryLock(m: *Mutex) bool {
            return m.state.cmpxchgStrong(unlocked, locked, .acquire, .monotonic) == null;
        }

        /// Acquire the lock, blocking in the kernel while it is contended.
        pub fn lock(m: *Mutex) void {
            if (m.state.cmpxchgStrong(unlocked, locked, .acquire, .monotonic) != null) m.lockSlow();
        }

        fn lockSlow(m: *Mutex) void {
            @branchHint(.cold);
            // Mark the lock contended and take it as soon as it falls unlocked; park on
            // the futex while it stays contended. Marking contended may cause a spurious
            // wake on unlock (harmless), never a missed one.
            while (m.state.swap(contended, .acquire) != unlocked) {
                Futex.wait(&m.state, contended);
            }
        }

        /// Release the lock; wake one waiter if the lock was contended.
        pub fn unlock(m: *Mutex) void {
            if (m.state.swap(unlocked, .release) == contended) Futex.wake(&m.state, 1);
        }
    };

    /// A condition variable, `std.Thread.Condition`-shaped. Spurious wakeups are
    /// allowed — always wait in a predicate loop with the mutex held. Built on a futex
    /// sequence counter: a waiter samples the seq, drops the mutex, and parks until the
    /// seq changes (a signal that races the unlock bumps the seq, so it is not missed).
    pub const Condition = struct {
        seq: std.atomic.Value(u32) = std.atomic.Value(u32).init(0),

        /// Atomically release `mutex` and block until signalled, then re-acquire it.
        pub fn wait(c: *Condition, mutex: *Mutex) void {
            const seq = c.seq.load(.acquire);
            mutex.unlock();
            Futex.wait(&c.seq, seq);
            mutex.lock();
        }

        /// As `wait`, but returns `error.Timeout` if `timeout_ns` elapses first. The
        /// mutex is re-acquired either way.
        pub fn timedWait(c: *Condition, mutex: *Mutex, timeout_ns: u64) error{Timeout}!void {
            const seq = c.seq.load(.acquire);
            mutex.unlock();
            const timed_out = if (Futex.timedWait(&c.seq, seq, timeout_ns)) |_| false else |_| true;
            mutex.lock();
            if (timed_out) return error.Timeout;
        }

        /// Wake one waiter.
        pub fn signal(c: *Condition) void {
            _ = c.seq.fetchAdd(1, .release);
            Futex.wake(&c.seq, 1);
        }

        /// Wake all waiters.
        pub fn broadcast(c: *Condition) void {
            _ = c.seq.fetchAdd(1, .release);
            Futex.wake(&c.seq, std.math.maxInt(u32));
        }
    };

    /// A counting semaphore, `std.Thread.Semaphore`-shaped: a permit count guarded by a
    /// `Mutex` + `Condition`.
    pub const Semaphore = struct {
        mutex: Mutex = .{},
        cond: Condition = .{},
        permits: usize = 0,

        /// Take a permit, blocking until one is available.
        pub fn wait(s: *Semaphore) void {
            s.mutex.lock();
            defer s.mutex.unlock();
            while (s.permits == 0) s.cond.wait(&s.mutex);
            s.permits -= 1;
        }

        /// Return a permit and wake a waiter.
        pub fn post(s: *Semaphore) void {
            s.mutex.lock();
            defer s.mutex.unlock();
            s.permits += 1;
            s.cond.signal();
        }
    };
};

/// thread_spawn(entry, stack_top, arg, exit_endpoint) -> tid, or a wrapped error.
fn threadSpawn(entry: usize, stack_top: usize, arg: usize, exit_endpoint: ipc.Handle) usize {
    return sc.systemCall4(.thread_spawn, entry, stack_top, arg, exit_endpoint);
}

/// The kernel returns a real (small) task id on success and a wrapped `-1` on failure;
/// no valid task id ever exceeds a u32.
inline fn threadSpawnFailed(ret: usize) bool {
    return ret > std.math.maxInt(u32);
}

/// End the calling thread. Never returns.
fn exitThread() noreturn {
    _ = sc.systemCall0(.thread_exit);
    unreachable;
}

/// futex_wait(addr, expect, timeout_ns) -> status (abi.futex_*).
fn futexWait(addr: usize, expect: u32, timeout_ns: u64) usize {
    return sc.systemCall3(.futex_wait, addr, expect, timeout_ns);
}

/// futex_wake(addr, count) -> number woken.
fn futexWake(addr: usize, count: u32) usize {
    return sc.systemCall2(.futex_wake, addr, count);
}