From 68e0f7f8f7244e77112f6c6c344366207e2be11c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Lo=C3=AFc=20Gr=C3=A9goire?= Date: Fri, 12 Jun 2026 15:53:10 -0400 Subject: [PATCH] ke: add turnstile and mutex --- build.zig.zon | 4 +- src/ex/init.zig | 100 ++++++++++- src/kern/mutex.zig | 69 ++++++++ src/kern/private.zig | 3 + src/kern/sched.zig | 67 ++++---- src/kern/thread.zig | 31 +++- src/kern/turnstile.zig | 365 +++++++++++++++++++++++++++++++++++++++++ 7 files changed, 596 insertions(+), 43 deletions(-) create mode 100644 src/kern/mutex.zig create mode 100644 src/kern/turnstile.zig diff --git a/build.zig.zon b/build.zig.zon index c183c7c..829660b 100644 --- a/build.zig.zon +++ b/build.zig.zon @@ -13,8 +13,8 @@ .hash = "N-V-__8AAAAAjAwsopZIlbp4U6NN1Eh573uVdLTwl1mkFUug", }, .zonfig = .{ - .url = "git+https://github.com/rdmsr/zonfig?ref=HEAD#e181738efb500d2e902743a0b60d50d17a006259", - .hash = "zonfig-0.1.0-V0D8-XI4AQArg0SFDWoYjfJ_byhget22QM5jPPuo9mRi", + .url = "git+https://github.com/rdmsr/zonfig?ref=HEAD#ee612592ba79d99da3ddf715ffec1414f72609ca", + .hash = "zonfig-0.1.0-V0D8-WQ4AQAW5RNFJPxDNIpI7KB7eGPx0MlzGDM6HFbK", }, }, diff --git a/src/ex/init.zig b/src/ex/init.zig index edbc2b6..fd0dc45 100644 --- a/src/ex/init.zig +++ b/src/ex/init.zig @@ -5,22 +5,114 @@ const ke = r.ke; const mm = r.mm; const ex = r.ex; const amd64 = @import("arch"); +const turnstile = ke.private.turnstile; var thread0: ke.Thread = undefined; +var mtx: ke.private.mutex.Mutex = .init(); +var shared_counter: usize = 0; + +fn mutex_test(_: ?*anyopaque) void { + while (true) { + mtx.acquire(); + shared_counter += 1; + mtx.release(); + } +} + +var starts: [10]ke.Event = undefined; +var done: ke.Event = undefined; +var done_count: std.atomic.Value(usize) = .init(0); + +fn worker(arg: ?*anyopaque) void { + const idx: usize = @intFromPtr(arg); + while (true) { + _ = ke.wait.wait_one(&starts[idx].hdr, null) catch {}; + if (done_count.fetchAdd(1, .acq_rel) == 10 - 1) { + done_count.store(0, .release); + done.signal(); + } + } +} + +fn controller(_: ?*anyopaque) void { + var i: usize = 0; + while (true) { + for (&starts) |*s| s.signal(); + _ = ke.wait.wait_one(&done.hdr, null) catch {}; + + i += 1; + } +} + +fn make_thread(entry: *const fn (?*anyopaque) void, arg: ?*anyopaque) *ke.Thread { + var ret: *ke.Thread = mm.zone.gpa.create(ke.Thread) catch @panic("oom"); + const stack = mm.heap.alloc(r.kib(64)) catch @panic("wtf"); + + std.log.info("Stack = {*}", .{stack}); + + ret.init(@intFromPtr(stack), r.kib(64), entry, arg); + + ret.priority = ke.Thread.Priority.default; + ret.base_priority = ret.priority; + ret.turnstile = mm.zone.gpa.create(ke.private.turnstile.Turnstile) catch @panic("oom"); + ret.turnstile.* = .{ + .link = undefined, + .next_free = null, + .obj = undefined, + .owner = null, + .boost_link = undefined, + .donated = null, + .queues = undefined, + .waiters = 0, + }; + + ret.turnstile.queues[0].init(); + ret.turnstile.queues[1].init(); + ret.turnstile_waiter = null; + + return ret; +} + pub fn init(boot_info: *pl.BootInfo) void { mm.init(boot_info); pl.late_init(boot_info); mm.late_init(); ke.sched.late_init(); + ke.private.turnstile.init_turnstiles(); - if (boot_info.framebuffer != null) { - const sp = mm.heap.alloc(r.kib(16)) catch @panic("oom"); - thread0.init(@intFromPtr(sp), r.kib(16), ex.fireworks.start, boot_info); - ke.sched.enqueue(&thread0); + if (false) { + if (boot_info.framebuffer != null) { + const sp = mm.heap.alloc(r.kib(16)) catch @panic("oom"); + thread0.init(@intFromPtr(sp), r.kib(16), ex.fireworks.start, boot_info); + ke.sched.enqueue(&thread0); + } + + while (true) {} } + const ipl = ke.ipl.raise(.Dispatch); + + if (false) { + for (&starts) |*s| s.init(.Synchronization); + done.init(.Synchronization); + + for (0..10) |i| { + const t = make_thread(worker, @ptrFromInt(i)); + ke.sched.enqueue(t); + } + const t = make_thread(controller, null); + ke.sched.enqueue(t); + } + + for (0..10) |_| { + const t = make_thread(mutex_test, null); + ke.sched.enqueue(t); + } + + ke.ipl.lower(ipl); + while (true) { std.atomic.spinLoopHint(); } diff --git a/src/kern/mutex.zig b/src/kern/mutex.zig new file mode 100644 index 0000000..a71b9b5 --- /dev/null +++ b/src/kern/mutex.zig @@ -0,0 +1,69 @@ +const std = @import("std"); +const ke = @import("root").ke; +const ki = ke.private; + +pub const Mutex = struct { + /// The thread currently holding the mutex, null if unlocked. + owner: std.atomic.Value(?*ke.Thread), + + pub fn init() Mutex { + return .{ .owner = .init(null) }; + } + + pub fn acquire(m: *Mutex) void { + const ipl = ke.ipl.raise(.Dispatch); + const curtd = ki.sched.percpu.local().current_thread.?; + + // Fast path: try to acquire without a turnstile. + if (m.owner.cmpxchgStrong(null, curtd, .acquire, .monotonic) == null) { + ke.ipl.lower(ipl); + return; + } + + // Slow path: contended, block on a turnstile. + while (true) { + const ts = ki.turnstile.lookup(m); + const owner = m.owner.load(.monotonic); + + if (owner == null) { + // Lock was released between lookup and here. + ki.turnstile.exit(m); + if (m.owner.cmpxchgStrong(null, curtd, .acquire, .monotonic) == null) { + break; + } + continue; + } + + // Block until woken. + ki.turnstile.block(ts, m, owner.?, .Shared); + + // Re-try acquisition after wakeup. + if (m.owner.cmpxchgStrong(null, curtd, .acquire, .monotonic) == null) { + break; + } + } + + ke.ipl.lower(ipl); + } + + pub fn release(m: *Mutex) void { + const ipl = ke.ipl.raise(.Dispatch); + + const ts = ki.turnstile.lookup(m); + + m.owner.store(null, .release); + + if (ts == null) { + ki.turnstile.exit(m); + ke.ipl.lower(ipl); + return; + } + + // Note: wake up all waiters. This so-called "lock barging" (name from WTF::ParkingLot) + // has been shown to be better because this avoids lock convoys, see this mysterious 70s paper: + // https://dl.acm.org/doi/pdf/10.1145/850657.850659 + ki.turnstile.wakeup(ts.?, .Exlusive, ts.?.waiters, null); + + ke.ipl.lower(ipl); + } +}; diff --git a/src/kern/private.zig b/src/kern/private.zig index 40a5b94..9dd65f7 100644 --- a/src/kern/private.zig +++ b/src/kern/private.zig @@ -24,3 +24,6 @@ pub const sched = @import("sched.zig"); pub const wait = @import("wait.zig"); pub const timer = @import("timer.zig"); pub const log_ring = @import("log_ring.zig"); +pub const event = @import("event.zig"); +pub const turnstile = @import("turnstile.zig"); +pub const mutex = @import("mutex.zig"); diff --git a/src/kern/sched.zig b/src/kern/sched.zig index 3771293..d7f02f7 100644 --- a/src/kern/sched.zig +++ b/src/kern/sched.zig @@ -207,8 +207,7 @@ pub fn clock(_: ?*anyopaque) void { cpu.insidx = (cpu.insidx + 1) % runqueues_n; } - if (curtd.priority_class() == .Batch) { - // Update interactivity metrics. + if (curtd.base_priority_class() == .Batch) { curtd.run_time += config.sched_timeslice; clamp_time(curtd); @@ -265,7 +264,7 @@ pub fn unblock(td: *ke.Thread) void { td.sleep_time = (ke.time.read_time() - td.sleep_start) / std.time.ns_per_ms; - if (td.sleep_time >= config.sched_timeslice and td.priority_class() == .Batch) { + if (td.sleep_time >= config.sched_timeslice and td.base_priority_class() == .Batch) { // If we have slept for more than a tick, update interactivity. clamp_time(td); recompute_priority(td); @@ -554,7 +553,7 @@ fn remove_from_queue(cpu: *PerCpu, td: *ke.Thread) void { std.debug.assert(cpu.queues_lock.is_locked()); // Decrease load. - cpu.load.fetchSub(1, .monotonic); + _ = cpu.load.fetchSub(1, .monotonic); // Remove the thread from its queue. td.runq_link.remove(); @@ -627,7 +626,7 @@ fn interactive_score(td: *ke.Thread) usize { return score; } -// Recompute the priority for a thread. +// Recompute the base priority for a thread from interactivity and nice. fn recompute_priority(td: *ke.Thread) void { // Priority is computed only for time-shared (batch) thread based on @@ -637,41 +636,43 @@ fn recompute_priority(td: *ke.Thread) void { // priority is calculated based on recent CPU usage and nice. const score = interactive_score(td); - if (score < interactivity_threshold) { + const base: u8 = if (score < interactivity_threshold) // Choose a priority based on score, the lower the score, // the higher the priority it will be. // This is a simple formula I came up with that is probably good enough. - td.priority = @intCast(ke.Thread.Priority.low_interactive + ((interactivity_threshold - score) / 2)); - return; - } + @intCast(ke.Thread.Priority.low_interactive + ((interactivity_threshold - score) / 2)) + else blk: { + // Thread is not interactive, compute priority from recent CPU usage. + // cpu_pri_off is the fraction of the history window spent running, + // scaled to [0, batch_range). A thread that spent all its time running + // gets cpu_pri_off = batch_range - 1 (lowest priority). + // A thread that barely ran gets cpu_pri_off = 0 (highest batch priority). + // This is a pretty straightforward formula, which /should/ be good enough. + const cpu_range = ke.Thread.Priority.batch_range; + const window = if (td.run_time + td.sleep_time == 0) @as(u64, 1) else td.run_time + td.sleep_time; + const cpu_pri_off: u8 = @intCast((td.run_time * (cpu_range - 1)) / window); - // Thread is not interactive, compute priority from recent CPU usage. - // cpu_pri_off is the fraction of the history window spent running, - // scaled to [0, batch_range). A thread that spent all its time running - // gets cpu_pri_off = batch_range - 1 (lowest priority). - // A thread that barely ran gets cpu_pri_off = 0 (highest batch priority). - // This is a pretty straightforward formula, which /should/ be good enough. - const cpu_range = ke.Thread.Priority.batch_range; - const window = if (td.run_time + td.sleep_time == 0) @as(u64, 1) else td.run_time + td.sleep_time; - const cpu_pri_off: u8 = @intCast((td.run_time * (cpu_range - 1)) / window); + // nice offset: negative nice increases priority value, positive decreases it. + // Clamped to half the range to avoid pushing priority out of bounds before clamping. + const nice_off = std.math.clamp( + -@as(i32, td.nice), + -@as(i32, cpu_range / 2), + @as(i32, cpu_range / 2), + ); - // nice offset: negative nice increases priority value, positive decreases it. - // Clamped to half the range to avoid pushing priority out of bounds before clamping. - const nice_off = std.math.clamp( - -@as(i32, td.nice), - -@as(i32, cpu_range / 2), - @as(i32, cpu_range / 2), - ); + const raw = @as(i32, ke.Thread.Priority.low_batch) + + @as(i32, cpu_pri_off) + + nice_off; - const raw = @as(i32, ke.Thread.Priority.low_batch) + - @as(i32, cpu_pri_off) + - nice_off; + break :blk @intCast(std.math.clamp( + raw, + ke.Thread.Priority.low_batch, + ke.Thread.Priority.high_batch, + )); + }; - td.priority = @intCast(std.math.clamp( - raw, - ke.Thread.Priority.low_batch, - ke.Thread.Priority.high_batch, - )); + td.base_priority = base; + td.priority = td.effective_priority(); } // Return whether or not a thread with given priority and interactivity should get preempted by `td`. diff --git a/src/kern/thread.zig b/src/kern/thread.zig index aadb947..063686d 100644 --- a/src/kern/thread.zig +++ b/src/kern/thread.zig @@ -66,18 +66,18 @@ pub const Thread = struct { lock: ke.SpinLock, /// Niceness value. nice: i8, - /// Effective priority value of the thread. + /// Effective priority value of the thread: `max(base_priority, inherited_prio)`. priority: u8, /// Base priority value of the thread. base_priority: u8, + /// Priority inherited via priority donation. + inherited_prio: u8, /// When the thread started sleeping. sleep_start: u64, /// Ticks spent voluntarily sleeping recently. sleep_time: u64, /// Ticks spent running recently. run_time: u64, - /// Whether the thread is interactive or not. - interactive: bool, /// Whether the thread is pinned to this CPU. /// If it is pinned, then it can't be moved across another CPU. pinned: bool, @@ -101,6 +101,13 @@ pub const Thread = struct { /// Timer used for timeouts. timer: ke.Timer, + /// Turnstile. + turnstile: *ki.turnstile.Turnstile, + turnstile_waiter: ?*ki.turnstile.Waiter, + turnstiles_owned: rtl.List, + /// Object this thread is currently blocked on, or null. + waiting_on: ?*anyopaque, + /// Initialize a thread. /// - `stack`: Address of the **base** of the stack on which the initial context for the thread is built /// - `stack_size`: Size of the stack @@ -113,10 +120,10 @@ pub const Thread = struct { .nice = 0, .priority = Thread.Priority.default, .base_priority = Thread.Priority.default, + .inherited_prio = 0, .sleep_start = 0, .sleep_time = 0, .run_time = 0, - .interactive = false, .pinned = false, .state = .Ready, .runq_link = .{}, @@ -127,8 +134,14 @@ pub const Thread = struct { .wait_status = .init(.Satisfied), .waitblocks = undefined, .timer = undefined, + .turnstile = undefined, + .turnstile_waiter = null, + .turnstiles_owned = undefined, + .waiting_on = null, }; + thread.turnstiles_owned.init(); + thread.timer.init(); } @@ -136,6 +149,16 @@ pub const Thread = struct { return Priority.class_from_prio(self.priority); } + /// Class of the thread's *base* priority, ignoring any inherited boost. + pub fn base_priority_class(self: *Thread) Priority.Class { + return Priority.class_from_prio(self.base_priority); + } + + /// The priority the thread should run at. + pub fn effective_priority(self: *Thread) u8 { + return @max(self.base_priority, self.inherited_prio); + } + pub fn is_interactive(self: *Thread) bool { return self.priority >= Priority.low_interactive and self.priority <= Priority.high_interactive; } diff --git a/src/kern/turnstile.zig b/src/kern/turnstile.zig new file mode 100644 index 0000000..1dd47e3 --- /dev/null +++ b/src/kern/turnstile.zig @@ -0,0 +1,365 @@ +const ke = @import("root").ke; +const ki = ke.private; +const rtl = @import("rtl"); +const std = @import("std"); + +const num_chains = 128; +const hash_mask = num_chains - 1; + +pub const Waiter = struct { + link: rtl.List.Entry, + event: ke.Event, + thread: *ke.Thread, +}; + +pub const Turnstile = struct { + /// Linkage for hash chain. + link: rtl.List.Entry, + /// Linkage on freelist. + next_free: ?*Turnstile, + /// Waiter queues, indexed by the queue type specified in `Queue`. + queues: [2]rtl.List, + /// Number of waiters on this turnstile. + waiters: usize, + /// Object threads are waiting on. + obj: *anyopaque, + /// Thread currently holding the object, and the target of any donation. + owner: ?*ke.Thread, + /// Linkage into `owner`'s `turnstiles_owned` list. + boost_link: rtl.List.Entry, + /// Priority currently donated to `owner`, null means this turnstile is not + /// boosting anyone. + donated: ?u8, +}; + +const Chain = struct { + /// List of turnstiles on the chain. + list: rtl.List, + /// Lock for the chain. + lock: ke.SpinLock, +}; + +/// Which queue to block or wakeup threads on. +pub const Queue = enum(u1) { + /// Only one waiter will be woken up at once. + Exclusive = 0, + /// All waiters will be woken up. + Shared = 1, +}; + +var chains: [num_chains]Chain = undefined; + +fn hash_obj(obj: *const anyopaque) usize { + return (@intFromPtr(obj) >> 6) & hash_mask; +} + +/// Highest effective priority among all waiters on `ts`, or 0 if none. +/// The caller must hold `ts`'s chain lock so the waiter set is stable. +fn highest_waiter(ts: *Turnstile) ?u8 { + var best: ?u8 = null; + for (&ts.queues) |*q| { + var it = q.iterator(); + while (it.next()) : (it.advance()) { + const w: *Waiter = @fieldParentPtr("link", it.get()); + best = @max(best orelse 0, w.thread.priority); + } + } + return best; +} + +/// Compute the inherited priority of `td` from the turnstiles it +/// owns. `td.lock` must be held. +fn recompute_inherited(td: *ke.Thread) void { + var pri: u8 = 0; + var it = td.turnstiles_owned.iterator(); + while (it.next()) : (it.advance()) { + const ts: *Turnstile = @fieldParentPtr("boost_link", it.get()); + pri = @max(pri, ts.donated orelse 0); + } + td.inherited_prio = pri; +} + +fn update_prio(td: *ke.Thread) void { + const pri = td.effective_priority(); + if (pri != td.priority) ki.sched.update_priority_locked(td, pri); +} + +/// Donate priority `pri` to `to` through turnstile `ts`. The caller holds +/// `to.lock` and `ts`'s chain lock. Returns whether `to`'s effective priority +/// actually rose (false means the chain stops here). +fn donate_to(ts: *Turnstile, to: *ke.Thread, pri: u8) bool { + std.debug.assert(ts.owner == to); + + // A lend only ever raises. + if (ts.donated != null and pri <= ts.donated.?) return false; + + const before = to.priority; + + if (ts.donated == null) to.turnstiles_owned.insert_head(&ts.boost_link); + ts.donated = pri; + if (pri > to.inherited_prio) to.inherited_prio = pri; + + update_prio(to); + return to.priority > before; +} + +/// Detach `ts`'s donation from whatever thread it is currently boosting and +/// recompute that thread's priority (which may lower it). Caller holds `ts`'s +/// chain lock. +fn revoke(ts: *Turnstile) void { + // Not boosting anyone. + if (ts.donated == null) return; + + const to = ts.owner.?; + to.lock.acquire_no_ipl(); + ts.boost_link.remove(); + ts.donated = null; + // Recompute the priority as the floor may drop now. + recompute_inherited(to); + update_prio(to); + to.lock.release_no_ipl(); +} + +/// (Re)establish `ts`'s donation to `to` from its current set of waiters. +/// Caller holds `ts`'s chain lock. +fn reboost(ts: *Turnstile, to: *ke.Thread) void { + const pri = highest_waiter(ts); + if (pri == null) return; + + to.lock.acquire_no_ipl(); + _ = donate_to(ts, to, pri.?); + to.lock.release_no_ipl(); +} + +/// Lend `curtd`'s effective priority down the blocking chain, boosting each +/// successive owner until one is already at least as high. +/// +/// Locking: +/// - `root` is held by the caller throughout and never dropped here. +/// - Any other chain bucket is taken with trylock, failure means restart. At +/// most one such bucket is held at a time. +fn propagate(curtd: *ke.Thread, root: *Chain) void { + var donate = curtd.priority; + var thread = curtd; + var held: ?*Chain = null; + + while (true) { + // We hold `thread.lock` (curtd's on the first + // pass, the previous owner's after a hop) and `root`, `held` is null. + const obj = thread.waiting_on orelse break; + const bucket = &chains[hash_obj(obj)]; + + // Lock `bucket` and confirm `thread` didn't move before we got it. + // If either fails, restart below. + const restart = blk: { + if (bucket != root) { + if (!bucket.lock.try_acquire_no_ipl()) break :blk true; + held = bucket; + } + break :blk thread.waiting_on != obj; + }; + + // Try again. + if (restart) { + if (held) |h| h.lock.release_no_ipl(); + held = null; + if (thread != curtd) { + thread.lock.release_no_ipl(); + curtd.lock.acquire_no_ipl(); + thread = curtd; + } + donate = curtd.priority; + continue; + } + + const ts = thread.turnstile; + const owner = ts.owner orelse break; + if (owner == curtd) @panic("turnstile: cycle in blocking chain"); + + owner.lock.acquire_no_ipl(); + if (!donate_to(ts, owner, donate)) { + owner.lock.release_no_ipl(); + break; + } + + // Keep the owner locked, drop everything below it. + thread.lock.release_no_ipl(); + if (held) |h| h.lock.release_no_ipl(); + held = null; + thread = owner; + } + + // Leave the caller holding exactly curtd.lock and root. + if (held) |h| h.lock.release_no_ipl(); + if (thread != curtd) { + thread.lock.release_no_ipl(); + curtd.lock.acquire_no_ipl(); + } +} + +pub fn init_turnstiles() void { + for (&chains) |*chain| { + chain.list.init(); + chain.lock = .init(); + } +} + +/// Look up the turnstile for the specified object. +/// This acquires the turnstile chain lock and must be called at IPL dispatch. +/// Returns null if no turnstile is found. +pub fn lookup(obj: *const anyopaque) ?*Turnstile { + const chain = &chains[hash_obj(obj)]; + + chain.lock.acquire_no_ipl(); + + var it = chain.list.iterator(); + while (it.next()) : (it.advance()) { + const turnstile: *Turnstile = @fieldParentPtr("link", it.get()); + if (turnstile.obj == obj) { + return turnstile; + } + } + + return null; +} + +/// Drop the lock protecting the chain for obj. +pub fn exit(obj: *const anyopaque) void { + chains[hash_obj(obj)].lock.release_no_ipl(); +} + +/// Block the current thread on a synchronization object backed by a turnstile. +/// This must be called with the appropriate turnstile chain lock held, +/// and returns with the chain lock released. IPL is kept as before on return. +pub fn block(turnstile: ?*Turnstile, obj: *anyopaque, owner: *ke.Thread, queue: Queue) void { + const chain = &chains[hash_obj(obj)]; + const curtd = ki.sched.percpu.local().current_thread.?; + const ipl = ke.ipl.current(); + + var ts = turnstile; + const queue_idx = @intFromEnum(queue); + + std.debug.assert(chain.lock.is_locked()); + + if (ts) |turn| { + // Another thread already donated its turnstile, + // Put our turnstile on the freelist. + curtd.turnstile.next_free = turn.next_free; + turn.next_free = curtd.turnstile; + curtd.turnstile = turn; + } else { + // This is the first thread to block on this object. + // Lend its turnstile and add it to the hash chain. + ts = curtd.turnstile; + ts.?.obj = obj; + ts.?.owner = owner; + chain.list.insert_head(&ts.?.link); + } + + // Initialize a waiter struct for this thread. + var waiter: Waiter = .{ + .event = undefined, + .link = undefined, + .thread = curtd, + }; + + waiter.event.init(.Synchronization); + + curtd.turnstile_waiter = &waiter; + + ts.?.queues[queue_idx].insert_tail(&waiter.link); + ts.?.waiters += 1; + + // Record what we block on (the object) and lend our priority down the + // blocking chain. + curtd.lock.acquire_no_ipl(); + curtd.waiting_on = obj; + propagate(curtd, chain); + chain.lock.release_no_ipl(); + curtd.lock.release_no_ipl(); + + // Now block on the event. + _ = ke.ipl.lower(.Passive); + _ = ke.wait.wait_one(&waiter.event.hdr, null) catch { + _ = ke.ipl.raise(ipl); + + // If wait_one aborted (e.g. due to a timeout), remove the waiter and fix + // up the owner's inheritance. + chain.lock.acquire_no_ipl(); + + if (curtd.turnstile_waiter == &waiter) { + const active = ts.?; + const holder = active.owner; + revoke(active); + dequeue(active, curtd); + if (active.waiters > 0) { + if (holder) |o| reboost(active, o); + } + } + + chain.lock.release_no_ipl(); + return; + }; + _ = ke.ipl.raise(ipl); +} + +/// Wake whomever was waiting on the turnstile. +/// Do hand-off to new_owner if specified. +pub fn wakeup(turnstile: *Turnstile, queue: Queue, count: usize, new_owner: ?*ke.Thread) void { + const chain = &chains[hash_obj(turnstile.obj)]; + const queue_idx = @intFromEnum(queue); + + // The current owner is giving up the object, so it stops inheriting from + // this turnstile entirely. + revoke(turnstile); + + if (new_owner) |no| { + // Hand the object to a specific waiter, the rest now boost it. + dequeue(turnstile, no); + if (turnstile.waiters > 0) { + turnstile.owner = no; + reboost(turnstile, no); + } + } else { + for (0..count) |_| { + if (turnstile.queues[queue_idx].is_empty()) break; + const waiter: *Waiter = @fieldParentPtr("link", turnstile.queues[queue_idx].first()); + dequeue(turnstile, waiter.thread); + } + if (turnstile.waiters > 0) turnstile.owner = null; + } + + chain.lock.release_no_ipl(); +} + +/// Remove a single waiter from the turnstile and wake it. Caller holds the +/// chain lock. +fn dequeue(ts: *Turnstile, td: *ke.Thread) void { + std.debug.assert(td.turnstile == ts); + std.debug.assert(td.turnstile_waiter != null); + std.debug.assert(!ts.queues[0].is_empty() or !ts.queues[1].is_empty()); + + if (ts.next_free) |free| { + // Steal a turnstile from the freelist. + td.turnstile = free; + ts.next_free = free.next_free; + free.next_free = null; + } else { + // Last waiter, pull the turnstile off the chain and keep it for ourselves. + ts.link.remove(); + ts.queues[0].init(); + ts.queues[1].init(); + ts.next_free = null; + ts.owner = null; + ts.donated = null; + } + + const waiter = td.turnstile_waiter.?; + waiter.link.remove(); + td.turnstile_waiter = null; + ts.waiters -= 1; + + td.waiting_on = null; + + waiter.event.signal(); +}