various tests for synchronization primitives arbitrary sleep to wait until n threads are blocked like waiting for a lock a bunch of threads construct a bunch of nthread threads running the same function func if waitbeforeexit is true the threads won t terminate until dofinish is called wait until threads completely exit according to thread count break reference cycle the waiting andor time monotonic can be imprecise which is why comparing to the expected value would sometimes fail especially under windows test nothing insane happened tests for both recursive and nonrecursive locks threads block on lock acquire threads unblocked acquire the lock do nothing with releases the lock check that the lock is unacquired acquire the lock raise an exception with releases the lock check that the lock is unacquired even if after an exception was raised in the previous with lock block the lock shouldn t leak a thread instance when used from a foreign nonthreading thread we run many threads in the hope that existing threads ids won t be recycled can t set timeout if not blocking invalid timeout values timeoutmax is ok just a sanity test that it didn t actually wait for the timeout tests for nonrecursive weak locks which can be acquired and released from different threads lock needs to be released before reacquiring thread blocked on lock acquire thread unblocked lock can be released from a different thread issue 11618 check that lock is in a proper state after a nonzero timeout make sure that the lock still works normally after atforkreinit unlocked locked atforkreinit resets the lock to the unlocked state tests for recursive locks cannot release an unacquired lock cannot releasesave an unacquired lock thread blocked on lock acquire thread unblocked cannot release from a different thread tests for event objects all threads get notified threads blocked on first evt wait threads unblocked another time after an explicit clear the event is set gh57711 check that wait returns true even when the event is cleared before the waiting thread is woken up threads blocked on event wait threads unblocked ensure that condition is still using a lock after reset tests for condition variables be default we have an rlock the condition can be acquired multiple times 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 waitthreadsblocked 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 first wait to ensure all workers settle into cond wait before we continue see issues 8799 and 30727 notify 3 threads at first phase 1 wait until awaken workers are blocked on cond wait notify 5 threads they might be in their first or second wait phase 2 make sure all workers settle into cond wait notify all threads they are all in their second wait phase 3 a second time to check internal state is still ok note that conceptually thats 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 only increment 3 times so state 4 is never reached common tests for bounded unbounded semaphore objects phase 0 phase 1 phase 2 the semaphore is still locked final release to let the last thread finish phase 0 phase 1 phase 2 the semaphore is still locked final release to let the last thread finish there can be a thread switch between acquiring the semaphore and appending the result therefore results will not necessarily be ordered the default initial value is 1 thread blocked on sem acquire thread unblocked tests for unbounded semaphores unbounded releases are allowed and increment the semaphore s value tests for bounded semaphores cannot go past the initial value tests for barrier objects test that a barrier is passed in lockstep test that a barrier works for 10 consecutive runs test the return value from barrier wait test the action callback test that an abort will put the barrier in a broken state test that a reset on a barrier frees the waiting threads wait until the other threads are all in the barrier now pass the barrier again test that a barrier can be reset after being broken 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 test waittimeout one thread is late default timeout is 2 0 so this is shorter test the barrier s default timeout threads blocked on barrier wait threads unblocked abort the barrier arbitrary sleep to wait until n threads are blocked like waiting for a lock a bunch of threads 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 wait until threads completely exit according to _thread _count break reference cycle the waiting and or time monotonic can be imprecise which is why comparing to the expected value would sometimes fail especially under windows test nothing insane happened tests for both recursive and non recursive locks threads block on lock acquire threads unblocked acquire the lock do nothing with releases the lock check that the lock is unacquired acquire the lock raise an exception with releases the lock check that the lock is unacquired even if after an exception was raised in the previous with lock block the lock shouldn t leak a thread instance when used from a foreign non threading thread we run many threads in the hope that existing threads ids won t be recycled can t set timeout if not blocking invalid timeout values timeout_max is ok just a sanity test that it didn t actually wait for the timeout for pypy or other gcs tests for non recursive weak locks which can be acquired and released from different threads lock needs to be released before re acquiring thread blocked on lock acquire thread unblocked lock can be released from a different thread issue 11618 check that lock is in a proper state after a non zero timeout make sure that the lock still works normally after _at_fork_reinit unlocked locked _at_fork_reinit resets the lock to the unlocked state tests for recursive locks cannot release an unacquired lock cannot _release_save an unacquired lock thread blocked on lock acquire thread unblocked cannot release from a different thread tests for event objects all threads get notified threads blocked on first evt wait threads unblocked another time after an explicit clear the event is set gh 57711 check that wait returns true even when the event is cleared before the waiting thread is woken up threads blocked on event wait threads unblocked ensure that condition is still using a lock after reset tests for condition variables be default we have an rlock the condition can be acquired multiple times 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 first wait to ensure all workers settle into cond wait before we continue see issues 8799 and 30727 notify 3 threads at first phase 1 wait until awaken workers are blocked on cond wait notify 5 threads they might be in their first or second wait phase 2 make sure all workers settle into cond wait notify all threads they are all in their second wait phase 3 a second time to check internal state is still ok 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 only increment 3 times so state 4 is never reached common tests for bounded unbounded semaphore objects phase 0 phase 1 phase 2 the semaphore is still locked final release to let the last thread finish phase 0 phase 1 phase 2 the semaphore is still locked final release to let the last thread finish there can be a thread switch between acquiring the semaphore and appending the result therefore results will not necessarily be ordered the default initial value is 1 thread blocked on sem acquire thread unblocked tests for unbounded semaphores unbounded releases are allowed and increment the semaphore s value tests for bounded semaphores cannot go past the initial value tests for barrier objects test that a barrier is passed in lockstep test that a barrier works for 10 consecutive runs test the return value from barrier wait test the action callback test that an abort will put the barrier in a broken state test that a reset on a barrier frees the waiting threads wait until the other threads are all in the barrier now pass the barrier again test that a barrier can be reset after being broken 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 test wait timeout one thread is late default timeout is 2 0 so this is shorter test the barrier s default timeout threads blocked on barrier wait threads unblocked abort the barrier
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):
    
    
    time.sleep(0.010 * nthread)


class Bunch(object):
    
    def __init__(self, func, nthread, wait_before_exit=False):
        
        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

        
        self._wait_thread.__exit__(None, None, None)

        
        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):
        
        
        
        self.assertGreaterEqual(actual, expected * 0.6)
        
        self.assertLess(actual, expected * 10.0)


class BaseLockTests(BaseTestCase):
    

    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_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:
            
            wait_threads_blocked(N)
            self.assertEqual(len(bunch.finished), 0)

            
            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

        
        with lock:
            pass

        
        with Bunch(f, 1):
            pass

        
        with self.assertRaises(TypeError):
            with lock:
                raise TypeError

        
        
        with Bunch(f, 1):
            pass

    def test_thread_leak(self):
        
        
        lock = self.locktype()
        def f():
            lock.acquire()
            lock.release()

        
        
        with Bunch(f, 15):
            pass

    def test_timeout(self):
        lock = self.locktype()
        
        self.assertRaises(ValueError, lock.acquire, False, 1)
        
        self.assertRaises(ValueError, lock.acquire, timeout=-100)
        self.assertRaises(OverflowError, lock.acquire, timeout=1e100)
        self.assertRaises(OverflowError, lock.acquire, timeout=TIMEOUT_MAX + 1)
        
        lock.acquire(timeout=TIMEOUT_MAX)
        lock.release()
        t1 = time.monotonic()
        self.assertTrue(lock.acquire(timeout=5))
        t2 = time.monotonic()
        
        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()  
        self.assertIsNone(ref())


class LockTests(BaseLockTests):
    
    def test_reacquire(self):
        
        lock = self.locktype()
        phase = []

        def f():
            lock.acquire()
            phase.append(None)
            lock.acquire()
            phase.append(None)

        with threading_helper.wait_threads_exit():
            
            start_new_thread(f, ())
            self.wait_phase(phase, 1)

            
            lock.release()
            self.wait_phase(phase, 2)

    def test_different_thread(self):
        
        lock = self.locktype()
        lock.acquire()
        def f():
            lock.release()
        with Bunch(f, 1):
            pass
        lock.acquire()
        lock.release()

    def test_state_after_timeout(self):
        
        
        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):
            
            
            lock.acquire()
            lock.release()

        
        lock = self.locktype()
        lock._at_fork_reinit()
        use_lock(lock)

        
        lock2 = self.locktype()
        lock2.acquire()
        lock2._at_fork_reinit()
        use_lock(lock2)


class RLockTests(BaseLockTests):
    
    def test_reacquire(self):
        lock = self.locktype()
        lock.acquire()
        lock.acquire()
        lock.release()
        lock.acquire()
        lock.release()
        lock.release()

    def test_release_unacquired(self):
        
        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_release_save_unacquired(self):
        
        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():
            
            start_new_thread(f, ())
            self.wait_phase(phase, 1)
            self.assertEqual(0, lock._recursion_count())

            
            phase.append(None)
            self.wait_phase(phase, 3)
            self.assertEqual(0, lock._recursion_count())

    def test_different_thread(self):
        
        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):
    

    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):
        
        N = 5
        results1 = []
        results2 = []
        def f():
            results1.append(evt.wait())
            results2.append(evt.wait())

        with Bunch(f, N):
            
            wait_threads_blocked(N)
            self.assertEqual(len(results1), 0)

            
            evt.set()

        self.assertEqual(results1, [True] * N)
        self.assertEqual(results2, [True] * N)

    def test_notify(self):
        evt = self.eventtype()
        self._check_notify(evt)
        
        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)

        
        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):
        
        
        event = self.eventtype()
        results = []
        def f():
            results.append(event.wait(support.LONG_TIMEOUT))

        N = 5
        with Bunch(f, N):
            
            wait_threads_blocked(N)

            
            event.set()
            event.clear()

        self.assertEqual(results, [True] * N)

    @requires_fork
    def test_at_fork_reinit(self):
        
        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):
    

    def test_acquire(self):
        cond = self.condtype()
        
        
        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):
        
        
        
        
        
        
        
        
        
        
        
        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):
            
            
            for _ in support.sleeping_retry(support.SHORT_TIMEOUT):
                if len(ready) >= N:
                    break

            ready.clear()
            self.assertEqual(results1, [])

            
            count1 = 3
            cond.acquire()
            cond.notify(count1)
            wait_threads_blocked(count1)

            
            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, [])

            
            for _ in support.sleeping_retry(support.SHORT_TIMEOUT):
                if len(ready) >= count1 :
                    break

            
            cond.acquire()
            cond.notify(5)
            wait_threads_blocked(N)

            
            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)

            
            for _ in support.sleeping_retry(support.SHORT_TIMEOUT):
                if len(ready) >= N:
                    break

            
            cond.acquire()
            cond.notify_all()
            wait_threads_blocked(N)

            
            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)
        
        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)
            
            
            
            
            
            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):
            
            for i in range(3):
                time.sleep(0.010)
                with cond:
                    state += 1
                    cond.notify()

        self.assertEqual(len(success), 1)


class BaseSemaphoreTests(BaseTestCase):
    

    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):
            
            count1 = sem_value - 1
            wait_count(count1)
            self.assertEqual(results1 + results2, [0] * count1)

            
            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_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)
            
            self.assertFalse(sem.acquire(False))

            
            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):
            
            count1 = sem_value - 1
            wait_count(count1)
            self.assertEqual(results1 + results2, [0] * count1)

            
            phase_num = 1
            count2 = sem_value
            sem.release(count2)
            wait_count(count1 + count2)
            self.assertEqual(sorted(results1 + results2),
                             [0] * count1 + [1] * count2)

            
            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)
            
            self.assertFalse(sem.acquire(False))

            
            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
        
        
        
        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):
        
        sem = self.semtype()
        sem.acquire()
        def f():
            sem.acquire()
            sem.release()

        with Bunch(f, 1) as bunch:
            
            wait_threads_blocked(1)
            self.assertFalse(bunch.finished)

            
            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):
    

    def test_release_unacquired(self):
        
        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):
    

    def test_release_unacquired(self):
        
        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):
    
    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_barrier(self, passes=1):
        
        results = [[],[]]
        def f():
            self.multipass(results, passes)
        self.run_threads(f)

    def test_barrier_10(self):
        
        return self.test_barrier(10)

    def test_wait_return(self):
        
        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):
        
        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):
        
        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):
        
        results1 = []
        results2 = []
        results3 = []
        def f():
            i = self.barrier.wait()
            if i == self.N//2:
                
                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)
            
            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):
        
        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
            
            
            
            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):
        
        def f():
            i = self.barrier.wait()
            if i == self.N // 2:
                
                time.sleep(self.defaultTimeout / 2)
            
            self.assertRaises(threading.BrokenBarrierError,
                              self.barrier.wait, self.defaultTimeout / 4)
        self.run_threads(f)

    def test_default_timeout(self):
        
        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):
            
            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>")

            
            barrier.wait(timeout)

        self.assertRegex(repr(barrier),
                         r"<\w+\.Barrier at .*: waiters=0/3>")

        
        barrier.abort()
        self.assertRegex(repr(barrier),
                         r"<\w+\.Barrier at .*: broken>")