threading support to prevent reporting refleaks when running regrtest py r note we use thread count rather than threading enumerate or the moral equivalent thereof because a threading thread object is still alive until its bootstrap method has returned even after it has been unregistered from the threading module thread count on the other hand only gets decremented after the bootstrap method has returned which gives us reliable reference counts at the end of a test run copy the thread list to get a consistent output threading dangling is a weakset its value changes when it s read timeout the warning happens when a test spawns threads and some of these threads are still running after the test completes to fix this warning join threads explicitly to wait until they complete to make the warning more likely reduce the timeout use this function when threads are being used this will ensure that the threads are cleaned up even when the test fails bpo31234 context manager to wait until all threads created in the with statement exit use thread count to check if threads exited indirectly wait until threads exit the internal tbootstrap c function of the thread module threadingsetup and threadingcleanup are designed to emit a warning if a test leaves running threads in the background this context manager is designed to cleanup threads started by the thread startnewthread which doesn t allow to wait for thread exit whereas thread thread has a join method join a thread raise an assertionerror if the thread is still alive after timeout seconds it isn t supported on subinterpreters yet context manager catching threading thread exception using threading excepthook attributes set when an exception is caught exctype excvalue exctraceback thread see threading excepthook documentation for these attributes these attributes are deleted at the context manager exit usage with threadinghelper catchthreadingexception as cm code spawning a thread which raises an exception check the thread exception use cm attributes exctype excvalue exctraceback thread exctype excvalue exctraceback thread attributes of cm no longer exists at this point to avoid reference cycles detect whether python can start new threads some webassembly platforms do not provide a working pthread implementation thread support is stubbed and any attempt to create a new thread fails wasm32wasi does not have threading wasm32emscripten can be compiled with or without pthread support s usepthreads emscriptenpthreads assume all other platforms have working thread support skip tests or modules that require working threading can be used as a functionclass decorator or to skip an entire module threading support to prevent reporting refleaks when running regrtest py r note we use thread _count rather than threading enumerate or the moral equivalent thereof because a threading thread object is still alive until its __bootstrap method has returned even after it has been unregistered from the threading module thread _count on the other hand only gets decremented after the __bootstrap method has returned which gives us reliable reference counts at the end of a test run copy the thread list to get a consistent output threading _dangling is a weakset its value changes when it s read timeout the warning happens when a test spawns threads and some of these threads are still running after the test completes to fix this warning join threads explicitly to wait until they complete to make the warning more likely reduce the timeout use this function when threads are being used this will ensure that the threads are cleaned up even when the test fails bpo 31234 context manager to wait until all threads created in the with statement exit use _thread count to check if threads exited indirectly wait until threads exit the internal t_bootstrap c function of the _thread module threading_setup and threading_cleanup are designed to emit a warning if a test leaves running threads in the background this context manager is designed to cleanup threads started by the _thread start_new_thread which doesn t allow to wait for thread exit whereas thread thread has a join method join a thread raise an assertionerror if the thread is still alive after timeout seconds it isn t supported on subinterpreters yet context manager catching threading thread exception using threading excepthook attributes set when an exception is caught exc_type exc_value exc_traceback thread see threading excepthook documentation for these attributes these attributes are deleted at the context manager exit usage with threading_helper catch_threading_exception as cm code spawning a thread which raises an exception check the thread exception use cm attributes exc_type exc_value exc_traceback thread exc_type exc_value exc_traceback thread attributes of cm no longer exists at this point to avoid reference cycles detect whether python can start new threads some webassembly platforms do not provide a working pthread implementation thread support is stubbed and any attempt to create a new thread fails wasm32 wasi does not have threading wasm32 emscripten can be compiled with or without pthread support s use_pthreads __emscripten_pthreads__ assume all other platforms have working thread support skip tests or modules that require working threading can be used as a function class decorator or to skip an entire module
import _thread
import contextlib
import functools
import sys
import threading
import time
import unittest

from test import support














def threading_setup():
    return _thread._count(), len(threading._dangling)


def threading_cleanup(*original_values):
    orig_count, orig_ndangling = original_values

    timeout = 1.0
    for _ in support.sleeping_retry(timeout, error=False):
        
        
        dangling_threads = list(threading._dangling)
        count = _thread._count()

        if count <= orig_count:
            return

    
    support.environment_altered = True
    support.print_warning(
        f"threading_cleanup() failed to clean up threads "
        f"in {timeout:.1f} seconds\n"
        f"  before: thread count={orig_count}, dangling={orig_ndangling}\n"
        f"  after: thread count={count}, dangling={len(dangling_threads)}")
    for thread in dangling_threads:
        support.print_warning(f"Dangling thread: {thread!r}")

    
    
    
    
    


def reap_threads(func):
    
    @functools.wraps(func)
    def decorator(*args):
        key = threading_setup()
        try:
            return func(*args)
        finally:
            threading_cleanup(*key)
    return decorator


@contextlib.contextmanager
def wait_threads_exit(timeout=None):
    
    if timeout is None:
        timeout = support.SHORT_TIMEOUT
    old_count = _thread._count()
    try:
        yield
    finally:
        start_time = time.monotonic()
        for _ in support.sleeping_retry(timeout, error=False):
            support.gc_collect()
            count = _thread._count()
            if count <= old_count:
                break
        else:
            dt = time.monotonic() - start_time
            msg = (f"wait_threads() failed to cleanup {count - old_count} "
                   f"threads after {dt:.1f} seconds "
                   f"(count: {count}, old count: {old_count})")
            raise AssertionError(msg)


def join_thread(thread, timeout=None):
    
    if timeout is None:
        timeout = support.SHORT_TIMEOUT
    thread.join(timeout)
    if thread.is_alive():
        msg = f"failed to join the thread in {timeout:.1f} seconds"
        raise AssertionError(msg)


@contextlib.contextmanager
def start_threads(threads, unlock=None):
    try:
        import faulthandler
    except ImportError:
        
        faulthandler = None
    threads = list(threads)
    started = []
    try:
        try:
            for t in threads:
                t.start()
                started.append(t)
        except:
            if support.verbose:
                print("Can't start %d threads, only %d threads started" %
                      (len(threads), len(started)))
            raise
        yield
    finally:
        try:
            if unlock:
                unlock()
            endtime = time.monotonic()
            for timeout in range(1, 16):
                endtime += 60
                for t in started:
                    t.join(max(endtime - time.monotonic(), 0.01))
                started = [t for t in started if t.is_alive()]
                if not started:
                    break
                if support.verbose:
                    print('Unable to join %d threads during a period of '
                          '%d minutes' % (len(started), timeout))
        finally:
            started = [t for t in started if t.is_alive()]
            if started:
                if faulthandler is not None:
                    faulthandler.dump_traceback(sys.stdout)
                raise AssertionError('Unable to join %d threads' % len(started))


class catch_threading_exception:
    

    def __init__(self):
        self.exc_type = None
        self.exc_value = None
        self.exc_traceback = None
        self.thread = None
        self._old_hook = None

    def _hook(self, args):
        self.exc_type = args.exc_type
        self.exc_value = args.exc_value
        self.exc_traceback = args.exc_traceback
        self.thread = args.thread

    def __enter__(self):
        self._old_hook = threading.excepthook
        threading.excepthook = self._hook
        return self

    def __exit__(self, *exc_info):
        threading.excepthook = self._old_hook
        del self.exc_type
        del self.exc_value
        del self.exc_traceback
        del self.thread


def _can_start_thread() -> bool:
    
    if sys.platform == "emscripten":
        return sys._emscripten_info.pthreads
    elif sys.platform == "wasi":
        return False
    else:
        
        return True

can_start_thread = _can_start_thread()

def requires_working_threading(*, module=False):
    
    msg = "requires threading support"
    if module:
        if not can_start_thread:
            raise unittest.SkipTest(msg)
    else:
        return unittest.skipUnless(can_start_thread, msg)