ke: add turnstile and mutex

build.zig.zon

@@ -13,8 +13,8 @@
             .hash = "N-V-__8AAAAAjAwsopZIlbp4U6NN1Eh573uVdLTwl1mkFUug",
         },
         .zonfig = .{
-            .url = "git+https://github.com/rdmsr/zonfig?ref=HEAD#e181738efb500d2e902743a0b60d50d17a006259",
+            .url = "git+https://github.com/rdmsr/zonfig?ref=HEAD#ee612592ba79d99da3ddf715ffec1414f72609ca",
-            .hash = "zonfig-0.1.0-V0D8-XI4AQArg0SFDWoYjfJ_byhget22QM5jPPuo9mRi",
+            .hash = "zonfig-0.1.0-V0D8-WQ4AQAW5RNFJPxDNIpI7KB7eGPx0MlzGDM6HFbK",
         },
     },
 

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) {
+    if (false) {
-        const sp = mm.heap.alloc(r.kib(16)) catch @panic("oom");
+        if (boot_info.framebuffer != null) {
-        thread0.init(@intFromPtr(sp), r.kib(16), ex.fireworks.start, boot_info);
+            const sp = mm.heap.alloc(r.kib(16)) catch @panic("oom");
-        ke.sched.enqueue(&thread0);
+            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();
     }

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);
+    }
+};

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");

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) {
+    if (curtd.base_priority_class() == .Batch) {
-        // Update interactivity metrics.
         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));
+        @intCast(ke.Thread.Priority.low_interactive + ((interactivity_threshold - score) / 2))
-        return;
+    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.
+        // nice offset: negative nice increases priority value, positive decreases it.
-    // cpu_pri_off is the fraction of the history window spent running,
+        // Clamped to half the range to avoid pushing priority out of bounds before clamping.
-    // scaled to [0, batch_range). A thread that spent all its time running
+        const nice_off = std.math.clamp(
-    // gets cpu_pri_off = batch_range - 1 (lowest priority).
+            -@as(i32, td.nice),
-    // A thread that barely ran gets cpu_pri_off = 0 (highest batch priority).
+            -@as(i32, cpu_range / 2),
-    // This is a pretty straightforward formula, which /should/ be good enough.
+            @as(i32, cpu_range / 2),
-    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.
+        const raw = @as(i32, ke.Thread.Priority.low_batch) +
-    // Clamped to half the range to avoid pushing priority out of bounds before clamping.
+            @as(i32, cpu_pri_off) +
-    const nice_off = std.math.clamp(
+            nice_off;
-        -@as(i32, td.nice),
-        -@as(i32, cpu_range / 2),
-        @as(i32, cpu_range / 2),
-    );
 
-    const raw = @as(i32, ke.Thread.Priority.low_batch) +
+        break :blk @intCast(std.math.clamp(
-        @as(i32, cpu_pri_off) +
+            raw,
-        nice_off;
+            ke.Thread.Priority.low_batch,
+            ke.Thread.Priority.high_batch,
+        ));
+    };
 
-    td.priority = @intCast(std.math.clamp(
+    td.base_priority = base;
-        raw,
+    td.priority = td.effective_priority();
-        ke.Thread.Priority.low_batch,
-        ke.Thread.Priority.high_batch,
-    ));
 }
 
 // Return whether or not a thread with given priority and interactivity should get preempted by `td`.

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;
     }

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();
+}