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-29 06:20:48-07:00
log5469db66e4b3f1ffe2ed9b58d5e12dc5a9c815f0
treebb100140598eb34559a68d41831bceb14cd57a82
parentf9d976a4e1616ab49664e5293508a151f11c1e08

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) {
...@@ -1780,3 +1900,125 @@ fn testTls() !void {...@@ -1780,3 +1900,125 @@ fn testTls() !void {
1780 x += 1;1900 x += 1;
1781 if (x != 1235) return error.TlsBadEndValue;1901 if (x != 1235) return error.TlsBadEndValue;
1782}1902}
1903
1904test "ResetEvent smoke test" {
1905 // make sure the event is unset
1906 var event = ResetEvent{};
1907 try testing.expectEqual(false, event.isSet());
1908
1909 // make sure the event gets set
1910 event.set();
1911 try testing.expectEqual(true, event.isSet());
1912
1913 // make sure the event gets unset again
1914 event.reset();
1915 try testing.expectEqual(false, event.isSet());
1916
1917 // waits should timeout as there's no other thread to set the event
1918 try testing.expectError(error.Timeout, event.timedWait(0));
1919 try testing.expectError(error.Timeout, event.timedWait(std.time.ns_per_ms));
1920
1921 // set the event again and make sure waits complete
1922 event.set();
1923 event.wait();
1924 try event.timedWait(std.time.ns_per_ms);
1925 try testing.expectEqual(true, event.isSet());
1926}
1927
1928test "ResetEvent signaling" {
1929 // This test requires spawning threads
1930 if (builtin.single_threaded) {
1931 return error.SkipZigTest;
1932 }
1933
1934 const Context = struct {
1935 in: ResetEvent = .{},
1936 out: ResetEvent = .{},
1937 value: usize = 0,
1938
1939 fn input(self: *@This()) !void {
1940 // wait for the value to become 1
1941 self.in.wait();
1942 self.in.reset();
1943 try testing.expectEqual(self.value, 1);
1944
1945 // bump the value and wake up output()
1946 self.value = 2;
1947 self.out.set();
1948
1949 // wait for output to receive 2, bump the value and wake us up with 3
1950 self.in.wait();
1951 self.in.reset();
1952 try testing.expectEqual(self.value, 3);
1953
1954 // bump the value and wake up output() for it to see 4
1955 self.value = 4;
1956 self.out.set();
1957 }
1958
1959 fn output(self: *@This()) !void {
1960 // start with 0 and bump the value for input to see 1
1961 try testing.expectEqual(self.value, 0);
1962 self.value = 1;
1963 self.in.set();
1964
1965 // wait for input to receive 1, bump the value to 2 and wake us up
1966 self.out.wait();
1967 self.out.reset();
1968 try testing.expectEqual(self.value, 2);
1969
1970 // bump the value to 3 for input to see (rhymes)
1971 self.value = 3;
1972 self.in.set();
1973
1974 // wait for input to bump the value to 4 and receive no more (rhymes)
1975 self.out.wait();
1976 self.out.reset();
1977 try testing.expectEqual(self.value, 4);
1978 }
1979 };
1980
1981 var ctx = Context{};
1982
1983 const thread = try std.Thread.spawn(.{}, Context.output, .{&ctx});
1984 defer thread.join();
1985
1986 try ctx.input();
1987}
1988
1989test "ResetEvent broadcast" {
1990 // This test requires spawning threads
1991 if (builtin.single_threaded) {
1992 return error.SkipZigTest;
1993 }
1994
1995 const num_threads = 10;
1996 const Barrier = struct {
1997 event: ResetEvent = .{},
1998 counter: std.atomic.Value(usize) = std.atomic.Value(usize).init(num_threads),
1999
2000 fn wait(self: *@This()) void {
2001 if (self.counter.fetchSub(1, .acq_rel) == 1) {
2002 self.event.set();
2003 }
2004 }
2005 };
2006
2007 const Context = struct {
2008 start_barrier: Barrier = .{},
2009 finish_barrier: Barrier = .{},
2010
2011 fn run(self: *@This()) void {
2012 self.start_barrier.wait();
2013 self.finish_barrier.wait();
2014 }
2015 };
2016
2017 var ctx = Context{};
2018 var threads: [num_threads - 1]std.Thread = undefined;
2019
2020 for (&threads) |*t| t.* = try std.Thread.spawn(.{}, Context.run, .{&ctx});
2021 defer for (threads) |t| t.join();
2022
2023 ctx.run();
2024}
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 {