authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-09-26 21:58:06-07:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-10-02 16:30:59-07:00
logd37d795cba27c9793436449bfd783fc504c69cbb
tree40f3d0daa07ae816564761ad653b2317d4b270f8
parent96dc44b5eadd5104783c6cc8f60032e4221ca20a

std.Thread.ResetEvent: make it more reusable

5 files changed, 264 insertions(+), 290 deletions(-)

CMakeLists.txt-1
...@@ -413,7 +413,6 @@ set(ZIG_STAGE2_SOURCES...@@ -413,7 +413,6 @@ set(ZIG_STAGE2_SOURCES
413 lib/std/Thread/Futex.zig413 lib/std/Thread/Futex.zig
414 lib/std/Thread/Mutex.zig414 lib/std/Thread/Mutex.zig
415 lib/std/Thread/Pool.zig415 lib/std/Thread/Pool.zig
416 lib/std/Thread/ResetEvent.zig
417 lib/std/Thread/WaitGroup.zig416 lib/std/Thread/WaitGroup.zig
418 lib/std/array_hash_map.zig417 lib/std/array_hash_map.zig
419 lib/std/array_list.zig418 lib/std/array_list.zig
lib/std/Progress.zig+1-1
...@@ -392,7 +392,7 @@ var global_progress: Progress = .{...@@ -392,7 +392,7 @@ var global_progress: Progress = .{
392 .terminal = undefined,392 .terminal = undefined,
393 .terminal_mode = .off,393 .terminal_mode = .off,
394 .update_thread = null,394 .update_thread = null,
395 .redraw_event = .{},395 .redraw_event = .unset,
396 .refresh_rate_ns = undefined,396 .refresh_rate_ns = undefined,
397 .initial_delay_ns = undefined,397 .initial_delay_ns = undefined,
398 .rows = 0,398 .rows = 0,
lib/std/Thread.zig+243-1
...@@ -10,9 +10,9 @@ const target = builtin.target;...@@ -10,9 +10,9 @@ const target = builtin.target;
10const native_os = builtin.os.tag;10const native_os = builtin.os.tag;
11const posix = std.posix;11const posix = std.posix;
12const windows = std.os.windows;12const windows = std.os.windows;
13const testing = std.testing;
1314
14pub const Futex = @import("Thread/Futex.zig");15pub const Futex = @import("Thread/Futex.zig");
15pub const ResetEvent = @import("Thread/ResetEvent.zig");
16pub const Mutex = @import("Thread/Mutex.zig");16pub const Mutex = @import("Thread/Mutex.zig");
17pub const Semaphore = @import("Thread/Semaphore.zig");17pub const Semaphore = @import("Thread/Semaphore.zig");
18pub const Condition = @import("Thread/Condition.zig");18pub const Condition = @import("Thread/Condition.zig");
...@@ -22,6 +22,126 @@ pub const WaitGroup = @import("Thread/WaitGroup.zig");...@@ -22,6 +22,126 @@ pub const WaitGroup = @import("Thread/WaitGroup.zig");
2222
23pub const use_pthreads = native_os != .windows and native_os != .wasi and builtin.link_libc;23pub const use_pthreads = native_os != .windows and native_os != .wasi and builtin.link_libc;
2424
25/// A thread-safe logical boolean value which can be `set` and `unset`.
26///
27/// It can also block threads until the value is set with cancelation via timed
28/// waits. Statically initializable; four bytes on all targets.
29pub const ResetEvent = enum(u32) {
30 unset = 0,
31 waiting = 1,
32 is_set = 2,
33
34 /// Returns whether the logical boolean is `set`.
35 ///
36 /// Once `reset` is called, this returns false until the next `set`.
37 ///
38 /// The memory accesses before the `set` can be said to happen before
39 /// `isSet` returns true.
40 pub fn isSet(re: *const ResetEvent) bool {
41 if (builtin.single_threaded) return switch (re.*) {
42 .unset => false,
43 .waiting => unreachable,
44 .is_set => true,
45 };
46 // Acquire barrier ensures memory accesses before `set` happen before
47 // returning true.
48 return @atomicLoad(ResetEvent, re, .acquire) == .is_set;
49 }
50
51 /// Blocks the calling thread until `set` is called.
52 ///
53 /// This is effectively a more efficient version of `while (!isSet()) {}`.
54 ///
55 /// The memory accesses before the `set` can be said to happen before `wait` returns.
56 pub fn wait(re: *ResetEvent) void {
57 if (builtin.single_threaded) switch (re.*) {
58 .unset => unreachable, // Deadlock, no other threads to wake us up.
59 .waiting => unreachable, // Invalid state.
60 .is_set => return,
61 };
62 if (!re.isSet()) return timedWaitInner(re, null) catch |err| switch (err) {
63 error.Timeout => unreachable, // No timeout specified.
64 };
65 }
66
67 /// Blocks the calling thread until `set` is called, or until the
68 /// corresponding timeout expires, returning `error.Timeout`.
69 ///
70 /// This is effectively a more efficient version of `while (!isSet()) {}`.
71 ///
72 /// The memory accesses before the set() can be said to happen before
73 /// timedWait() returns without error.
74 pub fn timedWait(re: *ResetEvent, timeout_ns: u64) void {
75 if (builtin.single_threaded) switch (re.*) {
76 .unset => {
77 sleep(timeout_ns);
78 return error.Timeout;
79 },
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
25/// Spurious wakeups are possible and no precision of timing is guaranteed.145/// Spurious wakeups are possible and no precision of timing is guaranteed.
26pub fn sleep(nanoseconds: u64) void {146pub fn sleep(nanoseconds: u64) void {
27 if (builtin.os.tag == .windows) {147 if (builtin.os.tag == .windows) {
...@@ -1681,3 +1801,125 @@ fn testTls() !void {...@@ -1681,3 +1801,125 @@ fn testTls() !void {
1681 x += 1;1801 x += 1;
1682 if (x != 1235) return error.TlsBadEndValue;1802 if (x != 1235) return error.TlsBadEndValue;
1683}1803}
1804
1805test "ResetEvent smoke test" {
1806 // make sure the event is unset
1807 var event = ResetEvent{};
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 = .{},
1837 out: ResetEvent = .{},
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 = .{},
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}
lib/std/Thread/ResetEvent.zig deleted-278
...@@ -1,278 +0,0 @@
1//! ResetEvent is a thread-safe bool which can be set to true/false ("set"/"unset").
2//! It can also block threads until the "bool" is set with cancellation via timed waits.
3//! ResetEvent can be statically initialized and is at most `@sizeOf(u64)` large.
4
5const std = @import("../std.zig");
6const builtin = @import("builtin");
7const ResetEvent = @This();
8
9const os = std.os;
10const assert = std.debug.assert;
11const testing = std.testing;
12const Futex = std.Thread.Futex;
13
14impl: Impl = .{},
15
16/// Returns if the ResetEvent was set().
17/// Once reset() is called, this returns false until the next set().
18/// The memory accesses before the set() can be said to happen before isSet() returns true.
19pub fn isSet(self: *const ResetEvent) bool {
20 return self.impl.isSet();
21}
22
23/// Block's the callers thread until the ResetEvent is set().
24/// This is effectively a more efficient version of `while (!isSet()) {}`.
25/// The memory accesses before the set() can be said to happen before wait() returns.
26pub fn wait(self: *ResetEvent) void {
27 self.impl.wait(null) catch |err| switch (err) {
28 error.Timeout => unreachable, // no timeout provided so we shouldn't have timed-out
29 };
30}
31
32/// Block's the callers thread until the ResetEvent is set(), or until the corresponding timeout expires.
33/// If the timeout expires before the ResetEvent is set, `error.Timeout` is returned.
34/// This is effectively a more efficient version of `while (!isSet()) {}`.
35/// The memory accesses before the set() can be said to happen before timedWait() returns without error.
36pub fn timedWait(self: *ResetEvent, timeout_ns: u64) error{Timeout}!void {
37 return self.impl.wait(timeout_ns);
38}
39
40/// Marks the ResetEvent as "set" and unblocks any threads in `wait()` or `timedWait()` to observe the new state.
41/// The ResetEvent says "set" until reset() is called, making future set() calls do nothing semantically.
42/// The memory accesses before set() can be said to happen before isSet() returns true or wait()/timedWait() return successfully.
43pub fn set(self: *ResetEvent) void {
44 self.impl.set();
45}
46
47/// Unmarks the ResetEvent from its "set" state if set() was called previously.
48/// It is undefined behavior is reset() is called while threads are blocked in wait() or timedWait().
49/// Concurrent calls to set(), isSet() and reset() are allowed.
50pub fn reset(self: *ResetEvent) void {
51 self.impl.reset();
52}
53
54const Impl = if (builtin.single_threaded)
55 SingleThreadedImpl
56else
57 FutexImpl;
58
59const SingleThreadedImpl = struct {
60 is_set: bool = false,
61
62 fn isSet(self: *const Impl) bool {
63 return self.is_set;
64 }
65
66 fn wait(self: *Impl, timeout: ?u64) error{Timeout}!void {
67 if (self.isSet()) {
68 return;
69 }
70
71 // There are no other threads to wake us up.
72 // So if we wait without a timeout we would never wake up.
73 const timeout_ns = timeout orelse {
74 unreachable; // deadlock detected
75 };
76
77 std.Thread.sleep(timeout_ns);
78 return error.Timeout;
79 }
80
81 fn set(self: *Impl) void {
82 self.is_set = true;
83 }
84
85 fn reset(self: *Impl) void {
86 self.is_set = false;
87 }
88};
89
90const FutexImpl = struct {
91 state: std.atomic.Value(u32) = std.atomic.Value(u32).init(unset),
92
93 const unset = 0;
94 const waiting = 1;
95 const is_set = 2;
96
97 fn isSet(self: *const Impl) bool {
98 // Acquire barrier ensures memory accesses before set() happen before we return true.
99 return self.state.load(.acquire) == is_set;
100 }
101
102 fn wait(self: *Impl, timeout: ?u64) error{Timeout}!void {
103 // Outline the slow path to allow isSet() to be inlined
104 if (!self.isSet()) {
105 return self.waitUntilSet(timeout);
106 }
107 }
108
109 fn waitUntilSet(self: *Impl, timeout: ?u64) error{Timeout}!void {
110 @branchHint(.cold);
111
112 // Try to set the state from `unset` to `waiting` to indicate
113 // to the set() thread that others are blocked on the ResetEvent.
114 // We avoid using any strict barriers until the end when we know the ResetEvent is set.
115 var state = self.state.load(.acquire);
116 if (state == unset) {
117 state = self.state.cmpxchgStrong(state, waiting, .acquire, .acquire) orelse waiting;
118 }
119
120 // Wait until the ResetEvent is set since the state is waiting.
121 if (state == waiting) {
122 var futex_deadline = Futex.Deadline.init(timeout);
123 while (true) {
124 const wait_result = futex_deadline.wait(&self.state, waiting);
125
126 // Check if the ResetEvent was set before possibly reporting error.Timeout below.
127 state = self.state.load(.acquire);
128 if (state != waiting) {
129 break;
130 }
131
132 try wait_result;
133 }
134 }
135
136 assert(state == is_set);
137 }
138
139 fn set(self: *Impl) void {
140 // Quick check if the ResetEvent is already set before doing the atomic swap below.
141 // set() could be getting called quite often and multiple threads calling swap() increases contention unnecessarily.
142 if (self.state.load(.monotonic) == is_set) {
143 return;
144 }
145
146 // Mark the ResetEvent as set and unblock all waiters waiting on it if any.
147 // Release barrier ensures memory accesses before set() happen before the ResetEvent is observed to be "set".
148 if (self.state.swap(is_set, .release) == waiting) {
149 Futex.wake(&self.state, std.math.maxInt(u32));
150 }
151 }
152
153 fn reset(self: *Impl) void {
154 self.state.store(unset, .monotonic);
155 }
156};
157
158test "smoke test" {
159 // make sure the event is unset
160 var event = ResetEvent{};
161 try testing.expectEqual(false, event.isSet());
162
163 // make sure the event gets set
164 event.set();
165 try testing.expectEqual(true, event.isSet());
166
167 // make sure the event gets unset again
168 event.reset();
169 try testing.expectEqual(false, event.isSet());
170
171 // waits should timeout as there's no other thread to set the event
172 try testing.expectError(error.Timeout, event.timedWait(0));
173 try testing.expectError(error.Timeout, event.timedWait(std.time.ns_per_ms));
174
175 // set the event again and make sure waits complete
176 event.set();
177 event.wait();
178 try event.timedWait(std.time.ns_per_ms);
179 try testing.expectEqual(true, event.isSet());
180}
181
182test "signaling" {
183 // This test requires spawning threads
184 if (builtin.single_threaded) {
185 return error.SkipZigTest;
186 }
187
188 const Context = struct {
189 in: ResetEvent = .{},
190 out: ResetEvent = .{},
191 value: usize = 0,
192
193 fn input(self: *@This()) !void {
194 // wait for the value to become 1
195 self.in.wait();
196 self.in.reset();
197 try testing.expectEqual(self.value, 1);
198
199 // bump the value and wake up output()
200 self.value = 2;
201 self.out.set();
202
203 // wait for output to receive 2, bump the value and wake us up with 3
204 self.in.wait();
205 self.in.reset();
206 try testing.expectEqual(self.value, 3);
207
208 // bump the value and wake up output() for it to see 4
209 self.value = 4;
210 self.out.set();
211 }
212
213 fn output(self: *@This()) !void {
214 // start with 0 and bump the value for input to see 1
215 try testing.expectEqual(self.value, 0);
216 self.value = 1;
217 self.in.set();
218
219 // wait for input to receive 1, bump the value to 2 and wake us up
220 self.out.wait();
221 self.out.reset();
222 try testing.expectEqual(self.value, 2);
223
224 // bump the value to 3 for input to see (rhymes)
225 self.value = 3;
226 self.in.set();
227
228 // wait for input to bump the value to 4 and receive no more (rhymes)
229 self.out.wait();
230 self.out.reset();
231 try testing.expectEqual(self.value, 4);
232 }
233 };
234
235 var ctx = Context{};
236
237 const thread = try std.Thread.spawn(.{}, Context.output, .{&ctx});
238 defer thread.join();
239
240 try ctx.input();
241}
242
243test "broadcast" {
244 // This test requires spawning threads
245 if (builtin.single_threaded) {
246 return error.SkipZigTest;
247 }
248
249 const num_threads = 10;
250 const Barrier = struct {
251 event: ResetEvent = .{},
252 counter: std.atomic.Value(usize) = std.atomic.Value(usize).init(num_threads),
253
254 fn wait(self: *@This()) void {
255 if (self.counter.fetchSub(1, .acq_rel) == 1) {
256 self.event.set();
257 }
258 }
259 };
260
261 const Context = struct {
262 start_barrier: Barrier = .{},
263 finish_barrier: Barrier = .{},
264
265 fn run(self: *@This()) void {
266 self.start_barrier.wait();
267 self.finish_barrier.wait();
268 }
269 };
270
271 var ctx = Context{};
272 var threads: [num_threads - 1]std.Thread = undefined;
273
274 for (&threads) |*t| t.* = try std.Thread.spawn(.{}, Context.run, .{&ctx});
275 defer for (threads) |t| t.join();
276
277 ctx.run();
278}
lib/std/Thread/WaitGroup.zig+20-9
...@@ -7,11 +7,15 @@ const is_waiting: usize = 1 << 0;...@@ -7,11 +7,15 @@ const is_waiting: usize = 1 << 0;
7const one_pending: usize = 1 << 1;7const one_pending: usize = 1 << 1;
88
9state: std.atomic.Value(usize) = std.atomic.Value(usize).init(0),9state: std.atomic.Value(usize) = std.atomic.Value(usize).init(0),
10event: std.Thread.ResetEvent = .{},10event: std.Thread.ResetEvent = .unset,
1111
12pub fn start(self: *WaitGroup) void {12pub fn start(self: *WaitGroup) void {
13 const state = self.state.fetchAdd(one_pending, .monotonic);13 return startStateless(&self.state);
14 assert((state / one_pending) < (std.math.maxInt(usize) / one_pending));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));
15}19}
1620
17pub fn startMany(self: *WaitGroup, n: usize) void {21pub fn startMany(self: *WaitGroup, n: usize) void {
...@@ -28,13 +32,20 @@ pub fn finish(self: *WaitGroup) void {...@@ -28,13 +32,20 @@ pub fn finish(self: *WaitGroup) void {
28 }32 }
29}33}
3034
31pub fn wait(self: *WaitGroup) void {35pub fn finishStateless(state: *std.atomic.Value(usize), event: *std.Thread.ResetEvent) void {
32 const state = self.state.fetchAdd(is_waiting, .acquire);36 const prev_state = state.fetchSub(one_pending, .acq_rel);
33 assert(state & is_waiting == 0);37 assert((prev_state / one_pending) > 0);
38 if (prev_state == (one_pending | is_waiting)) event.set();
39}
3440
35 if ((state / one_pending) > 0) {41pub fn wait(wg: *WaitGroup) void {
36 self.event.wait();42 return waitStateless(&wg.state, &wg.event);
37 }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();
38}49}
3950
40pub fn reset(self: *WaitGroup) void {51pub fn reset(self: *WaitGroup) void {