authorgravatar for igor.anic@gmail.comIgor Anić <igor.anic@gmail.com> 2022-11-17 20:58:45+01:00
committergravatar for igor.anic@gmail.comIgor Anić <igor.anic@gmail.com> 2022-11-17 20:58:45+01:00
logf229b740999b58432dc49e3aa412fac14e3781f3
tree5d457a6c2468723a9be5102ea6567f9cecc7e0b5
parent88a0f3d0e55691aa2b341cfa6efc53f2540b22f9

stdlib: fix condition variable broadcast FutexImpl

fixes #12877 Current implementation (before this fix) observes number of waiters when broadcast occurs and then makes that number of wakeups. If we have multiple threads waiting for wakeup which immediately go into wait if wakeup is not for that thread (as described in the issue). The same thread can get multiple wakeups while some got none. That is not consistent with documented behavior for condition variable broadcast: `Unblocks all threads currently blocked in a call to wait() or timedWait() with a given Mutex.`. This fix ensures that the thread waiting on futext is woken up on futex wake.

1 files changed, 105 insertions(+), 43 deletions(-)

lib/std/Thread/Condition.zig+105-43
...@@ -194,59 +194,50 @@ const FutexImpl = struct {...@@ -194,59 +194,50 @@ const FutexImpl = struct {
194 const signal_mask = 0xffff << 16;194 const signal_mask = 0xffff << 16;
195195
196 fn wait(self: *Impl, mutex: *Mutex, timeout: ?u64) error{Timeout}!void {196 fn wait(self: *Impl, mutex: *Mutex, timeout: ?u64) error{Timeout}!void {
197 // Register that we're waiting on the state by incrementing the wait count.197 // Observe the epoch, then check the state again to see if we should wake up.
198 // This assumes that there can be at most ((1<<16)-1) or 65,355 threads concurrently waiting on the same Condvar.198 // The epoch must be observed before we check the state or we could potentially miss a wake() and deadlock:
199 // If this is hit in practice, then this condvar not working is the least of your concerns.199 //
200 // - T1: s = LOAD(&state)
201 // - T2: UPDATE(&s, signal)
202 // - T2: UPDATE(&epoch, 1) + FUTEX_WAKE(&epoch)
203 // - T1: e = LOAD(&epoch) (was reordered after the state load)
204 // - T1: s & signals == 0 -> FUTEX_WAIT(&epoch, e) (missed the state update + the epoch change)
205 //
206 // Acquire barrier to ensure the epoch load happens before the state load.
207 const epoch = self.epoch.load(.Acquire);
200 var state = self.state.fetchAdd(one_waiter, .Monotonic);208 var state = self.state.fetchAdd(one_waiter, .Monotonic);
201 assert(state & waiter_mask != waiter_mask);209 assert(state & waiter_mask != waiter_mask);
202 state += one_waiter;210 state += one_waiter;
211 var futex_deadline = Futex.Deadline.init(timeout);
203212
204 // Temporarily release the mutex in order to block on the condition variable.
205 mutex.unlock();213 mutex.unlock();
206 defer mutex.lock();214 defer mutex.lock();
207215
208 var futex_deadline = Futex.Deadline.init(timeout);216 futex_deadline.wait(&self.epoch, epoch) catch |err| switch (err) {
209 while (true) {217 // On timeout, we must decrement the waiter we added above.
210 // Try to wake up by consuming a signal and decremented the waiter we added previously.218 error.Timeout => {
211 // Acquire barrier ensures code before the wake() which added the signal happens before we decrement it and return.219 while (true) {
212 while (state & signal_mask != 0) {220 // If there's a signal when we're timing out, consume it and report being woken up instead.
213 const new_state = state - one_waiter - one_signal;221 // Acquire barrier ensures code before the wake() which added the signal happens before we decrement it and return.
214 state = self.state.tryCompareAndSwap(state, new_state, .Acquire, .Monotonic) orelse return;222 while (state & signal_mask != 0) {
215 }223 const new_state = state - one_waiter - one_signal;
224 state = self.state.tryCompareAndSwap(state, new_state, .Acquire, .Monotonic) orelse return;
225 }
216226
217 // Observe the epoch, then check the state again to see if we should wake up.227 // Remove the waiter we added and officially return timed out.
218 // The epoch must be observed before we check the state or we could potentially miss a wake() and deadlock:228 const new_state = state - one_waiter;
219 //229 state = self.state.tryCompareAndSwap(state, new_state, .Monotonic, .Monotonic) orelse return err;
220 // - T1: s = LOAD(&state)230 }
221 // - T2: UPDATE(&s, signal)231 },
222 // - T2: UPDATE(&epoch, 1) + FUTEX_WAKE(&epoch)232 };
223 // - T1: e = LOAD(&epoch) (was reordered after the state load)233
224 // - T1: s & signals == 0 -> FUTEX_WAIT(&epoch, e) (missed the state update + the epoch change)234 while (true) {
225 //235 // Wait thread, decrement waiter and consume signal if exists.
226 // Acquire barrier to ensure the epoch load happens before the state load.236 var new_state = state - one_waiter;
227 const epoch = self.epoch.load(.Acquire);
228 state = self.state.load(.Monotonic);
229 if (state & signal_mask != 0) {237 if (state & signal_mask != 0) {
230 continue;238 new_state = state - one_signal;
231 }239 }
232240 state = self.state.tryCompareAndSwap(state, new_state, .Acquire, .Monotonic) orelse return;
233 futex_deadline.wait(&self.epoch, epoch) catch |err| switch (err) {
234 // On timeout, we must decrement the waiter we added above.
235 error.Timeout => {
236 while (true) {
237 // If there's a signal when we're timing out, consume it and report being woken up instead.
238 // Acquire barrier ensures code before the wake() which added the signal happens before we decrement it and return.
239 while (state & signal_mask != 0) {
240 const new_state = state - one_waiter - one_signal;
241 state = self.state.tryCompareAndSwap(state, new_state, .Acquire, .Monotonic) orelse return;
242 }
243
244 // Remove the waiter we added and officially return timed out.
245 const new_state = state - one_waiter;
246 state = self.state.tryCompareAndSwap(state, new_state, .Monotonic, .Monotonic) orelse return err;
247 }
248 },
249 };
250 }241 }
251 }242 }
252243
...@@ -536,3 +527,74 @@ test "Condition - broadcasting" {...@@ -536,3 +527,74 @@ test "Condition - broadcasting" {
536 t.join();527 t.join();
537 }528 }
538}529}
530
531test "Condition - broadcasting - wake all threads" {
532 // Tests issue #12877
533 // This test requires spawning threads
534 if (builtin.single_threaded) {
535 return error.SkipZigTest;
536 }
537
538 const num_threads = 10;
539
540 const BroadcastTest = struct {
541 mutex: Mutex = .{},
542 cond: Condition = .{},
543 completed: Condition = .{},
544 count: usize = 0,
545 thread_id_to_wake: usize = 0,
546 threads: [num_threads]std.Thread = undefined,
547 wakeups: usize = 0,
548
549 fn run(self: *@This(), thread_id: usize) void {
550 self.mutex.lock();
551 defer self.mutex.unlock();
552
553 // The last broadcast thread to start tells the main test thread it's completed.
554 self.count += 1;
555 if (self.count == num_threads) {
556 self.completed.signal();
557 }
558
559 while (self.thread_id_to_wake != thread_id) {
560 self.cond.timedWait(&self.mutex, 1 * std.time.ns_per_s) catch std.debug.panic("thread_id {d} timeout {d}", .{ thread_id, self.thread_id_to_wake });
561 self.wakeups += 1;
562 }
563 if (self.thread_id_to_wake <= num_threads) {
564 // Signal next thread to wake up.
565 self.thread_id_to_wake += 1;
566 self.cond.broadcast();
567 }
568 }
569 };
570
571 var broadcast_test = BroadcastTest{};
572 var thread_id: usize = 1;
573 for (broadcast_test.threads) |*t| {
574 t.* = try std.Thread.spawn(.{}, BroadcastTest.run, .{ &broadcast_test, thread_id });
575 thread_id += 1;
576 }
577
578 {
579 broadcast_test.mutex.lock();
580 defer broadcast_test.mutex.unlock();
581
582 // Wait for all the broadcast threads to spawn.
583 // timedWait() to detect any potential deadlocks.
584 while (broadcast_test.count != num_threads) {
585 try broadcast_test.completed.timedWait(
586 &broadcast_test.mutex,
587 1 * std.time.ns_per_s,
588 );
589 }
590
591 // Signal thread 1 to wake up
592 broadcast_test.thread_id_to_wake = 1;
593 broadcast_test.cond.broadcast();
594 }
595
596 for (broadcast_test.threads) |t| {
597 t.join();
598 }
599 //std.debug.print("wakeups {d}\n", .{broadcast_test.wakeups});
600}