(self, cond)
| 628 | self.assertRaises(RuntimeError, cond.notify) |
| 629 | |
| 630 | def _check_notify(self, cond): |
| 631 | # Note that this test is sensitive to timing. If the worker threads |
| 632 | # don't execute in a timely fashion, the main thread may think they |
| 633 | # are further along then they are. The main thread therefore issues |
| 634 | # wait_threads_blocked() statements to try to make sure that it doesn't |
| 635 | # race ahead of the workers. |
| 636 | # Secondly, this test assumes that condition variables are not subject |
| 637 | # to spurious wakeups. The absence of spurious wakeups is an implementation |
| 638 | # detail of Condition Variables in current CPython, but in general, not |
| 639 | # a guaranteed property of condition variables as a programming |
| 640 | # construct. In particular, it is possible that this can no longer |
| 641 | # be conveniently guaranteed should their implementation ever change. |
| 642 | ready = [] |
| 643 | results1 = [] |
| 644 | results2 = [] |
| 645 | phase_num = 0 |
| 646 | def f(): |
| 647 | cond.acquire() |
| 648 | ready.append(phase_num) |
| 649 | result = cond.wait() |
| 650 | |
| 651 | cond.release() |
| 652 | results1.append((result, phase_num)) |
| 653 | |
| 654 | cond.acquire() |
| 655 | ready.append(phase_num) |
| 656 | |
| 657 | result = cond.wait() |
| 658 | cond.release() |
| 659 | results2.append((result, phase_num)) |
| 660 | |
| 661 | N = 5 |
| 662 | with Bunch(f, N): |
| 663 | # first wait, to ensure all workers settle into cond.wait() before |
| 664 | # we continue. See issues #8799 and #30727. |
| 665 | for _ in support.sleeping_retry(support.SHORT_TIMEOUT): |
| 666 | if len(ready) >= N: |
| 667 | break |
| 668 | |
| 669 | ready.clear() |
| 670 | self.assertEqual(results1, []) |
| 671 | |
| 672 | # Notify 3 threads at first |
| 673 | count1 = 3 |
| 674 | cond.acquire() |
| 675 | cond.notify(count1) |
| 676 | wait_threads_blocked(count1) |
| 677 | |
| 678 | # Phase 1 |
| 679 | phase_num = 1 |
| 680 | cond.release() |
| 681 | for _ in support.sleeping_retry(support.SHORT_TIMEOUT): |
| 682 | if len(results1) >= count1: |
| 683 | break |
| 684 | |
| 685 | self.assertEqual(results1, [(True, 1)] * count1) |
| 686 | self.assertEqual(results2, []) |
| 687 |
no test coverage detected