helper function for printing out debugging output if support verbose with printmutex printarg class basicthreadtestunittest testcase def setupself self donemutex thread allocatelock self donemutex acquire self runningmutex thread allocatelock self randommutex thread allocatelock self created 0 self running 0 self nextident 0 key threadinghelper threadingsetup self addcleanupthreadinghelper threadingcleanup key class threadrunningtestsbasicthreadtest def newtaskself with self runningmutex self nextident 1 verboseprintcreating task s self nextident thread startnewthreadself task self nextident self created 1 self running 1 def taskself ident with self randommutex delay random random 10000 0 verboseprinttask s will run for sus ident rounddelay1e6 time sleepdelay verboseprinttask s done ident with self runningmutex self running 1 if self created numtasks and self running 0 self donemutex release def teststartingthreadsself with threadinghelper waitthreadsexit basic test for thread creation for i in rangenumtasks self newtask verboseprintwaiting for tasks to complete self donemutex acquire verboseprintall tasks done def teststacksizeself various stack size tests self assertequalthread stacksize 0 initial stack size is not 0 thread stacksize0 self assertequalthread stacksize 0 stacksize not reset to default unittest skipifos name not in nt posix test meant for nt and posix def testntandposixstacksizeself try thread stacksize4096 except valueerror verboseprintcaught expected valueerror setting stacksize4096 except thread error self skiptestplatform does not support changing thread stack size failmsg stacksized failed should succeed for tss in 262144 0x100000 0 thread stacksizetss self assertequalthread stacksize tss failmsg tss verboseprintsuccessfully set stacksized tss for tss in 262144 0x100000 verboseprinttrying stacksize d tss self nextident 0 self created 0 with threadinghelper waitthreadsexit for i in rangenumtasks self newtask verboseprintwaiting for all tasks to complete self donemutex acquire verboseprintall tasks done thread stacksize0 def testcountself test the count function orig thread count mut thread allocatelock mut acquire started def task started appendnone mut acquire mut release with threadinghelper waitthreadsexit thread startnewthreadtask for in support sleepingretrysupport longtimeout if started break self assertequalthread 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 reftask lambda done appendnone del task for in support sleepingretrysupport longtimeout if done break support gccollect for pypy or other gcs self assertequalthread count orig def testunraisableexceptionself def task started release raise valueerrortask failed started thread allocatelock with support catchunraisableexception as cm with threadinghelper waitthreadsexit started acquire thread startnewthreadtask started acquire self assertequalstrcm unraisable excvalue task failed self assertisnonecm unraisable object self assertequalcm unraisable errmsg fexception ignored in thread started by task r self assertisnotnonecm unraisable exctraceback def testjointhreadself finished def task time sleep0 05 finished appendthread getident with threadinghelper waitthreadsexit handle thread startjoinablethreadtask handle join self assertequallenfinished 1 self assertequalhandle ident finished0 def testjointhreadalreadyexitedself def task pass with threadinghelper waitthreadsexit handle thread startjoinablethreadtask time sleep0 05 handle join def testjoinseveraltimesself def task pass with threadinghelper waitthreadsexit handle thread startjoinablethreadtask handle join with self assertraisesregexvalueerror not joinable handle join def testjoinablenotjoinedself handledestroyed thread allocatelock handledestroyed acquire def task handledestroyed acquire with threadinghelper waitthreadsexit handle thread startjoinablethreadtask del handle handledestroyed release def testjoinfromselfself errors handles startjoinablethreadreturned thread allocatelock startjoinablethreadreturned acquire tasktriedtojoin thread allocatelock tasktriedtojoin acquire def task startjoinablethreadreturned acquire try handles0 join except exception as e errors appende finally tasktriedtojoin release with threadinghelper waitthreadsexit handle thread startjoinablethreadtask handles appendhandle startjoinablethreadreturned release can still join after joining failed in other thread tasktriedtojoin acquire handle join assert lenerrors 1 with self assertraisesregexruntimeerror cannot join current thread raise errors0 def testdetachfromselfself errors handles startjoinablethreadreturned thread allocatelock startjoinablethreadreturned acquire threaddetached thread allocatelock threaddetached acquire def task startjoinablethreadreturned acquire try handles0 detach except exception as e errors appende finally threaddetached release with threadinghelper waitthreadsexit handle thread startjoinablethreadtask handles appendhandle startjoinablethreadreturned release threaddetached acquire with self assertraisesregexvalueerror not joinable handle join assert lenerrors 0 def testdetachthenjoinself lock thread allocatelock lock acquire def task lock acquire with threadinghelper waitthreadsexit handle thread startjoinablethreadtask detach returns even though the thread is blocked on lock handle detach join then cannot be called anymore with self assertraisesregexvalueerror not joinable handle join lock release def testjointhendetachself def task pass with threadinghelper waitthreadsexit handle thread startjoinablethreadtask handle join with self assertraisesregexvalueerror not joinable handle detach class barrier def initself numthreads self numthreads numthreads self waiting 0 self checkinmutex thread allocatelock self checkoutmutex thread allocatelock self checkoutmutex acquire def enterself self checkinmutex acquire self waiting self waiting 1 if self waiting self numthreads self waiting self numthreads 1 self checkoutmutex release return self checkinmutex release self checkoutmutex acquire self waiting self waiting 1 if self waiting 0 self checkinmutex release return self checkoutmutex release class barriertestbasicthreadtest def testbarrierself with threadinghelper waitthreadsexit self bar barriernumtasks self running numtasks for i in rangenumtasks thread startnewthreadself task2 i verboseprintwaiting for tasks to end self donemutex acquire verboseprinttasks done def task2self ident for i in rangenumtrips 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 randommutex delay random random 10000 0 verboseprinttask s will run for sus ident rounddelay 1e6 time sleepdelay verboseprinttask s entering s ident i self bar enter verboseprinttask s leaving barrier ident with self runningmutex 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 donemutex release class locktestslocktests locktests locktype thread allocatelock class testforkinthreadunittest testcase def setupself self readfd self writefd os pipe support requiresfork threadinghelper reapthreads def testforkinthreadself pid none def forkthreadreadfd writefd nonlocal pid ignore the warning about fork with threads with warnings catchwarningscategorydeprecationwarning actionignore fork in a thread danger undefined per posix if pid os fork parent process return child process try os closereadfd os writewritefd bok finally os exit0 with threadinghelper waitthreadsexit thread startnewthreadforkthread self readfd self writefd self assertequalos readself readfd 2 bok os closeself writefd self assertisnotnonepid support waitprocesspid exitcode0 def teardownself try os closeself readfd except oserror pass try os closeself writefd except oserror pass if name main unittest main helper function for printing out debugging output basic test for thread creation various stack size tests test the _count function allow the task to finish 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 for pypy or other gcs can still join after joining failed in other thread detach returns even though the thread is blocked on lock join then cannot be called anymore give it a good chance to enter the next barrier before the others are all out of the current one 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 ignore the warning about fork with threads fork in a thread danger undefined per posix parent process child process
import os
import unittest
import random
from test import support
from test.support import threading_helper
import _thread as thread
import time
import warnings
import weakref

from test import lock_tests

threading_helper.requires_working_threading(module=True)

NUMTASKS = 10
NUMTRIPS = 3

_print_mutex = thread.allocate_lock()

def verbose_print(arg):
    
    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 = threading_helper.threading_setup()
        self.addCleanup(threading_helper.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 threading_helper.wait_threads_exit():
            
            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):
        
        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 threading_helper.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):
        
        orig = thread._count()
        mut = thread.allocate_lock()
        mut.acquire()
        started = []

        def task():
            started.append(None)
            mut.acquire()
            mut.release()

        with threading_helper.wait_threads_exit():
            thread.start_new_thread(task, ())
            for _ in support.sleeping_retry(support.LONG_TIMEOUT):
                if started:
                    break
            self.assertEqual(thread._count(), orig + 1)

            
            mut.release()

            
            
            
            done = []
            wr = weakref.ref(task, lambda _: done.append(None))
            del task

            for _ in support.sleeping_retry(support.LONG_TIMEOUT):
                if done:
                    break
                support.gc_collect()  
            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 threading_helper.wait_threads_exit():
                started.acquire()
                thread.start_new_thread(task, ())
                started.acquire()

            self.assertEqual(str(cm.unraisable.exc_value), "task failed")
            self.assertIsNone(cm.unraisable.object)
            self.assertEqual(cm.unraisable.err_msg,
                             f"Exception ignored in thread started by {task!r}")
            self.assertIsNotNone(cm.unraisable.exc_traceback)

    def test_join_thread(self):
        finished = []

        def task():
            time.sleep(0.05)
            finished.append(thread.get_ident())

        with threading_helper.wait_threads_exit():
            handle = thread.start_joinable_thread(task)
            handle.join()
            self.assertEqual(len(finished), 1)
            self.assertEqual(handle.ident, finished[0])

    def test_join_thread_already_exited(self):
        def task():
            pass

        with threading_helper.wait_threads_exit():
            handle = thread.start_joinable_thread(task)
            time.sleep(0.05)
            handle.join()

    def test_join_several_times(self):
        def task():
            pass

        with threading_helper.wait_threads_exit():
            handle = thread.start_joinable_thread(task)
            handle.join()
            with self.assertRaisesRegex(ValueError, "not joinable"):
                handle.join()

    def test_joinable_not_joined(self):
        handle_destroyed = thread.allocate_lock()
        handle_destroyed.acquire()

        def task():
            handle_destroyed.acquire()

        with threading_helper.wait_threads_exit():
            handle = thread.start_joinable_thread(task)
            del handle
            handle_destroyed.release()

    def test_join_from_self(self):
        errors = []
        handles = []
        start_joinable_thread_returned = thread.allocate_lock()
        start_joinable_thread_returned.acquire()
        task_tried_to_join = thread.allocate_lock()
        task_tried_to_join.acquire()

        def task():
            start_joinable_thread_returned.acquire()
            try:
                handles[0].join()
            except Exception as e:
                errors.append(e)
            finally:
                task_tried_to_join.release()

        with threading_helper.wait_threads_exit():
            handle = thread.start_joinable_thread(task)
            handles.append(handle)
            start_joinable_thread_returned.release()
            
            task_tried_to_join.acquire()
            handle.join()

        assert len(errors) == 1
        with self.assertRaisesRegex(RuntimeError, "Cannot join current thread"):
            raise errors[0]

    def test_detach_from_self(self):
        errors = []
        handles = []
        start_joinable_thread_returned = thread.allocate_lock()
        start_joinable_thread_returned.acquire()
        thread_detached = thread.allocate_lock()
        thread_detached.acquire()

        def task():
            start_joinable_thread_returned.acquire()
            try:
                handles[0].detach()
            except Exception as e:
                errors.append(e)
            finally:
                thread_detached.release()

        with threading_helper.wait_threads_exit():
            handle = thread.start_joinable_thread(task)
            handles.append(handle)
            start_joinable_thread_returned.release()
            thread_detached.acquire()
            with self.assertRaisesRegex(ValueError, "not joinable"):
                handle.join()

        assert len(errors) == 0

    def test_detach_then_join(self):
        lock = thread.allocate_lock()
        lock.acquire()

        def task():
            lock.acquire()

        with threading_helper.wait_threads_exit():
            handle = thread.start_joinable_thread(task)
            
            handle.detach()
            
            with self.assertRaisesRegex(ValueError, "not joinable"):
                handle.join()
            lock.release()

    def test_join_then_detach(self):
        def task():
            pass

        with threading_helper.wait_threads_exit():
            handle = thread.start_joinable_thread(task)
            handle.join()
            with self.assertRaisesRegex(ValueError, "not joinable"):
                handle.detach()


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 threading_helper.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:
                
                
                
                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
            
            
            
            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()

    @support.requires_fork()
    @threading_helper.reap_threads
    def test_forkinthread(self):
        pid = None

        def fork_thread(read_fd, write_fd):
            nonlocal pid

            
            with warnings.catch_warnings(category=DeprecationWarning,
                                         action="ignore"):
                
                if (pid := os.fork()):
                    
                    return

            
            try:
                os.close(read_fd)
                os.write(write_fd, b"OK")
            finally:
                os._exit(0)

        with threading_helper.wait_threads_exit():
            thread.start_new_thread(fork_thread, (self.read_fd, self.write_fd))
            self.assertEqual(os.read(self.read_fd, 2), b"OK")
            os.close(self.write_fd)

        self.assertIsNotNone(pid)
        support.wait_process(pid, exitcode=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()