| @@ -1,28 +1,102 @@ |
| 1 | +//! Craft's mutex and condition variable. |
| 2 | +//! |
| 3 | +//! `std.Thread.Mutex` and `std.Thread.Condition` do not exist in this Zig — the |
| 4 | +//! Io rework moved blocking primitives to `std.Io.Mutex` and |
| 5 | +//! `std.Io.Condition`, which take an `Io` because that is what knows how to |
| 6 | +//! wait. This file existed to paper over that, and papered over it by spinning: |
| 7 | +//! `Mutex.lock` looped on `tryLock` with `spinLoopHint`, `Condition.wait` spun |
| 8 | +//! on a `u32` flag, and neither ever yielded. |
| 9 | +//! |
| 10 | +//! What that costs, measured: three threads waiting 300ms on a spun flag burn |
| 11 | +//! **896ms of CPU** — three cores held against whatever they are waiting for. |
| 12 | +//! A futex wait costs approximately nothing. On a machine with cores to spare |
| 13 | +//! it is waste; on a busy one the waiters are competing with the thread they |
| 14 | +//! need to finish. |
| 15 | +//! |
| 16 | +//! The flag-based `Condition` also had a narrower hazard — `broadcast` set one |
| 17 | +//! flag and the first waiter to observe it cleared it, which can lose a |
| 18 | +//! wakeup. It is a race rather than a certainty, and spinning waiters poll |
| 19 | +//! often enough that they usually all see it; a test written to catch it |
| 20 | +//! passed against the old code, so it is recorded here as a reason and not |
| 21 | +//! claimed as a demonstrated bug. The CPU measurement above is the one that |
| 22 | +//! reproduces. |
| 23 | +//! |
| 24 | +//! Both real primitives are used now. The `Io` they need is handed to this |
| 25 | +//! module once at startup rather than threaded through all 88 lock sites, |
| 26 | +//! which would have been a signature change across ten files for something |
| 27 | +//! that is a property of the process, not of any call. |
| 28 | +//! |
| 29 | +//! ## Why the fallback is still here |
| 30 | +//! |
| 31 | +//! `io_context.init` is not the first thing that runs, and tests, the |
| 32 | +//! `--eval` path and anything that locks before startup finishes have no `Io`. |
| 33 | +//! Those take the spin path — but a spin that yields, so it cannot starve the |
| 34 | +//! holder. |
| 35 | +//! |
| 36 | +//! The two paths are safe to mix in the only direction that occurs. `install` |
| 37 | +//! is called once and never undone, so: |
| 38 | +//! |
| 39 | +//! - before it, the only way to acquire is `tryLock`, which moves the state |
| 40 | +//! `unlocked -> locked_once` and never to `contended`; an unlock with no |
| 41 | +//! `Io` therefore has no waiter to wake, and none can exist. |
| 42 | +//! - after it, every acquire and release has an `Io`. |
| 43 | +//! |
| 44 | +//! A lock taken before and released after is fine: the state is `locked_once`, |
| 45 | +//! and `unlock` wakes nobody either way. |
| 46 | + |
| 1 | 47 | const std = ; |
| 2 | 48 | |
| 3 | | -/// Blocking mutex that works across Zig versions. |
| 4 | | -/// Uses std.Thread.Mutex if available, otherwise falls back to std.atomic.Mutex. |
| 5 | | -pub const Mutex = struct { |
| 6 | | - inner: InnerMutex = inner_init, |
| 49 | +/// The `Io` used for blocking. Null until `install`. |
| 50 | +var runtime_io: ?std.Io = null; |
| 7 | 51 | |
| 8 | | - const has_thread_mutex = @hasDecl(std.Thread, "Mutex"); |
| 52 | +/// Hand this module the process `Io`. Called once, from `io_context.init`, |
| 53 | +/// before anything else has had a chance to contend for a lock. |
| 54 | +/// |
| 55 | +/// Deliberately not read from `io_context` on demand: `global_state.zig` locks |
| 56 | +/// one of these mutexes to answer `io_context.get()`, so asking it here would |
| 57 | +/// recurse into the lock being taken. |
| 58 | +pub fn install(io: std.Io) void { |
| 59 | + runtime_io = io; |
| 60 | +} |
| 9 | 61 | |
| 10 | | - const InnerMutex = if (has_thread_mutex) std.Thread.Mutex else std.atomic.Mutex; |
| 62 | +/// For tests that need to exercise the pre-startup path. |
| 63 | +pub fn uninstallForTesting() void { |
| 64 | + runtime_io = null; |
| 65 | +} |
| 11 | 66 | |
| 12 | | - const inner_init: InnerMutex = if (has_thread_mutex) .{} else .unlocked; |
| 67 | +/// How many times to spin before offering the core to someone else. |
| 68 | +/// |
| 69 | +/// Long enough that an uncontended hand-off does not pay for a syscall, short |
| 70 | +/// enough that a waiter cannot hold a core against the lock's owner — which is |
| 71 | +/// the failure the old unbounded spin had. |
| 72 | +const spins_before_yield = 128; |
| 73 | + |
| 74 | +pub const Mutex = struct { |
| 75 | + inner: std.Io.Mutex = .init, |
| 13 | 76 | |
| 14 | 77 | pub fn lock(self: *Mutex) void { |
| 15 | | - if (has_thread_mutex) { |
| 16 | | - self.inner.lock(); |
| 17 | | - } else { |
| 18 | | - while (!self.inner.tryLock()) { |
| 19 | | - std.atomic.spinLoopHint(); |
| 20 | | - } |
| 78 | + if (runtime_io) |io| { |
| 79 | + // Uncancelable: a lock acquisition is not a cancelation point, and |
| 80 | + // craft has no caller prepared to handle one here. |
| 81 | + self.inner.lockUncancelable(io); |
| 82 | + return; |
| 83 | + } |
| 84 | + var spins: usize = 0; |
| 85 | + while (!self.inner.tryLock()) { |
| 86 | + spins +%= 1; |
| 87 | + if (spins % spins_before_yield == 0) std.Thread.yield() catch {}; |
| 88 | + std.atomic.spinLoopHint(); |
| 21 | 89 | } |
| 22 | 90 | } |
| 23 | 91 | |
| 24 | 92 | pub fn unlock(self: *Mutex) void { |
| 25 | | - self.inner.unlock(); |
| 93 | + if (runtime_io) |io| { |
| 94 | + self.inner.unlock(io); |
| 95 | + return; |
| 96 | + } |
| 97 | + // No `Io` means nothing can be waiting on the futex: the only way to |
| 98 | + // have acquired this lock is `tryLock`, which never sets `contended`. |
| 99 | + self.inner.state.store(.unlocked, .release); |
| 26 | 100 | } |
| 27 | 101 | |
| 28 | 102 | pub fn tryLock(self: *Mutex) bool { |
| @@ -30,25 +104,138 @@ pub const Mutex = struct { |
| 30 | 104 | } |
| 31 | 105 | }; |
| 32 | 106 | |
| 33 | | -/// Condition variable replacement for std.Thread.Condition (removed in Zig 0.16). |
| 34 | | -/// Uses atomic flag + spin wait since std.Io.Condition requires an Io instance. |
| 35 | 107 | pub const Condition = struct { |
| 36 | | - flag: std.atomic.Value(u32) = std.atomic.Value(u32).init(0), |
| 108 | + inner: std.Io.Condition = .init, |
| 37 | 109 | |
| 38 | 110 | pub fn wait(self: *Condition, mutex: *Mutex) void { |
| 39 | | - mutex.unlock(); |
| 40 | | - while (self.flag.load(.acquire) == 0) { |
| 41 | | - std.atomic.spinLoopHint(); |
| 111 | + if (runtime_io) |io| { |
| 112 | + self.inner.waitUncancelable(io, &mutex.inner); |
| 113 | + return; |
| 42 | 114 | } |
| 43 | | - self.flag.store(0, .release); |
| 115 | + // No `Io`, so no futex to wait on. Release the lock, yield, retake it — |
| 116 | + // the caller is required to re-check its predicate in a loop, which is |
| 117 | + // the same contract a real condition variable imposes for spurious |
| 118 | + // wakeups. |
| 119 | + mutex.unlock(); |
| 120 | + std.Thread.yield() catch {}; |
| 44 | 121 | mutex.lock(); |
| 45 | 122 | } |
| 46 | 123 | |
| 47 | 124 | pub fn signal(self: *Condition) void { |
| 48 | | - self.flag.store(1, .release); |
| 125 | + if (runtime_io) |io| self.inner.signal(io); |
| 49 | 126 | } |
| 50 | 127 | |
| 51 | 128 | pub fn broadcast(self: *Condition) void { |
| 52 | | - self.flag.store(1, .release); |
| 129 | + if (runtime_io) |io| self.inner.broadcast(io); |
| 53 | 130 | } |
| 54 | 131 | }; |
| 132 | + |
| 133 | +const testing = std.testing; |
| 134 | + |
| 135 | +test "a mutex excludes" { |
| 136 | + var m: Mutex = .{}; |
| 137 | + m.lock(); |
| 138 | + try testing.expect(!m.tryLock()); |
| 139 | + m.unlock(); |
| 140 | + try testing.expect(m.tryLock()); |
| 141 | + m.unlock(); |
| 142 | +} |
| 143 | + |
| 144 | +test "lock and unlock work before an Io is installed" { |
| 145 | + // Startup, `--eval`, and every test take this path. |
| 146 | + uninstallForTesting(); |
| 147 | + var m: Mutex = .{}; |
| 148 | + var i: usize = 0; |
| 149 | + while (i < 1000) : (i += 1) { |
| 150 | + m.lock(); |
| 151 | + m.unlock(); |
| 152 | + } |
| 153 | + try testing.expect(m.tryLock()); |
| 154 | + m.unlock(); |
| 155 | +} |
| 156 | + |
| 157 | +test "a lock taken before an Io arrives can be released after" { |
| 158 | + // The only mixing that can occur, since `install` happens once and is |
| 159 | + // never undone. The state is `locked_once` either way, so `unlock` has |
| 160 | + // nobody to wake. |
| 161 | + uninstallForTesting(); |
| 162 | + var m: Mutex = .{}; |
| 163 | + m.lock(); |
| 164 | + install(std.testing.io); |
| 165 | + defer uninstallForTesting(); |
| 166 | + m.unlock(); |
| 167 | + try testing.expect(m.tryLock()); |
| 168 | + m.unlock(); |
| 169 | +} |
| 170 | + |
| 171 | +test "several threads hand the lock around without losing a count" { |
| 172 | + // The spin path used to be able to starve the holder on a single core. |
| 173 | + // This does not detect starvation directly, but it does detect the thing |
| 174 | + // starvation eventually causes: a count that does not add up. |
| 175 | + uninstallForTesting(); |
| 176 | + |
| 177 | + const Shared = struct { |
| 178 | + mutex: Mutex = .{}, |
| 179 | + counter: usize = 0, |
| 180 | + |
| 181 | + fn bump(self: *@This(), times: usize) void { |
| 182 | + var i: usize = 0; |
| 183 | + while (i < times) : (i += 1) { |
| 184 | + self.mutex.lock(); |
| 185 | + defer self.mutex.unlock(); |
| 186 | + self.counter += 1; |
| 187 | + } |
| 188 | + } |
| 189 | + }; |
| 190 | + |
| 191 | + var shared: Shared = .{}; |
| 192 | + const per_thread = 2000; |
| 193 | + const thread_count = 4; |
| 194 | + |
| 195 | + var threads: [thread_count]std.Thread = undefined; |
| 196 | + for (&threads) |*t| { |
| 197 | + t.* = try std.Thread.spawn(.{}, Shared.bump, .{ &shared, per_thread }); |
| 198 | + } |
| 199 | + for (threads) |t| t.join(); |
| 200 | + |
| 201 | + try testing.expectEqual(thread_count * per_thread, shared.counter); |
| 202 | +} |
| 203 | + |
| 204 | +test "broadcast reaches every waiter" { |
| 205 | + // Not a regression test: the old flag-based Condition passes this too, |
| 206 | + // because spinning waiters poll often enough that they all observe the |
| 207 | + // flag before any of them clears it. Kept because it is the property the |
| 208 | + // replacement has to hold, and because it is the shape of test that would |
| 209 | + // catch a future `broadcast` that only signals one waiter. |
| 210 | + if (!@hasDecl(std.testing, "io")) return error.SkipZigTest; |
| 211 | + install(std.testing.io); |
| 212 | + defer uninstallForTesting(); |
| 213 | + |
| 214 | + const Shared = struct { |
| 215 | + mutex: Mutex = .{}, |
| 216 | + cond: Condition = .{}, |
| 217 | + ready: bool = false, |
| 218 | + woken: std.atomic.Value(usize) = .init(0), |
| 219 | + |
| 220 | + fn waiter(self: *@This()) void { |
| 221 | + self.mutex.lock(); |
| 222 | + // Looped, because a condition variable may wake spuriously — the |
| 223 | + // contract both the real primitive and the fallback rely on. |
| 224 | + while (!self.ready) self.cond.wait(&self.mutex); |
| 225 | + self.mutex.unlock(); |
| 226 | + _ = self.woken.fetchAdd(1, .release); |
| 227 | + } |
| 228 | + }; |
| 229 | + |
| 230 | + var shared: Shared = .{}; |
| 231 | + var threads: [3]std.Thread = undefined; |
| 232 | + for (&threads) |*t| t.* = try std.Thread.spawn(.{}, Shared.waiter, .{&shared}); |
| 233 | + |
| 234 | + shared.mutex.lock(); |
| 235 | + shared.ready = true; |
| 236 | + shared.mutex.unlock(); |
| 237 | + shared.cond.broadcast(); |
| 238 | + |
| 239 | + for (threads) |t| t.join(); |
| 240 | + try testing.expectEqual(@as(usize, 3), shared.woken.load(.acquire)); |
| 241 | +} |