| ... | ... | @@ -194,42 +194,27 @@ const FutexImpl = struct { |
| 194 | 194 | const signal_mask = 0xffff << 16; |
| 195 | 195 | |
| 196 | 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. |
| 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); |
| 200 | 208 | var state = self.state.fetchAdd(one_waiter, .Monotonic); |
| 201 | 209 | assert(state & waiter_mask != waiter_mask); |
| 202 | 210 | state += one_waiter; |
| 203 | 211 | |
| 204 | | // Temporarily release the mutex in order to block on the condition variable. |
| 205 | 212 | mutex.unlock(); |
| 206 | 213 | defer mutex.lock(); |
| 207 | 214 | |
| 208 | 215 | 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 | | } |
| 232 | 216 | |
| 217 | while (true) { |
| 233 | 218 | futex_deadline.wait(&self.epoch, epoch) catch |err| switch (err) { |
| 234 | 219 | // On timeout, we must decrement the waiter we added above. |
| 235 | 220 | error.Timeout => { |
| ... | ... | @@ -247,6 +232,16 @@ const FutexImpl = struct { |
| 247 | 232 | } |
| 248 | 233 | }, |
| 249 | 234 | }; |
| 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 | } |
| 250 | 245 | } |
| 251 | 246 | } |
| 252 | 247 | |
| ... | ... | @@ -536,3 +531,150 @@ test "Condition - broadcasting" { |
| 536 | 531 | t.join(); |
| 537 | 532 | } |
| 538 | 533 | } |
| 534 | |
| 535 | test "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 | |
| 608 | test "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 | } |
Andrew Kelley <andrew@ziglang.org> 2022-11-23 16:24:55-05:00
lib