authorgravatar for mlugg@mlugg.co.ukMatthew Lugg <mlugg@mlugg.co.uk> 2025-12-20 13:05:35+00:00
committergravatar for mlugg@mlugg.co.ukMatthew Lugg <mlugg@mlugg.co.uk> 2025-12-21 13:07:03+00:00
log6ed5b620507aa4589d6c6fe3a5e48e166d48b3d2
treeb1293f8558cd79837c8a7a699020f75af0e51977
parent330e295bc46a074cac4b9aa1f42f4678426936c5
signaturelock-open Commit is signed but in an unrecognized format.

std.Io: introduce futex primitives

Co-authored-by: Andrew Kelley <andrew@ziglang.org>

2 files changed, 305 insertions(+), 262 deletions(-)

lib/std/Io.zig+40
......@@ -654,6 +654,10 @@ pub const VTable = struct {
654654 /// that awaiting it will not block. Returns that index.
655655 select: *const fn (?*anyopaque, futures: []const *AnyFuture) Cancelable!usize,
656656
657 futexWait: *const fn (?*anyopaque, ptr: *const u32, expected: u32, Timeout) Cancelable!void,
658 futexWaitUncancelable: *const fn (?*anyopaque, ptr: *const u32, expected: u32) void,
659 futexWake: *const fn (?*anyopaque, ptr: *const u32, max_waiters: u32) void,
660
657661 mutexLock: *const fn (?*anyopaque, prev_state: Mutex.State, mutex: *Mutex) Cancelable!void,
658662 mutexLockUncancelable: *const fn (?*anyopaque, prev_state: Mutex.State, mutex: *Mutex) void,
659663 mutexUnlock: *const fn (?*anyopaque, prev_state: Mutex.State, mutex: *Mutex) void,
......@@ -1174,6 +1178,42 @@ pub fn Select(comptime U: type) type {
11741178 };
11751179}
11761180
1181/// Atomically checks if the value at `ptr` equals `expected`, and if so, blocks until either:
1182///
1183/// * a matching (same `ptr` argument) `futexWake` call occurs, or
1184/// * a spurious ("random") wakeup occurs.
1185///
1186/// Typically, `futexWake` should be called immediately after updating the value at `ptr.*`, to
1187/// unblock tasks using `futexWait` to wait for the value to change from what it previously was.
1188///
1189/// The caller is responsible for identifying spurious wakeups if necessary, typically by checking
1190/// the value at `ptr.*`.
1191///
1192/// Asserts that `T` is 4 bytes in length and has a well-defined layout with no padding bits.
1193pub fn futexWait(io: Io, comptime T: type, ptr: *align(@alignOf(u32)) const T, expected: T) Cancelable!void {
1194 return futexWaitTimeout(io, T, ptr, expected, .none);
1195}
1196/// Same as `futexWait`, except also unblocks if `timeout` expires. As with `futexWait`, spurious
1197/// wakeups are possible. It remains the caller's responsibility to differentiate between these
1198/// three possible wake-up reasons if necessary.
1199pub fn futexWaitTimeout(io: Io, comptime T: type, ptr: *align(@alignOf(u32)) const T, expected: T, timeout: Timeout) Cancelable!void {
1200 comptime assert(@sizeOf(T) == 4);
1201 const expected_raw: *align(1) const u32 = @ptrCast(&expected);
1202 return io.vtable.futexWait(io.userdata, @ptrCast(ptr), expected_raw.*, timeout);
1203}
1204/// Same as `futexWait`, except is not affected by task cancelation.
1205pub fn futexWaitUncancelable(io: Io, comptime T: type, ptr: *align(@alignOf(u32)) const T, expected: T) void {
1206 comptime assert(@sizeOf(T) == @sizeOf(u32));
1207 const expected_raw: *align(1) const u32 = @ptrCast(&expected);
1208 io.vtable.futexWaitUncancelable(io.userdata, @ptrCast(ptr), expected_raw.*);
1209}
1210/// Unblocks pending futex waits on `ptr`, up to a limit of `max_waiters` calls.
1211pub fn futexWake(io: Io, comptime T: type, ptr: *align(@alignOf(u32)) const T, max_waiters: u32) void {
1212 comptime assert(@sizeOf(T) == @sizeOf(u32));
1213 if (max_waiters == 0) return;
1214 return io.vtable.futexWake(io.userdata, @ptrCast(ptr), max_waiters);
1215}
1216
11771217pub const Mutex = struct {
11781218 state: State,
11791219
lib/std/Io/Threaded.zig+265-262
......@@ -155,6 +155,220 @@ const Thread = struct {
155155 fn currentSignalId() SignaleeId {
156156 return if (std.Thread.use_pthreads) std.c.pthread_self() else std.Thread.getCurrentId();
157157 }
158
159 fn futexWaitUncancelable(ptr: *const u32, expect: u32) void {
160 return Thread.futexWaitTimed(null, ptr, expect, null) catch unreachable;
161 }
162
163 fn futexWait(thread: *Thread, ptr: *const u32, expect: u32) Io.Cancelable!void {
164 return Thread.futexWaitTimed(thread, ptr, expect, null) catch |err| switch (err) {
165 error.Canceled => return error.Canceled,
166 error.Timeout => unreachable,
167 };
168 }
169
170 fn futexWaitTimed(thread: ?*Thread, ptr: *const u32, expect: u32, timeout_ns: ?u64) Io.Cancelable!void {
171 @branchHint(.cold);
172
173 if (builtin.single_threaded) unreachable; // nobody would ever wake us
174
175 if (builtin.cpu.arch.isWasm()) {
176 comptime assert(builtin.cpu.has(.wasm, .atomics));
177 if (thread) |t| try t.checkCancel();
178 const to: i64 = if (timeout_ns) |ns| ns else -1;
179 const signed_expect: i32 = @bitCast(expect);
180 const result = asm volatile (
181 \\local.get %[ptr]
182 \\local.get %[expected]
183 \\local.get %[timeout]
184 \\memory.atomic.wait32 0
185 \\local.set %[ret]
186 : [ret] "=r" (-> u32),
187 : [ptr] "r" (ptr),
188 [expected] "r" (signed_expect),
189 [timeout] "r" (to),
190 );
191 switch (result) {
192 0 => {}, // ok
193 1 => {}, // expected != loaded
194 2 => {}, // timeout
195 else => assert(!is_debug),
196 }
197 } else switch (native_os) {
198 .linux => {
199 const linux = std.os.linux;
200 var ts_buffer: linux.timespec = undefined;
201 const ts: ?*linux.timespec = if (timeout_ns) |ns| ts: {
202 ts_buffer = timestampToPosix(ns);
203 break :ts &ts_buffer;
204 } else null;
205 if (thread) |t| try t.beginSyscall();
206 const rc = linux.futex_4arg(ptr, .{ .cmd = .WAIT, .private = true }, expect, ts);
207 if (thread) |t| t.endSyscall();
208 switch (linux.errno(rc)) {
209 .SUCCESS => {}, // notified by `wake()`
210 .INTR => {}, // caller's responsibility to retry
211 .AGAIN => {}, // ptr.* != expect
212 .INVAL => {}, // possibly timeout overflow
213 .TIMEDOUT => {}, // timeout
214 .FAULT => recoverableOsBugDetected(), // ptr was invalid
215 else => recoverableOsBugDetected(),
216 }
217 },
218 .driverkit, .ios, .maccatalyst, .macos, .tvos, .visionos, .watchos => {
219 const c = std.c;
220 const flags: c.UL = .{
221 .op = .COMPARE_AND_WAIT,
222 .NO_ERRNO = true,
223 };
224 if (thread) |t| try t.beginSyscall();
225 const status = switch (darwin_supports_ulock_wait2) {
226 true => c.__ulock_wait2(flags, ptr, expect, ns: {
227 const ns = timeout_ns orelse break :ns 0;
228 if (ns == 0) break :ns 1;
229 break :ns ns;
230 }, 0),
231 false => c.__ulock_wait(flags, ptr, expect, us: {
232 const ns = timeout_ns orelse break :us 0;
233 const us = std.math.lossyCast(u32, ns / std.time.ns_per_us);
234 if (us == 0) break :us 1;
235 break :us us;
236 }),
237 };
238 if (thread) |t| t.endSyscall();
239 if (status >= 0) return;
240 switch (@as(c.E, @enumFromInt(-status))) {
241 .INTR => {}, // spurious wake
242 // Address of the futex was paged out. This is unlikely, but possible in theory, and
243 // pthread/libdispatch on darwin bother to handle it. In this case we'll return
244 // without waiting, but the caller should retry anyway.
245 .FAULT => {},
246 .TIMEDOUT => {}, // timeout
247 else => recoverableOsBugDetected(),
248 }
249 },
250 .windows => {
251 var timeout_value: windows.LARGE_INTEGER = undefined;
252 var timeout_ptr: ?*const windows.LARGE_INTEGER = null;
253 // NTDLL functions work with time in units of 100 nanoseconds.
254 // Positive values are absolute deadlines while negative values are relative durations.
255 if (timeout_ns) |delay| {
256 timeout_value = @as(windows.LARGE_INTEGER, @intCast(delay / 100));
257 timeout_value = -timeout_value;
258 timeout_ptr = &timeout_value;
259 }
260 if (thread) |t| try t.checkCancel();
261 switch (windows.ntdll.RtlWaitOnAddress(ptr, &expect, @sizeOf(@TypeOf(expect)), timeout_ptr)) {
262 .SUCCESS => {},
263 .CANCELLED => {},
264 .TIMEOUT => {}, // timeout
265 else => recoverableOsBugDetected(),
266 }
267 },
268 .freebsd => {
269 const flags = @intFromEnum(std.c.UMTX_OP.WAIT_UINT_PRIVATE);
270 var tm_size: usize = 0;
271 var tm: std.c._umtx_time = undefined;
272 var tm_ptr: ?*const std.c._umtx_time = null;
273 if (timeout_ns) |ns| {
274 tm_ptr = &tm;
275 tm_size = @sizeOf(@TypeOf(tm));
276 tm.flags = 0; // use relative time not UMTX_ABSTIME
277 tm.clockid = .MONOTONIC;
278 tm.timeout = timestampToPosix(ns);
279 }
280 if (thread) |t| try t.beginSyscall();
281 const rc = std.c._umtx_op(@intFromPtr(ptr), flags, @as(c_ulong, expect), tm_size, @intFromPtr(tm_ptr));
282 if (thread) |t| t.endSyscall();
283 if (is_debug) switch (posix.errno(rc)) {
284 .SUCCESS => {},
285 .FAULT => unreachable, // one of the args points to invalid memory
286 .INVAL => unreachable, // arguments should be correct
287 .TIMEDOUT => {}, // timeout
288 .INTR => {}, // spurious wake
289 else => unreachable,
290 };
291 },
292 else => @compileError("unimplemented: futexWait"),
293 }
294 }
295
296 fn futexWake(ptr: *const u32, max_waiters: u32) void {
297 @branchHint(.cold);
298
299 if (builtin.single_threaded) return; // nothing to wake up
300
301 if (builtin.cpu.arch.isWasm()) {
302 comptime assert(builtin.cpu.has(.wasm, .atomics));
303 assert(max_waiters != 0);
304 const woken_count = asm volatile (
305 \\local.get %[ptr]
306 \\local.get %[waiters]
307 \\memory.atomic.notify 0
308 \\local.set %[ret]
309 : [ret] "=r" (-> u32),
310 : [ptr] "r" (ptr),
311 [waiters] "r" (max_waiters),
312 );
313 _ = woken_count; // can be 0 when linker flag 'shared-memory' is not enabled
314 } else switch (native_os) {
315 .linux => {
316 const linux = std.os.linux;
317 switch (linux.errno(linux.futex_3arg(
318 ptr,
319 .{ .cmd = .WAKE, .private = true },
320 @min(max_waiters, std.math.maxInt(i32)),
321 ))) {
322 .SUCCESS => return, // successful wake up
323 .INVAL => return, // invalid futex_wait() on ptr done elsewhere
324 .FAULT => return, // pointer became invalid while doing the wake
325 else => return recoverableOsBugDetected(), // deadlock due to operating system bug
326 }
327 },
328 .driverkit, .ios, .maccatalyst, .macos, .tvos, .visionos, .watchos => {
329 const c = std.c;
330 const flags: c.UL = .{
331 .op = .COMPARE_AND_WAIT,
332 .NO_ERRNO = true,
333 .WAKE_ALL = max_waiters > 1,
334 };
335 while (true) {
336 const status = c.__ulock_wake(flags, ptr, 0);
337 if (status >= 0) return;
338 switch (@as(c.E, @enumFromInt(-status))) {
339 .INTR, .CANCELED => continue, // spurious wake()
340 .FAULT => unreachable, // __ulock_wake doesn't generate EFAULT according to darwin pthread_cond_t
341 .NOENT => return, // nothing was woken up
342 .ALREADY => unreachable, // only for UL.Op.WAKE_THREAD
343 else => unreachable, // deadlock due to operating system bug
344 }
345 }
346 },
347 .windows => {
348 assert(max_waiters != 0);
349 switch (max_waiters) {
350 1 => windows.ntdll.RtlWakeAddressSingle(ptr),
351 else => windows.ntdll.RtlWakeAddressAll(ptr),
352 }
353 },
354 .freebsd => {
355 const rc = std.c._umtx_op(
356 @intFromPtr(ptr),
357 @intFromEnum(std.c.UMTX_OP.WAKE_PRIVATE),
358 @as(c_ulong, max_waiters),
359 0, // there is no timeout struct
360 0, // there is no timeout struct pointer
361 );
362 switch (posix.errno(rc)) {
363 .SUCCESS => {},
364 .FAULT => {}, // it's ok if the ptr doesn't point to valid memory
365 .INVAL => unreachable, // arguments should be correct
366 else => unreachable, // deadlock due to operating system bug
367 }
368 },
369 else => @compileError("unimplemented: futexWake"),
370 }
371 }
158372};
159373
160374const max_iovecs_len = 8;
......@@ -403,6 +617,10 @@ pub fn io(t: *Threaded) Io {
403617 .groupWait = groupWait,
404618 .groupCancel = groupCancel,
405619
620 .futexWait = futexWait,
621 .futexWaitUncancelable = futexWaitUncancelable,
622 .futexWake = futexWake,
623
406624 .mutexLock = mutexLock,
407625 .mutexLockUncancelable = mutexLockUncancelable,
408626 .mutexUnlock = mutexUnlock,
......@@ -499,6 +717,10 @@ pub fn ioBasic(t: *Threaded) Io {
499717 .groupWait = groupWait,
500718 .groupCancel = groupCancel,
501719
720 .futexWait = futexWait,
721 .futexWaitUncancelable = futexWaitUncancelable,
722 .futexWake = futexWake,
723
502724 .mutexLock = mutexLock,
503725 .mutexLockUncancelable = mutexLockUncancelable,
504726 .mutexUnlock = mutexUnlock,
......@@ -1020,6 +1242,38 @@ fn cancel(
10201242 ac.waitAndDeinit(t, result);
10211243}
10221244
1245fn futexWait(userdata: ?*anyopaque, ptr: *const u32, expected: u32, timeout: Io.Timeout) Io.Cancelable!void {
1246 const t: *Threaded = @ptrCast(@alignCast(userdata));
1247 const current_thread = Thread.getCurrent(t);
1248 const t_io = ioBasic(t);
1249 const timeout_ns: ?u64 = ns: {
1250 const d = (timeout.toDurationFromNow(t_io) catch break :ns 10) orelse break :ns null;
1251 break :ns std.math.lossyCast(u64, d.raw.toNanoseconds());
1252 };
1253 switch (native_os) {
1254 .illumos, .netbsd, .openbsd => @panic("TODO"),
1255 else => try current_thread.futexWaitTimed(ptr, expected, timeout_ns),
1256 }
1257}
1258
1259fn futexWaitUncancelable(userdata: ?*anyopaque, ptr: *const u32, expected: u32) void {
1260 const t: *Threaded = @ptrCast(@alignCast(userdata));
1261 _ = t;
1262 switch (native_os) {
1263 .illumos, .netbsd, .openbsd => @panic("TODO"),
1264 else => Thread.futexWaitUncancelable(ptr, expected),
1265 }
1266}
1267
1268fn futexWake(userdata: ?*anyopaque, ptr: *const u32, max_waiters: u32) void {
1269 const t: *Threaded = @ptrCast(@alignCast(userdata));
1270 _ = t;
1271 switch (native_os) {
1272 .illumos, .netbsd, .openbsd => @panic("TODO"),
1273 else => Thread.futexWake(ptr, max_waiters),
1274 }
1275}
1276
10231277fn mutexLock(userdata: ?*anyopaque, prev_state: Io.Mutex.State, mutex: *Io.Mutex) Io.Cancelable!void {
10241278 if (builtin.single_threaded) unreachable; // Interface should have prevented this.
10251279 if (native_os == .netbsd) @panic("TODO");
......@@ -1027,10 +1281,10 @@ fn mutexLock(userdata: ?*anyopaque, prev_state: Io.Mutex.State, mutex: *Io.Mutex
10271281 const t: *Threaded = @ptrCast(@alignCast(userdata));
10281282 const current_thread = Thread.getCurrent(t);
10291283 if (prev_state == .contended) {
1030 try futexWait(current_thread, @ptrCast(&mutex.state), @intFromEnum(Io.Mutex.State.contended));
1284 try current_thread.futexWait(@ptrCast(&mutex.state), @intFromEnum(Io.Mutex.State.contended));
10311285 }
10321286 while (@atomicRmw(Io.Mutex.State, &mutex.state, .Xchg, .contended, .acquire) != .unlocked) {
1033 try futexWait(current_thread, @ptrCast(&mutex.state), @intFromEnum(Io.Mutex.State.contended));
1287 try current_thread.futexWait(@ptrCast(&mutex.state), @intFromEnum(Io.Mutex.State.contended));
10341288 }
10351289}
10361290
......@@ -1040,10 +1294,10 @@ fn mutexLockUncancelable(userdata: ?*anyopaque, prev_state: Io.Mutex.State, mute
10401294 if (native_os == .openbsd) @panic("TODO");
10411295 _ = userdata;
10421296 if (prev_state == .contended) {
1043 futexWaitUncancelable(@ptrCast(&mutex.state), @intFromEnum(Io.Mutex.State.contended));
1297 Thread.futexWaitUncancelable(@ptrCast(&mutex.state), @intFromEnum(Io.Mutex.State.contended));
10441298 }
10451299 while (@atomicRmw(Io.Mutex.State, &mutex.state, .Xchg, .contended, .acquire) != .unlocked) {
1046 futexWaitUncancelable(@ptrCast(&mutex.state), @intFromEnum(Io.Mutex.State.contended));
1300 Thread.futexWaitUncancelable(@ptrCast(&mutex.state), @intFromEnum(Io.Mutex.State.contended));
10471301 }
10481302}
10491303
......@@ -1054,7 +1308,7 @@ fn mutexUnlock(userdata: ?*anyopaque, prev_state: Io.Mutex.State, mutex: *Io.Mut
10541308 _ = userdata;
10551309 _ = prev_state;
10561310 if (@atomicRmw(Io.Mutex.State, &mutex.state, .Xchg, .unlocked, .release) == .contended) {
1057 futexWake(@ptrCast(&mutex.state), 1);
1311 Thread.futexWake(@ptrCast(&mutex.state), 1);
10581312 }
10591313}
10601314
......@@ -1081,7 +1335,7 @@ fn conditionWaitUncancelable(userdata: ?*anyopaque, cond: *Io.Condition, mutex:
10811335 defer mutex.lockUncancelable(t_io);
10821336
10831337 while (true) {
1084 futexWaitUncancelable(cond_epoch, epoch);
1338 Thread.futexWaitUncancelable(@ptrCast(cond_epoch), epoch);
10851339 epoch = cond_epoch.load(.acquire);
10861340 state = cond_state.load(.monotonic);
10871341 while (state & signal_mask != 0) {
......@@ -1125,7 +1379,7 @@ fn conditionWait(userdata: ?*anyopaque, cond: *Io.Condition, mutex: *Io.Mutex) I
11251379 defer mutex.lockUncancelable(t_io);
11261380
11271381 while (true) {
1128 try futexWait(current_thread, cond_epoch, epoch);
1382 try current_thread.futexWait(@ptrCast(cond_epoch), epoch);
11291383
11301384 epoch = cond_epoch.load(.acquire);
11311385 state = cond_state.load(.monotonic);
......@@ -1198,7 +1452,7 @@ fn conditionWake(userdata: ?*anyopaque, cond: *Io.Condition, wake: Io.Condition.
11981452 _ = cond_epoch.fetchAdd(1, .release);
11991453 if (native_os == .netbsd) @panic("TODO");
12001454 if (native_os == .openbsd) @panic("TODO");
1201 futexWake(cond_epoch, to_wake);
1455 Thread.futexWake(@ptrCast(cond_epoch), to_wake);
12021456 return;
12031457 };
12041458 }
......@@ -6612,257 +6866,6 @@ fn copyCanon(canonical_name_buffer: *[HostName.max_len]u8, name: []const u8) Hos
66126866/// ulock_wait2() uses 64-bit nano-second timeouts (with the same convention)
66136867const darwin_supports_ulock_wait2 = builtin.os.version_range.semver.min.major >= 11;
66146868
6615fn futexWait(current_thread: *Thread, ptr: *const std.atomic.Value(u32), expect: u32) Io.Cancelable!void {
6616 @branchHint(.cold);
6617
6618 if (builtin.cpu.arch.isWasm()) {
6619 comptime assert(builtin.cpu.has(.wasm, .atomics));
6620 try current_thread.checkCancel();
6621 const timeout: i64 = -1;
6622 const signed_expect: i32 = @bitCast(expect);
6623 const result = asm volatile (
6624 \\local.get %[ptr]
6625 \\local.get %[expected]
6626 \\local.get %[timeout]
6627 \\memory.atomic.wait32 0
6628 \\local.set %[ret]
6629 : [ret] "=r" (-> u32),
6630 : [ptr] "r" (&ptr.raw),
6631 [expected] "r" (signed_expect),
6632 [timeout] "r" (timeout),
6633 );
6634 switch (result) {
6635 0 => {}, // ok
6636 1 => {}, // expected != loaded
6637 2 => assert(!is_debug), // timeout
6638 else => assert(!is_debug),
6639 }
6640 } else switch (native_os) {
6641 .linux => {
6642 const linux = std.os.linux;
6643 try current_thread.beginSyscall();
6644 const rc = linux.futex_4arg(ptr, .{ .cmd = .WAIT, .private = true }, expect, null);
6645 current_thread.endSyscall();
6646 switch (linux.errno(rc)) {
6647 .SUCCESS => {}, // notified by `wake()`
6648 .INTR => {}, // caller's responsibility to retry
6649 .AGAIN => {}, // ptr.* != expect
6650 .INVAL => {}, // possibly timeout overflow
6651 .TIMEDOUT => recoverableOsBugDetected(),
6652 .FAULT => recoverableOsBugDetected(), // ptr was invalid
6653 else => recoverableOsBugDetected(),
6654 }
6655 },
6656 .driverkit, .ios, .maccatalyst, .macos, .tvos, .visionos, .watchos => {
6657 const c = std.c;
6658 const flags: c.UL = .{
6659 .op = .COMPARE_AND_WAIT,
6660 .NO_ERRNO = true,
6661 };
6662 try current_thread.beginSyscall();
6663 const status = if (darwin_supports_ulock_wait2)
6664 c.__ulock_wait2(flags, ptr, expect, 0, 0)
6665 else
6666 c.__ulock_wait(flags, ptr, expect, 0);
6667 current_thread.endSyscall();
6668
6669 if (status >= 0) return;
6670
6671 if (is_debug) switch (@as(c.E, @enumFromInt(-status))) {
6672 .INTR => {}, // spurious wake
6673 // Address of the futex was paged out. This is unlikely, but possible in theory, and
6674 // pthread/libdispatch on darwin bother to handle it. In this case we'll return
6675 // without waiting, but the caller should retry anyway.
6676 .FAULT => {},
6677 .TIMEDOUT => unreachable,
6678 else => unreachable,
6679 };
6680 },
6681 .windows => {
6682 try current_thread.checkCancel();
6683 switch (windows.ntdll.RtlWaitOnAddress(ptr, &expect, @sizeOf(@TypeOf(expect)), null)) {
6684 .SUCCESS => {},
6685 .CANCELLED => return error.Canceled,
6686 else => recoverableOsBugDetected(),
6687 }
6688 },
6689 .freebsd => {
6690 const flags = @intFromEnum(std.c.UMTX_OP.WAIT_UINT_PRIVATE);
6691 try current_thread.beginSyscall();
6692 const rc = std.c._umtx_op(@intFromPtr(&ptr.raw), flags, @as(c_ulong, expect), 0, 0);
6693 current_thread.endSyscall();
6694 if (is_debug) switch (posix.errno(rc)) {
6695 .SUCCESS => {},
6696 .FAULT => unreachable, // one of the args points to invalid memory
6697 .INVAL => unreachable, // arguments should be correct
6698 .TIMEDOUT => unreachable, // no timeout provided
6699 .INTR => {}, // spurious wake
6700 else => unreachable,
6701 };
6702 },
6703 else => @compileError("unimplemented: futexWait"),
6704 }
6705}
6706
6707pub fn futexWaitUncancelable(ptr: *const std.atomic.Value(u32), expect: u32) void {
6708 @branchHint(.cold);
6709
6710 if (builtin.cpu.arch.isWasm()) {
6711 comptime assert(builtin.cpu.has(.wasm, .atomics));
6712 const timeout: i64 = -1;
6713 const signed_expect: i32 = @bitCast(expect);
6714 const result = asm volatile (
6715 \\local.get %[ptr]
6716 \\local.get %[expected]
6717 \\local.get %[timeout]
6718 \\memory.atomic.wait32 0
6719 \\local.set %[ret]
6720 : [ret] "=r" (-> u32),
6721 : [ptr] "r" (&ptr.raw),
6722 [expected] "r" (signed_expect),
6723 [timeout] "r" (timeout),
6724 );
6725 switch (result) {
6726 0 => {}, // ok
6727 1 => {}, // expected != loaded
6728 2 => recoverableOsBugDetected(), // timeout
6729 else => recoverableOsBugDetected(),
6730 }
6731 } else switch (native_os) {
6732 .linux => {
6733 const linux = std.os.linux;
6734 const rc = linux.futex_4arg(ptr, .{ .cmd = .WAIT, .private = true }, expect, null);
6735 switch (linux.errno(rc)) {
6736 .SUCCESS => {}, // notified by `wake()`
6737 .INTR => {}, // caller's responsibility to repeat
6738 .AGAIN => {}, // ptr.* != expect
6739 .INVAL => {}, // possibly timeout overflow
6740 .TIMEDOUT => recoverableOsBugDetected(),
6741 .FAULT => recoverableOsBugDetected(), // ptr was invalid
6742 else => recoverableOsBugDetected(),
6743 }
6744 },
6745 .driverkit, .ios, .maccatalyst, .macos, .tvos, .visionos, .watchos => {
6746 const c = std.c;
6747 const flags: c.UL = .{
6748 .op = .COMPARE_AND_WAIT,
6749 .NO_ERRNO = true,
6750 };
6751 const status = if (darwin_supports_ulock_wait2)
6752 c.__ulock_wait2(flags, ptr, expect, 0, 0)
6753 else
6754 c.__ulock_wait(flags, ptr, expect, 0);
6755
6756 if (status >= 0) return;
6757
6758 switch (@as(c.E, @enumFromInt(-status))) {
6759 // Wait was interrupted by the OS or other spurious signalling.
6760 .INTR => {},
6761 // Address of the futex was paged out. This is unlikely, but possible in theory, and
6762 // pthread/libdispatch on darwin bother to handle it. In this case we'll return
6763 // without waiting, but the caller should retry anyway.
6764 .FAULT => {},
6765 .TIMEDOUT => recoverableOsBugDetected(),
6766 else => recoverableOsBugDetected(),
6767 }
6768 },
6769 .windows => {
6770 switch (windows.ntdll.RtlWaitOnAddress(ptr, &expect, @sizeOf(@TypeOf(expect)), null)) {
6771 .SUCCESS, .CANCELLED => {},
6772 else => recoverableOsBugDetected(),
6773 }
6774 },
6775 .freebsd => {
6776 const flags = @intFromEnum(std.c.UMTX_OP.WAIT_UINT_PRIVATE);
6777 const rc = std.c._umtx_op(@intFromPtr(&ptr.raw), flags, @as(c_ulong, expect), 0, 0);
6778 switch (posix.errno(rc)) {
6779 .SUCCESS => {},
6780 .INTR => {}, // spurious wake
6781 .FAULT => recoverableOsBugDetected(), // one of the args points to invalid memory
6782 .INVAL => recoverableOsBugDetected(), // arguments should be correct
6783 .TIMEDOUT => recoverableOsBugDetected(), // no timeout provided
6784 else => recoverableOsBugDetected(),
6785 }
6786 },
6787 else => @compileError("unimplemented: futexWaitUncancelable"),
6788 }
6789}
6790
6791pub fn futexWake(ptr: *const std.atomic.Value(u32), max_waiters: u32) void {
6792 @branchHint(.cold);
6793
6794 if (builtin.cpu.arch.isWasm()) {
6795 comptime assert(builtin.cpu.has(.wasm, .atomics));
6796 assert(max_waiters != 0);
6797 const woken_count = asm volatile (
6798 \\local.get %[ptr]
6799 \\local.get %[waiters]
6800 \\memory.atomic.notify 0
6801 \\local.set %[ret]
6802 : [ret] "=r" (-> u32),
6803 : [ptr] "r" (&ptr.raw),
6804 [waiters] "r" (max_waiters),
6805 );
6806 _ = woken_count; // can be 0 when linker flag 'shared-memory' is not enabled
6807 } else switch (native_os) {
6808 .linux => {
6809 const linux = std.os.linux;
6810 switch (linux.errno(linux.futex_3arg(
6811 &ptr.raw,
6812 .{ .cmd = .WAKE, .private = true },
6813 @min(max_waiters, std.math.maxInt(i32)),
6814 ))) {
6815 .SUCCESS => return, // successful wake up
6816 .INVAL => return, // invalid futex_wait() on ptr done elsewhere
6817 .FAULT => return, // pointer became invalid while doing the wake
6818 else => return recoverableOsBugDetected(), // deadlock due to operating system bug
6819 }
6820 },
6821 .driverkit, .ios, .maccatalyst, .macos, .tvos, .visionos, .watchos => {
6822 const c = std.c;
6823 const flags: c.UL = .{
6824 .op = .COMPARE_AND_WAIT,
6825 .NO_ERRNO = true,
6826 .WAKE_ALL = max_waiters > 1,
6827 };
6828 while (true) {
6829 const status = c.__ulock_wake(flags, ptr, 0);
6830 if (status >= 0) return;
6831 switch (@as(c.E, @enumFromInt(-status))) {
6832 .INTR, .CANCELED => continue, // spurious wake()
6833 .FAULT => unreachable, // __ulock_wake doesn't generate EFAULT according to darwin pthread_cond_t
6834 .NOENT => return, // nothing was woken up
6835 .ALREADY => unreachable, // only for UL.Op.WAKE_THREAD
6836 else => unreachable, // deadlock due to operating system bug
6837 }
6838 }
6839 },
6840 .windows => {
6841 assert(max_waiters != 0);
6842 switch (max_waiters) {
6843 1 => windows.ntdll.RtlWakeAddressSingle(ptr),
6844 else => windows.ntdll.RtlWakeAddressAll(ptr),
6845 }
6846 },
6847 .freebsd => {
6848 const rc = std.c._umtx_op(
6849 @intFromPtr(&ptr.raw),
6850 @intFromEnum(std.c.UMTX_OP.WAKE_PRIVATE),
6851 @as(c_ulong, max_waiters),
6852 0, // there is no timeout struct
6853 0, // there is no timeout struct pointer
6854 );
6855 switch (posix.errno(rc)) {
6856 .SUCCESS => {},
6857 .FAULT => {}, // it's ok if the ptr doesn't point to valid memory
6858 .INVAL => unreachable, // arguments should be correct
6859 else => unreachable, // deadlock due to operating system bug
6860 }
6861 },
6862 else => @compileError("unimplemented: futexWake"),
6863 }
6864}
6865
68666869/// A thread-safe logical boolean value which can be `set` and `unset`.
68676870///
68686871/// It can also block threads until the value is set with cancelation via timed
......@@ -6919,7 +6922,7 @@ const ResetEventFutex = enum(u32) {
69196922 }
69206923 const current_thread = Thread.getCurrent(t);
69216924 while (state == .waiting) {
6922 try futexWait(current_thread, @ptrCast(ref), @intFromEnum(ResetEventFutex.waiting));
6925 try current_thread.futexWait(@ptrCast(ref), @intFromEnum(ResetEventFutex.waiting));
69236926 state = @atomicLoad(ResetEventFutex, ref, .acquire);
69246927 }
69256928 assert(state == .is_set);
......@@ -6944,7 +6947,7 @@ const ResetEventFutex = enum(u32) {
69446947 state = @cmpxchgStrong(ResetEventFutex, ref, state, .waiting, .acquire, .acquire) orelse .waiting;
69456948 }
69466949 while (state == .waiting) {
6947 futexWaitUncancelable(@ptrCast(ref), @intFromEnum(ResetEventFutex.waiting));
6950 Thread.futexWaitUncancelable(@ptrCast(ref), @intFromEnum(ResetEventFutex.waiting));
69486951 state = @atomicLoad(ResetEventFutex, ref, .acquire);
69496952 }
69506953 assert(state == .is_set);
......@@ -6964,7 +6967,7 @@ const ResetEventFutex = enum(u32) {
69646967 return;
69656968 }
69666969 if (@atomicRmw(ResetEventFutex, ref, .Xchg, .is_set, .release) == .waiting) {
6967 futexWake(@ptrCast(ref), std.math.maxInt(u32));
6970 Thread.futexWake(@ptrCast(ref), std.math.maxInt(u32));
69686971 }
69696972 }
69706973