authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2022-11-23 16:24:55-05:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2022-11-23 16:24:55-05:00
log1d7faf30f9c40cd1d4cb74480d0cc4e51c43ff0b
tree6f9a5b4b626a2e759e38ff3062953f3bb98b105d
parent258bee41bf58c9001ceaefc974fa594a26ac0fc5
parent9947b47d803415f40c82b6cbb510f47bc800658d
signaturebadge-question-mark Signed by PGP key 4AEE18F83AFDEB23

Merge pull request #13577 from ianic/issue-12877

stdlib: fix condition variable broadcast FutexImpl

1 files changed, 169 insertions(+), 27 deletions(-)

lib/std/Thread/Condition.zig+169-27
......@@ -194,42 +194,27 @@ const FutexImpl = struct {
194194 const signal_mask = 0xffff << 16;
195195
196196 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.
198 // This assumes that there can be at most ((1<<16)-1) or 65,355 threads concurrently waiting on the same Condvar.
199 // If this is hit in practice, then this condvar not working is the least of your concerns.
197 // Observe the epoch, then check the state again to see if we should wake up.
198 // The epoch must be observed before we check the state or we could potentially miss a wake() and deadlock:
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 var epoch = self.epoch.load(.Acquire);
200208 var state = self.state.fetchAdd(one_waiter, .Monotonic);
201209 assert(state & waiter_mask != waiter_mask);
202210 state += one_waiter;
203211
204 // Temporarily release the mutex in order to block on the condition variable.
205212 mutex.unlock();
206213 defer mutex.lock();
207214
208215 var futex_deadline = Futex.Deadline.init(timeout);
209 while (true) {
210 // Try to wake up by consuming a signal and decremented the waiter we added previously.
211 // Acquire barrier ensures code before the wake() which added the signal happens before we decrement it and return.
212 while (state & signal_mask != 0) {
213 const new_state = state - one_waiter - one_signal;
214 state = self.state.tryCompareAndSwap(state, new_state, .Acquire, .Monotonic) orelse return;
215 }
216
217 // Observe the epoch, then check the state again to see if we should wake up.
218 // The epoch must be observed before we check the state or we could potentially miss a wake() and deadlock:
219 //
220 // - T1: s = LOAD(&state)
221 // - T2: UPDATE(&s, signal)
222 // - T2: UPDATE(&epoch, 1) + FUTEX_WAKE(&epoch)
223 // - T1: e = LOAD(&epoch) (was reordered after the state load)
224 // - T1: s & signals == 0 -> FUTEX_WAIT(&epoch, e) (missed the state update + the epoch change)
225 //
226 // Acquire barrier to ensure the epoch load happens before the state load.
227 const epoch = self.epoch.load(.Acquire);
228 state = self.state.load(.Monotonic);
229 if (state & signal_mask != 0) {
230 continue;
231 }
232216
217 while (true) {
233218 futex_deadline.wait(&self.epoch, epoch) catch |err| switch (err) {
234219 // On timeout, we must decrement the waiter we added above.
235220 error.Timeout => {
......@@ -247,6 +232,16 @@ const FutexImpl = struct {
247232 }
248233 },
249234 };
235
236 epoch = self.epoch.load(.Acquire);
237 state = self.state.load(.Monotonic);
238
239 // Try to wake up by consuming a signal and decremented the waiter we added previously.
240 // Acquire barrier ensures code before the wake() which added the signal happens before we decrement it and return.
241 while (state & signal_mask != 0) {
242 const new_state = state - one_waiter - one_signal;
243 state = self.state.tryCompareAndSwap(state, new_state, .Acquire, .Monotonic) orelse return;
244 }
250245 }
251246 }
252247
......@@ -536,3 +531,150 @@ test "Condition - broadcasting" {
536531 t.join();
537532 }
538533}
534
535test "Condition - broadcasting - wake all threads" {
536 // Tests issue #12877
537 // This test requires spawning threads
538 if (builtin.single_threaded) {
539 return error.SkipZigTest;
540 }
541
542 var num_runs: usize = 1;
543 const num_threads = 10;
544
545 while (num_runs > 0) : (num_runs -= 1) {
546 const BroadcastTest = struct {
547 mutex: Mutex = .{},
548 cond: Condition = .{},
549 completed: Condition = .{},
550 count: usize = 0,
551 thread_id_to_wake: usize = 0,
552 threads: [num_threads]std.Thread = undefined,
553 wakeups: usize = 0,
554
555 fn run(self: *@This(), thread_id: usize) void {
556 self.mutex.lock();
557 defer self.mutex.unlock();
558
559 // The last broadcast thread to start tells the main test thread it's completed.
560 self.count += 1;
561 if (self.count == num_threads) {
562 self.completed.signal();
563 }
564
565 while (self.thread_id_to_wake != thread_id) {
566 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 });
567 self.wakeups += 1;
568 }
569 if (self.thread_id_to_wake <= num_threads) {
570 // Signal next thread to wake up.
571 self.thread_id_to_wake += 1;
572 self.cond.broadcast();
573 }
574 }
575 };
576
577 var broadcast_test = BroadcastTest{};
578 var thread_id: usize = 1;
579 for (broadcast_test.threads) |*t| {
580 t.* = try std.Thread.spawn(.{}, BroadcastTest.run, .{ &broadcast_test, thread_id });
581 thread_id += 1;
582 }
583
584 {
585 broadcast_test.mutex.lock();
586 defer broadcast_test.mutex.unlock();
587
588 // Wait for all the broadcast threads to spawn.
589 // timedWait() to detect any potential deadlocks.
590 while (broadcast_test.count != num_threads) {
591 try broadcast_test.completed.timedWait(
592 &broadcast_test.mutex,
593 1 * std.time.ns_per_s,
594 );
595 }
596
597 // Signal thread 1 to wake up
598 broadcast_test.thread_id_to_wake = 1;
599 broadcast_test.cond.broadcast();
600 }
601
602 for (broadcast_test.threads) |t| {
603 t.join();
604 }
605 }
606}
607
608test "Condition - signal wakes one" {
609 // This test requires spawning threads
610 if (builtin.single_threaded) {
611 return error.SkipZigTest;
612 }
613
614 var num_runs: usize = 1;
615 const num_threads = 3;
616 const timeoutDelay = 10 * std.time.ns_per_ms;
617
618 while (num_runs > 0) : (num_runs -= 1) {
619
620 // Start multiple runner threads, wait for them to start and send the signal
621 // then. Expect that one thread wake up and all other times out.
622 //
623 // Test depends on delay in timedWait! If too small all threads can timeout
624 // before any one gets wake up.
625
626 const Runner = struct {
627 mutex: Mutex = .{},
628 cond: Condition = .{},
629 completed: Condition = .{},
630 count: usize = 0,
631 threads: [num_threads]std.Thread = undefined,
632 wakeups: usize = 0,
633 timeouts: usize = 0,
634
635 fn run(self: *@This()) void {
636 self.mutex.lock();
637 defer self.mutex.unlock();
638
639 // The last started thread tells the main test thread it's completed.
640 self.count += 1;
641 if (self.count == num_threads) {
642 self.completed.signal();
643 }
644
645 self.cond.timedWait(&self.mutex, timeoutDelay) catch {
646 self.timeouts += 1;
647 return;
648 };
649 self.wakeups += 1;
650 }
651 };
652
653 // Start threads
654 var runner = Runner{};
655 for (runner.threads) |*t| {
656 t.* = try std.Thread.spawn(.{}, Runner.run, .{&runner});
657 }
658
659 {
660 runner.mutex.lock();
661 defer runner.mutex.unlock();
662
663 // Wait for all the threads to spawn.
664 // timedWait() to detect any potential deadlocks.
665 while (runner.count != num_threads) {
666 try runner.completed.timedWait(&runner.mutex, 1 * std.time.ns_per_s);
667 }
668 // Signal one thread, the others should get timeout.
669 runner.cond.signal();
670 }
671
672 for (runner.threads) |t| {
673 t.join();
674 }
675
676 // Expect that only one got singal
677 try std.testing.expectEqual(runner.wakeups, 1);
678 try std.testing.expectEqual(runner.timeouts, num_threads - 1);
679 }
680}