also looking at this
fix(sync): stop craft's locks busy-waiting
#80
3 files
+229
-23
| @@ -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 | } | |