authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2026-02-04 05:06:17+01:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2026-02-04 05:06:17+01:00
logd1e01e94311ce4423f65d0d102d169ed59f1ca9d
tree4999446e0560021edef1d54508b91d7d2dfbf7d2
parent1ab5a58474d69ba8b9cc63a7ce3eb9b5338c5d2d
parent6d6532dd9eb862dfd6e59ceed5c762342d2cc0d5

Merge pull request 'std.Io.Threaded: Windows *cannot* spuriously unpark, and introduce ParkingMutex' (#31102) from parking-futex-parking-mutex into master

Reviewed-on: https://codeberg.org/ziglang/zig/pulls/31102 Reviewed-by: Andrew Kelley <andrew@ziglang.org>

1 files changed, 376 insertions(+), 247 deletions(-)

lib/std/Io/Threaded.zig+376-247
...@@ -29,8 +29,8 @@ const ws2_32 = std.os.windows.ws2_32;...@@ -29,8 +29,8 @@ const ws2_32 = std.os.windows.ws2_32;
29/// * scanning environment variables on some targets29/// * scanning environment variables on some targets
30/// * memory-mapping when mmap or equivalent is not available30/// * memory-mapping when mmap or equivalent is not available
31allocator: Allocator,31allocator: Allocator,
32mutex: Mutex = .init,32mutex: Io.Mutex = .init,
33cond: Condition = .init,33cond: Io.Condition = .init,
34run_queue: std.SinglyLinkedList = .{},34run_queue: std.SinglyLinkedList = .{},
35join_requested: bool = false,35join_requested: bool = false,
36stack_size: usize,36stack_size: usize,
...@@ -1505,8 +1505,8 @@ var global_single_threaded_instance: Threaded = .init_single_threaded;...@@ -1505,8 +1505,8 @@ var global_single_threaded_instance: Threaded = .init_single_threaded;
1505pub const global_single_threaded: *Threaded = &global_single_threaded_instance;1505pub const global_single_threaded: *Threaded = &global_single_threaded_instance;
15061506
1507pub fn setAsyncLimit(t: *Threaded, new_limit: Io.Limit) void {1507pub fn setAsyncLimit(t: *Threaded, new_limit: Io.Limit) void {
1508 mutexLockInternal(&t.mutex);1508 mutexLock(&t.mutex);
1509 defer mutexUnlockInternal(&t.mutex);1509 defer mutexUnlock(&t.mutex);
1510 t.async_limit = new_limit;1510 t.async_limit = new_limit;
1511}1511}
15121512
...@@ -1527,8 +1527,8 @@ pub fn deinit(t: *Threaded) void {...@@ -1527,8 +1527,8 @@ pub fn deinit(t: *Threaded) void {
1527fn join(t: *Threaded) void {1527fn join(t: *Threaded) void {
1528 if (builtin.single_threaded) return;1528 if (builtin.single_threaded) return;
1529 {1529 {
1530 mutexLockInternal(&t.mutex);1530 mutexLock(&t.mutex);
1531 defer mutexUnlockInternal(&t.mutex);1531 defer mutexUnlock(&t.mutex);
1532 t.join_requested = true;1532 t.join_requested = true;
1533 }1533 }
1534 condBroadcast(&t.cond);1534 condBroadcast(&t.cond);
...@@ -1593,16 +1593,16 @@ fn worker(t: *Threaded) void {...@@ -1593,16 +1593,16 @@ fn worker(t: *Threaded) void {
15931593
1594 defer t.wait_group.finish();1594 defer t.wait_group.finish();
15951595
1596 mutexLockInternal(&t.mutex);1596 mutexLock(&t.mutex);
1597 defer mutexUnlockInternal(&t.mutex);1597 defer mutexUnlock(&t.mutex);
15981598
1599 while (true) {1599 while (true) {
1600 while (t.run_queue.popFirst()) |runnable_node| {1600 while (t.run_queue.popFirst()) |runnable_node| {
1601 mutexUnlockInternal(&t.mutex);1601 mutexUnlock(&t.mutex);
1602 thread.cancel_protection = .unblocked;1602 thread.cancel_protection = .unblocked;
1603 const runnable: *Runnable = @fieldParentPtr("node", runnable_node);1603 const runnable: *Runnable = @fieldParentPtr("node", runnable_node);
1604 runnable.startFn(runnable, &thread, t);1604 runnable.startFn(runnable, &thread, t);
1605 mutexLockInternal(&t.mutex);1605 mutexLock(&t.mutex);
1606 t.busy_count -= 1;1606 t.busy_count -= 1;
1607 }1607 }
1608 if (t.join_requested) break;1608 if (t.join_requested) break;
...@@ -2025,12 +2025,12 @@ fn async(...@@ -2025,12 +2025,12 @@ fn async(
2025 },2025 },
2026 };2026 };
20272027
2028 mutexLockInternal(&t.mutex);2028 mutexLock(&t.mutex);
20292029
2030 const busy_count = t.busy_count;2030 const busy_count = t.busy_count;
20312031
2032 if (busy_count >= @intFromEnum(t.async_limit)) {2032 if (busy_count >= @intFromEnum(t.async_limit)) {
2033 mutexUnlockInternal(&t.mutex);2033 mutexUnlock(&t.mutex);
2034 future.destroy(gpa);2034 future.destroy(gpa);
2035 start(context.ptr, result.ptr);2035 start(context.ptr, result.ptr);
2036 return null;2036 return null;
...@@ -2044,7 +2044,7 @@ fn async(...@@ -2044,7 +2044,7 @@ fn async(
2044 const thread = std.Thread.spawn(.{ .stack_size = t.stack_size }, worker, .{t}) catch {2044 const thread = std.Thread.spawn(.{ .stack_size = t.stack_size }, worker, .{t}) catch {
2045 t.wait_group.finish();2045 t.wait_group.finish();
2046 t.busy_count = busy_count;2046 t.busy_count = busy_count;
2047 mutexUnlockInternal(&t.mutex);2047 mutexUnlock(&t.mutex);
2048 future.destroy(gpa);2048 future.destroy(gpa);
2049 start(context.ptr, result.ptr);2049 start(context.ptr, result.ptr);
2050 return null;2050 return null;
...@@ -2054,7 +2054,7 @@ fn async(...@@ -2054,7 +2054,7 @@ fn async(
20542054
2055 t.run_queue.prepend(&future.runnable.node);2055 t.run_queue.prepend(&future.runnable.node);
20562056
2057 mutexUnlockInternal(&t.mutex);2057 mutexUnlock(&t.mutex);
2058 condSignal(&t.cond);2058 condSignal(&t.cond);
2059 return @ptrCast(future);2059 return @ptrCast(future);
2060}2060}
...@@ -2077,8 +2077,8 @@ fn concurrent(...@@ -2077,8 +2077,8 @@ fn concurrent(
2077 };2077 };
2078 errdefer future.destroy(gpa);2078 errdefer future.destroy(gpa);
20792079
2080 mutexLockInternal(&t.mutex);2080 mutexLock(&t.mutex);
2081 defer mutexUnlockInternal(&t.mutex);2081 defer mutexUnlock(&t.mutex);
20822082
2083 const busy_count = t.busy_count;2083 const busy_count = t.busy_count;
20842084
...@@ -2122,12 +2122,12 @@ fn groupAsync(...@@ -2122,12 +2122,12 @@ fn groupAsync(
2122 error.OutOfMemory => return groupAsyncEager(start, context.ptr),2122 error.OutOfMemory => return groupAsyncEager(start, context.ptr),
2123 };2123 };
21242124
2125 mutexLockInternal(&t.mutex);2125 mutexLock(&t.mutex);
21262126
2127 const busy_count = t.busy_count;2127 const busy_count = t.busy_count;
21282128
2129 if (busy_count >= @intFromEnum(t.async_limit)) {2129 if (busy_count >= @intFromEnum(t.async_limit)) {
2130 mutexUnlockInternal(&t.mutex);2130 mutexUnlock(&t.mutex);
2131 task.destroy(gpa);2131 task.destroy(gpa);
2132 return groupAsyncEager(start, context.ptr);2132 return groupAsyncEager(start, context.ptr);
2133 }2133 }
...@@ -2140,7 +2140,7 @@ fn groupAsync(...@@ -2140,7 +2140,7 @@ fn groupAsync(
2140 const thread = std.Thread.spawn(.{ .stack_size = t.stack_size }, worker, .{t}) catch {2140 const thread = std.Thread.spawn(.{ .stack_size = t.stack_size }, worker, .{t}) catch {
2141 t.wait_group.finish();2141 t.wait_group.finish();
2142 t.busy_count = busy_count;2142 t.busy_count = busy_count;
2143 mutexUnlockInternal(&t.mutex);2143 mutexUnlock(&t.mutex);
2144 task.destroy(gpa);2144 task.destroy(gpa);
2145 return groupAsyncEager(start, context.ptr);2145 return groupAsyncEager(start, context.ptr);
2146 };2146 };
...@@ -2157,7 +2157,7 @@ fn groupAsync(...@@ -2157,7 +2157,7 @@ fn groupAsync(
2157 }, .monotonic);2157 }, .monotonic);
2158 t.run_queue.prepend(&task.runnable.node);2158 t.run_queue.prepend(&task.runnable.node);
21592159
2160 mutexUnlockInternal(&t.mutex);2160 mutexUnlock(&t.mutex);
2161 condSignal(&t.cond);2161 condSignal(&t.cond);
2162}2162}
2163fn groupAsyncEager(2163fn groupAsyncEager(
...@@ -2222,8 +2222,8 @@ fn groupConcurrent(...@@ -2222,8 +2222,8 @@ fn groupConcurrent(
2222 };2222 };
2223 errdefer task.destroy(gpa);2223 errdefer task.destroy(gpa);
22242224
2225 mutexLockInternal(&t.mutex);2225 mutexLock(&t.mutex);
2226 defer mutexUnlockInternal(&t.mutex);2226 defer mutexUnlock(&t.mutex);
22272227
2228 const busy_count = t.busy_count;2228 const busy_count = t.busy_count;
22292229
...@@ -2662,7 +2662,7 @@ fn batchAwaitConcurrent(userdata: ?*anyopaque, b: *Io.Batch, timeout: Io.Timeout...@@ -2662,7 +2662,7 @@ fn batchAwaitConcurrent(userdata: ?*anyopaque, b: *Io.Batch, timeout: Io.Timeout
2662 while (b.pending.head != .none and b.completions.head == .none) {2662 while (b.pending.head != .none and b.completions.head == .none) {
2663 var delay_interval: windows.LARGE_INTEGER = interval: {2663 var delay_interval: windows.LARGE_INTEGER = interval: {
2664 const d = deadline orelse break :interval std.math.minInt(windows.LARGE_INTEGER);2664 const d = deadline orelse break :interval std.math.minInt(windows.LARGE_INTEGER);
2665 break :interval t.deadlineToWindowsInterval(d);2665 break :interval timeoutToWindowsInterval(.{ .deadline = d }).?;
2666 };2666 };
2667 const alertable_syscall = try AlertableSyscall.start();2667 const alertable_syscall = try AlertableSyscall.start();
2668 const delay_rc = windows.ntdll.NtDelayExecution(windows.TRUE, &delay_interval);2668 const delay_rc = windows.ntdll.NtDelayExecution(windows.TRUE, &delay_interval);
...@@ -3847,8 +3847,8 @@ fn fileStatWindows(userdata: ?*anyopaque, file: File) File.StatError!File.Stat {...@@ -3847,8 +3847,8 @@ fn fileStatWindows(userdata: ?*anyopaque, file: File) File.StatError!File.Stat {
38473847
3848fn systemBasicInformation(t: *Threaded) ?*const windows.SYSTEM_BASIC_INFORMATION {3848fn systemBasicInformation(t: *Threaded) ?*const windows.SYSTEM_BASIC_INFORMATION {
3849 if (!t.system_basic_information.initialized.load(.acquire)) {3849 if (!t.system_basic_information.initialized.load(.acquire)) {
3850 mutexLockInternal(&t.mutex);3850 mutexLock(&t.mutex);
3851 defer mutexUnlockInternal(&t.mutex);3851 defer mutexUnlock(&t.mutex);
38523852
3853 switch (windows.ntdll.NtQuerySystemInformation(3853 switch (windows.ntdll.NtQuerySystemInformation(
3854 .SystemBasicInformation,3854 .SystemBasicInformation,
...@@ -4339,7 +4339,10 @@ fn dirCreateFileWindows(...@@ -4339,7 +4339,10 @@ fn dirCreateFileWindows(
4339 // kernel bug with retry attempts.4339 // kernel bug with retry attempts.
4340 syscall.finish();4340 syscall.finish();
4341 if (max_attempts - attempt == 0) return error.FileBusy;4341 if (max_attempts - attempt == 0) return error.FileBusy;
4342 try parking_sleep.windowsRetrySleep((@as(u32, 1) << attempt) >> 1);4342 try parking_sleep.sleep(.{ .duration = .{
4343 .raw = .fromMilliseconds((@as(u32, 1) << attempt) >> 1),
4344 .clock = .awake,
4345 } });
4343 attempt += 1;4346 attempt += 1;
4344 syscall = try .start();4347 syscall = try .start();
4345 continue;4348 continue;
...@@ -4352,7 +4355,10 @@ fn dirCreateFileWindows(...@@ -4352,7 +4355,10 @@ fn dirCreateFileWindows(
4352 // fixed by sleeping and retrying until the error goes away.4355 // fixed by sleeping and retrying until the error goes away.
4353 syscall.finish();4356 syscall.finish();
4354 if (max_attempts - attempt == 0) return error.FileBusy;4357 if (max_attempts - attempt == 0) return error.FileBusy;
4355 try parking_sleep.windowsRetrySleep((@as(u32, 1) << attempt) >> 1);4358 try parking_sleep.sleep(.{ .duration = .{
4359 .raw = .fromMilliseconds((@as(u32, 1) << attempt) >> 1),
4360 .clock = .awake,
4361 } });
4356 attempt += 1;4362 attempt += 1;
4357 syscall = try .start();4363 syscall = try .start();
4358 continue;4364 continue;
...@@ -4955,7 +4961,10 @@ pub fn dirOpenFileWtf16(...@@ -4955,7 +4961,10 @@ pub fn dirOpenFileWtf16(
4955 // kernel bug with retry attempts.4961 // kernel bug with retry attempts.
4956 syscall.finish();4962 syscall.finish();
4957 if (max_attempts - attempt == 0) return error.FileBusy;4963 if (max_attempts - attempt == 0) return error.FileBusy;
4958 try parking_sleep.windowsRetrySleep((@as(u32, 1) << attempt) >> 1);4964 try parking_sleep.sleep(.{ .duration = .{
4965 .raw = .fromMilliseconds((@as(u32, 1) << attempt) >> 1),
4966 .clock = .awake,
4967 } });
4959 attempt += 1;4968 attempt += 1;
4960 syscall = try .start();4969 syscall = try .start();
4961 continue;4970 continue;
...@@ -4977,7 +4986,10 @@ pub fn dirOpenFileWtf16(...@@ -4977,7 +4986,10 @@ pub fn dirOpenFileWtf16(
4977 // fixed by sleeping and retrying until the error goes away.4986 // fixed by sleeping and retrying until the error goes away.
4978 syscall.finish();4987 syscall.finish();
4979 if (max_attempts - attempt == 0) return error.FileBusy;4988 if (max_attempts - attempt == 0) return error.FileBusy;
4980 try parking_sleep.windowsRetrySleep((@as(u32, 1) << attempt) >> 1);4989 try parking_sleep.sleep(.{ .duration = .{
4990 .raw = .fromMilliseconds((@as(u32, 1) << attempt) >> 1),
4991 .clock = .awake,
4992 } });
4981 attempt += 1;4993 attempt += 1;
4982 syscall = try .start();4994 syscall = try .start();
4983 continue;4995 continue;
...@@ -7376,7 +7388,10 @@ fn dirReadLinkWindows(dir: Dir, sub_path: []const u8, buffer: []u8) Dir.ReadLink...@@ -7376,7 +7388,10 @@ fn dirReadLinkWindows(dir: Dir, sub_path: []const u8, buffer: []u8) Dir.ReadLink
7376 // kernel bug with retry attempts.7388 // kernel bug with retry attempts.
7377 syscall.finish();7389 syscall.finish();
7378 if (max_attempts - attempt == 0) return error.FileBusy;7390 if (max_attempts - attempt == 0) return error.FileBusy;
7379 try parking_sleep.windowsRetrySleep((@as(u32, 1) << attempt) >> 1);7391 try parking_sleep.sleep(.{ .duration = .{
7392 .raw = .fromMilliseconds((@as(u32, 1) << attempt) >> 1),
7393 .clock = .awake,
7394 } });
7380 attempt += 1;7395 attempt += 1;
7381 syscall = try .start();7396 syscall = try .start();
7382 continue;7397 continue;
...@@ -7389,7 +7404,10 @@ fn dirReadLinkWindows(dir: Dir, sub_path: []const u8, buffer: []u8) Dir.ReadLink...@@ -7389,7 +7404,10 @@ fn dirReadLinkWindows(dir: Dir, sub_path: []const u8, buffer: []u8) Dir.ReadLink
7389 // fixed by sleeping and retrying until the error goes away.7404 // fixed by sleeping and retrying until the error goes away.
7390 syscall.finish();7405 syscall.finish();
7391 if (max_attempts - attempt == 0) return error.FileBusy;7406 if (max_attempts - attempt == 0) return error.FileBusy;
7392 try parking_sleep.windowsRetrySleep((@as(u32, 1) << attempt) >> 1);7407 try parking_sleep.sleep(.{ .duration = .{
7408 .raw = .fromMilliseconds((@as(u32, 1) << attempt) >> 1),
7409 .clock = .awake,
7410 } });
7393 attempt += 1;7411 attempt += 1;
7394 syscall = try .start();7412 syscall = try .start();
7395 continue;7413 continue;
...@@ -10823,9 +10841,6 @@ fn nowPosix(clock: Io.Clock) Io.Timestamp {...@@ -10823,9 +10841,6 @@ fn nowPosix(clock: Io.Clock) Io.Timestamp {
10823fn now(userdata: ?*anyopaque, clock: Io.Clock) Io.Timestamp {10841fn now(userdata: ?*anyopaque, clock: Io.Clock) Io.Timestamp {
10824 const t: *Threaded = @ptrCast(@alignCast(userdata));10842 const t: *Threaded = @ptrCast(@alignCast(userdata));
10825 _ = t;10843 _ = t;
10826 return nowInner(clock);
10827}
10828fn nowInner(clock: Io.Clock) Io.Timestamp {
10829 return switch (native_os) {10844 return switch (native_os) {
10830 .windows => nowWindows(clock),10845 .windows => nowWindows(clock),
10831 .wasi => nowWasi(clock),10846 .wasi => nowWasi(clock),
...@@ -10955,7 +10970,7 @@ fn nowWasi(clock: Io.Clock) Io.Timestamp {...@@ -10955,7 +10970,7 @@ fn nowWasi(clock: Io.Clock) Io.Timestamp {
10955fn sleep(userdata: ?*anyopaque, timeout: Io.Timeout) Io.Cancelable!void {10970fn sleep(userdata: ?*anyopaque, timeout: Io.Timeout) Io.Cancelable!void {
10956 const t: *Threaded = @ptrCast(@alignCast(userdata));10971 const t: *Threaded = @ptrCast(@alignCast(userdata));
10957 if (timeout == .none) return;10972 if (timeout == .none) return;
10958 if (use_parking_sleep) return parking_sleep.sleep(timeout.toTimestamp(ioBasic(t)));10973 if (use_parking_sleep) return parking_sleep.sleep(timeout);
10959 if (native_os == .wasi) return sleepWasi(t, timeout);10974 if (native_os == .wasi) return sleepWasi(t, timeout);
10960 if (@TypeOf(posix.system.clock_nanosleep) != void) return sleepPosix(timeout);10975 if (@TypeOf(posix.system.clock_nanosleep) != void) return sleepPosix(timeout);
10961 return sleepNanosleep(t, timeout);10976 return sleepNanosleep(t, timeout);
...@@ -14361,10 +14376,9 @@ const Wsa = struct {...@@ -14361,10 +14376,9 @@ const Wsa = struct {
14361};14376};
1436214377
14363fn initializeWsa(t: *Threaded) error{ NetworkDown, Canceled }!void {14378fn initializeWsa(t: *Threaded) error{ NetworkDown, Canceled }!void {
14364 const t_io = io(t);
14365 const wsa = &t.wsa;14379 const wsa = &t.wsa;
14366 try wsa.mutex.lock(t_io);14380 mutexLock(&wsa.mutex);
14367 defer wsa.mutex.unlock(t_io);14381 defer mutexUnlock(&wsa.mutex);
14368 switch (wsa.status) {14382 switch (wsa.status) {
14369 .uninitialized => {14383 .uninitialized => {
14370 var wsa_data: ws2_32.WSADATA = undefined;14384 var wsa_data: ws2_32.WSADATA = undefined;
...@@ -14435,8 +14449,8 @@ const WindowsEnvironStrings = struct {...@@ -14435,8 +14449,8 @@ const WindowsEnvironStrings = struct {
14435};14449};
1443614450
14437fn scanEnviron(t: *Threaded) void {14451fn scanEnviron(t: *Threaded) void {
14438 mutexLockInternal(&t.mutex);14452 mutexLock(&t.mutex);
14439 defer mutexUnlockInternal(&t.mutex);14453 defer mutexUnlock(&t.mutex);
1444014454
14441 if (t.environ.initialized) return;14455 if (t.environ.initialized) return;
14442 t.environ.initialized = true;14456 t.environ.initialized = true;
...@@ -14791,8 +14805,8 @@ fn spawnPosix(t: *Threaded, options: process.SpawnOptions) process.SpawnError!Sp...@@ -14791,8 +14805,8 @@ fn spawnPosix(t: *Threaded, options: process.SpawnOptions) process.SpawnError!Sp
1479114805
14792fn getDevNullFd(t: *Threaded) !posix.fd_t {14806fn getDevNullFd(t: *Threaded) !posix.fd_t {
14793 {14807 {
14794 mutexLockInternal(&t.mutex);14808 mutexLock(&t.mutex);
14795 defer mutexUnlockInternal(&t.mutex);14809 defer mutexUnlock(&t.mutex);
14796 if (t.null_file.fd != -1) return t.null_file.fd;14810 if (t.null_file.fd != -1) return t.null_file.fd;
14797 }14811 }
14798 const mode: u32 = 0;14812 const mode: u32 = 0;
...@@ -14803,8 +14817,8 @@ fn getDevNullFd(t: *Threaded) !posix.fd_t {...@@ -14803,8 +14817,8 @@ fn getDevNullFd(t: *Threaded) !posix.fd_t {
14803 .SUCCESS => {14817 .SUCCESS => {
14804 syscall.finish();14818 syscall.finish();
14805 const fresh_fd: posix.fd_t = @intCast(rc);14819 const fresh_fd: posix.fd_t = @intCast(rc);
14806 mutexLockInternal(&t.mutex); // Another thread might have won the race.14820 mutexLock(&t.mutex); // Another thread might have won the race.
14807 defer mutexUnlockInternal(&t.mutex);14821 defer mutexUnlock(&t.mutex);
14808 if (t.null_file.fd != -1) {14822 if (t.null_file.fd != -1) {
14809 posix.close(fresh_fd);14823 posix.close(fresh_fd);
14810 return t.null_file.fd;14824 return t.null_file.fd;
...@@ -15464,8 +15478,8 @@ fn processSpawnWindows(userdata: ?*anyopaque, options: process.SpawnOptions) pro...@@ -15464,8 +15478,8 @@ fn processSpawnWindows(userdata: ?*anyopaque, options: process.SpawnOptions) pro
1546415478
15465fn getCngHandle(t: *Threaded) Io.RandomSecureError!windows.HANDLE {15479fn getCngHandle(t: *Threaded) Io.RandomSecureError!windows.HANDLE {
15466 {15480 {
15467 mutexLockInternal(&t.mutex);15481 mutexLock(&t.mutex);
15468 defer mutexUnlockInternal(&t.mutex);15482 defer mutexUnlock(&t.mutex);
15469 if (t.random_file.handle) |handle| return handle;15483 if (t.random_file.handle) |handle| return handle;
15470 }15484 }
1547115485
...@@ -15499,8 +15513,8 @@ fn getCngHandle(t: *Threaded) Io.RandomSecureError!windows.HANDLE {...@@ -15499,8 +15513,8 @@ fn getCngHandle(t: *Threaded) Io.RandomSecureError!windows.HANDLE {
15499 )) {15513 )) {
15500 .SUCCESS => {15514 .SUCCESS => {
15501 syscall.finish();15515 syscall.finish();
15502 mutexLockInternal(&t.mutex); // Another thread might have won the race.15516 mutexLock(&t.mutex); // Another thread might have won the race.
15503 defer mutexUnlockInternal(&t.mutex);15517 defer mutexUnlock(&t.mutex);
15504 if (t.random_file.handle) |prev_handle| {15518 if (t.random_file.handle) |prev_handle| {
15505 windows.CloseHandle(fresh_handle);15519 windows.CloseHandle(fresh_handle);
15506 return prev_handle;15520 return prev_handle;
...@@ -15520,8 +15534,8 @@ fn getCngHandle(t: *Threaded) Io.RandomSecureError!windows.HANDLE {...@@ -15520,8 +15534,8 @@ fn getCngHandle(t: *Threaded) Io.RandomSecureError!windows.HANDLE {
1552015534
15521fn getNulHandle(t: *Threaded) !windows.HANDLE {15535fn getNulHandle(t: *Threaded) !windows.HANDLE {
15522 {15536 {
15523 mutexLockInternal(&t.mutex);15537 mutexLock(&t.mutex);
15524 defer mutexUnlockInternal(&t.mutex);15538 defer mutexUnlock(&t.mutex);
15525 if (t.null_file.handle) |handle| return handle;15539 if (t.null_file.handle) |handle| return handle;
15526 }15540 }
1552715541
...@@ -15567,8 +15581,8 @@ fn getNulHandle(t: *Threaded) !windows.HANDLE {...@@ -15567,8 +15581,8 @@ fn getNulHandle(t: *Threaded) !windows.HANDLE {
15567 )) {15581 )) {
15568 .SUCCESS => {15582 .SUCCESS => {
15569 syscall.finish();15583 syscall.finish();
15570 mutexLockInternal(&t.mutex); // Another thread might have won the race.15584 mutexLock(&t.mutex); // Another thread might have won the race.
15571 defer mutexUnlockInternal(&t.mutex);15585 defer mutexUnlock(&t.mutex);
15572 if (t.null_file.handle) |prev_handle| {15586 if (t.null_file.handle) |prev_handle| {
15573 windows.CloseHandle(fresh_handle);15587 windows.CloseHandle(fresh_handle);
15574 return prev_handle;15588 return prev_handle;
...@@ -15585,7 +15599,10 @@ fn getNulHandle(t: *Threaded) !windows.HANDLE {...@@ -15585,7 +15599,10 @@ fn getNulHandle(t: *Threaded) !windows.HANDLE {
15585 // this other than retrying the creation after the OS finishes15599 // this other than retrying the creation after the OS finishes
15586 // the deletion.15600 // the deletion.
15587 syscall.finish();15601 syscall.finish();
15588 try parking_sleep.windowsRetrySleep(1);15602 try parking_sleep.sleep(.{ .duration = .{
15603 .raw = .fromMilliseconds(1),
15604 .clock = .awake,
15605 } });
15589 syscall = try .start();15606 syscall = try .start();
15590 continue;15607 continue;
15591 },15608 },
...@@ -16613,15 +16630,15 @@ fn random(userdata: ?*anyopaque, buffer: []u8) void {...@@ -16613,15 +16630,15 @@ fn random(userdata: ?*anyopaque, buffer: []u8) void {
16613}16630}
1661416631
16615fn randomMainThread(t: *Threaded, buffer: []u8) void {16632fn randomMainThread(t: *Threaded, buffer: []u8) void {
16616 mutexLockInternal(&t.mutex);16633 mutexLock(&t.mutex);
16617 defer mutexUnlockInternal(&t.mutex);16634 defer mutexUnlock(&t.mutex);
1661816635
16619 if (!t.csprng.isInitialized()) {16636 if (!t.csprng.isInitialized()) {
16620 @branchHint(.unlikely);16637 @branchHint(.unlikely);
16621 var seed: [Csprng.seed_len]u8 = undefined;16638 var seed: [Csprng.seed_len]u8 = undefined;
16622 {16639 {
16623 mutexUnlockInternal(&t.mutex);16640 mutexUnlock(&t.mutex);
16624 defer mutexLockInternal(&t.mutex);16641 defer mutexLock(&t.mutex);
1662516642
16626 const prev = swapCancelProtection(t, .blocked);16643 const prev = swapCancelProtection(t, .blocked);
16627 defer _ = swapCancelProtection(t, prev);16644 defer _ = swapCancelProtection(t, prev);
...@@ -16806,8 +16823,8 @@ fn randomSecure(userdata: ?*anyopaque, buffer: []u8) Io.RandomSecureError!void {...@@ -16806,8 +16823,8 @@ fn randomSecure(userdata: ?*anyopaque, buffer: []u8) Io.RandomSecureError!void {
1680616823
16807fn getRandomFd(t: *Threaded) Io.RandomSecureError!posix.fd_t {16824fn getRandomFd(t: *Threaded) Io.RandomSecureError!posix.fd_t {
16808 {16825 {
16809 mutexLockInternal(&t.mutex);16826 mutexLock(&t.mutex);
16810 defer mutexUnlockInternal(&t.mutex);16827 defer mutexUnlock(&t.mutex);
1681116828
16812 if (t.random_file.fd == -2) return error.EntropyUnavailable;16829 if (t.random_file.fd == -2) return error.EntropyUnavailable;
16813 if (t.random_file.fd != -1) return t.random_file.fd;16830 if (t.random_file.fd != -1) return t.random_file.fd;
...@@ -16847,8 +16864,8 @@ fn getRandomFd(t: *Threaded) Io.RandomSecureError!posix.fd_t {...@@ -16847,8 +16864,8 @@ fn getRandomFd(t: *Threaded) Io.RandomSecureError!posix.fd_t {
16847 .SUCCESS => {16864 .SUCCESS => {
16848 syscall.finish();16865 syscall.finish();
16849 if (!statx.mask.TYPE) return error.EntropyUnavailable;16866 if (!statx.mask.TYPE) return error.EntropyUnavailable;
16850 mutexLockInternal(&t.mutex); // Another thread might have won the race.16867 mutexLock(&t.mutex); // Another thread might have won the race.
16851 defer mutexUnlockInternal(&t.mutex);16868 defer mutexUnlock(&t.mutex);
16852 if (t.random_file.fd >= 0) {16869 if (t.random_file.fd >= 0) {
16853 posix.close(fd);16870 posix.close(fd);
16854 return t.random_file.fd;16871 return t.random_file.fd;
...@@ -16875,8 +16892,8 @@ fn getRandomFd(t: *Threaded) Io.RandomSecureError!posix.fd_t {...@@ -16875,8 +16892,8 @@ fn getRandomFd(t: *Threaded) Io.RandomSecureError!posix.fd_t {
16875 switch (posix.errno(fstat_sym(fd, &stat))) {16892 switch (posix.errno(fstat_sym(fd, &stat))) {
16876 .SUCCESS => {16893 .SUCCESS => {
16877 syscall.finish();16894 syscall.finish();
16878 mutexLockInternal(&t.mutex); // Another thread might have won the race.16895 mutexLock(&t.mutex); // Another thread might have won the race.
16879 defer mutexUnlockInternal(&t.mutex);16896 defer mutexUnlock(&t.mutex);
16880 if (t.random_file.fd >= 0) {16897 if (t.random_file.fd >= 0) {
16881 posix.close(fd);16898 posix.close(fd);
16882 return t.random_file.fd;16899 return t.random_file.fd;
...@@ -16940,7 +16957,7 @@ const parking_futex = struct {...@@ -16940,7 +16957,7 @@ const parking_futex = struct {
16940 /// avoid a race.16957 /// avoid a race.
16941 num_waiters: std.atomic.Value(u32),16958 num_waiters: std.atomic.Value(u32),
16942 /// Protects `waiters`.16959 /// Protects `waiters`.
16943 mutex: Mutex,16960 mutex: ParkingMutex,
16944 waiters: std.DoublyLinkedList,16961 waiters: std.DoublyLinkedList,
1694516962
16946 /// Prevent false sharing between buckets.16963 /// Prevent false sharing between buckets.
...@@ -16958,13 +16975,9 @@ const parking_futex = struct {...@@ -16958,13 +16975,9 @@ const parking_futex = struct {
16958 ///16975 ///
16959 /// * Removing the `Waiter` from `Bucket.waiters`16976 /// * Removing the `Waiter` from `Bucket.waiters`
16960 /// * Decrementing `Bucket.num_waiters`16977 /// * Decrementing `Bucket.num_waiters`
16961 /// * Atomically setting `done` (after this, the `Waiter` may go out of scope at any time,16978 /// * Unparking the thread (*after* the above, so that the `Waiter` does not go out of scope
16962 /// so must not be referenced again)16979 /// while it is still in the `Bucket`).
16963 /// * Unparking the thread (last, so that the unparked thread definitely sees `done`)
16964 thread_status: *std.atomic.Value(Thread.Status),16980 thread_status: *std.atomic.Value(Thread.Status),
16965 /// Initially `false`. Whoever updates `thread_status` to `.none`/`.canceling` will update
16966 /// this to `true` once they are done with the `Waiter`, just before unparking `tid`.
16967 done: std.atomic.Value(bool),
16968 };16981 };
1696916982
16970 fn bucketForAddress(address: usize) *Bucket {16983 fn bucketForAddress(address: usize) *Bucket {
...@@ -17003,14 +17016,13 @@ const parking_futex = struct {...@@ -17003,14 +17016,13 @@ const parking_futex = struct {
17003 .address = @intFromPtr(ptr),17016 .address = @intFromPtr(ptr),
17004 .tid = self_tid,17017 .tid = self_tid,
17005 .thread_status = undefined, // populated in critical section17018 .thread_status = undefined, // populated in critical section
17006 .done = .init(false),
17007 };17019 };
1700817020
17009 var status_buf: std.atomic.Value(Thread.Status) = undefined;17021 var status_buf: std.atomic.Value(Thread.Status) = undefined;
1701017022
17011 {17023 {
17012 mutexLockInternal(&bucket.mutex);17024 bucket.mutex.lock();
17013 defer mutexUnlockInternal(&bucket.mutex);17025 defer bucket.mutex.unlock();
1701417026
17015 _ = bucket.num_waiters.fetchAdd(1, .acquire);17027 _ = bucket.num_waiters.fetchAdd(1, .acquire);
1701617028
...@@ -17061,44 +17073,41 @@ const parking_futex = struct {...@@ -17061,44 +17073,41 @@ const parking_futex = struct {
17061 bucket.waiters.append(&waiter.node);17073 bucket.waiters.append(&waiter.node);
17062 }17074 }
1706317075
17064 const deadline: ?Io.Clock.Timestamp = switch (timeout) {17076 if (park(timeout, ptr, waiter.thread_status)) {
17065 .none => null,17077 // We were unparked by either `wake` or cancelation, so our current status is either
17066 .duration => |d| .{17078 // `.none` or `.canceling`. In either case, they've already removed `waiter` from
17067 .raw = nowInner(d.clock).addDuration(d.raw),17079 // `bucket`, so we have nothing more to do!
17068 .clock = d.clock,
17069 },
17070 .deadline => |d| d,
17071 };
17072 while (park(deadline, ptr)) {
17073 if (waiter.done.load(.acquire)) return; // all done!
17074 } else |err| switch (err) {17080 } else |err| switch (err) {
17075 error.Timeout => switch (waiter.thread_status.fetchAnd(17081 error.Timeout => {
17076 .{ .cancelation = @enumFromInt(0b110), .awaitable = .all_ones },17082 // We're not out of the woods yet: an unpark could race with the timeout.
17077 .monotonic,17083 const old_status = waiter.thread_status.fetchAnd(
17078 ).cancelation) {17084 .{ .cancelation = @enumFromInt(0b110), .awaitable = .all_ones },
17079 .parked => {17085 .monotonic,
17080 // We saw a timeout and updated our own status from `.parked` to `.none`. It is17086 );
17081 // our responsibility to remove `waiter` from `bucket`.17087 switch (old_status.cancelation) {
17082 mutexLockInternal(&bucket.mutex);17088 .parked => {
17083 defer mutexUnlockInternal(&bucket.mutex);17089 // No race. It is our responsibility to remove `waiter` from `bucket`.
17084 bucket.waiters.remove(&waiter.node);17090 // New status is `.none`.
17085 assert(bucket.num_waiters.fetchSub(1, .monotonic) > 0);17091 bucket.mutex.lock();
17086 },17092 defer bucket.mutex.unlock();
17087 .none, .canceling => {17093 bucket.waiters.remove(&waiter.node);
17088 // Race condition: the timeout was reached, then `wake` or a cancelation tried17094 assert(bucket.num_waiters.fetchSub(1, .monotonic) > 0);
17089 // to update our status. They won the race, so wait for them to do the cleanup.17095 },
17090 // They'll tell us by setting `waiter.done` and unparking us.17096 .none, .canceling => {
17091 while (!waiter.done.load(.acquire)) {17097 // Race condition: the timeout was reached, then `wake` or a canceler tried
17092 park(null, ptr) catch |e| switch (e) {17098 // to unpark us. Whoever did that will remove us from `bucket`. Wait for
17099 // that (and drop the unpark request in doing so).
17100 // New status is `.none` or `.canceling` respectively.
17101 park(.none, ptr, waiter.thread_status) catch |e| switch (e) {
17093 error.Timeout => unreachable,17102 error.Timeout => unreachable,
17094 };17103 };
17095 }17104 },
17096 },17105 .canceled => unreachable,
17097 .canceled => unreachable,17106 .blocked => unreachable,
17098 .blocked => unreachable,17107 .blocked_alertable => unreachable,
17099 .blocked_alertable => unreachable,17108 .blocked_canceling => unreachable,
17100 .blocked_alertable_canceling => unreachable,17109 .blocked_alertable_canceling => unreachable,
17101 .blocked_canceling => unreachable,17110 }
17102 },17111 },
17103 }17112 }
17104 }17113 }
...@@ -17119,8 +17128,8 @@ const parking_futex = struct {...@@ -17119,8 +17128,8 @@ const parking_futex = struct {
17119 // of the critical section. This forms a singly-linked list of waiters using `Waiter.node.next`.17128 // of the critical section. This forms a singly-linked list of waiters using `Waiter.node.next`.
17120 var waking_head: ?*std.DoublyLinkedList.Node = null;17129 var waking_head: ?*std.DoublyLinkedList.Node = null;
17121 {17130 {
17122 mutexLockInternal(&bucket.mutex);17131 bucket.mutex.lock();
17123 defer mutexUnlockInternal(&bucket.mutex);17132 defer bucket.mutex.unlock();
1712417133
17125 var num_removed: u32 = 0;17134 var num_removed: u32 = 0;
17126 var it = bucket.waiters.first;17135 var it = bucket.waiters.first;
...@@ -17147,6 +17156,9 @@ const parking_futex = struct {...@@ -17147,6 +17156,9 @@ const parking_futex = struct {
17147 waiter.node.next = waking_head;17156 waiter.node.next = waking_head;
17148 waking_head = &waiter.node;17157 waking_head = &waiter.node;
17149 num_removed += 1;17158 num_removed += 1;
17159 // Signal to `waiter` that they're about to be unparked, in case we're racing with their
17160 // timeout. See corresponding logic in `wake`.
17161 waiter.address = 0;
17150 }17162 }
1715117163
17152 _ = bucket.num_waiters.fetchSub(num_removed, .monotonic);17164 _ = bucket.num_waiters.fetchSub(num_removed, .monotonic);
...@@ -17161,8 +17173,6 @@ const parking_futex = struct {...@@ -17161,8 +17173,6 @@ const parking_futex = struct {
17161 const waiter: *Waiter = @fieldParentPtr("node", node);17173 const waiter: *Waiter = @fieldParentPtr("node", node);
17162 unpark_buf[unpark_len] = waiter.tid;17174 unpark_buf[unpark_len] = waiter.tid;
17163 unpark_len += 1;17175 unpark_len += 1;
17164 waiter.done.store(true, .release);
17165 // `waiter.*` is now potentially invalid so must not be referenced again.
17166 if (unpark_len == unpark_buf.len) {17176 if (unpark_len == unpark_buf.len) {
17167 unpark(&unpark_buf, ptr);17177 unpark(&unpark_buf, ptr);
17168 unpark_len = 0;17178 unpark_len = 0;
...@@ -17175,18 +17185,17 @@ const parking_futex = struct {...@@ -17175,18 +17185,17 @@ const parking_futex = struct {
1717517185
17176 fn removeCanceledWaiter(waiter: *Waiter) void {17186 fn removeCanceledWaiter(waiter: *Waiter) void {
17177 const bucket = bucketForAddress(waiter.address);17187 const bucket = bucketForAddress(waiter.address);
17178 mutexLockInternal(&bucket.mutex);17188 bucket.mutex.lock();
17179 defer mutexUnlockInternal(&bucket.mutex);17189 defer bucket.mutex.unlock();
17180 bucket.waiters.remove(&waiter.node);17190 bucket.waiters.remove(&waiter.node);
17181 assert(bucket.num_waiters.fetchSub(1, .monotonic) > 0);17191 assert(bucket.num_waiters.fetchSub(1, .monotonic) > 0);
17182 waiter.done.store(true, .release); // potentially invalidates `waiter.*`
17183 }17192 }
17184};17193};
17185const parking_sleep = struct {17194const parking_sleep = struct {
17186 comptime {17195 comptime {
17187 assert(use_parking_sleep);17196 assert(use_parking_sleep);
17188 }17197 }
17189 fn sleep(deadline: ?Io.Clock.Timestamp) Io.Cancelable!void {17198 fn sleep(timeout: Io.Timeout) Io.Cancelable!void {
17190 const opt_thread = Thread.current;17199 const opt_thread = Thread.current;
17191 cancelable: {17200 cancelable: {
17192 const thread = opt_thread orelse break :cancelable;17201 const thread = opt_thread orelse break :cancelable;
...@@ -17195,90 +17204,238 @@ const parking_sleep = struct {...@@ -17195,90 +17204,238 @@ const parking_sleep = struct {
17195 .unblocked => {},17204 .unblocked => {},
17196 }17205 }
17197 thread.futex_waiter = null;17206 thread.futex_waiter = null;
17198 const orig_status = thread.status.fetchOr(17207 {
17199 .{ .cancelation = @enumFromInt(0b001), .awaitable = .null },17208 const old_status = thread.status.fetchOr(
17200 .release, // release `thread.futex_waiter`17209 .{ .cancelation = @enumFromInt(0b001), .awaitable = .null },
17201 );17210 .release, // release `thread.futex_waiter`
17202 switch (orig_status.cancelation) {17211 );
17203 .none => {}, // status is now `.parked`17212 switch (old_status.cancelation) {
17204 .canceling => return error.Canceled, // status is now `.canceled`17213 .none => {}, // status is now `.parked`
17205 .canceled => break :cancelable, // status is still `.canceled`17214 .canceling => return error.Canceled, // status is now `.canceled`
17206 .parked => unreachable,17215 .canceled => break :cancelable, // status is still `.canceled`
17207 .blocked => unreachable,17216 .parked => unreachable,
17208 .blocked_alertable => unreachable,
17209 .blocked_alertable_canceling => unreachable,
17210 .blocked_canceling => unreachable,
17211 }
17212 while (park(deadline, null)) {
17213 // Either a cancelation or a spurious unpark; let's see which!
17214 switch (thread.status.load(.monotonic).cancelation) {
17215 .parked => continue, // spurious unpark; keep sleeping
17216 .canceling => {
17217 // We got canceled; update our state and return.
17218 thread.status.store(
17219 .{ .cancelation = .canceled, .awaitable = orig_status.awaitable },
17220 .monotonic,
17221 );
17222 return error.Canceled;
17223 },
17224 .none => unreachable,
17225 .canceled => unreachable,
17226 .blocked => unreachable,17217 .blocked => unreachable,
17227 .blocked_alertable => unreachable,17218 .blocked_alertable => unreachable,
17228 .blocked_alertable_canceling => unreachable,17219 .blocked_alertable_canceling => unreachable,
17229 .blocked_canceling => unreachable,17220 .blocked_canceling => unreachable,
17230 }17221 }
17231 } else |err| switch (err) {17222 }
17232 error.Timeout => switch (thread.status.fetchAnd(17223 if (park(timeout, null, &thread.status)) {
17233 .{ .cancelation = @enumFromInt(0b110), .awaitable = .all_ones },17224 // The only reason this could possibly happen is cancelation.
17225 const old_status = thread.status.load(.monotonic);
17226 assert(old_status.cancelation == .canceling);
17227 thread.status.store(
17228 .{ .cancelation = .canceled, .awaitable = old_status.awaitable },
17234 .monotonic,17229 .monotonic,
17235 ).cancelation) {17230 );
17236 // We updated our own status from `.parked` to `.none`.17231 return error.Canceled;
17237 .parked => return, // new status is `.none`17232 } else |err| switch (err) {
17238 .canceling => {17233 error.Timeout => {
17239 // Timeout raced with a cancelation. We don't need to do anything, but17234 // We're not out of the woods yet: an unpark could race with the timeout.
17240 // the next `park` on this thread will see a spurious unpark.17235 const old_status = thread.status.fetchAnd(
17241 // Status is still `.canceling`.17236 .{ .cancelation = @enumFromInt(0b110), .awaitable = .all_ones },
17242 return;17237 .monotonic,
17243 },17238 );
17244 .none => unreachable,17239 switch (old_status.cancelation) {
17245 .canceled => unreachable,17240 .parked => return, // No race; new status is `.none`
17246 .blocked => unreachable,17241 .canceling => {
17247 .blocked_alertable => unreachable,17242 // Race condition: the timeout was reached, then someone tried to unpark
17248 .blocked_alertable_canceling => unreachable,17243 // us for a cancelation. Whoever did that will have called `unpark`, so
17249 .blocked_canceling => unreachable,17244 // drop that unpark request by waiting for it.
17245 // Status is still `.canceling`.
17246 park(.none, null, &thread.status) catch |e| switch (e) {
17247 error.Timeout => unreachable,
17248 };
17249 return;
17250 },
17251 .none => unreachable,
17252 .canceled => unreachable,
17253 .blocked => unreachable,
17254 .blocked_alertable => unreachable,
17255 .blocked_canceling => unreachable,
17256 .blocked_alertable_canceling => unreachable,
17257 }
17250 },17258 },
17251 }17259 }
17252 }17260 }
17253 // Uncancelable sleep; this case is very simple.17261 // Uncancelable sleep; we expect not to be manually unparked.
17254 while (park(deadline, null)) {17262 var dummy_status: std.atomic.Value(Thread.Status) = .init(.{ .cancelation = .parked, .awaitable = .null });
17255 // Definitely spurious; nothing to do.17263 if (park(timeout, null, &dummy_status)) {
17264 unreachable; // unexpected unpark
17256 } else |err| switch (err) {17265 } else |err| switch (err) {
17257 error.Timeout => return,17266 error.Timeout => return,
17258 }17267 }
17259 }17268 }
17260 /// Sleep for approximately `ms` awake milliseconds in an attempt to work around Windows kernel bugs.17269};
17261 fn windowsRetrySleep(ms: u32) (Io.Cancelable || Io.UnexpectedError)!void {17270const ParkingMutex = struct {
17262 const now_timestamp = nowWindows(.awake); // '.awake' is supported on Windows17271 state: std.atomic.Value(State),
17263 const deadline = now_timestamp.addDuration(.fromMilliseconds(ms));17272
17264 try parking_sleep.sleep(.{ .raw = deadline, .clock = .awake });17273 const init: ParkingMutex = .{ .state = .init(.unlocked) };
17274
17275 comptime {
17276 assert(use_parking_futex);
17277 }
17278
17279 const State = enum(usize) {
17280 unlocked = 1,
17281 /// This value is intentionally 0 so that `waiter` returns `null`.
17282 locked_once = 0,
17283 /// Contended; value is a `*Waiter`.
17284 _,
17285 /// Returns the head of the waiter list. Illegal to call if `s == .unlocked`.
17286 fn waiter(s: State) ?*Waiter {
17287 return @ptrFromInt(@intFromEnum(s));
17288 }
17289 /// Returns a locked state where `w` is contending the lock.
17290 /// If `w` is `null`, returns `.locked_once`.
17291 fn fromWaiter(w: ?*Waiter) State {
17292 return @enumFromInt(@intFromPtr(w));
17293 }
17294 };
17295 const Waiter = struct {
17296 status: std.atomic.Value(Thread.Status),
17297 /// Never modified once the `Waiter` is in the linked list.
17298 next: ?*Waiter,
17299 /// Never modified once the `Waiter` is in the linked list.
17300 tid: std.Thread.Id,
17301 };
17302 fn lock(m: *ParkingMutex) void {
17303 state: switch (State.unlocked) { // assume 'unlocked' to optimize for uncontended case
17304 .unlocked => continue :state m.state.cmpxchgWeak(
17305 .unlocked,
17306 .locked_once,
17307 .acquire, // acquire lock
17308 .monotonic,
17309 ) orelse {
17310 @branchHint(.likely);
17311 return;
17312 },
17313
17314 .locked_once, _ => |last_state| {
17315 const old_waiter = last_state.waiter();
17316 const self_tid = if (Thread.current) |t| t.id else std.Thread.getCurrentId();
17317 var waiter: Waiter = .{
17318 .next = old_waiter,
17319 .status = .init(.{ .cancelation = .parked, .awaitable = .null }),
17320 .tid = self_tid,
17321 };
17322 if (m.state.cmpxchgWeak(
17323 .fromWaiter(old_waiter),
17324 .fromWaiter(&waiter),
17325 .release, // release `waiter`
17326 .monotonic,
17327 )) |new_state| {
17328 continue :state new_state;
17329 }
17330 // We're now in the list of waiters---park until we're given the lock.
17331 park(.none, m, &waiter.status) catch |err| switch (err) {
17332 error.Timeout => unreachable,
17333 };
17334 // We now hold the lock.
17335 assert(waiter.status.load(.monotonic).cancelation == .none);
17336 return;
17337 },
17338 }
17339 }
17340 fn unlock(m: *ParkingMutex) void {
17341 state: switch (State.locked_once) { // assume 'locked_once' to optimize for uncontended case
17342 .unlocked => unreachable, // we hold the lock
17343
17344 .locked_once => continue :state m.state.cmpxchgWeak(
17345 .locked_once,
17346 .unlocked,
17347 .release, // release lock
17348 .acquire, // acquire any `Waiter` memory
17349 ) orelse {
17350 @branchHint(.likely);
17351 return;
17352 },
17353
17354 _ => |last_state| {
17355 // The logic here does not have ABA problems, and does some accesses non-atomically,
17356 // because `Waiter.next` is owned by the lock holder (that's us!) once the waiter is
17357 // in the linked list, up until we set `Waiter.status` to `.none`.
17358
17359 // Run through the waiter list to the end to ensure fairness. This is obviously not
17360 // ideal, but it shouldn't be a big deal in practice provided the critical section
17361 // is fairly small (so we won't get too many threads contending the mutex at once).
17362 // There's a *chance* we could get away with a LIFO queue for our use case, but I
17363 // don't wanna risk that.
17364 var parent: ?*Waiter = null;
17365 var waiter: *Waiter = last_state.waiter().?;
17366 while (waiter.next) |next| {
17367 parent = waiter;
17368 waiter = next;
17369 }
17370 // `waiter` is next in line for the lock. Remove them from the list.
17371 if (parent) |p| {
17372 assert(p.next == waiter);
17373 p.next = null;
17374 } else {
17375 // We're waking the last waiter, so clear the list head.
17376 if (m.state.cmpxchgWeak(
17377 .fromWaiter(last_state.waiter().?),
17378 .locked_once,
17379 .acquire,
17380 .acquire, // acquire any new `Waiter` memory
17381 )) |new_state| {
17382 continue :state new_state;
17383 }
17384 }
17385 // Now we're ready to actually hand the lock over to them.
17386 const tid = waiter.tid; // load this before the store below potentially invalidates `waiter`
17387 waiter.status.store(.{ .cancelation = .none, .awaitable = .null }, .release); // release lock
17388 unpark(&.{tid}, m);
17389 return;
17390 },
17391 }
17265 }17392 }
17266};17393};
1726717394
17268/// Spurious wakeups are possible.17395fn timeoutToWindowsInterval(timeout: Io.Timeout) ?windows.LARGE_INTEGER {
17269///17396 // ntdll only supports two combinations:
17270/// `addr_hint` has no semantic effect, but may allow the OS to optimize this operation.17397 // * real-time (`.real`) sleeps with absolute deadlines
17271fn park(opt_deadline: ?Io.Clock.Timestamp, addr_hint: ?*const anyopaque) error{Timeout}!void {17398 // * monotonic (`.awake`/`.boot`) sleeps with relative durations
17399 const clock = switch (timeout) {
17400 .none => return null,
17401 .duration => |d| d.clock,
17402 .deadline => |d| d.clock,
17403 };
17404 switch (clock) {
17405 .cpu_process, .cpu_thread => unreachable, // cannot sleep for CPU time
17406 .real => {
17407 const deadline = switch (timeout) {
17408 .none => unreachable,
17409 .duration => |d| nowWindows(clock).addDuration(d.raw),
17410 .deadline => |d| d.raw,
17411 };
17412 return @intCast(@max(@divTrunc(deadline.nanoseconds, 100), 0));
17413 },
17414 .awake, .boot => {
17415 const duration = switch (timeout) {
17416 .none => unreachable,
17417 .duration => |d| d.raw,
17418 .deadline => |d| nowWindows(clock).durationTo(d.raw),
17419 };
17420 return @intCast(@min(@divTrunc(-duration.nanoseconds, 100), -1));
17421 },
17422 }
17423}
17424
17425fn park(
17426 timeout: Io.Timeout,
17427 /// This value has no semantic effect, but may allow the OS to optimize the operation.
17428 addr_hint: ?*const anyopaque,
17429 /// The API on NetBSD and Illumos sucks and can unpark spuriously (well, it *can't*, but signals
17430 /// cause an indistinguishable unblock, and libpthread really likes to leave unparks pending).
17431 /// As such, on these targets only, this `status` is checked to determine if an unpark is real.
17432 /// no way to differentiate
17433 status: *std.atomic.Value(Thread.Status),
17434) error{Timeout}!void {
17272 comptime assert(use_parking_futex or use_parking_sleep);17435 comptime assert(use_parking_futex or use_parking_sleep);
17273 switch (native_os) {17436 switch (native_os) {
17274 .windows => {17437 .windows => {
17275 var timeout_buf: windows.LARGE_INTEGER = undefined;17438 const raw_timeout = timeoutToWindowsInterval(timeout);
17276 const raw_timeout: ?*windows.LARGE_INTEGER = if (opt_deadline) |deadline| timeout: {
17277 const now_timestamp = nowWindows(deadline.clock);
17278 const nanoseconds = now_timestamp.durationTo(deadline.raw).nanoseconds;
17279 timeout_buf = @intCast(@divTrunc(-nanoseconds, 100));
17280 break :timeout &timeout_buf;
17281 } else null;
17282 // `RtlWaitOnAddress` passes the futex address in as the first argument to this call,17439 // `RtlWaitOnAddress` passes the futex address in as the first argument to this call,
17283 // but it's unclear what that actually does, especially since `NtAlertThreadByThreadId`17440 // but it's unclear what that actually does, especially since `NtAlertThreadByThreadId`
17284 // does *not* accept the address so the kernel can't really be using it as a hint. An17441 // does *not* accept the address so the kernel can't really be using it as a hint. An
...@@ -17292,7 +17449,10 @@ fn park(opt_deadline: ?Io.Clock.Timestamp, addr_hint: ?*const anyopaque) error{T...@@ -17292,7 +17449,10 @@ fn park(opt_deadline: ?Io.Clock.Timestamp, addr_hint: ?*const anyopaque) error{T
17292 // this parameter). However, to err on the side of caution, let's match the behavior of17449 // this parameter). However, to err on the side of caution, let's match the behavior of
17293 // `RtlWaitOnAddress` and pass the pointer, in case the kernel ever does something17450 // `RtlWaitOnAddress` and pass the pointer, in case the kernel ever does something
17294 // stupid such as trying to dereference it.17451 // stupid such as trying to dereference it.
17295 switch (windows.ntdll.NtWaitForAlertByThreadId(addr_hint, raw_timeout)) {17452 switch (windows.ntdll.NtWaitForAlertByThreadId(
17453 addr_hint,
17454 if (raw_timeout) |*t| t else null,
17455 )) {
17296 .ALERTED => return,17456 .ALERTED => return,
17297 .TIMEOUT => return error.Timeout,17457 .TIMEOUT => return error.Timeout,
17298 else => unreachable,17458 else => unreachable,
...@@ -17300,23 +17460,34 @@ fn park(opt_deadline: ?Io.Clock.Timestamp, addr_hint: ?*const anyopaque) error{T...@@ -17300,23 +17460,34 @@ fn park(opt_deadline: ?Io.Clock.Timestamp, addr_hint: ?*const anyopaque) error{T
17300 },17460 },
17301 .netbsd => {17461 .netbsd => {
17302 var ts_buf: posix.timespec = undefined;17462 var ts_buf: posix.timespec = undefined;
17303 const ts: ?*posix.timespec, const clock_real: bool = if (opt_deadline) |deadline| timeout: {17463 const ts: ?*posix.timespec, const abstime: bool, const clock_real: bool = switch (timeout) {
17304 ts_buf = timestampToPosix(deadline.raw.nanoseconds);17464 .none => .{ null, false, false },
17305 break :timeout .{ &ts_buf, deadline.clock == .real };17465 .deadline => |timestamp| timeout: {
17306 } else .{ null, true };17466 ts_buf = timestampToPosix(timestamp.raw.nanoseconds);
17307 switch (posix.errno(std.c._lwp_park(17467 break :timeout .{ &ts_buf, true, timestamp.clock == .real };
17308 if (clock_real) .REALTIME else .MONOTONIC,17468 },
17309 .{ .ABSTIME = true },17469 .duration => |duration| timeout: {
17310 ts,17470 ts_buf = timestampToPosix(duration.raw.nanoseconds);
17311 0,17471 break :timeout .{ &ts_buf, false, duration.clock == .real };
17312 addr_hint,17472 },
17313 null,17473 };
17314 ))) {17474 // It's okay to pass the same timeout in a loop. If it's a duration, the OS actually
17315 .SUCCESS, .ALREADY, .INTR => return,17475 // writes the remaining time into the buffer when the syscall returns.
17316 .TIMEDOUT => return error.Timeout,17476 while (status.load(.monotonic).cancelation == .parked) {
17317 .INVAL => unreachable,17477 switch (posix.errno(std.c._lwp_park(
17318 .SRCH => unreachable,17478 if (clock_real) .REALTIME else .MONOTONIC,
17319 else => unreachable,17479 .{ .ABSTIME = abstime },
17480 ts,
17481 0,
17482 addr_hint,
17483 null,
17484 ))) {
17485 .SUCCESS, .ALREADY, .INTR => {},
17486 .TIMEDOUT => return error.Timeout,
17487 .INVAL => unreachable,
17488 .SRCH => unreachable,
17489 else => unreachable,
17490 }
17320 }17491 }
17321 },17492 },
17322 .illumos => @panic("TODO: illumos lwp_park"),17493 .illumos => @panic("TODO: illumos lwp_park"),
...@@ -17324,24 +17495,8 @@ fn park(opt_deadline: ?Io.Clock.Timestamp, addr_hint: ?*const anyopaque) error{T...@@ -17324,24 +17495,8 @@ fn park(opt_deadline: ?Io.Clock.Timestamp, addr_hint: ?*const anyopaque) error{T
17324 }17495 }
17325}17496}
1732617497
17327fn deadlineToWindowsInterval(t: *Io.Threaded, deadline: Io.Clock.Timestamp) windows.LARGE_INTEGER {
17328 // ntdll only supports two combinations:
17329 // * real-time (`.real`) sleeps with absolute deadlines
17330 // * monotonic (`.awake`/`.boot`) sleeps with relative durations
17331 switch (deadline.clock) {
17332 .cpu_process, .cpu_thread => return 0,
17333 .real => {
17334 return @intCast(@max(@divTrunc(deadline.raw.nanoseconds, 100), 0));
17335 },
17336 .awake, .boot => {
17337 const duration = deadline.durationFromNow(ioBasic(t));
17338 return @intCast(@min(@divTrunc(-duration.raw.nanoseconds, 100), -1));
17339 },
17340 }
17341}
17342
17343const UnparkTid = switch (native_os) {17498const UnparkTid = switch (native_os) {
17344 // `NtAlertMultipleThreadByThreadId` is weird and wants 64-bit thread handles?17499 // `NtAlertMultipleThreadByThreadId` is weird and wants 64-bit thread IDs?
17345 .windows => usize,17500 .windows => usize,
17346 else => std.Thread.Id,17501 else => std.Thread.Id,
17347};17502};
...@@ -18162,14 +18317,8 @@ fn eventSet(event: *Io.Event) void {...@@ -18162,14 +18317,8 @@ fn eventSet(event: *Io.Event) void {
18162 }18317 }
18163}18318}
1816418319
18165const Condition = if (!is_windows) Io.Condition else struct {
18166 condition: windows.CONDITION_VARIABLE,
18167 const init: @This() = .{ .condition = .{} };
18168};
18169
18170/// Same as `Io.Condition.broadcast` but avoids the VTable.18320/// Same as `Io.Condition.broadcast` but avoids the VTable.
18171fn condBroadcast(cond: *Condition) void {18321fn condBroadcast(cond: *Io.Condition) void {
18172 if (is_windows) return windows.ntdll.RtlWakeAllConditionVariable(&cond.condition);
18173 var prev_state = cond.state.load(.monotonic);18322 var prev_state = cond.state.load(.monotonic);
18174 while (prev_state.waiters > prev_state.signals) {18323 while (prev_state.waiters > prev_state.signals) {
18175 @branchHint(.unlikely);18324 @branchHint(.unlikely);
...@@ -18189,8 +18338,7 @@ fn condBroadcast(cond: *Condition) void {...@@ -18189,8 +18338,7 @@ fn condBroadcast(cond: *Condition) void {
18189}18338}
1819018339
18191/// Same as `Io.Condition.signal` but avoids the VTable.18340/// Same as `Io.Condition.signal` but avoids the VTable.
18192fn condSignal(cond: *Condition) void {18341fn condSignal(cond: *Io.Condition) void {
18193 if (is_windows) return windows.ntdll.RtlWakeConditionVariable(&cond.condition);
18194 var prev_state = cond.state.load(.monotonic);18342 var prev_state = cond.state.load(.monotonic);
18195 while (prev_state.waiters > prev_state.signals) {18343 while (prev_state.waiters > prev_state.signals) {
18196 @branchHint(.unlikely);18344 @branchHint(.unlikely);
...@@ -18210,11 +18358,7 @@ fn condSignal(cond: *Condition) void {...@@ -18210,11 +18358,7 @@ fn condSignal(cond: *Condition) void {
18210}18358}
1821118359
18212/// Same as `Io.Condition.waitUncancelable` but avoids the VTable.18360/// Same as `Io.Condition.waitUncancelable` but avoids the VTable.
18213fn condWait(cond: *Condition, mutex: *Mutex) void {18361fn condWait(cond: *Io.Condition, mutex: *Io.Mutex) void {
18214 if (is_windows) {
18215 _ = windows.kernel32.SleepConditionVariableSRW(&cond.condition, &mutex.srwlock, windows.INFINITE, 0);
18216 return;
18217 }
18218 var epoch = cond.epoch.load(.acquire); // `.acquire` to ensure ordered before state load18362 var epoch = cond.epoch.load(.acquire); // `.acquire` to ensure ordered before state load
1821918363
18220 {18364 {
...@@ -18222,8 +18366,8 @@ fn condWait(cond: *Condition, mutex: *Mutex) void {...@@ -18222,8 +18366,8 @@ fn condWait(cond: *Condition, mutex: *Mutex) void {
18222 assert(prev_state.waiters < std.math.maxInt(u16)); // overflow caused by too many waiters18366 assert(prev_state.waiters < std.math.maxInt(u16)); // overflow caused by too many waiters
18223 }18367 }
1822418368
18225 mutexUnlockInternal(mutex);18369 mutexUnlock(mutex);
18226 defer mutexLockInternal(mutex);18370 defer mutexLock(mutex);
1822718371
18228 while (true) {18372 while (true) {
18229 Thread.futexWaitUncancelable(&cond.epoch.raw, epoch, null);18373 Thread.futexWaitUncancelable(&cond.epoch.raw, epoch, null);
...@@ -18243,16 +18387,6 @@ fn condWait(cond: *Condition, mutex: *Mutex) void {...@@ -18243,16 +18387,6 @@ fn condWait(cond: *Condition, mutex: *Mutex) void {
18243 }18387 }
18244}18388}
1824518389
18246const Mutex = if (!is_windows) Io.Mutex else struct {
18247 srwlock: windows.SRWLOCK,
18248 const init: @This() = .{ .srwlock = .{} };
18249};
18250
18251fn mutexLockInternal(m: *Mutex) void {
18252 if (is_windows) return windows.ntdll.RtlAcquireSRWLockExclusive(&m.srwlock);
18253 return mutexLock(m);
18254}
18255
18256/// Same as `Io.Mutex.lockUncancelable` but avoids the VTable.18390/// Same as `Io.Mutex.lockUncancelable` but avoids the VTable.
18257pub fn mutexLock(m: *Io.Mutex) void {18391pub fn mutexLock(m: *Io.Mutex) void {
18258 const initial_state = m.state.cmpxchgWeak(18392 const initial_state = m.state.cmpxchgWeak(
...@@ -18272,11 +18406,6 @@ pub fn mutexLock(m: *Io.Mutex) void {...@@ -18272,11 +18406,6 @@ pub fn mutexLock(m: *Io.Mutex) void {
18272 }18406 }
18273}18407}
1827418408
18275fn mutexUnlockInternal(m: *Mutex) void {
18276 if (is_windows) return windows.ntdll.RtlReleaseSRWLockExclusive(&m.srwlock);
18277 return mutexUnlock(m);
18278}
18279
18280/// Same as `Io.Mutex.unlock` but avoids the VTable.18409/// Same as `Io.Mutex.unlock` but avoids the VTable.
18281pub fn mutexUnlock(m: *Io.Mutex) void {18410pub fn mutexUnlock(m: *Io.Mutex) void {
18282 switch (m.state.swap(.unlocked, .release)) {18411 switch (m.state.swap(.unlocked, .release)) {