| """ | |
| Various tests for synchronization primitives. | |
| """ | |
| import gc | |
| import sys | |
| import time | |
| from _thread import start_new_thread, TIMEOUT_MAX | |
| import threading | |
| import unittest | |
| import weakref | |
| from test import support | |
| from test.support import threading_helper | |
| requires_fork = unittest.skipUnless(support.has_fork_support, | |
| "platform doesn't support fork " | |
| "(no _at_fork_reinit method)") | |
| def wait_threads_blocked(nthread): | |
| # Arbitrary sleep to wait until N threads are blocked, | |
| # like waiting for a lock. | |
| time.sleep(0.010 * nthread) | |
| class Bunch(object): | |
| """ | |
| A bunch of threads. | |
| """ | |
| def __init__(self, func, nthread, wait_before_exit=False): | |
| """ | |
| Construct a bunch of `nthread` threads running the same function `func`. | |
| If `wait_before_exit` is True, the threads won't terminate until | |
| do_finish() is called. | |
| """ | |
| self.func = func | |
| self.nthread = nthread | |
| self.started = [] | |
| self.finished = [] | |
| self.exceptions = [] | |
| self._can_exit = not wait_before_exit | |
| self._wait_thread = None | |
| def task(self): | |
| tid = threading.get_ident() | |
| self.started.append(tid) | |
| try: | |
| self.func() | |
| except BaseException as exc: | |
| self.exceptions.append(exc) | |
| finally: | |
| self.finished.append(tid) | |
| for _ in support.sleeping_retry(support.SHORT_TIMEOUT): | |
| if self._can_exit: | |
| break | |
| def __enter__(self): | |
| self._wait_thread = threading_helper.wait_threads_exit(support.SHORT_TIMEOUT) | |
| self._wait_thread.__enter__() | |
| try: | |
| for _ in range(self.nthread): | |
| start_new_thread(self.task, ()) | |
| except: | |
| self._can_exit = True | |
| raise | |
| for _ in support.sleeping_retry(support.SHORT_TIMEOUT): | |
| if len(self.started) >= self.nthread: | |
| break | |
| return self | |
| def __exit__(self, exc_type, exc_value, traceback): | |
| for _ in support.sleeping_retry(support.SHORT_TIMEOUT): | |
| if len(self.finished) >= self.nthread: | |
| break | |
| # Wait until threads completely exit according to _thread._count() | |
| self._wait_thread.__exit__(None, None, None) | |
| # Break reference cycle | |
| exceptions = self.exceptions | |
| self.exceptions = None | |
| if exceptions: | |
| raise ExceptionGroup(f"{self.func} threads raised exceptions", | |
| exceptions) | |
| def do_finish(self): | |
| self._can_exit = True | |
| class BaseTestCase(unittest.TestCase): | |
| def setUp(self): | |
| self._threads = threading_helper.threading_setup() | |
| def tearDown(self): | |
| threading_helper.threading_cleanup(*self._threads) | |
| support.reap_children() | |
| def assertTimeout(self, actual, expected): | |
| # The waiting and/or time.monotonic() can be imprecise, which | |
| # is why comparing to the expected value would sometimes fail | |
| # (especially under Windows). | |
| self.assertGreaterEqual(actual, expected * 0.6) | |
| # Test nothing insane happened | |
| self.assertLess(actual, expected * 10.0) | |
| class BaseLockTests(BaseTestCase): | |
| """ | |
| Tests for both recursive and non-recursive locks. | |
| """ | |
| def wait_phase(self, phase, expected): | |
| for _ in support.sleeping_retry(support.SHORT_TIMEOUT): | |
| if len(phase) >= expected: | |
| break | |
| self.assertEqual(len(phase), expected) | |
| def test_constructor(self): | |
| lock = self.locktype() | |
| del lock | |
| def test_constructor_noargs(self): | |
| self.assertRaises(TypeError, self.locktype, 1) | |
| self.assertRaises(TypeError, self.locktype, x=1) | |
| self.assertRaises(TypeError, self.locktype, 1, x=2) | |
| def test_repr(self): | |
| lock = self.locktype() | |
| self.assertRegex(repr(lock), "<unlocked .* object (.*)?at .*>") | |
| del lock | |
| def test_locked_repr(self): | |
| lock = self.locktype() | |
| lock.acquire() | |
| self.assertRegex(repr(lock), "<locked .* object (.*)?at .*>") | |
| del lock | |
| def test_acquire_destroy(self): | |
| lock = self.locktype() | |
| lock.acquire() | |
| del lock | |
| def test_acquire_release(self): | |
| lock = self.locktype() | |
| lock.acquire() | |
| lock.release() | |
| del lock | |
| def test_try_acquire(self): | |
| lock = self.locktype() | |
| self.assertTrue(lock.acquire(False)) | |
| lock.release() | |
| def test_try_acquire_contended(self): | |
| lock = self.locktype() | |
| lock.acquire() | |
| result = [] | |
| def f(): | |
| result.append(lock.acquire(False)) | |
| with Bunch(f, 1): | |
| pass | |
| self.assertFalse(result[0]) | |
| lock.release() | |
| def test_acquire_contended(self): | |
| lock = self.locktype() | |
| lock.acquire() | |
| def f(): | |
| lock.acquire() | |
| lock.release() | |
| N = 5 | |
| with Bunch(f, N) as bunch: | |
| # Threads block on lock.acquire() | |
| wait_threads_blocked(N) | |
| self.assertEqual(len(bunch.finished), 0) | |
| # Threads unblocked | |
| lock.release() | |
| self.assertEqual(len(bunch.finished), N) | |
| def test_with(self): | |
| lock = self.locktype() | |
| def f(): | |
| lock.acquire() | |
| lock.release() | |
| def with_lock(err=None): | |
| with lock: | |
| if err is not None: | |
| raise err | |
| # Acquire the lock, do nothing, with releases the lock | |
| with lock: | |
| pass | |
| # Check that the lock is unacquired | |
| with Bunch(f, 1): | |
| pass | |
| # Acquire the lock, raise an exception, with releases the lock | |
| with self.assertRaises(TypeError): | |
| with lock: | |
| raise TypeError | |
| # Check that the lock is unacquired even if after an exception | |
| # was raised in the previous "with lock:" block | |
| with Bunch(f, 1): | |
| pass | |
| def test_thread_leak(self): | |
| # The lock shouldn't leak a Thread instance when used from a foreign | |
| # (non-threading) thread. | |
| lock = self.locktype() | |
| def f(): | |
| lock.acquire() | |
| lock.release() | |
| # We run many threads in the hope that existing threads ids won't | |
| # be recycled. | |
| with Bunch(f, 15): | |
| pass | |
| def test_timeout(self): | |
| lock = self.locktype() | |
| # Can't set timeout if not blocking | |
| self.assertRaises(ValueError, lock.acquire, False, 1) | |
| # Invalid timeout values | |
| self.assertRaises(ValueError, lock.acquire, timeout=-100) | |
| self.assertRaises(OverflowError, lock.acquire, timeout=1e100) | |
| self.assertRaises(OverflowError, lock.acquire, timeout=TIMEOUT_MAX + 1) | |
| # TIMEOUT_MAX is ok | |
| lock.acquire(timeout=TIMEOUT_MAX) | |
| lock.release() | |
| t1 = time.monotonic() | |
| self.assertTrue(lock.acquire(timeout=5)) | |
| t2 = time.monotonic() | |
| # Just a sanity test that it didn't actually wait for the timeout. | |
| self.assertLess(t2 - t1, 5) | |
| results = [] | |
| def f(): | |
| t1 = time.monotonic() | |
| results.append(lock.acquire(timeout=0.5)) | |
| t2 = time.monotonic() | |
| results.append(t2 - t1) | |
| with Bunch(f, 1): | |
| pass | |
| self.assertFalse(results[0]) | |
| self.assertTimeout(results[1], 0.5) | |
| def test_weakref_exists(self): | |
| lock = self.locktype() | |
| ref = weakref.ref(lock) | |
| self.assertIsNotNone(ref()) | |
| def test_weakref_deleted(self): | |
| lock = self.locktype() | |
| ref = weakref.ref(lock) | |
| del lock | |
| gc.collect() # For PyPy or other GCs. | |
| self.assertIsNone(ref()) | |
| class LockTests(BaseLockTests): | |
| """ | |
| Tests for non-recursive, weak locks | |
| (which can be acquired and released from different threads). | |
| """ | |
| def test_reacquire(self): | |
| # Lock needs to be released before re-acquiring. | |
| lock = self.locktype() | |
| phase = [] | |
| def f(): | |
| lock.acquire() | |
| phase.append(None) | |
| lock.acquire() | |
| phase.append(None) | |
| with threading_helper.wait_threads_exit(): | |
| # Thread blocked on lock.acquire() | |
| start_new_thread(f, ()) | |
| self.wait_phase(phase, 1) | |
| # Thread unblocked | |
| lock.release() | |
| self.wait_phase(phase, 2) | |
| def test_different_thread(self): | |
| # Lock can be released from a different thread. | |
| lock = self.locktype() | |
| lock.acquire() | |
| def f(): | |
| lock.release() | |
| with Bunch(f, 1): | |
| pass | |
| lock.acquire() | |
| lock.release() | |
| def test_state_after_timeout(self): | |
| # Issue #11618: check that lock is in a proper state after a | |
| # (non-zero) timeout. | |
| lock = self.locktype() | |
| lock.acquire() | |
| self.assertFalse(lock.acquire(timeout=0.01)) | |
| lock.release() | |
| self.assertFalse(lock.locked()) | |
| self.assertTrue(lock.acquire(blocking=False)) | |
| @requires_fork | |
| def test_at_fork_reinit(self): | |
| def use_lock(lock): | |
| # make sure that the lock still works normally | |
| # after _at_fork_reinit() | |
| lock.acquire() | |
| lock.release() | |
| # unlocked | |
| lock = self.locktype() | |
| lock._at_fork_reinit() | |
| use_lock(lock) | |
| # locked: _at_fork_reinit() resets the lock to the unlocked state | |
| lock2 = self.locktype() | |
| lock2.acquire() | |
| lock2._at_fork_reinit() | |
| use_lock(lock2) | |
| class RLockTests(BaseLockTests): | |
| """ | |
| Tests for recursive locks. | |
| """ | |
| def test_repr_count(self): | |
| # see gh-134322: check that count values are correct: | |
| # when a rlock is just created, | |
| # in a second thread when rlock is acquired in the main thread. | |
| lock = self.locktype() | |
| self.assertIn("count=0", repr(lock)) | |
| self.assertIn("<unlocked", repr(lock)) | |
| lock.acquire() | |
| lock.acquire() | |
| self.assertIn("count=2", repr(lock)) | |
| self.assertIn("<locked", repr(lock)) | |
| result = [] | |
| def call_repr(): | |
| result.append(repr(lock)) | |
| with Bunch(call_repr, 1): | |
| pass | |
| self.assertIn("count=2", result[0]) | |
| self.assertIn("<locked", result[0]) | |
| def test_reacquire(self): | |
| lock = self.locktype() | |
| lock.acquire() | |
| lock.acquire() | |
| lock.release() | |
| lock.acquire() | |
| lock.release() | |
| lock.release() | |
| def test_release_unacquired(self): | |
| # Cannot release an unacquired lock | |
| lock = self.locktype() | |
| self.assertRaises(RuntimeError, lock.release) | |
| lock.acquire() | |
| lock.acquire() | |
| lock.release() | |
| lock.acquire() | |
| lock.release() | |
| lock.release() | |
| self.assertRaises(RuntimeError, lock.release) | |
| def test_locked(self): | |
| lock = self.locktype() | |
| self.assertFalse(lock.locked()) | |
| lock.acquire() | |
| self.assertTrue(lock.locked()) | |
| lock.acquire() | |
| self.assertTrue(lock.locked()) | |
| lock.release() | |
| self.assertTrue(lock.locked()) | |
| lock.release() | |
| self.assertFalse(lock.locked()) | |
| def test_locked_with_2threads(self): | |
| # see gh-134323: check that a rlock which | |
| # is acquired in a different thread, | |
| # is still locked in the main thread. | |
| result = [] | |
| rlock = self.locktype() | |
| self.assertFalse(rlock.locked()) | |
| def acquire(): | |
| result.append(rlock.locked()) | |
| rlock.acquire() | |
| result.append(rlock.locked()) | |
| with Bunch(acquire, 1): | |
| pass | |
| self.assertTrue(rlock.locked()) | |
| self.assertFalse(result[0]) | |
| self.assertTrue(result[1]) | |
| def test_release_save_unacquired(self): | |
| # Cannot _release_save an unacquired lock | |
| lock = self.locktype() | |
| self.assertRaises(RuntimeError, lock._release_save) | |
| lock.acquire() | |
| lock.acquire() | |
| lock.release() | |
| lock.acquire() | |
| lock.release() | |
| lock.release() | |
| self.assertRaises(RuntimeError, lock._release_save) | |
| def test_recursion_count(self): | |
| lock = self.locktype() | |
| self.assertEqual(0, lock._recursion_count()) | |
| lock.acquire() | |
| self.assertEqual(1, lock._recursion_count()) | |
| lock.acquire() | |
| lock.acquire() | |
| self.assertEqual(3, lock._recursion_count()) | |
| lock.release() | |
| self.assertEqual(2, lock._recursion_count()) | |
| lock.release() | |
| lock.release() | |
| self.assertEqual(0, lock._recursion_count()) | |
| phase = [] | |
| def f(): | |
| lock.acquire() | |
| phase.append(None) | |
| self.wait_phase(phase, 2) | |
| lock.release() | |
| phase.append(None) | |
| with threading_helper.wait_threads_exit(): | |
| # Thread blocked on lock.acquire() | |
| start_new_thread(f, ()) | |
| self.wait_phase(phase, 1) | |
| self.assertEqual(0, lock._recursion_count()) | |
| # Thread unblocked | |
| phase.append(None) | |
| self.wait_phase(phase, 3) | |
| self.assertEqual(0, lock._recursion_count()) | |
| def test_different_thread(self): | |
| # Cannot release from a different thread | |
| lock = self.locktype() | |
| def f(): | |
| lock.acquire() | |
| with Bunch(f, 1, True) as bunch: | |
| try: | |
| self.assertRaises(RuntimeError, lock.release) | |
| finally: | |
| bunch.do_finish() | |
| def test__is_owned(self): | |
| lock = self.locktype() | |
| self.assertFalse(lock._is_owned()) | |
| lock.acquire() | |
| self.assertTrue(lock._is_owned()) | |
| lock.acquire() | |
| self.assertTrue(lock._is_owned()) | |
| result = [] | |
| def f(): | |
| result.append(lock._is_owned()) | |
| with Bunch(f, 1): | |
| pass | |
| self.assertFalse(result[0]) | |
| lock.release() | |
| self.assertTrue(lock._is_owned()) | |
| lock.release() | |
| self.assertFalse(lock._is_owned()) | |
| class EventTests(BaseTestCase): | |
| """ | |
| Tests for Event objects. | |
| """ | |
| def test_is_set(self): | |
| evt = self.eventtype() | |
| self.assertFalse(evt.is_set()) | |
| evt.set() | |
| self.assertTrue(evt.is_set()) | |
| evt.set() | |
| self.assertTrue(evt.is_set()) | |
| evt.clear() | |
| self.assertFalse(evt.is_set()) | |
| evt.clear() | |
| self.assertFalse(evt.is_set()) | |
| def _check_notify(self, evt): | |
| # All threads get notified | |
| N = 5 | |
| results1 = [] | |
| results2 = [] | |
| def f(): | |
| results1.append(evt.wait()) | |
| results2.append(evt.wait()) | |
| with Bunch(f, N): | |
| # Threads blocked on first evt.wait() | |
| wait_threads_blocked(N) | |
| self.assertEqual(len(results1), 0) | |
| # Threads unblocked | |
| evt.set() | |
| self.assertEqual(results1, [True] * N) | |
| self.assertEqual(results2, [True] * N) | |
| def test_notify(self): | |
| evt = self.eventtype() | |
| self._check_notify(evt) | |
| # Another time, after an explicit clear() | |
| evt.set() | |
| evt.clear() | |
| self._check_notify(evt) | |
| def test_timeout(self): | |
| evt = self.eventtype() | |
| results1 = [] | |
| results2 = [] | |
| N = 5 | |
| def f(): | |
| results1.append(evt.wait(0.0)) | |
| t1 = time.monotonic() | |
| r = evt.wait(0.5) | |
| t2 = time.monotonic() | |
| results2.append((r, t2 - t1)) | |
| with Bunch(f, N): | |
| pass | |
| self.assertEqual(results1, [False] * N) | |
| for r, dt in results2: | |
| self.assertFalse(r) | |
| self.assertTimeout(dt, 0.5) | |
| # The event is set | |
| results1 = [] | |
| results2 = [] | |
| evt.set() | |
| with Bunch(f, N): | |
| pass | |
| self.assertEqual(results1, [True] * N) | |
| for r, dt in results2: | |
| self.assertTrue(r) | |
| def test_set_and_clear(self): | |
| # gh-57711: check that wait() returns true even when the event is | |
| # cleared before the waiting thread is woken up. | |
| event = self.eventtype() | |
| results = [] | |
| def f(): | |
| results.append(event.wait(support.LONG_TIMEOUT)) | |
| N = 5 | |
| with Bunch(f, N): | |
| # Threads blocked on event.wait() | |
| wait_threads_blocked(N) | |
| # Threads unblocked | |
| event.set() | |
| event.clear() | |
| self.assertEqual(results, [True] * N) | |
| @requires_fork | |
| def test_at_fork_reinit(self): | |
| # ensure that condition is still using a Lock after reset | |
| evt = self.eventtype() | |
| with evt._cond: | |
| self.assertFalse(evt._cond.acquire(False)) | |
| evt._at_fork_reinit() | |
| with evt._cond: | |
| self.assertFalse(evt._cond.acquire(False)) | |
| def test_repr(self): | |
| evt = self.eventtype() | |
| self.assertRegex(repr(evt), r"<\w+\.Event at .*: unset>") | |
| evt.set() | |
| self.assertRegex(repr(evt), r"<\w+\.Event at .*: set>") | |
| class ConditionTests(BaseTestCase): | |
| """ | |
| Tests for condition variables. | |
| """ | |
| def test_acquire(self): | |
| cond = self.condtype() | |
| # Be default we have an RLock: the condition can be acquired multiple | |
| # times. | |
| cond.acquire() | |
| cond.acquire() | |
| cond.release() | |
| cond.release() | |
| lock = threading.Lock() | |
| cond = self.condtype(lock) | |
| cond.acquire() | |
| self.assertFalse(lock.acquire(False)) | |
| cond.release() | |
| self.assertTrue(lock.acquire(False)) | |
| self.assertFalse(cond.acquire(False)) | |
| lock.release() | |
| with cond: | |
| self.assertFalse(lock.acquire(False)) | |
| def test_unacquired_wait(self): | |
| cond = self.condtype() | |
| self.assertRaises(RuntimeError, cond.wait) | |
| def test_unacquired_notify(self): | |
| cond = self.condtype() | |
| self.assertRaises(RuntimeError, cond.notify) | |
| def _check_notify(self, cond): | |
| # Note that this test is sensitive to timing. If the worker threads | |
| # don't execute in a timely fashion, the main thread may think they | |
| # are further along then they are. The main thread therefore issues | |
| # wait_threads_blocked() statements to try to make sure that it doesn't | |
| # race ahead of the workers. | |
| # Secondly, this test assumes that condition variables are not subject | |
| # to spurious wakeups. The absence of spurious wakeups is an implementation | |
| # detail of Condition Variables in current CPython, but in general, not | |
| # a guaranteed property of condition variables as a programming | |
| # construct. In particular, it is possible that this can no longer | |
| # be conveniently guaranteed should their implementation ever change. | |
| ready = [] | |
| results1 = [] | |
| results2 = [] | |
| phase_num = 0 | |
| def f(): | |
| cond.acquire() | |
| ready.append(phase_num) | |
| result = cond.wait() | |
| cond.release() | |
| results1.append((result, phase_num)) | |
| cond.acquire() | |
| ready.append(phase_num) | |
| result = cond.wait() | |
| cond.release() | |
| results2.append((result, phase_num)) | |
| N = 5 | |
| with Bunch(f, N): | |
| # first wait, to ensure all workers settle into cond.wait() before | |
| # we continue. See issues #8799 and #30727. | |
| for _ in support.sleeping_retry(support.SHORT_TIMEOUT): | |
| if len(ready) >= N: | |
| break | |
| ready.clear() | |
| self.assertEqual(results1, []) | |
| # Notify 3 threads at first | |
| count1 = 3 | |
| cond.acquire() | |
| cond.notify(count1) | |
| wait_threads_blocked(count1) | |
| # Phase 1 | |
| phase_num = 1 | |
| cond.release() | |
| for _ in support.sleeping_retry(support.SHORT_TIMEOUT): | |
| if len(results1) >= count1: | |
| break | |
| self.assertEqual(results1, [(True, 1)] * count1) | |
| self.assertEqual(results2, []) | |
| # Wait until awaken workers are blocked on cond.wait() | |
| for _ in support.sleeping_retry(support.SHORT_TIMEOUT): | |
| if len(ready) >= count1 : | |
| break | |
| # Notify 5 threads: they might be in their first or second wait | |
| cond.acquire() | |
| cond.notify(5) | |
| wait_threads_blocked(N) | |
| # Phase 2 | |
| phase_num = 2 | |
| cond.release() | |
| for _ in support.sleeping_retry(support.SHORT_TIMEOUT): | |
| if len(results1) + len(results2) >= (N + count1): | |
| break | |
| count2 = N - count1 | |
| self.assertEqual(results1, [(True, 1)] * count1 + [(True, 2)] * count2) | |
| self.assertEqual(results2, [(True, 2)] * count1) | |
| # Make sure all workers settle into cond.wait() | |
| for _ in support.sleeping_retry(support.SHORT_TIMEOUT): | |
| if len(ready) >= N: | |
| break | |
| # Notify all threads: they are all in their second wait | |
| cond.acquire() | |
| cond.notify_all() | |
| wait_threads_blocked(N) | |
| # Phase 3 | |
| phase_num = 3 | |
| cond.release() | |
| for _ in support.sleeping_retry(support.SHORT_TIMEOUT): | |
| if len(results2) >= N: | |
| break | |
| self.assertEqual(results1, [(True, 1)] * count1 + [(True, 2)] * count2) | |
| self.assertEqual(results2, [(True, 2)] * count1 + [(True, 3)] * count2) | |
| def test_notify(self): | |
| cond = self.condtype() | |
| self._check_notify(cond) | |
| # A second time, to check internal state is still ok. | |
| self._check_notify(cond) | |
| def test_timeout(self): | |
| cond = self.condtype() | |
| timeout = 0.5 | |
| results = [] | |
| def f(): | |
| cond.acquire() | |
| t1 = time.monotonic() | |
| result = cond.wait(timeout) | |
| t2 = time.monotonic() | |
| cond.release() | |
| results.append((t2 - t1, result)) | |
| N = 5 | |
| with Bunch(f, N): | |
| pass | |
| self.assertEqual(len(results), N) | |
| for dt, result in results: | |
| self.assertTimeout(dt, timeout) | |
| # Note that conceptually (that"s the condition variable protocol) | |
| # a wait() may succeed even if no one notifies us and before any | |
| # timeout occurs. Spurious wakeups can occur. | |
| # This makes it hard to verify the result value. | |
| # In practice, this implementation has no spurious wakeups. | |
| self.assertFalse(result) | |
| def test_waitfor(self): | |
| cond = self.condtype() | |
| state = 0 | |
| def f(): | |
| with cond: | |
| result = cond.wait_for(lambda: state == 4) | |
| self.assertTrue(result) | |
| self.assertEqual(state, 4) | |
| with Bunch(f, 1): | |
| for i in range(4): | |
| time.sleep(0.010) | |
| with cond: | |
| state += 1 | |
| cond.notify() | |
| def test_waitfor_timeout(self): | |
| cond = self.condtype() | |
| state = 0 | |
| success = [] | |
| def f(): | |
| with cond: | |
| dt = time.monotonic() | |
| result = cond.wait_for(lambda : state==4, timeout=0.1) | |
| dt = time.monotonic() - dt | |
| self.assertFalse(result) | |
| self.assertTimeout(dt, 0.1) | |
| success.append(None) | |
| with Bunch(f, 1): | |
| # Only increment 3 times, so state == 4 is never reached. | |
| for i in range(3): | |
| time.sleep(0.010) | |
| with cond: | |
| state += 1 | |
| cond.notify() | |
| self.assertEqual(len(success), 1) | |
| class BaseSemaphoreTests(BaseTestCase): | |
| """ | |
| Common tests for {bounded, unbounded} semaphore objects. | |
| """ | |
| def test_constructor(self): | |
| self.assertRaises(ValueError, self.semtype, value = -1) | |
| self.assertRaises(ValueError, self.semtype, value = -sys.maxsize) | |
| def test_acquire(self): | |
| sem = self.semtype(1) | |
| sem.acquire() | |
| sem.release() | |
| sem = self.semtype(2) | |
| sem.acquire() | |
| sem.acquire() | |
| sem.release() | |
| sem.release() | |
| def test_acquire_destroy(self): | |
| sem = self.semtype() | |
| sem.acquire() | |
| del sem | |
| def test_acquire_contended(self): | |
| sem_value = 7 | |
| sem = self.semtype(sem_value) | |
| sem.acquire() | |
| sem_results = [] | |
| results1 = [] | |
| results2 = [] | |
| phase_num = 0 | |
| def func(): | |
| sem_results.append(sem.acquire()) | |
| results1.append(phase_num) | |
| sem_results.append(sem.acquire()) | |
| results2.append(phase_num) | |
| def wait_count(count): | |
| for _ in support.sleeping_retry(support.SHORT_TIMEOUT): | |
| if len(results1) + len(results2) >= count: | |
| break | |
| N = 10 | |
| with Bunch(func, N): | |
| # Phase 0 | |
| count1 = sem_value - 1 | |
| wait_count(count1) | |
| self.assertEqual(results1 + results2, [0] * count1) | |
| # Phase 1 | |
| phase_num = 1 | |
| for i in range(sem_value): | |
| sem.release() | |
| count2 = sem_value | |
| wait_count(count1 + count2) | |
| self.assertEqual(sorted(results1 + results2), | |
| [0] * count1 + [1] * count2) | |
| # Phase 2 | |
| phase_num = 2 | |
| count3 = (sem_value - 1) | |
| for i in range(count3): | |
| sem.release() | |
| wait_count(count1 + count2 + count3) | |
| self.assertEqual(sorted(results1 + results2), | |
| [0] * count1 + [1] * count2 + [2] * count3) | |
| # The semaphore is still locked | |
| self.assertFalse(sem.acquire(False)) | |
| # Final release, to let the last thread finish | |
| count4 = 1 | |
| sem.release() | |
| self.assertEqual(sem_results, | |
| [True] * (count1 + count2 + count3 + count4)) | |
| def test_multirelease(self): | |
| sem_value = 7 | |
| sem = self.semtype(sem_value) | |
| sem.acquire() | |
| results1 = [] | |
| results2 = [] | |
| phase_num = 0 | |
| def func(): | |
| sem.acquire() | |
| results1.append(phase_num) | |
| sem.acquire() | |
| results2.append(phase_num) | |
| def wait_count(count): | |
| for _ in support.sleeping_retry(support.SHORT_TIMEOUT): | |
| if len(results1) + len(results2) >= count: | |
| break | |
| with Bunch(func, 10): | |
| # Phase 0 | |
| count1 = sem_value - 1 | |
| wait_count(count1) | |
| self.assertEqual(results1 + results2, [0] * count1) | |
| # Phase 1 | |
| phase_num = 1 | |
| count2 = sem_value | |
| sem.release(count2) | |
| wait_count(count1 + count2) | |
| self.assertEqual(sorted(results1 + results2), | |
| [0] * count1 + [1] * count2) | |
| # Phase 2 | |
| phase_num = 2 | |
| count3 = sem_value - 1 | |
| sem.release(count3) | |
| wait_count(count1 + count2 + count3) | |
| self.assertEqual(sorted(results1 + results2), | |
| [0] * count1 + [1] * count2 + [2] * count3) | |
| # The semaphore is still locked | |
| self.assertFalse(sem.acquire(False)) | |
| # Final release, to let the last thread finish | |
| sem.release() | |
| def test_try_acquire(self): | |
| sem = self.semtype(2) | |
| self.assertTrue(sem.acquire(False)) | |
| self.assertTrue(sem.acquire(False)) | |
| self.assertFalse(sem.acquire(False)) | |
| sem.release() | |
| self.assertTrue(sem.acquire(False)) | |
| def test_try_acquire_contended(self): | |
| sem = self.semtype(4) | |
| sem.acquire() | |
| results = [] | |
| def f(): | |
| results.append(sem.acquire(False)) | |
| results.append(sem.acquire(False)) | |
| with Bunch(f, 5): | |
| pass | |
| # There can be a thread switch between acquiring the semaphore and | |
| # appending the result, therefore results will not necessarily be | |
| # ordered. | |
| self.assertEqual(sorted(results), [False] * 7 + [True] * 3 ) | |
| def test_acquire_timeout(self): | |
| sem = self.semtype(2) | |
| self.assertRaises(ValueError, sem.acquire, False, timeout=1.0) | |
| self.assertTrue(sem.acquire(timeout=0.005)) | |
| self.assertTrue(sem.acquire(timeout=0.005)) | |
| self.assertFalse(sem.acquire(timeout=0.005)) | |
| sem.release() | |
| self.assertTrue(sem.acquire(timeout=0.005)) | |
| t = time.monotonic() | |
| self.assertFalse(sem.acquire(timeout=0.5)) | |
| dt = time.monotonic() - t | |
| self.assertTimeout(dt, 0.5) | |
| def test_default_value(self): | |
| # The default initial value is 1. | |
| sem = self.semtype() | |
| sem.acquire() | |
| def f(): | |
| sem.acquire() | |
| sem.release() | |
| with Bunch(f, 1) as bunch: | |
| # Thread blocked on sem.acquire() | |
| wait_threads_blocked(1) | |
| self.assertFalse(bunch.finished) | |
| # Thread unblocked | |
| sem.release() | |
| def test_with(self): | |
| sem = self.semtype(2) | |
| def _with(err=None): | |
| with sem: | |
| self.assertTrue(sem.acquire(False)) | |
| sem.release() | |
| with sem: | |
| self.assertFalse(sem.acquire(False)) | |
| if err: | |
| raise err | |
| _with() | |
| self.assertTrue(sem.acquire(False)) | |
| sem.release() | |
| self.assertRaises(TypeError, _with, TypeError) | |
| self.assertTrue(sem.acquire(False)) | |
| sem.release() | |
| class SemaphoreTests(BaseSemaphoreTests): | |
| """ | |
| Tests for unbounded semaphores. | |
| """ | |
| def test_release_unacquired(self): | |
| # Unbounded releases are allowed and increment the semaphore's value | |
| sem = self.semtype(1) | |
| sem.release() | |
| sem.acquire() | |
| sem.acquire() | |
| sem.release() | |
| def test_repr(self): | |
| sem = self.semtype(3) | |
| self.assertRegex(repr(sem), r"<\w+\.Semaphore at .*: value=3>") | |
| sem.acquire() | |
| self.assertRegex(repr(sem), r"<\w+\.Semaphore at .*: value=2>") | |
| sem.release() | |
| sem.release() | |
| self.assertRegex(repr(sem), r"<\w+\.Semaphore at .*: value=4>") | |
| class BoundedSemaphoreTests(BaseSemaphoreTests): | |
| """ | |
| Tests for bounded semaphores. | |
| """ | |
| def test_release_unacquired(self): | |
| # Cannot go past the initial value | |
| sem = self.semtype() | |
| self.assertRaises(ValueError, sem.release) | |
| sem.acquire() | |
| sem.release() | |
| self.assertRaises(ValueError, sem.release) | |
| def test_repr(self): | |
| sem = self.semtype(3) | |
| self.assertRegex(repr(sem), r"<\w+\.BoundedSemaphore at .*: value=3/3>") | |
| sem.acquire() | |
| self.assertRegex(repr(sem), r"<\w+\.BoundedSemaphore at .*: value=2/3>") | |
| class BarrierTests(BaseTestCase): | |
| """ | |
| Tests for Barrier objects. | |
| """ | |
| N = 5 | |
| defaultTimeout = 2.0 | |
| def setUp(self): | |
| self.barrier = self.barriertype(self.N, timeout=self.defaultTimeout) | |
| def tearDown(self): | |
| self.barrier.abort() | |
| def run_threads(self, f): | |
| with Bunch(f, self.N): | |
| pass | |
| def multipass(self, results, n): | |
| m = self.barrier.parties | |
| self.assertEqual(m, self.N) | |
| for i in range(n): | |
| results[0].append(True) | |
| self.assertEqual(len(results[1]), i * m) | |
| self.barrier.wait() | |
| results[1].append(True) | |
| self.assertEqual(len(results[0]), (i + 1) * m) | |
| self.barrier.wait() | |
| self.assertEqual(self.barrier.n_waiting, 0) | |
| self.assertFalse(self.barrier.broken) | |
| def test_constructor(self): | |
| self.assertRaises(ValueError, self.barriertype, parties=0) | |
| self.assertRaises(ValueError, self.barriertype, parties=-1) | |
| def test_barrier(self, passes=1): | |
| """ | |
| Test that a barrier is passed in lockstep | |
| """ | |
| results = [[],[]] | |
| def f(): | |
| self.multipass(results, passes) | |
| self.run_threads(f) | |
| def test_barrier_10(self): | |
| """ | |
| Test that a barrier works for 10 consecutive runs | |
| """ | |
| return self.test_barrier(10) | |
| def test_wait_return(self): | |
| """ | |
| test the return value from barrier.wait | |
| """ | |
| results = [] | |
| def f(): | |
| r = self.barrier.wait() | |
| results.append(r) | |
| self.run_threads(f) | |
| self.assertEqual(sum(results), sum(range(self.N))) | |
| def test_action(self): | |
| """ | |
| Test the 'action' callback | |
| """ | |
| results = [] | |
| def action(): | |
| results.append(True) | |
| barrier = self.barriertype(self.N, action) | |
| def f(): | |
| barrier.wait() | |
| self.assertEqual(len(results), 1) | |
| self.run_threads(f) | |
| def test_abort(self): | |
| """ | |
| Test that an abort will put the barrier in a broken state | |
| """ | |
| results1 = [] | |
| results2 = [] | |
| def f(): | |
| try: | |
| i = self.barrier.wait() | |
| if i == self.N//2: | |
| raise RuntimeError | |
| self.barrier.wait() | |
| results1.append(True) | |
| except threading.BrokenBarrierError: | |
| results2.append(True) | |
| except RuntimeError: | |
| self.barrier.abort() | |
| pass | |
| self.run_threads(f) | |
| self.assertEqual(len(results1), 0) | |
| self.assertEqual(len(results2), self.N-1) | |
| self.assertTrue(self.barrier.broken) | |
| def test_reset(self): | |
| """ | |
| Test that a 'reset' on a barrier frees the waiting threads | |
| """ | |
| results1 = [] | |
| results2 = [] | |
| results3 = [] | |
| def f(): | |
| i = self.barrier.wait() | |
| if i == self.N//2: | |
| # Wait until the other threads are all in the barrier. | |
| for _ in support.sleeping_retry(support.SHORT_TIMEOUT): | |
| if self.barrier.n_waiting >= (self.N - 1): | |
| break | |
| self.barrier.reset() | |
| else: | |
| try: | |
| self.barrier.wait() | |
| results1.append(True) | |
| except threading.BrokenBarrierError: | |
| results2.append(True) | |
| # Now, pass the barrier again | |
| self.barrier.wait() | |
| results3.append(True) | |
| self.run_threads(f) | |
| self.assertEqual(len(results1), 0) | |
| self.assertEqual(len(results2), self.N-1) | |
| self.assertEqual(len(results3), self.N) | |
| def test_abort_and_reset(self): | |
| """ | |
| Test that a barrier can be reset after being broken. | |
| """ | |
| results1 = [] | |
| results2 = [] | |
| results3 = [] | |
| barrier2 = self.barriertype(self.N) | |
| def f(): | |
| try: | |
| i = self.barrier.wait() | |
| if i == self.N//2: | |
| raise RuntimeError | |
| self.barrier.wait() | |
| results1.append(True) | |
| except threading.BrokenBarrierError: | |
| results2.append(True) | |
| except RuntimeError: | |
| self.barrier.abort() | |
| pass | |
| # Synchronize and reset the barrier. Must synchronize first so | |
| # that everyone has left it when we reset, and after so that no | |
| # one enters it before the reset. | |
| if barrier2.wait() == self.N//2: | |
| self.barrier.reset() | |
| barrier2.wait() | |
| self.barrier.wait() | |
| results3.append(True) | |
| self.run_threads(f) | |
| self.assertEqual(len(results1), 0) | |
| self.assertEqual(len(results2), self.N-1) | |
| self.assertEqual(len(results3), self.N) | |
| def test_timeout(self): | |
| """ | |
| Test wait(timeout) | |
| """ | |
| def f(): | |
| i = self.barrier.wait() | |
| if i == self.N // 2: | |
| # One thread is late! | |
| time.sleep(self.defaultTimeout / 2) | |
| # Default timeout is 2.0, so this is shorter. | |
| self.assertRaises(threading.BrokenBarrierError, | |
| self.barrier.wait, self.defaultTimeout / 4) | |
| self.run_threads(f) | |
| def test_default_timeout(self): | |
| """ | |
| Test the barrier's default timeout | |
| """ | |
| timeout = 0.100 | |
| barrier = self.barriertype(2, timeout=timeout) | |
| def f(): | |
| self.assertRaises(threading.BrokenBarrierError, | |
| barrier.wait) | |
| start_time = time.monotonic() | |
| with Bunch(f, 1): | |
| pass | |
| dt = time.monotonic() - start_time | |
| self.assertGreaterEqual(dt, timeout) | |
| def test_single_thread(self): | |
| b = self.barriertype(1) | |
| b.wait() | |
| b.wait() | |
| def test_repr(self): | |
| barrier = self.barriertype(3) | |
| timeout = support.LONG_TIMEOUT | |
| self.assertRegex(repr(barrier), r"<\w+\.Barrier at .*: waiters=0/3>") | |
| def f(): | |
| barrier.wait(timeout) | |
| N = 2 | |
| with Bunch(f, N): | |
| # Threads blocked on barrier.wait() | |
| for _ in support.sleeping_retry(support.SHORT_TIMEOUT): | |
| if barrier.n_waiting >= N: | |
| break | |
| self.assertRegex(repr(barrier), | |
| r"<\w+\.Barrier at .*: waiters=2/3>") | |
| # Threads unblocked | |
| barrier.wait(timeout) | |
| self.assertRegex(repr(barrier), | |
| r"<\w+\.Barrier at .*: waiters=0/3>") | |
| # Abort the barrier | |
| barrier.abort() | |
| self.assertRegex(repr(barrier), | |
| r"<\w+\.Barrier at .*: broken>") |