authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2026-02-02 23:14:05+01:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2026-02-02 23:14:05+01:00
logc2d4806d659abf8c4c0ab989eae225303de57af3
treed298fac5d9517452511e58fc7a987b7d44785264
parent1cd3af43fd74a8481b35d58c351ae056c1cba362
parent633eb247ab819d7120fea2e0178100cc8fca8ac3

Merge pull request 'std.Thread: remove redundant sync APIs ResetEvent and WaitGroup' (#31088) from sync-cleanup-smaller into master

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

7 files changed, 231 insertions(+), 365 deletions(-)

CMakeLists.txt-1
...@@ -410,7 +410,6 @@ set(ZIG_STAGE2_SOURCES...@@ -410,7 +410,6 @@ set(ZIG_STAGE2_SOURCES
410 lib/std/Thread.zig410 lib/std/Thread.zig
411 lib/std/Thread/Futex.zig411 lib/std/Thread/Futex.zig
412 lib/std/Thread/Mutex.zig412 lib/std/Thread/Mutex.zig
413 lib/std/Thread/WaitGroup.zig
414 lib/std/array_hash_map.zig413 lib/std/array_hash_map.zig
415 lib/std/array_list.zig414 lib/std/array_list.zig
416 lib/std/ascii.zig415 lib/std/ascii.zig
lib/std/Io.zig+2-10
...@@ -1236,15 +1236,7 @@ pub const Mutex = extern struct {...@@ -1236,15 +1236,7 @@ pub const Mutex = extern struct {
1236 };1236 };
12371237
1238 pub fn tryLock(m: *Mutex) bool {1238 pub fn tryLock(m: *Mutex) bool {
1239 switch (m.state.cmpxchgWeak(1239 return m.state.cmpxchgWeak(.unlocked, .locked_once, .acquire, .monotonic) == null;
1240 .unlocked,
1241 .locked_once,
1242 .acquire,
1243 .monotonic,
1244 ) orelse return true) {
1245 .unlocked => unreachable,
1246 .locked_once, .contended => return false,
1247 }
1248 }1240 }
12491241
1250 pub fn lock(m: *Mutex, io: Io) Cancelable!void {1242 pub fn lock(m: *Mutex, io: Io) Cancelable!void {
...@@ -1475,7 +1467,7 @@ pub const Event = enum(u32) {...@@ -1475,7 +1467,7 @@ pub const Event = enum(u32) {
1475 pub fn waitTimeout(event: *Event, io: Io, timeout: Timeout) WaitTimeoutError!void {1467 pub fn waitTimeout(event: *Event, io: Io, timeout: Timeout) WaitTimeoutError!void {
1476 if (@cmpxchgStrong(Event, event, .unset, .waiting, .acquire, .acquire)) |prev| switch (prev) {1468 if (@cmpxchgStrong(Event, event, .unset, .waiting, .acquire, .acquire)) |prev| switch (prev) {
1477 .unset => unreachable,1469 .unset => unreachable,
1478 .waiting => assert(!builtin.single_threaded), // invalid state1470 .waiting => {},
1479 .is_set => return,1471 .is_set => return,
1480 };1472 };
1481 errdefer {1473 errdefer {
lib/std/Io/Threaded.zig+65-3
...@@ -35,7 +35,7 @@ run_queue: std.SinglyLinkedList = .{},...@@ -35,7 +35,7 @@ run_queue: std.SinglyLinkedList = .{},
35join_requested: bool = false,35join_requested: bool = false,
36stack_size: usize,36stack_size: usize,
37/// All threads are spawned detached; this is how we wait until they all exit.37/// All threads are spawned detached; this is how we wait until they all exit.
38wait_group: std.Thread.WaitGroup = .{},38wait_group: WaitGroup = .init,
39async_limit: Io.Limit,39async_limit: Io.Limit,
40concurrent_limit: Io.Limit = .unlimited,40concurrent_limit: Io.Limit = .unlimited,
41/// Error from calling `std.Thread.getCpuCount` in `init`.41/// Error from calling `std.Thread.getCpuCount` in `init`.
...@@ -2461,7 +2461,10 @@ fn cancel(...@@ -2461,7 +2461,10 @@ fn cancel(
2461}2461}
24622462
2463fn futexWait(userdata: ?*anyopaque, ptr: *const u32, expected: u32, timeout: Io.Timeout) Io.Cancelable!void {2463fn futexWait(userdata: ?*anyopaque, ptr: *const u32, expected: u32, timeout: Io.Timeout) Io.Cancelable!void {
2464 if (builtin.single_threaded) unreachable; // Deadlock.2464 if (builtin.single_threaded) {
2465 assert(timeout != .none); // Deadlock.
2466 return;
2467 }
2465 const t: *Threaded = @ptrCast(@alignCast(userdata));2468 const t: *Threaded = @ptrCast(@alignCast(userdata));
2466 const t_io = ioBasic(t);2469 const t_io = ioBasic(t);
2467 const timeout_ns: ?u64 = ns: {2470 const timeout_ns: ?u64 = ns: {
...@@ -2479,7 +2482,7 @@ fn futexWaitUncancelable(userdata: ?*anyopaque, ptr: *const u32, expected: u32)...@@ -2479,7 +2482,7 @@ fn futexWaitUncancelable(userdata: ?*anyopaque, ptr: *const u32, expected: u32)
2479}2482}
24802483
2481fn futexWake(userdata: ?*anyopaque, ptr: *const u32, max_waiters: u32) void {2484fn futexWake(userdata: ?*anyopaque, ptr: *const u32, max_waiters: u32) void {
2482 if (builtin.single_threaded) unreachable; // Nothing to wake up.2485 if (builtin.single_threaded) return; // Nothing to wake up.
2483 const t: *Threaded = @ptrCast(@alignCast(userdata));2486 const t: *Threaded = @ptrCast(@alignCast(userdata));
2484 _ = t;2487 _ = t;
2485 Thread.futexWake(ptr, max_waiters);2488 Thread.futexWake(ptr, max_waiters);
...@@ -18040,3 +18043,62 @@ fn deviceIoControl(t: *Threaded, o: *const Io.Operation.DeviceIoControl) Io.Canc...@@ -18040,3 +18043,62 @@ fn deviceIoControl(t: *Threaded, o: *const Io.Operation.DeviceIoControl) Io.Canc
18040 }18043 }
18041 }18044 }
18042}18045}
18046
18047const WaitGroup = struct {
18048 state: std.atomic.Value(usize),
18049 event: Io.Event,
18050
18051 const init: WaitGroup = .{ .state = .{ .raw = 0 }, .event = .unset };
18052
18053 const is_waiting: usize = 1 << 0;
18054 const one_pending: usize = 1 << 1;
18055
18056 fn start(wg: *WaitGroup) void {
18057 const prev_state = wg.state.fetchAdd(one_pending, .monotonic);
18058 assert((prev_state / one_pending) < (std.math.maxInt(usize) / one_pending));
18059 }
18060
18061 fn value(wg: *WaitGroup) usize {
18062 return wg.state.load(.monotonic) / one_pending;
18063 }
18064
18065 fn wait(wg: *WaitGroup) void {
18066 const prev_state = wg.state.fetchAdd(is_waiting, .acquire);
18067 assert(prev_state & is_waiting == 0);
18068 if ((prev_state / one_pending) > 0) eventWait(&wg.event);
18069 }
18070
18071 fn finish(wg: *WaitGroup) void {
18072 const state = wg.state.fetchSub(one_pending, .acq_rel);
18073 assert((state / one_pending) > 0);
18074
18075 if (state == (one_pending | is_waiting)) {
18076 eventSet(&wg.event);
18077 }
18078 }
18079};
18080
18081/// Same as `Io.Event.wait` but avoids the VTable.
18082fn eventWait(event: *Io.Event) void {
18083 if (@cmpxchgStrong(Io.Event, event, .unset, .waiting, .acquire, .acquire)) |prev| switch (prev) {
18084 .unset => unreachable,
18085 .waiting => {},
18086 .is_set => return,
18087 };
18088 while (true) {
18089 Thread.futexWaitUncancelable(@ptrCast(event), @intFromEnum(Io.Event.waiting), null);
18090 switch (@atomicLoad(Io.Event, event, .acquire)) {
18091 .unset => unreachable, // `reset` called before pending `wait` returned
18092 .waiting => continue,
18093 .is_set => return,
18094 }
18095 }
18096}
18097
18098/// Same as `Io.Event.set` but avoids the VTable.
18099fn eventSet(event: *Io.Event) void {
18100 switch (@atomicRmw(Io.Event, event, .Xchg, .is_set, .release)) {
18101 .unset, .is_set => {},
18102 .waiting => Thread.futexWake(@ptrCast(event), std.math.maxInt(u32)),
18103 }
18104}
lib/std/Io/test.zig+133
...@@ -716,3 +716,136 @@ test "read from a file using Batch.awaitAsync API" {...@@ -716,3 +716,136 @@ test "read from a file using Batch.awaitAsync API" {
716 }716 }
717 }717 }
718}718}
719
720test "Event smoke test" {
721 const io = testing.io;
722
723 var event: Io.Event = .unset;
724 try testing.expectEqual(false, event.isSet());
725
726 // make sure the event gets set
727 event.set(io);
728 try testing.expectEqual(true, event.isSet());
729
730 // make sure the event gets unset again
731 event.reset();
732 try testing.expectEqual(false, event.isSet());
733
734 // waits should timeout as there's no other thread to set the event
735 try testing.expectError(error.Timeout, event.waitTimeout(io, .{ .duration = .{
736 .raw = .zero,
737 .clock = .awake,
738 } }));
739 try testing.expectError(error.Timeout, event.waitTimeout(io, .{ .duration = .{
740 .raw = .fromMilliseconds(1),
741 .clock = .awake,
742 } }));
743
744 // set the event again and make sure waits complete
745 event.set(io);
746 try event.wait(io);
747 try event.waitTimeout(io, .{ .duration = .{ .raw = .fromMilliseconds(1), .clock = .awake } });
748 try testing.expectEqual(true, event.isSet());
749}
750
751test "Event signaling" {
752 if (builtin.single_threaded) {
753 // This test requires spawning threads.
754 return error.SkipZigTest;
755 }
756
757 const io = testing.io;
758
759 const Context = struct {
760 in: Io.Event = .unset,
761 out: Io.Event = .unset,
762 value: usize = 0,
763
764 fn input(self: *@This()) !void {
765 // wait for the value to become 1
766 try self.in.wait(io);
767 self.in.reset();
768 try testing.expectEqual(self.value, 1);
769
770 // bump the value and wake up output()
771 self.value = 2;
772 self.out.set(io);
773
774 // wait for output to receive 2, bump the value and wake us up with 3
775 try self.in.wait(io);
776 self.in.reset();
777 try testing.expectEqual(self.value, 3);
778
779 // bump the value and wake up output() for it to see 4
780 self.value = 4;
781 self.out.set(io);
782 }
783
784 fn output(self: *@This()) !void {
785 // start with 0 and bump the value for input to see 1
786 try testing.expectEqual(self.value, 0);
787 self.value = 1;
788 self.in.set(io);
789
790 // wait for input to receive 1, bump the value to 2 and wake us up
791 try self.out.wait(io);
792 self.out.reset();
793 try testing.expectEqual(self.value, 2);
794
795 // bump the value to 3 for input to see (rhymes)
796 self.value = 3;
797 self.in.set(io);
798
799 // wait for input to bump the value to 4 and receive no more (rhymes)
800 try self.out.wait(io);
801 self.out.reset();
802 try testing.expectEqual(self.value, 4);
803 }
804 };
805
806 var ctx = Context{};
807
808 const thread = try std.Thread.spawn(.{}, Context.output, .{&ctx});
809 defer thread.join();
810
811 try ctx.input();
812}
813
814test "Event broadcast" {
815 if (builtin.single_threaded) {
816 // This test requires spawning threads.
817 return error.SkipZigTest;
818 }
819
820 const io = testing.io;
821
822 const num_threads = 10;
823 const Barrier = struct {
824 event: Io.Event = .unset,
825 counter: std.atomic.Value(usize) = std.atomic.Value(usize).init(num_threads),
826
827 fn wait(self: *@This()) void {
828 if (self.counter.fetchSub(1, .acq_rel) == 1) {
829 self.event.set(io);
830 }
831 }
832 };
833
834 const Context = struct {
835 start_barrier: Barrier = .{},
836 finish_barrier: Barrier = .{},
837
838 fn run(self: *@This()) void {
839 self.start_barrier.wait();
840 self.finish_barrier.wait();
841 }
842 };
843
844 var ctx = Context{};
845 var threads: [num_threads - 1]std.Thread = undefined;
846
847 for (&threads) |*t| t.* = try std.Thread.spawn(.{}, Context.run, .{&ctx});
848 defer for (threads) |t| t.join();
849
850 ctx.run();
851}
lib/std/Thread.zig+20-256
...@@ -19,129 +19,11 @@ pub const Mutex = @import("Thread/Mutex.zig");...@@ -19,129 +19,11 @@ pub const Mutex = @import("Thread/Mutex.zig");
19pub const Semaphore = @import("Thread/Semaphore.zig");19pub const Semaphore = @import("Thread/Semaphore.zig");
20pub const Condition = @import("Thread/Condition.zig");20pub const Condition = @import("Thread/Condition.zig");
21pub const RwLock = @import("Thread/RwLock.zig");21pub const RwLock = @import("Thread/RwLock.zig");
22pub const WaitGroup = @import("Thread/WaitGroup.zig");
2322
24pub const Pool = @compileError("deprecated; consider using 'std.Io.Group' with 'std.Io.Threaded'");23pub const Pool = @compileError("deprecated; consider using 'std.Io.Group' with 'std.Io.Threaded'");
2524
26pub const use_pthreads = native_os != .windows and native_os != .wasi and builtin.link_libc;25pub const use_pthreads = native_os != .windows and native_os != .wasi and builtin.link_libc;
2726
28/// A thread-safe logical boolean value which can be `set` and `unset`.
29///
30/// It can also block threads until the value is set with cancelation via timed
31/// waits. Statically initializable; four bytes on all targets.
32pub const ResetEvent = enum(u32) {
33 unset = 0,
34 waiting = 1,
35 is_set = 2,
36
37 /// Returns whether the logical boolean is `set`.
38 ///
39 /// Once `reset` is called, this returns false until the next `set`.
40 ///
41 /// The memory accesses before the `set` can be said to happen before
42 /// `isSet` returns true.
43 pub fn isSet(re: *const ResetEvent) bool {
44 if (builtin.single_threaded) return switch (re.*) {
45 .unset => false,
46 .waiting => unreachable,
47 .is_set => true,
48 };
49 // Acquire barrier ensures memory accesses before `set` happen before
50 // returning true.
51 return @atomicLoad(ResetEvent, re, .acquire) == .is_set;
52 }
53
54 /// Blocks the calling thread until `set` is called.
55 ///
56 /// This is effectively a more efficient version of `while (!isSet()) {}`.
57 ///
58 /// The memory accesses before the `set` can be said to happen before `wait` returns.
59 pub fn wait(re: *ResetEvent) void {
60 if (builtin.single_threaded) switch (re.*) {
61 .unset => unreachable, // Deadlock, no other threads to wake us up.
62 .waiting => unreachable, // Invalid state.
63 .is_set => return,
64 };
65 if (!re.isSet()) return timedWaitInner(re, null) catch |err| switch (err) {
66 error.Timeout => unreachable, // No timeout specified.
67 };
68 }
69
70 /// Blocks the calling thread until `set` is called, or until the
71 /// corresponding timeout expires, returning `error.Timeout`.
72 ///
73 /// This is effectively a more efficient version of `while (!isSet()) {}`.
74 ///
75 /// The memory accesses before the set() can be said to happen before
76 /// timedWait() returns without error.
77 pub fn timedWait(re: *ResetEvent, timeout_ns: u64) error{Timeout}!void {
78 if (builtin.single_threaded) switch (re.*) {
79 .unset => return error.Timeout,
80 .waiting => unreachable, // Invalid state.
81 .is_set => return,
82 };
83 if (!re.isSet()) return timedWaitInner(re, timeout_ns);
84 }
85
86 fn timedWaitInner(re: *ResetEvent, timeout: ?u64) error{Timeout}!void {
87 @branchHint(.cold);
88
89 // Try to set the state from `unset` to `waiting` to indicate to the
90 // `set` thread that others are blocked on the ResetEvent. Avoid using
91 // any strict barriers until we know the ResetEvent is set.
92 var state = @atomicLoad(ResetEvent, re, .acquire);
93 if (state == .unset) {
94 state = @cmpxchgStrong(ResetEvent, re, state, .waiting, .acquire, .acquire) orelse .waiting;
95 }
96
97 // Wait until the ResetEvent is set since the state is waiting.
98 if (state == .waiting) {
99 var futex_deadline = Futex.Deadline.init(timeout);
100 while (true) {
101 const wait_result = futex_deadline.wait(@ptrCast(re), @intFromEnum(ResetEvent.waiting));
102
103 // Check if the ResetEvent was set before possibly reporting error.Timeout below.
104 state = @atomicLoad(ResetEvent, re, .acquire);
105 if (state != .waiting) break;
106
107 try wait_result;
108 }
109 }
110
111 assert(state == .is_set);
112 }
113
114 /// Marks the logical boolean as `set` and unblocks any threads in `wait`
115 /// or `timedWait` to observe the new state.
116 ///
117 /// The logical boolean stays `set` until `reset` is called, making future
118 /// `set` calls do nothing semantically.
119 ///
120 /// The memory accesses before `set` can be said to happen before `isSet`
121 /// returns true or `wait`/`timedWait` return successfully.
122 pub fn set(re: *ResetEvent) void {
123 if (builtin.single_threaded) {
124 re.* = .is_set;
125 return;
126 }
127 if (@atomicRmw(ResetEvent, re, .Xchg, .is_set, .release) == .waiting) {
128 Futex.wake(@ptrCast(re), std.math.maxInt(u32));
129 }
130 }
131
132 /// Unmarks the ResetEvent as if `set` was never called.
133 ///
134 /// Assumes no threads are blocked in `wait` or `timedWait`. Concurrent
135 /// calls to `set`, `isSet` and `reset` are allowed.
136 pub fn reset(re: *ResetEvent) void {
137 if (builtin.single_threaded) {
138 re.* = .unset;
139 return;
140 }
141 @atomicStore(ResetEvent, re, .unset, .monotonic);
142 }
143};
144
145const Thread = @This();27const Thread = @This();
146const Impl = if (native_os == .windows)28const Impl = if (native_os == .windows)
147 WindowsThreadImpl29 WindowsThreadImpl
...@@ -1676,16 +1558,16 @@ test "setName, getName" {...@@ -1676,16 +1558,16 @@ test "setName, getName" {
1676 const io = testing.io;1558 const io = testing.io;
16771559
1678 const Context = struct {1560 const Context = struct {
1679 start_wait_event: ResetEvent = .unset,1561 start_wait_event: Io.Event = .unset,
1680 test_done_event: ResetEvent = .unset,1562 test_done_event: Io.Event = .unset,
1681 thread_done_event: ResetEvent = .unset,1563 thread_done_event: Io.Event = .unset,
16821564
1683 done: std.atomic.Value(bool) = std.atomic.Value(bool).init(false),1565 done: std.atomic.Value(bool) = std.atomic.Value(bool).init(false),
1684 thread: Thread = undefined,1566 thread: Thread = undefined,
16851567
1686 pub fn run(ctx: *@This()) !void {1568 pub fn run(ctx: *@This()) !void {
1687 // Wait for the main thread to have set the thread field in the context.1569 // Wait for the main thread to have set the thread field in the context.
1688 ctx.start_wait_event.wait();1570 try ctx.start_wait_event.wait(io);
16891571
1690 switch (native_os) {1572 switch (native_os) {
1691 .windows => testThreadName(io, &ctx.thread) catch |err| switch (err) {1573 .windows => testThreadName(io, &ctx.thread) catch |err| switch (err) {
...@@ -1696,10 +1578,10 @@ test "setName, getName" {...@@ -1696,10 +1578,10 @@ test "setName, getName" {
1696 }1578 }
16971579
1698 // Signal our test is done1580 // Signal our test is done
1699 ctx.test_done_event.set();1581 ctx.test_done_event.set(io);
17001582
1701 // wait for the thread to property exit1583 // wait for the thread to property exit
1702 ctx.thread_done_event.wait();1584 try ctx.thread_done_event.wait(io);
1703 }1585 }
1704 };1586 };
17051587
...@@ -1707,8 +1589,8 @@ test "setName, getName" {...@@ -1707,8 +1589,8 @@ test "setName, getName" {
1707 var thread = try spawn(.{}, Context.run, .{&context});1589 var thread = try spawn(.{}, Context.run, .{&context});
17081590
1709 context.thread = thread;1591 context.thread = thread;
1710 context.start_wait_event.set();1592 context.start_wait_event.set(io);
1711 context.test_done_event.wait();1593 try context.test_done_event.wait(io);
17121594
1713 switch (native_os) {1595 switch (native_os) {
1714 .driverkit, .ios, .maccatalyst, .macos, .tvos, .visionos, .watchos => {1596 .driverkit, .ios, .maccatalyst, .macos, .tvos, .visionos, .watchos => {
...@@ -1722,31 +1604,32 @@ test "setName, getName" {...@@ -1722,31 +1604,32 @@ test "setName, getName" {
1722 else => try testThreadName(io, &thread),1604 else => try testThreadName(io, &thread),
1723 }1605 }
17241606
1725 context.thread_done_event.set();1607 context.thread_done_event.set(io);
1726 thread.join();1608 thread.join();
1727}1609}
17281610
1729test {1611test {
1730 _ = Futex;1612 _ = Futex;
1731 _ = ResetEvent;
1732 _ = Mutex;1613 _ = Mutex;
1733 _ = Semaphore;1614 _ = Semaphore;
1734 _ = Condition;1615 _ = Condition;
1735 _ = RwLock;1616 _ = RwLock;
1736}1617}
17371618
1738fn testIncrementNotify(value: *usize, event: *ResetEvent) void {1619fn testIncrementNotify(io: Io, value: *usize, event: *Io.Event) void {
1739 value.* += 1;1620 value.* += 1;
1740 event.set();1621 event.set(io);
1741}1622}
17421623
1743test join {1624test join {
1744 if (builtin.single_threaded) return error.SkipZigTest;1625 if (builtin.single_threaded) return error.SkipZigTest;
17451626
1627 const io = testing.io;
1628
1746 var value: usize = 0;1629 var value: usize = 0;
1747 var event: ResetEvent = .unset;1630 var event: Io.Event = .unset;
17481631
1749 const thread = try Thread.spawn(.{}, testIncrementNotify, .{ &value, &event });1632 const thread = try Thread.spawn(.{}, testIncrementNotify, .{ io, &value, &event });
1750 thread.join();1633 thread.join();
17511634
1752 try std.testing.expectEqual(value, 1);1635 try std.testing.expectEqual(value, 1);
...@@ -1755,13 +1638,15 @@ test join {...@@ -1755,13 +1638,15 @@ test join {
1755test detach {1638test detach {
1756 if (builtin.single_threaded) return error.SkipZigTest;1639 if (builtin.single_threaded) return error.SkipZigTest;
17571640
1641 const io = testing.io;
1642
1758 var value: usize = 0;1643 var value: usize = 0;
1759 var event: ResetEvent = .unset;1644 var event: Io.Event = .unset;
17601645
1761 const thread = try Thread.spawn(.{}, testIncrementNotify, .{ &value, &event });1646 const thread = try Thread.spawn(.{}, testIncrementNotify, .{ io, &value, &event });
1762 thread.detach();1647 thread.detach();
17631648
1764 event.wait();1649 try event.wait(io);
1765 try std.testing.expectEqual(value, 1);1650 try std.testing.expectEqual(value, 1);
1766}1651}
17671652
...@@ -1803,127 +1688,6 @@ fn testTls() !void {...@@ -1803,127 +1688,6 @@ fn testTls() !void {
1803 if (x != 1235) return error.TlsBadEndValue;1688 if (x != 1235) return error.TlsBadEndValue;
1804}1689}
18051690
1806test "ResetEvent smoke test" {
1807 var event: ResetEvent = .unset;
1808 try testing.expectEqual(false, event.isSet());
1809
1810 // make sure the event gets set
1811 event.set();
1812 try testing.expectEqual(true, event.isSet());
1813
1814 // make sure the event gets unset again
1815 event.reset();
1816 try testing.expectEqual(false, event.isSet());
1817
1818 // waits should timeout as there's no other thread to set the event
1819 try testing.expectError(error.Timeout, event.timedWait(0));
1820 try testing.expectError(error.Timeout, event.timedWait(std.time.ns_per_ms));
1821
1822 // set the event again and make sure waits complete
1823 event.set();
1824 event.wait();
1825 try event.timedWait(std.time.ns_per_ms);
1826 try testing.expectEqual(true, event.isSet());
1827}
1828
1829test "ResetEvent signaling" {
1830 // This test requires spawning threads
1831 if (builtin.single_threaded) {
1832 return error.SkipZigTest;
1833 }
1834
1835 const Context = struct {
1836 in: ResetEvent = .unset,
1837 out: ResetEvent = .unset,
1838 value: usize = 0,
1839
1840 fn input(self: *@This()) !void {
1841 // wait for the value to become 1
1842 self.in.wait();
1843 self.in.reset();
1844 try testing.expectEqual(self.value, 1);
1845
1846 // bump the value and wake up output()
1847 self.value = 2;
1848 self.out.set();
1849
1850 // wait for output to receive 2, bump the value and wake us up with 3
1851 self.in.wait();
1852 self.in.reset();
1853 try testing.expectEqual(self.value, 3);
1854
1855 // bump the value and wake up output() for it to see 4
1856 self.value = 4;
1857 self.out.set();
1858 }
1859
1860 fn output(self: *@This()) !void {
1861 // start with 0 and bump the value for input to see 1
1862 try testing.expectEqual(self.value, 0);
1863 self.value = 1;
1864 self.in.set();
1865
1866 // wait for input to receive 1, bump the value to 2 and wake us up
1867 self.out.wait();
1868 self.out.reset();
1869 try testing.expectEqual(self.value, 2);
1870
1871 // bump the value to 3 for input to see (rhymes)
1872 self.value = 3;
1873 self.in.set();
1874
1875 // wait for input to bump the value to 4 and receive no more (rhymes)
1876 self.out.wait();
1877 self.out.reset();
1878 try testing.expectEqual(self.value, 4);
1879 }
1880 };
1881
1882 var ctx = Context{};
1883
1884 const thread = try std.Thread.spawn(.{}, Context.output, .{&ctx});
1885 defer thread.join();
1886
1887 try ctx.input();
1888}
1889
1890test "ResetEvent broadcast" {
1891 // This test requires spawning threads
1892 if (builtin.single_threaded) {
1893 return error.SkipZigTest;
1894 }
1895
1896 const num_threads = 10;
1897 const Barrier = struct {
1898 event: ResetEvent = .unset,
1899 counter: std.atomic.Value(usize) = std.atomic.Value(usize).init(num_threads),
1900
1901 fn wait(self: *@This()) void {
1902 if (self.counter.fetchSub(1, .acq_rel) == 1) {
1903 self.event.set();
1904 }
1905 }
1906 };
1907
1908 const Context = struct {
1909 start_barrier: Barrier = .{},
1910 finish_barrier: Barrier = .{},
1911
1912 fn run(self: *@This()) void {
1913 self.start_barrier.wait();
1914 self.finish_barrier.wait();
1915 }
1916 };
1917
1918 var ctx = Context{};
1919 var threads: [num_threads - 1]std.Thread = undefined;
1920
1921 for (&threads) |*t| t.* = try std.Thread.spawn(.{}, Context.run, .{&ctx});
1922 defer for (threads) |t| t.join();
1923
1924 ctx.run();
1925}
1926
1927/// Configures the per-thread alternative signal stack requested by `std.options.signal_stack_size`.1691/// Configures the per-thread alternative signal stack requested by `std.options.signal_stack_size`.
1928pub fn maybeAttachSignalStack() void {1692pub fn maybeAttachSignalStack() void {
1929 const size = std.options.signal_stack_size orelse return;1693 const size = std.options.signal_stack_size orelse return;
lib/std/Thread/WaitGroup.zig deleted-87
...@@ -1,87 +0,0 @@
1const builtin = @import("builtin");
2const std = @import("std");
3const assert = std.debug.assert;
4const WaitGroup = @This();
5
6const is_waiting: usize = 1 << 0;
7const one_pending: usize = 1 << 1;
8
9state: std.atomic.Value(usize) = std.atomic.Value(usize).init(0),
10event: std.Thread.ResetEvent = .unset,
11
12pub fn start(self: *WaitGroup) void {
13 return startStateless(&self.state);
14}
15
16pub fn startStateless(state: *std.atomic.Value(usize)) void {
17 const prev_state = state.fetchAdd(one_pending, .monotonic);
18 assert((prev_state / one_pending) < (std.math.maxInt(usize) / one_pending));
19}
20
21pub fn startMany(self: *WaitGroup, n: usize) void {
22 const state = self.state.fetchAdd(one_pending * n, .monotonic);
23 assert((state / one_pending) < (std.math.maxInt(usize) / one_pending));
24}
25
26pub fn finish(self: *WaitGroup) void {
27 const state = self.state.fetchSub(one_pending, .acq_rel);
28 assert((state / one_pending) > 0);
29
30 if (state == (one_pending | is_waiting)) {
31 self.event.set();
32 }
33}
34
35pub fn finishStateless(state: *std.atomic.Value(usize), event: *std.Thread.ResetEvent) void {
36 const prev_state = state.fetchSub(one_pending, .acq_rel);
37 assert((prev_state / one_pending) > 0);
38 if (prev_state == (one_pending | is_waiting)) event.set();
39}
40
41pub fn wait(wg: *WaitGroup) void {
42 return waitStateless(&wg.state, &wg.event);
43}
44
45pub fn waitStateless(state: *std.atomic.Value(usize), event: *std.Thread.ResetEvent) void {
46 const prev_state = state.fetchAdd(is_waiting, .acquire);
47 assert(prev_state & is_waiting == 0);
48 if ((prev_state / one_pending) > 0) event.wait();
49}
50
51pub fn reset(self: *WaitGroup) void {
52 self.state.store(0, .monotonic);
53 self.event.reset();
54}
55
56pub fn isDone(wg: *WaitGroup) bool {
57 const state = wg.state.load(.acquire);
58 assert(state & is_waiting == 0);
59
60 return (state / one_pending) == 0;
61}
62
63pub fn value(wg: *WaitGroup) usize {
64 return wg.state.load(.monotonic) / one_pending;
65}
66
67// Spawns a new thread for the task. This is appropriate when the callee
68// delegates all work.
69pub fn spawnManager(
70 wg: *WaitGroup,
71 comptime func: anytype,
72 args: anytype,
73) void {
74 if (builtin.single_threaded) {
75 @call(.auto, func, args);
76 return;
77 }
78 const Manager = struct {
79 fn run(wg_inner: *WaitGroup, args_inner: @TypeOf(args)) void {
80 defer wg_inner.finish();
81 @call(.auto, func, args_inner);
82 }
83 };
84 wg.start();
85 const t = std.Thread.spawn(.{}, Manager.run, .{ wg, args }) catch return Manager.run(wg, args);
86 t.detach();
87}
lib/std/fs/test.zig+11-8
...@@ -1745,29 +1745,32 @@ test "open file with exclusive lock twice, make sure second lock waits" {...@@ -1745,29 +1745,32 @@ test "open file with exclusive lock twice, make sure second lock waits" {
1745 errdefer file.close(io);1745 errdefer file.close(io);
17461746
1747 const S = struct {1747 const S = struct {
1748 fn checkFn(inner_ctx: *TestContext, path: []const u8, started: *std.Thread.ResetEvent, locked: *std.Thread.ResetEvent) !void {1748 fn checkFn(inner_ctx: *TestContext, path: []const u8, started: *Io.Event, locked: *Io.Event) !void {
1749 started.set();1749 started.set(inner_ctx.io);
1750 const file1 = try inner_ctx.dir.createFile(inner_ctx.io, path, .{ .lock = .exclusive });1750 const file1 = try inner_ctx.dir.createFile(inner_ctx.io, path, .{ .lock = .exclusive });
17511751
1752 locked.set();1752 locked.set(inner_ctx.io);
1753 file1.close(inner_ctx.io);1753 file1.close(inner_ctx.io);
1754 }1754 }
1755 };1755 };
17561756
1757 var started: std.Thread.ResetEvent = .unset;1757 var started: Io.Event = .unset;
1758 var locked: std.Thread.ResetEvent = .unset;1758 var locked: Io.Event = .unset;
17591759
1760 const t = try std.Thread.spawn(.{}, S.checkFn, .{ ctx, filename, &started, &locked });1760 const t = try std.Thread.spawn(.{}, S.checkFn, .{ ctx, filename, &started, &locked });
1761 defer t.join();1761 defer t.join();
17621762
1763 // Wait for the spawned thread to start trying to acquire the exclusive file lock.1763 // Wait for the spawned thread to start trying to acquire the exclusive file lock.
1764 // Then wait a bit to make sure that can't acquire it since we currently hold the file lock.1764 // Then wait a bit to make sure that can't acquire it since we currently hold the file lock.
1765 started.wait();1765 try started.wait(io);
1766 try expectError(error.Timeout, locked.timedWait(10 * std.time.ns_per_ms));1766 try expectError(error.Timeout, locked.waitTimeout(io, .{ .duration = .{
1767 .raw = .fromMilliseconds(10),
1768 .clock = .awake,
1769 } }));
17671770
1768 // Release the file lock which should unlock the thread to lock it and set the locked event.1771 // Release the file lock which should unlock the thread to lock it and set the locked event.
1769 file.close(io);1772 file.close(io);
1770 locked.wait();1773 try locked.wait(io);
1771 }1774 }
1772 }.impl);1775 }.impl);
1773}1776}