authorgravatar for igor.anic@gmail.comIgor Anić <igor.anic@gmail.com> 2022-11-21 17:26:54+01:00
committergravatar for igor.anic@gmail.comIgor Anić <igor.anic@gmail.com> 2022-11-21 17:26:54+01:00
log9947b47d803415f40c82b6cbb510f47bc800658d
tree6976446b6e5f582a2469823e3a2b0b39978404ee
parentf229b740999b58432dc49e3aa412fac14e3781f3

stdlib: Thread.Condition wake only if signaled

Previous implementation didn't check whether there are pending signals after return from futex.wait. While it is ok for broadcast case it can result in multiple wakeups when only one thread is signaled. This implementation checks that there are pending signals before returning from wait. It is similar to the original implementation but the without initial signal check, here we first go to the futex and then check for pending signal.

1 files changed, 150 insertions(+), 70 deletions(-)

lib/std/Thread/Condition.zig+150-70
......@@ -204,40 +204,44 @@ const FutexImpl = struct {
204204 // - T1: s & signals == 0 -> FUTEX_WAIT(&epoch, e) (missed the state update + the epoch change)
205205 //
206206 // Acquire barrier to ensure the epoch load happens before the state load.
207 const epoch = self.epoch.load(.Acquire);
207 var epoch = self.epoch.load(.Acquire);
208208 var state = self.state.fetchAdd(one_waiter, .Monotonic);
209209 assert(state & waiter_mask != waiter_mask);
210210 state += one_waiter;
211 var futex_deadline = Futex.Deadline.init(timeout);
212211
213212 mutex.unlock();
214213 defer mutex.lock();
215214
216 futex_deadline.wait(&self.epoch, epoch) catch |err| switch (err) {
217 // On timeout, we must decrement the waiter we added above.
218 error.Timeout => {
219 while (true) {
220 // If there's a signal when we're timing out, consume it and report being woken up instead.
221 // Acquire barrier ensures code before the wake() which added the signal happens before we decrement it and return.
222 while (state & signal_mask != 0) {
223 const new_state = state - one_waiter - one_signal;
224 state = self.state.tryCompareAndSwap(state, new_state, .Acquire, .Monotonic) orelse return;
215 var futex_deadline = Futex.Deadline.init(timeout);
216
217 while (true) {
218 futex_deadline.wait(&self.epoch, epoch) catch |err| switch (err) {
219 // On timeout, we must decrement the waiter we added above.
220 error.Timeout => {
221 while (true) {
222 // If there's a signal when we're timing out, consume it and report being woken up instead.
223 // Acquire barrier ensures code before the wake() which added the signal happens before we decrement it and return.
224 while (state & signal_mask != 0) {
225 const new_state = state - one_waiter - one_signal;
226 state = self.state.tryCompareAndSwap(state, new_state, .Acquire, .Monotonic) orelse return;
227 }
228
229 // Remove the waiter we added and officially return timed out.
230 const new_state = state - one_waiter;
231 state = self.state.tryCompareAndSwap(state, new_state, .Monotonic, .Monotonic) orelse return err;
225232 }
233 },
234 };
226235
227 // Remove the waiter we added and officially return timed out.
228 const new_state = state - one_waiter;
229 state = self.state.tryCompareAndSwap(state, new_state, .Monotonic, .Monotonic) orelse return err;
230 }
231 },
232 };
236 epoch = self.epoch.load(.Acquire);
237 state = self.state.load(.Monotonic);
233238
234 while (true) {
235 // Wait thread, decrement waiter and consume signal if exists.
236 var new_state = state - one_waiter;
237 if (state & signal_mask != 0) {
238 new_state = state - one_signal;
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;
239244 }
240 state = self.state.tryCompareAndSwap(state, new_state, .Acquire, .Monotonic) orelse return;
241245 }
242246 }
243247
......@@ -535,66 +539,142 @@ test "Condition - broadcasting - wake all threads" {
535539 return error.SkipZigTest;
536540 }
537541
542 var num_runs: usize = 1;
538543 const num_threads = 10;
539544
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();
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 }
552564
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();
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 }
557574 }
575 };
558576
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();
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 );
567595 }
596
597 // Signal thread 1 to wake up
598 broadcast_test.thread_id_to_wake = 1;
599 broadcast_test.cond.broadcast();
568600 }
569 };
570601
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;
602 for (broadcast_test.threads) |t| {
603 t.join();
604 }
576605 }
606}
577607
578 {
579 broadcast_test.mutex.lock();
580 defer broadcast_test.mutex.unlock();
608test "Condition - signal wakes one" {
609 // This test requires spawning threads
610 if (builtin.single_threaded) {
611 return error.SkipZigTest;
612 }
581613
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 );
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});
589657 }
590658
591 // Signal thread 1 to wake up
592 broadcast_test.thread_id_to_wake = 1;
593 broadcast_test.cond.broadcast();
594 }
659 {
660 runner.mutex.lock();
661 defer runner.mutex.unlock();
595662
596 for (broadcast_test.threads) |t| {
597 t.join();
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);
598679 }
599 //std.debug.print("wakeups {d}\n", .{broadcast_test.wakeups});
600680}