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()); w.thread.lock.acquire_no_ipl(); best = @max(best orelse 0, w.thread.priority); w.thread.lock.release_no_ipl(); } } 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 on entry and is held again on return. /// - If another chain bucket cannot be trylocked, we drop the current thread /// lock and `root`, then reacquire `root` and restart from `curtd`. /// - Any other chain bucket is taken with trylock. 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)]; if (bucket != root) { if (!bucket.lock.try_acquire_no_ipl()) { std.debug.assert(held == null); // Drop everything and try again. This avoids a livelock // between two PI walks holding different roots and trying to // acquire each other. thread.lock.release_no_ipl(); root.lock.release_no_ipl(); root.lock.acquire_no_ipl(); curtd.lock.acquire_no_ipl(); thread = curtd; donate = curtd.priority; continue; } held = bucket; } std.debug.assert(thread.waiting_on == obj); 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; } // The next hop blocks owner, so lend the owner's priority // down the chain. donate = owner.priority; // 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; if (turn.owner) |cur| { std.debug.assert(cur == owner); } else { // Owner might be null because of a partial wakeup // where we cleared it, restore it here. turn.owner = owner; reboost(turn, owner); } } 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()); const waiter = td.turnstile_waiter.?; waiter.link.remove(); 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; } td.turnstile_waiter = null; ts.waiters -= 1; td.lock.acquire_no_ipl(); td.waiting_on = null; td.lock.release_no_ipl(); waiter.event.signal(); }