Add test.{lock_tests,test_thread} from CPython 3.8.2

Lib/test/lock_tests.py

@@ -0,0 +1,949 @@
+"""
+Various tests for synchronization primitives.
+"""
+
+import sys
+import time
+from _thread import start_new_thread, TIMEOUT_MAX
+import threading
+import unittest
+import weakref
+
+from test import support
+
+
+def _wait():
+    # A crude wait/yield function not relying on synchronization primitives.
+    time.sleep(0.01)
+
+class Bunch(object):
+    """
+    A bunch of threads.
+    """
+    def __init__(self, f, n, wait_before_exit=False):
+        """
+        Construct a bunch of `n` threads running the same function `f`.
+        If `wait_before_exit` is True, the threads won't terminate until
+        do_finish() is called.
+        """
+        self.f = f
+        self.n = n
+        self.started = []
+        self.finished = []
+        self._can_exit = not wait_before_exit
+        self.wait_thread = support.wait_threads_exit()
+        self.wait_thread.__enter__()
+
+        def task():
+            tid = threading.get_ident()
+            self.started.append(tid)
+            try:
+                f()
+            finally:
+                self.finished.append(tid)
+                while not self._can_exit:
+                    _wait()
+
+        try:
+            for i in range(n):
+                start_new_thread(task, ())
+        except:
+            self._can_exit = True
+            raise
+
+    def wait_for_started(self):
+        while len(self.started) < self.n:
+            _wait()
+
+    def wait_for_finished(self):
+        while len(self.finished) < self.n:
+            _wait()
+        # Wait for threads exit
+        self.wait_thread.__exit__(None, None, None)
+
+    def do_finish(self):
+        self._can_exit = True
+
+
+class BaseTestCase(unittest.TestCase):
+    def setUp(self):
+        self._threads = support.threading_setup()
+
+    def tearDown(self):
+        support.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 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))
+        Bunch(f, 1).wait_for_finished()
+        self.assertFalse(result[0])
+        lock.release()
+
+    def test_acquire_contended(self):
+        lock = self.locktype()
+        lock.acquire()
+        N = 5
+        def f():
+            lock.acquire()
+            lock.release()
+
+        b = Bunch(f, N)
+        b.wait_for_started()
+        _wait()
+        self.assertEqual(len(b.finished), 0)
+        lock.release()
+        b.wait_for_finished()
+        self.assertEqual(len(b.finished), N)
+
+    def test_with(self):
+        lock = self.locktype()
+        def f():
+            lock.acquire()
+            lock.release()
+        def _with(err=None):
+            with lock:
+                if err is not None:
+                    raise err
+        _with()
+        # Check the lock is unacquired
+        Bunch(f, 1).wait_for_finished()
+        self.assertRaises(TypeError, _with, TypeError)
+        # Check the lock is unacquired
+        Bunch(f, 1).wait_for_finished()
+
+    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()
+        n = len(threading.enumerate())
+        # We run many threads in the hope that existing threads ids won't
+        # be recycled.
+        Bunch(f, 15).wait_for_finished()
+        if len(threading.enumerate()) != n:
+            # There is a small window during which a Thread instance's
+            # target function has finished running, but the Thread is still
+            # alive and registered.  Avoid spurious failures by waiting a
+            # bit more (seen on a buildbot).
+            time.sleep(0.4)
+            self.assertEqual(n, len(threading.enumerate()))
+
+    def test_timeout(self):
+        lock = self.locktype()
+        # Can't set timeout if not blocking
+        self.assertRaises(ValueError, lock.acquire, 0, 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)
+        Bunch(f, 1).wait_for_finished()
+        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
+        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 support.wait_threads_exit():
+            start_new_thread(f, ())
+            while len(phase) == 0:
+                _wait()
+            _wait()
+            self.assertEqual(len(phase), 1)
+            lock.release()
+            while len(phase) == 1:
+                _wait()
+            self.assertEqual(len(phase), 2)
+
+    def test_different_thread(self):
+        # Lock can be released from a different thread.
+        lock = self.locktype()
+        lock.acquire()
+        def f():
+            lock.release()
+        b = Bunch(f, 1)
+        b.wait_for_finished()
+        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))
+
+
+class RLockTests(BaseLockTests):
+    """
+    Tests for recursive locks.
+    """
+    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_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_different_thread(self):
+        # Cannot release from a different thread
+        lock = self.locktype()
+        def f():
+            lock.acquire()
+        b = Bunch(f, 1, True)
+        try:
+            self.assertRaises(RuntimeError, lock.release)
+        finally:
+            b.do_finish()
+        b.wait_for_finished()
+
+    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())
+        Bunch(f, 1).wait_for_finished()
+        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())
+        b = Bunch(f, N)
+        b.wait_for_started()
+        _wait()
+        self.assertEqual(len(results1), 0)
+        evt.set()
+        b.wait_for_finished()
+        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))
+        Bunch(f, N).wait_for_finished()
+        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()
+        Bunch(f, N).wait_for_finished()
+        self.assertEqual(results1, [True] * N)
+        for r, dt in results2:
+            self.assertTrue(r)
+
+    def test_set_and_clear(self):
+        # Issue #13502: check that wait() returns true even when the event is
+        # cleared before the waiting thread is woken up.
+        evt = self.eventtype()
+        results = []
+        timeout = 0.250
+        N = 5
+        def f():
+            results.append(evt.wait(timeout * 4))
+        b = Bunch(f, N)
+        b.wait_for_started()
+        time.sleep(timeout)
+        evt.set()
+        evt.clear()
+        b.wait_for_finished()
+        self.assertEqual(results, [True] * N)
+
+    def test_reset_internal_locks(self):
+        # ensure that condition is still using a Lock after reset
+        evt = self.eventtype()
+        with evt._cond:
+            self.assertFalse(evt._cond.acquire(False))
+        evt._reset_internal_locks()
+        with evt._cond:
+            self.assertFalse(evt._cond.acquire(False))
+
+
+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() 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.
+        N = 5
+        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))
+        b = Bunch(f, N)
+        b.wait_for_started()
+        # first wait, to ensure all workers settle into cond.wait() before
+        # we continue. See issues #8799 and #30727.
+        while len(ready) < 5:
+            _wait()
+        ready.clear()
+        self.assertEqual(results1, [])
+        # Notify 3 threads at first
+        cond.acquire()
+        cond.notify(3)
+        _wait()
+        phase_num = 1
+        cond.release()
+        while len(results1) < 3:
+            _wait()
+        self.assertEqual(results1, [(True, 1)] * 3)
+        self.assertEqual(results2, [])
+        # make sure all awaken workers settle into cond.wait()
+        while len(ready) < 3:
+            _wait()
+        # Notify 5 threads: they might be in their first or second wait
+        cond.acquire()
+        cond.notify(5)
+        _wait()
+        phase_num = 2
+        cond.release()
+        while len(results1) + len(results2) < 8:
+            _wait()
+        self.assertEqual(results1, [(True, 1)] * 3 + [(True, 2)] * 2)
+        self.assertEqual(results2, [(True, 2)] * 3)
+        # make sure all workers settle into cond.wait()
+        while len(ready) < 5:
+            _wait()
+        # Notify all threads: they are all in their second wait
+        cond.acquire()
+        cond.notify_all()
+        _wait()
+        phase_num = 3
+        cond.release()
+        while len(results2) < 5:
+            _wait()
+        self.assertEqual(results1, [(True, 1)] * 3 + [(True,2)] * 2)
+        self.assertEqual(results2, [(True, 2)] * 3 + [(True, 3)] * 2)
+        b.wait_for_finished()
+
+    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()
+        results = []
+        N = 5
+        def f():
+            cond.acquire()
+            t1 = time.monotonic()
+            result = cond.wait(0.5)
+            t2 = time.monotonic()
+            cond.release()
+            results.append((t2 - t1, result))
+        Bunch(f, N).wait_for_finished()
+        self.assertEqual(len(results), N)
+        for dt, result in results:
+            self.assertTimeout(dt, 0.5)
+            # 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)
+        b = Bunch(f, 1)
+        b.wait_for_started()
+        for i in range(4):
+            time.sleep(0.01)
+            with cond:
+                state += 1
+                cond.notify()
+        b.wait_for_finished()
+
+    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)
+        b = Bunch(f, 1)
+        b.wait_for_started()
+        # Only increment 3 times, so state == 4 is never reached.
+        for i in range(3):
+            time.sleep(0.01)
+            with cond:
+                state += 1
+                cond.notify()
+        b.wait_for_finished()
+        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 = self.semtype(7)
+        sem.acquire()
+        N = 10
+        sem_results = []
+        results1 = []
+        results2 = []
+        phase_num = 0
+        def f():
+            sem_results.append(sem.acquire())
+            results1.append(phase_num)
+            sem_results.append(sem.acquire())
+            results2.append(phase_num)
+        b = Bunch(f, 10)
+        b.wait_for_started()
+        while len(results1) + len(results2) < 6:
+            _wait()
+        self.assertEqual(results1 + results2, [0] * 6)
+        phase_num = 1
+        for i in range(7):
+            sem.release()
+        while len(results1) + len(results2) < 13:
+            _wait()
+        self.assertEqual(sorted(results1 + results2), [0] * 6 + [1] * 7)
+        phase_num = 2
+        for i in range(6):
+            sem.release()
+        while len(results1) + len(results2) < 19:
+            _wait()
+        self.assertEqual(sorted(results1 + results2), [0] * 6 + [1] * 7 + [2] * 6)
+        # The semaphore is still locked
+        self.assertFalse(sem.acquire(False))
+        # Final release, to let the last thread finish
+        sem.release()
+        b.wait_for_finished()
+        self.assertEqual(sem_results, [True] * (6 + 7 + 6 + 1))
+
+    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))
+        Bunch(f, 5).wait_for_finished()
+        # 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()
+        b = Bunch(f, 1)
+        b.wait_for_started()
+        _wait()
+        self.assertFalse(b.finished)
+        sem.release()
+        b.wait_for_finished()
+
+    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()
+
+
+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)
+
+
+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):
+        b = Bunch(f, self.N-1)
+        f()
+        b.wait_for_finished()
+
+    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):
+        """
+        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.
+                while self.barrier.n_waiting < self.N-1:
+                    time.sleep(0.001)
+                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(1.0)
+            # Default timeout is 2.0, so this is shorter.
+            self.assertRaises(threading.BrokenBarrierError,
+                              self.barrier.wait, 0.5)
+        self.run_threads(f)
+
+    def test_default_timeout(self):
+        """
+        Test the barrier's default timeout
+        """
+        # create a barrier with a low default timeout
+        barrier = self.barriertype(self.N, timeout=0.3)
+        def f():
+            i = barrier.wait()
+            if i == self.N // 2:
+                # One thread is later than the default timeout of 0.3s.
+                time.sleep(1.0)
+            self.assertRaises(threading.BrokenBarrierError, barrier.wait)
+        self.run_threads(f)
+
+    def test_single_thread(self):
+        b = self.barriertype(1)
+        b.wait()
+        b.wait()

Lib/test/test_thread.py

@@ -0,0 +1,266 @@
+import os
+import unittest
+import random
+from test import support
+import _thread as thread
+import time
+import weakref
+
+from test import lock_tests
+
+NUMTASKS = 10
+NUMTRIPS = 3
+POLL_SLEEP = 0.010 # seconds = 10 ms
+
+_print_mutex = thread.allocate_lock()
+
+def verbose_print(arg):
+    """Helper function for printing out debugging output."""
+    if support.verbose:
+        with _print_mutex:
+            print(arg)
+
+
+class BasicThreadTest(unittest.TestCase):
+
+    def setUp(self):
+        self.done_mutex = thread.allocate_lock()
+        self.done_mutex.acquire()
+        self.running_mutex = thread.allocate_lock()
+        self.random_mutex = thread.allocate_lock()
+        self.created = 0
+        self.running = 0
+        self.next_ident = 0
+
+        key = support.threading_setup()
+        self.addCleanup(support.threading_cleanup, *key)
+
+
+class ThreadRunningTests(BasicThreadTest):
+
+    def newtask(self):
+        with self.running_mutex:
+            self.next_ident += 1
+            verbose_print("creating task %s" % self.next_ident)
+            thread.start_new_thread(self.task, (self.next_ident,))
+            self.created += 1
+            self.running += 1
+
+    def task(self, ident):
+        with self.random_mutex:
+            delay = random.random() / 10000.0
+        verbose_print("task %s will run for %sus" % (ident, round(delay*1e6)))
+        time.sleep(delay)
+        verbose_print("task %s done" % ident)
+        with self.running_mutex:
+            self.running -= 1
+            if self.created == NUMTASKS and self.running == 0:
+                self.done_mutex.release()
+
+    def test_starting_threads(self):
+        with support.wait_threads_exit():
+            # Basic test for thread creation.
+            for i in range(NUMTASKS):
+                self.newtask()
+            verbose_print("waiting for tasks to complete...")
+            self.done_mutex.acquire()
+            verbose_print("all tasks done")
+
+    def test_stack_size(self):
+        # Various stack size tests.
+        self.assertEqual(thread.stack_size(), 0, "initial stack size is not 0")
+
+        thread.stack_size(0)
+        self.assertEqual(thread.stack_size(), 0, "stack_size not reset to default")
+
+    @unittest.skipIf(os.name not in ("nt", "posix"), 'test meant for nt and posix')
+    def test_nt_and_posix_stack_size(self):
+        try:
+            thread.stack_size(4096)
+        except ValueError:
+            verbose_print("caught expected ValueError setting "
+                            "stack_size(4096)")
+        except thread.error:
+            self.skipTest("platform does not support changing thread stack "
+                          "size")
+
+        fail_msg = "stack_size(%d) failed - should succeed"
+        for tss in (262144, 0x100000, 0):
+            thread.stack_size(tss)
+            self.assertEqual(thread.stack_size(), tss, fail_msg % tss)
+            verbose_print("successfully set stack_size(%d)" % tss)
+
+        for tss in (262144, 0x100000):
+            verbose_print("trying stack_size = (%d)" % tss)
+            self.next_ident = 0
+            self.created = 0
+            with support.wait_threads_exit():
+                for i in range(NUMTASKS):
+                    self.newtask()
+
+                verbose_print("waiting for all tasks to complete")
+                self.done_mutex.acquire()
+                verbose_print("all tasks done")
+
+        thread.stack_size(0)
+
+    def test__count(self):
+        # Test the _count() function.
+        orig = thread._count()
+        mut = thread.allocate_lock()
+        mut.acquire()
+        started = []
+
+        def task():
+            started.append(None)
+            mut.acquire()
+            mut.release()
+
+        with support.wait_threads_exit():
+            thread.start_new_thread(task, ())
+            while not started:
+                time.sleep(POLL_SLEEP)
+            self.assertEqual(thread._count(), orig + 1)
+            # Allow the task to finish.
+            mut.release()
+            # The only reliable way to be sure that the thread ended from the
+            # interpreter's point of view is to wait for the function object to be
+            # destroyed.
+            done = []
+            wr = weakref.ref(task, lambda _: done.append(None))
+            del task
+            while not done:
+                time.sleep(POLL_SLEEP)
+            self.assertEqual(thread._count(), orig)
+
+    def test_unraisable_exception(self):
+        def task():
+            started.release()
+            raise ValueError("task failed")
+
+        started = thread.allocate_lock()
+        with support.catch_unraisable_exception() as cm:
+            with support.wait_threads_exit():
+                started.acquire()
+                thread.start_new_thread(task, ())
+                started.acquire()
+
+            self.assertEqual(str(cm.unraisable.exc_value), "task failed")
+            self.assertIs(cm.unraisable.object, task)
+            self.assertEqual(cm.unraisable.err_msg,
+                             "Exception ignored in thread started by")
+            self.assertIsNotNone(cm.unraisable.exc_traceback)
+
+
+class Barrier:
+    def __init__(self, num_threads):
+        self.num_threads = num_threads
+        self.waiting = 0
+        self.checkin_mutex  = thread.allocate_lock()
+        self.checkout_mutex = thread.allocate_lock()
+        self.checkout_mutex.acquire()
+
+    def enter(self):
+        self.checkin_mutex.acquire()
+        self.waiting = self.waiting + 1
+        if self.waiting == self.num_threads:
+            self.waiting = self.num_threads - 1
+            self.checkout_mutex.release()
+            return
+        self.checkin_mutex.release()
+
+        self.checkout_mutex.acquire()
+        self.waiting = self.waiting - 1
+        if self.waiting == 0:
+            self.checkin_mutex.release()
+            return
+        self.checkout_mutex.release()
+
+
+class BarrierTest(BasicThreadTest):
+
+    def test_barrier(self):
+        with support.wait_threads_exit():
+            self.bar = Barrier(NUMTASKS)
+            self.running = NUMTASKS
+            for i in range(NUMTASKS):
+                thread.start_new_thread(self.task2, (i,))
+            verbose_print("waiting for tasks to end")
+            self.done_mutex.acquire()
+            verbose_print("tasks done")
+
+    def task2(self, ident):
+        for i in range(NUMTRIPS):
+            if ident == 0:
+                # give it a good chance to enter the next
+                # barrier before the others are all out
+                # of the current one
+                delay = 0
+            else:
+                with self.random_mutex:
+                    delay = random.random() / 10000.0
+            verbose_print("task %s will run for %sus" %
+                          (ident, round(delay * 1e6)))
+            time.sleep(delay)
+            verbose_print("task %s entering %s" % (ident, i))
+            self.bar.enter()
+            verbose_print("task %s leaving barrier" % ident)
+        with self.running_mutex:
+            self.running -= 1
+            # Must release mutex before releasing done, else the main thread can
+            # exit and set mutex to None as part of global teardown; then
+            # mutex.release() raises AttributeError.
+            finished = self.running == 0
+        if finished:
+            self.done_mutex.release()
+
+class LockTests(lock_tests.LockTests):
+    locktype = thread.allocate_lock
+
+
+class TestForkInThread(unittest.TestCase):
+    def setUp(self):
+        self.read_fd, self.write_fd = os.pipe()
+
+    @unittest.skipUnless(hasattr(os, 'fork'), 'need os.fork')
+    @support.reap_threads
+    def test_forkinthread(self):
+        status = "not set"
+
+        def thread1():
+            nonlocal status
+
+            # fork in a thread
+            pid = os.fork()
+            if pid == 0:
+                # child
+                try:
+                    os.close(self.read_fd)
+                    os.write(self.write_fd, b"OK")
+                finally:
+                    os._exit(0)
+            else:
+                # parent
+                os.close(self.write_fd)
+                pid, status = os.waitpid(pid, 0)
+
+        with support.wait_threads_exit():
+            thread.start_new_thread(thread1, ())
+            self.assertEqual(os.read(self.read_fd, 2), b"OK",
+                             "Unable to fork() in thread")
+        self.assertEqual(status, 0)
+
+    def tearDown(self):
+        try:
+            os.close(self.read_fd)
+        except OSError:
+            pass
+
+        try:
+            os.close(self.write_fd)
+        except OSError:
+            pass
+
+
+if __name__ == "__main__":
+    unittest.main()