if not support.has_fork_support:

return unittest.skip("requires working os.fork()")(test)

if sys.platform in platforms_to_skip:

return unittest.skip("due to known OS bug related to thread+fork")(test)

if support.HAVE_ASAN_FORK_BUG:

return unittest.skip("libasan has a pthread_create() dead lock related to thread+fork")(test)

if support.check_sanitizer(thread=True):

return unittest.skip("TSAN doesn't support threads after fork")(test)

def __init__(self):

self.value = 0

def inc(self):

self.value += 1

def dec(self):

self.value -= 1

def get(self):

def __init__(self, name, testcase, sema, mutex, nrunning):

threading.Thread.__init__(self, name=name)

self.testcase = testcase

self.sema = sema

self.mutex = mutex

self.nrunning = nrunning

def run(self):

delay = random.random() / 10000.0

if verbose:

print('task %s will run for %.1f usec' %

(self.name, delay * 1e6))

with self.sema:

with self.mutex:

self.nrunning.inc()

if verbose:

print(self.nrunning.get(), 'tasks are running')

self.testcase.assertLessEqual(self.nrunning.get(), 3)

time.sleep(delay)

if verbose:

print('task', self.name, 'done')

with self.mutex:

self.nrunning.dec()

self.testcase.assertGreaterEqual(self.nrunning.get(), 0)

if verbose:

print('%s is finished. %d tasks are running' %

def setUp(self):

self._threads = threading_helper.threading_setup()

def tearDown(self):

threading_helper.threading_cleanup(*self._threads)

maxDiff = 9999

@cpython_only

def test_lazy_import(self):

ensure_lazy_imports("threading", {"functools", "warnings"})

@cpython_only

def test_name(self):

def func(): pass

thread = threading.Thread(name="myname1")

self.assertEqual(thread.name, "myname1")

# Convert int name to str

thread = threading.Thread(name=123)

self.assertEqual(thread.name, "123")

# target name is ignored if name is specified

thread = threading.Thread(target=func, name="myname2")

self.assertEqual(thread.name, "myname2")

with mock.patch.object(threading, '_counter', return_value=2):

thread = threading.Thread(name="")

self.assertEqual(thread.name, "Thread-2")

with mock.patch.object(threading, '_counter', return_value=3):

thread = threading.Thread()

self.assertEqual(thread.name, "Thread-3")

with mock.patch.object(threading, '_counter', return_value=5):

thread = threading.Thread(target=func)

self.assertEqual(thread.name, "Thread-5 (func)")

def test_args_argument(self):

# bpo-45735: Using list or tuple as *args* in constructor could

# achieve the same effect.

num_list = [1]

num_tuple = (1,)

str_list = ["str"]

str_tuple = ("str",)

list_in_tuple = ([1],)

tuple_in_list = [(1,)]

test_cases = (

(num_list, lambda arg: self.assertEqual(arg, 1)),

(num_tuple, lambda arg: self.assertEqual(arg, 1)),

(str_list, lambda arg: self.assertEqual(arg, "str")),

(str_tuple, lambda arg: self.assertEqual(arg, "str")),

(list_in_tuple, lambda arg: self.assertEqual(arg, [1])),

(tuple_in_list, lambda arg: self.assertEqual(arg, (1,)))

)

for args, target in test_cases:

with self.subTest(target=target, args=args):

t = threading.Thread(target=target, args=args)

t.start()

t.join()

def test_lock_no_args(self):

threading.Lock() # works

self.assertRaises(TypeError, threading.Lock, 1)

self.assertRaises(TypeError, threading.Lock, a=1)

self.assertRaises(TypeError, threading.Lock, 1, 2, a=1, b=2)

def test_lock_no_subclass(self):

# Intentionally disallow subclasses of threading.Lock because they have

# never been allowed, so why start now just because the type is public?

with self.assertRaises(TypeError):

class MyLock(threading.Lock): pass

def test_lock_or_none(self):

import types

self.assertIsInstance(threading.Lock | None, types.UnionType)

# Create a bunch of threads, let each do some work, wait until all are

# done.

def test_various_ops(self):

# This takes about n/3 seconds to run (about n/3 clumps of tasks,

# times about 1 second per clump).

NUMTASKS = 10

# no more than 3 of the 10 can run at once

sema = threading.BoundedSemaphore(value=3)

mutex = threading.RLock()

numrunning = Counter()

threads = []

for i in range(NUMTASKS):

t = TestThread("<thread %d>"%i, self, sema, mutex, numrunning)

threads.append(t)

self.assertIsNone(t.ident)

self.assertRegex(repr(t), r'^<TestThread\(.*, initial\)>$')

t.start()

if hasattr(threading, 'get_native_id'):

native_ids = set(t.native_id for t in threads) | {threading.get_native_id()}

self.assertNotIn(None, native_ids)

self.assertEqual(len(native_ids), NUMTASKS + 1)

if verbose:

print('waiting for all tasks to complete')

for t in threads:

t.join()

self.assertFalse(t.is_alive())

self.assertNotEqual(t.ident, 0)

self.assertIsNotNone(t.ident)

self.assertRegex(repr(t), r'^<TestThread\(.*, stopped -?\d+\)>$')

if verbose:

print('all tasks done')

self.assertEqual(numrunning.get(), 0)

def test_ident_of_no_threading_threads(self):

# The ident still must work for the main thread and dummy threads.

self.assertIsNotNone(threading.current_thread().ident)

def f():

ident.append(threading.current_thread().ident)

done.set()

done = threading.Event()

ident = []

with threading_helper.wait_threads_exit():

tid = _thread.start_new_thread(f, ())

done.wait()

self.assertEqual(ident[0], tid)

# run with a small(ish) thread stack size (256 KiB)

def test_various_ops_small_stack(self):

if verbose:

print('with 256 KiB thread stack size...')

try:

threading.stack_size(262144)

except _thread.error:

raise unittest.SkipTest(

'platform does not support changing thread stack size')

self.test_various_ops()

threading.stack_size(0)

# run with a large thread stack size (1 MiB)

def test_various_ops_large_stack(self):

if verbose:

print('with 1 MiB thread stack size...')

try:

threading.stack_size(0x100000)

except _thread.error:

raise unittest.SkipTest(

'platform does not support changing thread stack size')

self.test_various_ops()

threading.stack_size(0)

def test_foreign_thread(self):

# Check that a "foreign" thread can use the threading module.

dummy_thread = None

error = None

def f(mutex):

try:

nonlocal dummy_thread

nonlocal error

# Calling current_thread() forces an entry for the foreign

# thread to get made in the threading._active map.

dummy_thread = threading.current_thread()

tid = dummy_thread.ident

self.assertIn(tid, threading._active)

self.assertIsInstance(dummy_thread, threading._DummyThread)

self.assertIs(threading._active.get(tid), dummy_thread)

# gh-29376

self.assertTrue(

dummy_thread.is_alive(),

'Expected _DummyThread to be considered alive.'

)

self.assertIn('_DummyThread', repr(dummy_thread))

except BaseException as e:

error = e

finally:

mutex.release()

mutex = threading.Lock()

mutex.acquire()

with threading_helper.wait_threads_exit():

tid = _thread.start_new_thread(f, (mutex,))

# Wait for the thread to finish.

mutex.acquire()

if error is not None:

raise error

self.assertEqual(tid, dummy_thread.ident)

# Issue gh-106236:

with self.assertRaises(RuntimeError):

dummy_thread.join()

dummy_thread._started.clear()

with self.assertRaises(RuntimeError):

dummy_thread.is_alive()

# Busy wait for the following condition: after the thread dies, the

# related dummy thread must be removed from threading._active.

timeout = 5

timeout_at = time.monotonic() + timeout

while time.monotonic() < timeout_at:

if threading._active.get(dummy_thread.ident) is not dummy_thread:

break

time.sleep(.1)

else:

self.fail('It was expected that the created threading._DummyThread was removed from threading._active.')

# PyThreadState_SetAsyncExc() is a CPython-only gimmick, not (currently)

# exposed at the Python level. This test relies on ctypes to get at it.

@cpython_only

def test_PyThreadState_SetAsyncExc(self):

ctypes = import_module("ctypes")

import ctypes.util # noqa: F811

@ctypes.util.wrap_dll_function(ctypes.pythonapi)

def PyThreadState_SetAsyncExc(id: ctypes.c_ulong,

exc: ctypes.py_object) -> ctypes.c_int:

pass

set_async_exc = PyThreadState_SetAsyncExc

class AsyncExc(Exception):

pass

exception = ctypes.py_object(AsyncExc)

# First check it works when setting the exception from the same thread.

tid = threading.get_ident()

self.assertIsInstance(tid, int)

self.assertGreater(tid, 0)

try:

result = set_async_exc(tid, exception)

# The exception is async, so we might have to keep the VM busy until

# it notices.

while True:

pass

except AsyncExc:

pass

else:

# This code is unreachable but it reflects the intent. If we wanted

# to be smarter the above loop wouldn't be infinite.

self.fail("AsyncExc not raised")

try:

self.assertEqual(result, 1) # one thread state modified

except UnboundLocalError:

# The exception was raised too quickly for us to get the result.

pass

# `worker_started` is set by the thread when it's inside a try/except

# block waiting to catch the asynchronously set AsyncExc exception.

# `worker_saw_exception` is set by the thread upon catching that

# exception.

worker_started = threading.Event()

worker_saw_exception = threading.Event()

class Worker(threading.Thread):

def run(self):

self.id = threading.get_ident()

self.finished = False

try:

while True:

worker_started.set()

time.sleep(0.1)

except AsyncExc:

self.finished = True

worker_saw_exception.set()

t = Worker()

t.daemon = True # so if this fails, we don't hang Python at shutdown

t.start()

if verbose:

print(" started worker thread")

# Try a thread id that doesn't make sense.

if verbose:

print(" trying nonsensical thread id")

result = set_async_exc(-1, exception)

self.assertEqual(result, 0) # no thread states modified

# Now raise an exception in the worker thread.

if verbose:

print(" waiting for worker thread to get started")

ret = worker_started.wait()

self.assertTrue(ret)

if verbose:

print(" verifying worker hasn't exited")

self.assertFalse(t.finished)

if verbose:

print(" attempting to raise asynch exception in worker")

result = set_async_exc(t.id, exception)

self.assertEqual(result, 1) # one thread state modified

if verbose:

print(" waiting for worker to say it caught the exception")

worker_saw_exception.wait(timeout=support.SHORT_TIMEOUT)

self.assertTrue(t.finished)

if verbose:

print(" all OK -- joining worker")

if t.finished:

t.join()

# else the thread is still running, and we have no way to kill it

@cpython_only

@unittest.skipUnless(hasattr(signal, "pthread_kill"), "need pthread_kill")

@unittest.skipUnless(hasattr(signal, "SIGUSR1"), "need SIGUSR1")

def test_PyThreadState_SetAsyncExc_interrupts_sleep(self):

_testcapi = import_module("_testlimitedcapi")

worker_started = threading.Event()

class InjectedException(Exception):

"""Custom exception for testing"""

caught_exception = None

def catch_exception():

nonlocal caught_exception

day_as_seconds = 60 * 60 * 24

try:

worker_started.set()

time.sleep(day_as_seconds)

except InjectedException as exc:

caught_exception = exc

thread = threading.Thread(target=catch_exception)

thread.start()

worker_started.wait()

signal.signal(signal.SIGUSR1, lambda sig, frame: None)

result = _testcapi.threadstate_set_async_exc(

thread.ident, InjectedException)

self.assertEqual(result, 1)

for _ in support.sleeping_retry(support.SHORT_TIMEOUT):

if not thread.is_alive():

break

try:

signal.pthread_kill(thread.ident, signal.SIGUSR1)

except OSError:

# The thread might have terminated between the is_alive check

# and the pthread_kill

break

thread.join()

signal.signal(signal.SIGUSR1, signal.SIG_DFL)

self.assertIsInstance(caught_exception, InjectedException)

def test_limbo_cleanup(self):

# Issue 7481: Failure to start thread should cleanup the limbo map.

def fail_new_thread(*args, **kwargs):

raise threading.ThreadError()

_start_joinable_thread = threading._start_joinable_thread

threading._start_joinable_thread = fail_new_thread

try:

t = threading.Thread(target=lambda: None)

self.assertRaises(threading.ThreadError, t.start)

self.assertFalse(

t in threading._limbo,

"Failed to cleanup _limbo map on failure of Thread.start().")

finally:

threading._start_joinable_thread = _start_joinable_thread

def test_finalize_running_thread(self):

# Issue 1402: the PyGILState_Ensure / _Release functions may be called

# very late on python exit: on deallocation of a running thread for

# example.

if support.check_sanitizer(thread=True):

# the thread running `time.sleep(100)` below will still be alive

# at process exit

self.skipTest("TSAN would report thread leak")

import_module("ctypes")

rc, out, err = assert_python_failure("-c", """if 1:

""")

self.assertEqual(rc, 42)

def test_finalize_with_trace(self):

# Issue1733757

# Avoid a deadlock when sys.settrace steps into threading._shutdown

if support.check_sanitizer(thread=True):

# the thread running `time.sleep(2)` below will still be alive

# at process exit

self.skipTest("TSAN would report thread leak")

assert_python_ok("-c", """if 1:

""")

def test_join_nondaemon_on_shutdown(self):

# Issue 1722344

# Raising SystemExit skipped threading._shutdown

rc, out, err = assert_python_ok("-c", """if 1:

""")

self.assertEqual(out.strip(),

b"Woke up, sleep function is: <built-in function sleep>")

self.assertEqual(err, b"")

def test_enumerate_after_join(self):

# Try hard to trigger #1703448: a thread is still returned in

# threading.enumerate() after it has been join()ed.

enum = threading.enumerate

old_interval = sys.getswitchinterval()

try:

for i in range(1, 100):

support.setswitchinterval(i * 0.0002)

t = threading.Thread(target=lambda: None)

t.start()

t.join()

l = enum()

self.assertNotIn(t, l,

"#1703448 triggered after %d trials: %s" % (i, l))

finally:

sys.setswitchinterval(old_interval)

@support.bigmemtest(size=20, memuse=72*2**20, dry_run=False)

def test_join_from_multiple_threads(self, size):

# Thread.join() should be thread-safe

errors = []

def worker():

time.sleep(0.005)

def joiner(thread):

try:

thread.join()

except Exception as e:

errors.append(e)

for N in range(2, 20):

threads = [threading.Thread(target=worker)]

for i in range(N):

threads.append(threading.Thread(target=joiner,

args=(threads[0],)))

for t in threads:

t.start()

time.sleep(0.01)

for t in threads:

t.join()

if errors:

raise errors[0]

def test_join_with_timeout(self):

lock = _thread.allocate_lock()

lock.acquire()

def worker():

lock.acquire()

thread = threading.Thread(target=worker)

thread.start()

thread.join(timeout=0.01)

assert thread.is_alive()

lock.release()

thread.join()

assert not thread.is_alive()

def test_no_refcycle_through_target(self):

class RunSelfFunction(object):

def __init__(self, should_raise):

# The links in this refcycle from Thread back to self

# should be cleaned up when the thread completes.

self.should_raise = should_raise

self.thread = threading.Thread(target=self._run,

args=(self,),

kwargs={'yet_another':self})

self.thread.start()

def _run(self, other_ref, yet_another):

if self.should_raise:

raise SystemExit

restore_default_excepthook(self)

cyclic_object = RunSelfFunction(should_raise=False)

weak_cyclic_object = weakref.ref(cyclic_object)

cyclic_object.thread.join()

del cyclic_object

self.assertIsNone(weak_cyclic_object(),

msg=('%d references still around' %

sys.getrefcount(weak_cyclic_object())))

raising_cyclic_object = RunSelfFunction(should_raise=True)

weak_raising_cyclic_object = weakref.ref(raising_cyclic_object)

raising_cyclic_object.thread.join()

del raising_cyclic_object

self.assertIsNone(weak_raising_cyclic_object(),

msg=('%d references still around' %

sys.getrefcount(weak_raising_cyclic_object())))

def test_old_threading_api(self):

# Just a quick sanity check to make sure the old method names are

# still present

t = threading.Thread()

with self.assertWarnsRegex(DeprecationWarning,

r'get the daemon attribute'):

t.isDaemon()

with self.assertWarnsRegex(DeprecationWarning,

r'set the daemon attribute'):

t.setDaemon(True)

with self.assertWarnsRegex(DeprecationWarning,

r'get the name attribute'):

t.getName()

with self.assertWarnsRegex(DeprecationWarning,

r'set the name attribute'):

t.setName("name")

e = threading.Event()

with self.assertWarnsRegex(DeprecationWarning, 'use is_set()'):

e.isSet()

cond = threading.Condition()

cond.acquire()

with self.assertWarnsRegex(DeprecationWarning, 'use notify_all()'):

cond.notifyAll()

with self.assertWarnsRegex(DeprecationWarning, 'use active_count()'):

threading.activeCount()

with self.assertWarnsRegex(DeprecationWarning, 'use current_thread()'):

threading.currentThread()

def test_repr_daemon(self):

t = threading.Thread()

self.assertNotIn('daemon', repr(t))

t.daemon = True

self.assertIn('daemon', repr(t))

def test_daemon_param(self):

t = threading.Thread()

self.assertFalse(t.daemon)

t = threading.Thread(daemon=False)

self.assertFalse(t.daemon)

t = threading.Thread(daemon=True)

self.assertTrue(t.daemon)

@skip_unless_reliable_fork

def test_dummy_thread_after_fork(self):

# Issue #14308: a dummy thread in the active list doesn't mess up

# the after-fork mechanism.

code = """if 1:

_, out, err = assert_python_ok("-c", code)

self.assertEqual(out, b'')

self.assertEqual(err, b'')

@skip_unless_reliable_fork

def test_is_alive_after_fork(self):

# Try hard to trigger #18418: is_alive() could sometimes be True on

# threads that vanished after a fork.

old_interval = sys.getswitchinterval()

self.addCleanup(sys.setswitchinterval, old_interval)

# Make the bug more likely to manifest.

test.support.setswitchinterval(1e-6)

for i in range(20):

t = threading.Thread(target=lambda: None)

t.start()

# Ignore the warning about fork with threads.

with warnings.catch_warnings(category=DeprecationWarning,

action="ignore"):

if (pid := os.fork()) == 0:

os._exit(11 if t.is_alive() else 10)

else:

t.join()

support.wait_process(pid, exitcode=10)

def test_main_thread(self):

main = threading.main_thread()

self.assertEqual(main.name, 'MainThread')

self.assertEqual(main.ident, threading.current_thread().ident)

self.assertEqual(main.ident, threading.get_ident())

def f():

self.assertNotEqual(threading.main_thread().ident,

threading.current_thread().ident)

th = threading.Thread(target=f)

th.start()

th.join()

@skip_unless_reliable_fork

@unittest.skipUnless(hasattr(os, 'waitpid'), "test needs os.waitpid()")

def test_main_thread_after_fork(self):

code = """if 1:

_, out, err = assert_python_ok("-c", code)

data = out.decode().replace('\r', '')

self.assertEqual(err, b"")

self.assertEqual(data,

"current ident True\n"

"main MainThread\n"

"main ident True\n"

"current is main True\n")

@skip_unless_reliable_fork

@unittest.skipUnless(hasattr(os, 'waitpid'), "test needs os.waitpid()")

def test_main_thread_after_fork_from_nonmain_thread(self):

code = """if 1:

_, out, err = assert_python_ok("-c", code)

data = out.decode().replace('\r', '')

self.assertEqual(err.decode('utf-8'), "")

self.assertEqual(data,

"current ident True\n"

"main Thread-1 (func) Thread\n"

"main ident True\n"

"current is main True\n"

)

@skip_unless_reliable_fork

@unittest.skipUnless(hasattr(os, 'waitpid'), "test needs os.waitpid()")

def test_main_thread_after_fork_from_foreign_thread(self, create_dummy=False):

code = """if 1:

""" % create_dummy

# "DeprecationWarning: This process is multi-threaded, use of fork()

# may lead to deadlocks in the child"

_, out, err = assert_python_ok("-W", "ignore::DeprecationWarning", "-c", code)

data = out.decode().replace('\r', '')

self.assertEqual(err.decode(), "")

self.assertEqual(data,

("current Dummy-1 _DummyThread\n" if create_dummy else "") +

f"ident in _active {create_dummy!s}\n" +

"current ident True\n"

"main MainThread _MainThread\n"

"main ident True\n"

"current is main True\n"

"_dangling ['MainThread']\n")

def test_main_thread_after_fork_from_dummy_thread(self, create_dummy=False):

self.test_main_thread_after_fork_from_foreign_thread(create_dummy=True)

def test_main_thread_during_shutdown(self):

# bpo-31516: current_thread() should still point to the main thread

# at shutdown

code = """if 1:

_, out, err = assert_python_ok("-c", code)

data = out.decode()

self.assertEqual(err, b"")

self.assertEqual(data.splitlines(),

["GC: True True True"] * 2)

def test_finalization_shutdown(self):

# bpo-36402: Py_Finalize() calls threading._shutdown() which must wait

# until Python thread states of all non-daemon threads get deleted.

#

# Test similar to SubinterpThreadingTests.test_threads_join_2(), but

# test the finalization of the main interpreter.

code = """if 1:

rc, out, err = assert_python_ok("-c", code)

self.assertEqual(err, b"")

def test_repr_stopped(self):

# Verify that "stopped" shows up in repr(Thread) appropriately.

started = _thread.allocate_lock()

finish = _thread.allocate_lock()

started.acquire()

finish.acquire()

def f():

started.release()

finish.acquire()

t = threading.Thread(target=f)

t.start()

started.acquire()

self.assertIn("started", repr(t))

finish.release()

# "stopped" should appear in the repr in a reasonable amount of time.

# Implementation detail: as of this writing, that's trivially true

# if .join() is called, and almost trivially true if .is_alive() is

# called. The detail we're testing here is that "stopped" shows up

# "all on its own".