authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2026-02-03 03:30:42+01:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2026-02-03 03:30:42+01:00
loge9eadee00654f5f762abe3cdc596359b79893eab
treea3ef8619becb5a68fc44159583623c675b7d93d2
parentc2d4806d659abf8c4c0ab989eae225303de57af3
parent5312063138e787a09493e5f5affb5c8652b66dbc

Merge pull request 'std.Io.Threaded: sever dependency on std.Thread Mutex and Condition' (#31096) from severance into master

Reviewed-on: https://codeberg.org/ziglang/zig/pulls/31096

1 files changed, 202 insertions(+), 64 deletions(-)

lib/std/Io/Threaded.zig+202-64
......@@ -29,8 +29,8 @@ const ws2_32 = std.os.windows.ws2_32;
2929/// * scanning environment variables on some targets
3030/// * memory-mapping when mmap or equivalent is not available
3131allocator: Allocator,
32mutex: std.Thread.Mutex = .{},
33cond: std.Thread.Condition = .{},
32mutex: Mutex = .init,
33cond: Condition = .init,
3434run_queue: std.SinglyLinkedList = .{},
3535join_requested: bool = false,
3636stack_size: usize,
......@@ -1486,8 +1486,8 @@ var global_single_threaded_instance: Threaded = .init_single_threaded;
14861486pub const global_single_threaded: *Threaded = &global_single_threaded_instance;
14871487
14881488pub fn setAsyncLimit(t: *Threaded, new_limit: Io.Limit) void {
1489 t.mutex.lock();
1490 defer t.mutex.unlock();
1489 mutexLockUncancelable(&t.mutex);
1490 defer mutexUnlock(&t.mutex);
14911491 t.async_limit = new_limit;
14921492}
14931493
......@@ -1508,11 +1508,11 @@ pub fn deinit(t: *Threaded) void {
15081508fn join(t: *Threaded) void {
15091509 if (builtin.single_threaded) return;
15101510 {
1511 t.mutex.lock();
1512 defer t.mutex.unlock();
1511 mutexLockUncancelable(&t.mutex);
1512 defer mutexUnlock(&t.mutex);
15131513 t.join_requested = true;
15141514 }
1515 t.cond.broadcast();
1515 condBroadcast(&t.cond);
15161516 t.wait_group.wait();
15171517}
15181518
......@@ -1574,20 +1574,20 @@ fn worker(t: *Threaded) void {
15741574
15751575 defer t.wait_group.finish();
15761576
1577 t.mutex.lock();
1578 defer t.mutex.unlock();
1577 mutexLockUncancelable(&t.mutex);
1578 defer mutexUnlock(&t.mutex);
15791579
15801580 while (true) {
15811581 while (t.run_queue.popFirst()) |runnable_node| {
1582 t.mutex.unlock();
1582 mutexUnlock(&t.mutex);
15831583 thread.cancel_protection = .unblocked;
15841584 const runnable: *Runnable = @fieldParentPtr("node", runnable_node);
15851585 runnable.startFn(runnable, &thread, t);
1586 t.mutex.lock();
1586 mutexLockUncancelable(&t.mutex);
15871587 t.busy_count -= 1;
15881588 }
15891589 if (t.join_requested) break;
1590 t.cond.wait(&t.mutex);
1590 condWait(&t.cond, &t.mutex);
15911591 }
15921592}
15931593
......@@ -2004,12 +2004,12 @@ fn async(
20042004 },
20052005 };
20062006
2007 t.mutex.lock();
2007 mutexLockUncancelable(&t.mutex);
20082008
20092009 const busy_count = t.busy_count;
20102010
20112011 if (busy_count >= @intFromEnum(t.async_limit)) {
2012 t.mutex.unlock();
2012 mutexUnlock(&t.mutex);
20132013 future.destroy(gpa);
20142014 start(context.ptr, result.ptr);
20152015 return null;
......@@ -2023,7 +2023,7 @@ fn async(
20232023 const thread = std.Thread.spawn(.{ .stack_size = t.stack_size }, worker, .{t}) catch {
20242024 t.wait_group.finish();
20252025 t.busy_count = busy_count;
2026 t.mutex.unlock();
2026 mutexUnlock(&t.mutex);
20272027 future.destroy(gpa);
20282028 start(context.ptr, result.ptr);
20292029 return null;
......@@ -2033,8 +2033,8 @@ fn async(
20332033
20342034 t.run_queue.prepend(&future.runnable.node);
20352035
2036 t.mutex.unlock();
2037 t.cond.signal();
2036 mutexUnlock(&t.mutex);
2037 condSignal(&t.cond);
20382038 return @ptrCast(future);
20392039}
20402040
......@@ -2056,8 +2056,8 @@ fn concurrent(
20562056 };
20572057 errdefer future.destroy(gpa);
20582058
2059 t.mutex.lock();
2060 defer t.mutex.unlock();
2059 mutexLockUncancelable(&t.mutex);
2060 defer mutexUnlock(&t.mutex);
20612061
20622062 const busy_count = t.busy_count;
20632063
......@@ -2080,7 +2080,7 @@ fn concurrent(
20802080
20812081 t.run_queue.prepend(&future.runnable.node);
20822082
2083 t.cond.signal();
2083 condSignal(&t.cond);
20842084 return @ptrCast(future);
20852085}
20862086
......@@ -2101,12 +2101,12 @@ fn groupAsync(
21012101 error.OutOfMemory => return groupAsyncEager(start, context.ptr),
21022102 };
21032103
2104 t.mutex.lock();
2104 mutexLockUncancelable(&t.mutex);
21052105
21062106 const busy_count = t.busy_count;
21072107
21082108 if (busy_count >= @intFromEnum(t.async_limit)) {
2109 t.mutex.unlock();
2109 mutexUnlock(&t.mutex);
21102110 task.destroy(gpa);
21112111 return groupAsyncEager(start, context.ptr);
21122112 }
......@@ -2119,7 +2119,7 @@ fn groupAsync(
21192119 const thread = std.Thread.spawn(.{ .stack_size = t.stack_size }, worker, .{t}) catch {
21202120 t.wait_group.finish();
21212121 t.busy_count = busy_count;
2122 t.mutex.unlock();
2122 mutexUnlock(&t.mutex);
21232123 task.destroy(gpa);
21242124 return groupAsyncEager(start, context.ptr);
21252125 };
......@@ -2136,8 +2136,8 @@ fn groupAsync(
21362136 }, .monotonic);
21372137 t.run_queue.prepend(&task.runnable.node);
21382138
2139 t.mutex.unlock();
2140 t.cond.signal();
2139 mutexUnlock(&t.mutex);
2140 condSignal(&t.cond);
21412141}
21422142fn groupAsyncEager(
21432143 start: *const fn (context: *const anyopaque) Io.Cancelable!void,
......@@ -2201,8 +2201,8 @@ fn groupConcurrent(
22012201 };
22022202 errdefer task.destroy(gpa);
22032203
2204 t.mutex.lock();
2205 defer t.mutex.unlock();
2204 mutexLockUncancelable(&t.mutex);
2205 defer mutexUnlock(&t.mutex);
22062206
22072207 const busy_count = t.busy_count;
22082208
......@@ -2233,7 +2233,7 @@ fn groupConcurrent(
22332233 }, .monotonic);
22342234 t.run_queue.prepend(&task.runnable.node);
22352235
2236 t.cond.signal();
2236 condSignal(&t.cond);
22372237}
22382238
22392239fn groupAwait(userdata: ?*anyopaque, type_erased: *Io.Group, initial_token: *anyopaque) Io.Cancelable!void {
......@@ -3838,8 +3838,8 @@ fn fileStatWindows(userdata: ?*anyopaque, file: File) File.StatError!File.Stat {
38383838
38393839fn systemBasicInformation(t: *Threaded) ?*const windows.SYSTEM_BASIC_INFORMATION {
38403840 if (!t.system_basic_information.initialized.load(.acquire)) {
3841 t.mutex.lock();
3842 defer t.mutex.unlock();
3841 mutexLockUncancelable(&t.mutex);
3842 defer mutexUnlock(&t.mutex);
38433843
38443844 switch (windows.ntdll.NtQuerySystemInformation(
38453845 .SystemBasicInformation,
......@@ -14373,8 +14373,8 @@ const WindowsEnvironStrings = struct {
1437314373};
1437414374
1437514375fn scanEnviron(t: *Threaded) void {
14376 t.mutex.lock();
14377 defer t.mutex.unlock();
14376 mutexLockUncancelable(&t.mutex);
14377 defer mutexUnlock(&t.mutex);
1437814378
1437914379 if (t.environ.initialized) return;
1438014380 t.environ.initialized = true;
......@@ -14729,8 +14729,8 @@ fn spawnPosix(t: *Threaded, options: process.SpawnOptions) process.SpawnError!Sp
1472914729
1473014730fn getDevNullFd(t: *Threaded) !posix.fd_t {
1473114731 {
14732 t.mutex.lock();
14733 defer t.mutex.unlock();
14732 mutexLockUncancelable(&t.mutex);
14733 defer mutexUnlock(&t.mutex);
1473414734 if (t.null_file.fd != -1) return t.null_file.fd;
1473514735 }
1473614736 const mode: u32 = 0;
......@@ -14741,8 +14741,8 @@ fn getDevNullFd(t: *Threaded) !posix.fd_t {
1474114741 .SUCCESS => {
1474214742 syscall.finish();
1474314743 const fresh_fd: posix.fd_t = @intCast(rc);
14744 t.mutex.lock(); // Another thread might have won the race.
14745 defer t.mutex.unlock();
14744 mutexLockUncancelable(&t.mutex); // Another thread might have won the race.
14745 defer mutexUnlock(&t.mutex);
1474614746 if (t.null_file.fd != -1) {
1474714747 posix.close(fresh_fd);
1474814748 return t.null_file.fd;
......@@ -15402,8 +15402,8 @@ fn processSpawnWindows(userdata: ?*anyopaque, options: process.SpawnOptions) pro
1540215402
1540315403fn getCngHandle(t: *Threaded) Io.RandomSecureError!windows.HANDLE {
1540415404 {
15405 t.mutex.lock();
15406 defer t.mutex.unlock();
15405 mutexLockUncancelable(&t.mutex);
15406 defer mutexUnlock(&t.mutex);
1540715407 if (t.random_file.handle) |handle| return handle;
1540815408 }
1540915409
......@@ -15437,8 +15437,8 @@ fn getCngHandle(t: *Threaded) Io.RandomSecureError!windows.HANDLE {
1543715437 )) {
1543815438 .SUCCESS => {
1543915439 syscall.finish();
15440 t.mutex.lock(); // Another thread might have won the race.
15441 defer t.mutex.unlock();
15440 mutexLockUncancelable(&t.mutex); // Another thread might have won the race.
15441 defer mutexUnlock(&t.mutex);
1544215442 if (t.random_file.handle) |prev_handle| {
1544315443 windows.CloseHandle(fresh_handle);
1544415444 return prev_handle;
......@@ -15458,8 +15458,8 @@ fn getCngHandle(t: *Threaded) Io.RandomSecureError!windows.HANDLE {
1545815458
1545915459fn getNulHandle(t: *Threaded) !windows.HANDLE {
1546015460 {
15461 t.mutex.lock();
15462 defer t.mutex.unlock();
15461 mutexLockUncancelable(&t.mutex);
15462 defer mutexUnlock(&t.mutex);
1546315463 if (t.null_file.handle) |handle| return handle;
1546415464 }
1546515465
......@@ -15505,8 +15505,8 @@ fn getNulHandle(t: *Threaded) !windows.HANDLE {
1550515505 )) {
1550615506 .SUCCESS => {
1550715507 syscall.finish();
15508 t.mutex.lock(); // Another thread might have won the race.
15509 defer t.mutex.unlock();
15508 mutexLockUncancelable(&t.mutex); // Another thread might have won the race.
15509 defer mutexUnlock(&t.mutex);
1551015510 if (t.null_file.handle) |prev_handle| {
1551115511 windows.CloseHandle(fresh_handle);
1551215512 return prev_handle;
......@@ -16551,15 +16551,15 @@ fn random(userdata: ?*anyopaque, buffer: []u8) void {
1655116551}
1655216552
1655316553fn randomMainThread(t: *Threaded, buffer: []u8) void {
16554 t.mutex.lock();
16555 defer t.mutex.unlock();
16554 mutexLockUncancelable(&t.mutex);
16555 defer mutexUnlock(&t.mutex);
1655616556
1655716557 if (!t.csprng.isInitialized()) {
1655816558 @branchHint(.unlikely);
1655916559 var seed: [Csprng.seed_len]u8 = undefined;
1656016560 {
16561 t.mutex.unlock();
16562 defer t.mutex.lock();
16561 mutexUnlock(&t.mutex);
16562 defer mutexLockUncancelable(&t.mutex);
1656316563
1656416564 const prev = swapCancelProtection(t, .blocked);
1656516565 defer _ = swapCancelProtection(t, prev);
......@@ -16744,8 +16744,8 @@ fn randomSecure(userdata: ?*anyopaque, buffer: []u8) Io.RandomSecureError!void {
1674416744
1674516745fn getRandomFd(t: *Threaded) Io.RandomSecureError!posix.fd_t {
1674616746 {
16747 t.mutex.lock();
16748 defer t.mutex.unlock();
16747 mutexLockUncancelable(&t.mutex);
16748 defer mutexUnlock(&t.mutex);
1674916749
1675016750 if (t.random_file.fd == -2) return error.EntropyUnavailable;
1675116751 if (t.random_file.fd != -1) return t.random_file.fd;
......@@ -16785,8 +16785,8 @@ fn getRandomFd(t: *Threaded) Io.RandomSecureError!posix.fd_t {
1678516785 .SUCCESS => {
1678616786 syscall.finish();
1678716787 if (!statx.mask.TYPE) return error.EntropyUnavailable;
16788 t.mutex.lock(); // Another thread might have won the race.
16789 defer t.mutex.unlock();
16788 mutexLockUncancelable(&t.mutex); // Another thread might have won the race.
16789 defer mutexUnlock(&t.mutex);
1679016790 if (t.random_file.fd >= 0) {
1679116791 posix.close(fd);
1679216792 return t.random_file.fd;
......@@ -16813,8 +16813,8 @@ fn getRandomFd(t: *Threaded) Io.RandomSecureError!posix.fd_t {
1681316813 switch (posix.errno(fstat_sym(fd, &stat))) {
1681416814 .SUCCESS => {
1681516815 syscall.finish();
16816 t.mutex.lock(); // Another thread might have won the race.
16817 defer t.mutex.unlock();
16816 mutexLockUncancelable(&t.mutex); // Another thread might have won the race.
16817 defer mutexUnlock(&t.mutex);
1681816818 if (t.random_file.fd >= 0) {
1681916819 posix.close(fd);
1682016820 return t.random_file.fd;
......@@ -16878,13 +16878,13 @@ const parking_futex = struct {
1687816878 /// avoid a race.
1687916879 num_waiters: std.atomic.Value(u32),
1688016880 /// Protects `waiters`.
16881 mutex: std.Thread.Mutex,
16881 mutex: Mutex,
1688216882 waiters: std.DoublyLinkedList,
1688316883
1688416884 /// Prevent false sharing between buckets.
1688516885 _: void align(std.atomic.cache_line) = {},
1688616886
16887 const init: Bucket = .{ .num_waiters = .init(0), .mutex = .{}, .waiters = .{} };
16887 const init: Bucket = .{ .num_waiters = .init(0), .mutex = .init, .waiters = .{} };
1688816888 };
1688916889
1689016890 const Waiter = struct {
......@@ -16947,8 +16947,8 @@ const parking_futex = struct {
1694716947 var status_buf: std.atomic.Value(Thread.Status) = undefined;
1694816948
1694916949 {
16950 bucket.mutex.lock();
16951 defer bucket.mutex.unlock();
16950 mutexLockUncancelable(&bucket.mutex);
16951 defer mutexUnlock(&bucket.mutex);
1695216952
1695316953 _ = bucket.num_waiters.fetchAdd(1, .acquire);
1695416954
......@@ -17017,8 +17017,8 @@ const parking_futex = struct {
1701717017 .parked => {
1701817018 // We saw a timeout and updated our own status from `.parked` to `.none`. It is
1701917019 // our responsibility to remove `waiter` from `bucket`.
17020 bucket.mutex.lock();
17021 defer bucket.mutex.unlock();
17020 mutexLockUncancelable(&bucket.mutex);
17021 defer mutexUnlock(&bucket.mutex);
1702217022 bucket.waiters.remove(&waiter.node);
1702317023 assert(bucket.num_waiters.fetchSub(1, .monotonic) > 0);
1702417024 },
......@@ -17057,8 +17057,8 @@ const parking_futex = struct {
1705717057 // of the critical section. This forms a singly-linked list of waiters using `Waiter.node.next`.
1705817058 var waking_head: ?*std.DoublyLinkedList.Node = null;
1705917059 {
17060 bucket.mutex.lock();
17061 defer bucket.mutex.unlock();
17060 mutexLockUncancelable(&bucket.mutex);
17061 defer mutexUnlock(&bucket.mutex);
1706217062
1706317063 var num_removed: u32 = 0;
1706417064 var it = bucket.waiters.first;
......@@ -17113,8 +17113,8 @@ const parking_futex = struct {
1711317113
1711417114 fn removeCanceledWaiter(waiter: *Waiter) void {
1711517115 const bucket = bucketForAddress(waiter.address);
17116 bucket.mutex.lock();
17117 defer bucket.mutex.unlock();
17116 mutexLockUncancelable(&bucket.mutex);
17117 defer mutexUnlock(&bucket.mutex);
1711817118 bucket.waiters.remove(&waiter.node);
1711917119 assert(bucket.num_waiters.fetchSub(1, .monotonic) > 0);
1712017120 waiter.done.store(true, .release); // potentially invalidates `waiter.*`
......@@ -18102,3 +18102,141 @@ fn eventSet(event: *Io.Event) void {
1810218102 .waiting => Thread.futexWake(@ptrCast(event), std.math.maxInt(u32)),
1810318103 }
1810418104}
18105
18106const Condition = if (!is_windows) Io.Condition else struct {
18107 condition: windows.CONDITION_VARIABLE,
18108 const init: @This() = .{ .condition = .{} };
18109};
18110
18111/// Same as `Io.Condition.broadcast` but avoids the VTable.
18112fn condBroadcast(cond: *Condition) void {
18113 if (is_windows) return windows.ntdll.RtlWakeAllConditionVariable(&cond.condition);
18114 var prev_state = cond.state.load(.monotonic);
18115 while (prev_state.waiters > prev_state.signals) {
18116 @branchHint(.unlikely);
18117 prev_state = cond.state.cmpxchgWeak(prev_state, .{
18118 .waiters = prev_state.waiters,
18119 .signals = prev_state.waiters,
18120 }, .release, .monotonic) orelse {
18121 // Update the epoch to tell the waiting threads that there are new signals for them.
18122 // Note that a waiting thread could miss a take if *exactly* (1<<32)-1 wakes happen
18123 // between it observing the epoch and sleeping on it, but this is extraordinarily
18124 // unlikely due to the precise number of calls required.
18125 _ = cond.epoch.fetchAdd(1, .release); // `.release` to ensure ordered after `state` update
18126 Thread.futexWake(&cond.epoch.raw, prev_state.waiters - prev_state.signals);
18127 return;
18128 };
18129 }
18130}
18131
18132/// Same as `Io.Condition.signal` but avoids the VTable.
18133fn condSignal(cond: *Condition) void {
18134 if (is_windows) return windows.ntdll.RtlWakeConditionVariable(&cond.condition);
18135 var prev_state = cond.state.load(.monotonic);
18136 while (prev_state.waiters > prev_state.signals) {
18137 @branchHint(.unlikely);
18138 prev_state = cond.state.cmpxchgWeak(prev_state, .{
18139 .waiters = prev_state.waiters,
18140 .signals = prev_state.signals + 1,
18141 }, .release, .monotonic) orelse {
18142 // Update the epoch to tell the waiting threads that there are new signals for them.
18143 // Note that a waiting thread could miss a take if *exactly* (1<<32)-1 wakes happen
18144 // between it observing the epoch and sleeping on it, but this is extraordinarily
18145 // unlikely due to the precise number of calls required.
18146 _ = cond.epoch.fetchAdd(1, .release); // `.release` to ensure ordered after `state` update
18147 Thread.futexWake(&cond.epoch.raw, 1);
18148 return;
18149 };
18150 }
18151}
18152
18153/// Same as `Io.Condition.waitUncancelable` but avoids the VTable.
18154fn condWait(cond: *Condition, mutex: *Mutex) void {
18155 if (is_windows) {
18156 _ = windows.kernel32.SleepConditionVariableSRW(&cond.condition, &mutex.srwlock, windows.INFINITE, 0);
18157 return;
18158 }
18159 var epoch = cond.epoch.load(.acquire); // `.acquire` to ensure ordered before state load
18160
18161 {
18162 const prev_state = cond.state.fetchAdd(.{ .waiters = 1, .signals = 0 }, .monotonic);
18163 assert(prev_state.waiters < std.math.maxInt(u16)); // overflow caused by too many waiters
18164 }
18165
18166 mutexUnlock(mutex);
18167 defer mutexLockUncancelable(mutex);
18168
18169 while (true) {
18170 Thread.futexWaitUncancelable(&cond.epoch.raw, epoch, null);
18171
18172 epoch = cond.epoch.load(.acquire); // `.acquire` to ensure ordered before `state` laod
18173
18174 var prev_state = cond.state.load(.monotonic);
18175 while (prev_state.signals > 0) {
18176 prev_state = cond.state.cmpxchgWeak(prev_state, .{
18177 .waiters = prev_state.waiters - 1,
18178 .signals = prev_state.signals - 1,
18179 }, .acquire, .monotonic) orelse {
18180 // We successfully consumed a signal.
18181 return;
18182 };
18183 }
18184 }
18185}
18186
18187const Mutex = if (!is_windows) Io.Mutex else struct {
18188 srwlock: windows.SRWLOCK,
18189 const init: @This() = .{ .srwlock = .{} };
18190};
18191
18192/// Same as `Io.Mutex.lockUncancelable` but avoids the VTable.
18193fn mutexLock(m: *Io.Mutex) Io.Cancelable!void {
18194 const initial_state = m.state.cmpxchgWeak(
18195 .unlocked,
18196 .locked_once,
18197 .acquire,
18198 .monotonic,
18199 ) orelse {
18200 @branchHint(.likely);
18201 return;
18202 };
18203 if (initial_state == .contended) {
18204 try Thread.futexWait(@ptrCast(&m.state.raw), @intFromEnum(Io.Mutex.State.contended), null);
18205 }
18206 while (m.state.swap(.contended, .acquire) != .unlocked) {
18207 try Thread.futexWait(@ptrCast(&m.state.raw), @intFromEnum(Io.Mutex.State.contended), null);
18208 }
18209}
18210
18211/// Same as `Io.Mutex.lockUncancelable` but avoids the VTable.
18212fn mutexLockUncancelable(m: *Mutex) void {
18213 if (is_windows) return windows.ntdll.RtlAcquireSRWLockExclusive(&m.srwlock);
18214 const initial_state = m.state.cmpxchgWeak(
18215 .unlocked,
18216 .locked_once,
18217 .acquire,
18218 .monotonic,
18219 ) orelse {
18220 @branchHint(.likely);
18221 return;
18222 };
18223 if (initial_state == .contended) {
18224 Thread.futexWaitUncancelable(@ptrCast(&m.state.raw), @intFromEnum(Io.Mutex.State.contended), null);
18225 }
18226 while (m.state.swap(.contended, .acquire) != .unlocked) {
18227 Thread.futexWaitUncancelable(@ptrCast(&m.state.raw), @intFromEnum(Io.Mutex.State.contended), null);
18228 }
18229}
18230
18231/// Same as `Io.Mutex.unlock` but avoids the VTable.
18232fn mutexUnlock(m: *Mutex) void {
18233 if (is_windows) return windows.ntdll.RtlReleaseSRWLockExclusive(&m.srwlock);
18234 switch (m.state.swap(.unlocked, .release)) {
18235 .unlocked => unreachable,
18236 .locked_once => {},
18237 .contended => {
18238 @branchHint(.unlikely);
18239 Thread.futexWake(@ptrCast(&m.state.raw), 1);
18240 },
18241 }
18242}