//! `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); }