authorgravatar for cryptocode@zolo.iocryptocode <cryptocode@zolo.io> 2022-04-08 20:26:56+02:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2022-04-08 13:26:56-05:00
log531d5b213f5cc7450c56bea3fc2050dda5794067
treeba2ab7d1e4362d3642dab55ec9c05a7c7fccbdf0
parent6ae8fe193bd5397b9fceea98ba684d9d870286cf
signature Signed by PGP key 4AEE18F83AFDEB23

std: add Thread.Condition.timedWait (#11352)

* std: add Thread.Condition.timedWait I needed the equivalent of `std::condition_variable::wait_for`, but it's missing in std. This PR adds an implementation, following the status quo of using std.os.CLOCK.REALTIME in the pthread case (i.e. Futex) A follow-up patch moving futex/condition stuff to monotonic clocks where available seems like a good idea. This would involve conditionally exposing more functions and constants through std.c and std.os. For instance, Chromium picks `pthread_cond_timedwait_relative_np` on macOS and `clock_gettime(CLOCK_MONOTONIC...)` on BSD's. Tested on Windows 11, macOS 12.2.1 and Linux (with/without libc) * Sleep in the single threaded case, handle timeout overflow in the Windows case and address a race condition in the AtomicCondition case.

1 files changed, 173 insertions(+), 2 deletions(-)

lib/std/Thread/Condition.zig+173-2
...@@ -17,6 +17,10 @@ pub fn wait(cond: *Condition, mutex: *Mutex) void {...@@ -17,6 +17,10 @@ pub fn wait(cond: *Condition, mutex: *Mutex) void {
17 cond.impl.wait(mutex);17 cond.impl.wait(mutex);
18}18}
1919
20pub fn timedWait(cond: *Condition, mutex: *Mutex, timeout_ns: u64) error{TimedOut}!void {
21 try cond.impl.timedWait(mutex, timeout_ns);
22}
23
20pub fn signal(cond: *Condition) void {24pub fn signal(cond: *Condition) void {
21 cond.impl.signal();25 cond.impl.signal();
22}26}
...@@ -41,6 +45,14 @@ pub const SingleThreadedCondition = struct {...@@ -41,6 +45,14 @@ pub const SingleThreadedCondition = struct {
41 unreachable; // deadlock detected45 unreachable; // deadlock detected
42 }46 }
4347
48 pub fn timedWait(cond: *SingleThreadedCondition, mutex: *Mutex, timeout_ns: u64) error{TimedOut}!void {
49 _ = cond;
50 _ = mutex;
51 _ = timeout_ns;
52 std.time.sleep(timeout_ns);
53 return error.TimedOut;
54 }
55
44 pub fn signal(cond: *SingleThreadedCondition) void {56 pub fn signal(cond: *SingleThreadedCondition) void {
45 _ = cond;57 _ = cond;
46 }58 }
...@@ -63,6 +75,25 @@ pub const WindowsCondition = struct {...@@ -63,6 +75,25 @@ pub const WindowsCondition = struct {
63 assert(rc != windows.FALSE);75 assert(rc != windows.FALSE);
64 }76 }
6577
78 pub fn timedWait(cond: *WindowsCondition, mutex: *Mutex, timeout_ns: u64) error{TimedOut}!void {
79 var timeout_checked = std.math.cast(windows.DWORD, timeout_ns / std.time.ns_per_ms) catch overflow: {
80 break :overflow std.math.maxInt(windows.DWORD);
81 };
82
83 // Handle the case where timeout is INFINITE, otherwise SleepConditionVariableSRW's time-out never elapses
84 const timeout_overflowed = timeout_checked == windows.INFINITE;
85 timeout_checked -= @boolToInt(timeout_overflowed);
86
87 const rc = windows.kernel32.SleepConditionVariableSRW(
88 &cond.cond,
89 &mutex.impl.srwlock,
90 timeout_checked,
91 @as(windows.ULONG, 0),
92 );
93 if (rc == windows.FALSE and windows.kernel32.GetLastError() == windows.Win32Error.TIMEOUT) return error.TimedOut;
94 assert(rc != windows.FALSE);
95 }
96
66 pub fn signal(cond: *WindowsCondition) void {97 pub fn signal(cond: *WindowsCondition) void {
67 windows.kernel32.WakeConditionVariable(&cond.cond);98 windows.kernel32.WakeConditionVariable(&cond.cond);
68 }99 }
...@@ -80,6 +111,24 @@ pub const PthreadCondition = struct {...@@ -80,6 +111,24 @@ pub const PthreadCondition = struct {
80 assert(rc == .SUCCESS);111 assert(rc == .SUCCESS);
81 }112 }
82113
114 pub fn timedWait(cond: *PthreadCondition, mutex: *Mutex, timeout_ns: u64) error{TimedOut}!void {
115 var ts: std.os.timespec = undefined;
116 std.os.clock_gettime(std.os.CLOCK.REALTIME, &ts) catch unreachable;
117 ts.tv_sec += @intCast(@TypeOf(ts.tv_sec), timeout_ns / std.time.ns_per_s);
118 ts.tv_nsec += @intCast(@TypeOf(ts.tv_nsec), timeout_ns % std.time.ns_per_s);
119 if (ts.tv_nsec >= std.time.ns_per_s) {
120 ts.tv_sec += 1;
121 ts.tv_nsec -= std.time.ns_per_s;
122 }
123
124 const rc = std.c.pthread_cond_timedwait(&cond.cond, &mutex.impl.pthread_mutex, &ts);
125 return switch (rc) {
126 .SUCCESS => {},
127 .TIMEDOUT => error.TimedOut,
128 else => unreachable,
129 };
130 }
131
83 pub fn signal(cond: *PthreadCondition) void {132 pub fn signal(cond: *PthreadCondition) void {
84 const rc = std.c.pthread_cond_signal(&cond.cond);133 const rc = std.c.pthread_cond_signal(&cond.cond);
85 assert(rc == .SUCCESS);134 assert(rc == .SUCCESS);
...@@ -100,6 +149,7 @@ pub const AtomicCondition = struct {...@@ -100,6 +149,7 @@ pub const AtomicCondition = struct {
100149
101 pub const QueueItem = struct {150 pub const QueueItem = struct {
102 futex: i32 = 0,151 futex: i32 = 0,
152 dequeued: bool = false,
103153
104 fn wait(cond: *@This()) void {154 fn wait(cond: *@This()) void {
105 while (@atomicLoad(i32, &cond.futex, .Acquire) == 0) {155 while (@atomicLoad(i32, &cond.futex, .Acquire) == 0) {
...@@ -122,6 +172,39 @@ pub const AtomicCondition = struct {...@@ -122,6 +172,39 @@ pub const AtomicCondition = struct {
122 }172 }
123 }173 }
124174
175 pub fn timedWait(cond: *@This(), timeout_ns: u64) error{TimedOut}!void {
176 const start_time = std.time.nanoTimestamp();
177 while (@atomicLoad(i32, &cond.futex, .Acquire) == 0) {
178 switch (builtin.os.tag) {
179 .linux => {
180 var ts: std.os.timespec = undefined;
181 ts.tv_sec = @intCast(@TypeOf(ts.tv_sec), timeout_ns / std.time.ns_per_s);
182 ts.tv_nsec = @intCast(@TypeOf(ts.tv_nsec), timeout_ns % std.time.ns_per_s);
183 switch (linux.getErrno(linux.futex_wait(
184 &cond.futex,
185 linux.FUTEX.PRIVATE_FLAG | linux.FUTEX.WAIT,
186 0,
187 &ts,
188 ))) {
189 .SUCCESS => {},
190 .INTR => {},
191 .AGAIN => {},
192 .TIMEDOUT => return error.TimedOut,
193 .INVAL => {}, // possibly timeout overflow
194 .FAULT => unreachable,
195 else => unreachable,
196 }
197 },
198 else => {
199 if (std.time.nanoTimestamp() - start_time >= timeout_ns) {
200 return error.TimedOut;
201 }
202 std.atomic.spinLoopHint();
203 },
204 }
205 }
206 }
207
125 fn notify(cond: *@This()) void {208 fn notify(cond: *@This()) void {
126 @atomicStore(i32, &cond.futex, 1, .Release);209 @atomicStore(i32, &cond.futex, 1, .Release);
127210
...@@ -158,6 +241,41 @@ pub const AtomicCondition = struct {...@@ -158,6 +241,41 @@ pub const AtomicCondition = struct {
158 mutex.lock();241 mutex.lock();
159 }242 }
160243
244 pub fn timedWait(cond: *AtomicCondition, mutex: *Mutex, timeout_ns: u64) error{TimedOut}!void {
245 var waiter = QueueList.Node{ .data = .{} };
246
247 {
248 cond.queue_mutex.lock();
249 defer cond.queue_mutex.unlock();
250
251 cond.queue_list.prepend(&waiter);
252 @atomicStore(bool, &cond.pending, true, .SeqCst);
253 }
254
255 var timed_out = false;
256 mutex.unlock();
257 defer mutex.lock();
258 waiter.data.timedWait(timeout_ns) catch |err| switch (err) {
259 error.TimedOut => {
260 defer if (!timed_out) {
261 waiter.data.wait();
262 };
263 cond.queue_mutex.lock();
264 defer cond.queue_mutex.unlock();
265
266 if (!waiter.data.dequeued) {
267 timed_out = true;
268 cond.queue_list.remove(&waiter);
269 }
270 },
271 else => unreachable,
272 };
273
274 if (timed_out) {
275 return error.TimedOut;
276 }
277 }
278
161 pub fn signal(cond: *AtomicCondition) void {279 pub fn signal(cond: *AtomicCondition) void {
162 if (@atomicLoad(bool, &cond.pending, .SeqCst) == false)280 if (@atomicLoad(bool, &cond.pending, .SeqCst) == false)
163 return;281 return;
...@@ -167,12 +285,16 @@ pub const AtomicCondition = struct {...@@ -167,12 +285,16 @@ pub const AtomicCondition = struct {
167 defer cond.queue_mutex.unlock();285 defer cond.queue_mutex.unlock();
168286
169 const maybe_waiter = cond.queue_list.popFirst();287 const maybe_waiter = cond.queue_list.popFirst();
288 if (maybe_waiter) |waiter| {
289 waiter.data.dequeued = true;
290 }
170 @atomicStore(bool, &cond.pending, cond.queue_list.first != null, .SeqCst);291 @atomicStore(bool, &cond.pending, cond.queue_list.first != null, .SeqCst);
171 break :blk maybe_waiter;292 break :blk maybe_waiter;
172 };293 };
173294
174 if (maybe_waiter) |waiter|295 if (maybe_waiter) |waiter| {
175 waiter.data.notify();296 waiter.data.notify();
297 }
176 }298 }
177299
178 pub fn broadcast(cond: *AtomicCondition) void {300 pub fn broadcast(cond: *AtomicCondition) void {
...@@ -186,12 +308,19 @@ pub const AtomicCondition = struct {...@@ -186,12 +308,19 @@ pub const AtomicCondition = struct {
186 defer cond.queue_mutex.unlock();308 defer cond.queue_mutex.unlock();
187309
188 const waiters = cond.queue_list;310 const waiters = cond.queue_list;
311
312 var it = waiters.first;
313 while (it) |node| : (it = node.next) {
314 node.data.dequeued = true;
315 }
316
189 cond.queue_list = .{};317 cond.queue_list = .{};
190 break :blk waiters;318 break :blk waiters;
191 };319 };
192320
193 while (waiters.popFirst()) |waiter|321 while (waiters.popFirst()) |waiter| {
194 waiter.data.notify();322 waiter.data.notify();
323 }
195 }324 }
196};325};
197326
...@@ -238,3 +367,45 @@ test "Thread.Condition" {...@@ -238,3 +367,45 @@ test "Thread.Condition" {
238367
239 for (threads) |t| t.join();368 for (threads) |t| t.join();
240}369}
370
371test "Thread.Condition.timedWait" {
372 if (builtin.single_threaded) {
373 return error.SkipZigTest;
374 }
375
376 var cond = Condition{};
377 var mut = Mutex{};
378
379 // Expect a timeout, as the condition variable is never signaled
380 {
381 mut.lock();
382 defer mut.unlock();
383 try testing.expectError(error.TimedOut, cond.timedWait(&mut, 10 * std.time.ns_per_ms));
384 }
385
386 // Expect a signal before timeout
387 {
388 const TestContext = struct {
389 cond: *Condition,
390 mutex: *Mutex,
391 n: *u32,
392 fn worker(ctx: *@This()) void {
393 ctx.mutex.lock();
394 defer ctx.mutex.unlock();
395 ctx.n.* = 1;
396 ctx.cond.signal();
397 }
398 };
399
400 var n: u32 = 0;
401
402 var ctx = TestContext{ .cond = &cond, .mutex = &mut, .n = &n };
403 mut.lock();
404 var thread = try std.Thread.spawn(.{}, TestContext.worker, .{&ctx});
405 // Looped check to handle spurious wakeups
406 while (n != 1) try cond.timedWait(&mut, 500 * std.time.ns_per_ms);
407 mut.unlock();
408 try testing.expect(n == 1);
409 thread.join();
410 }
411}