diff options
Diffstat (limited to 'Lib/test/test_threading.py')
-rw-r--r-- | Lib/test/test_threading.py | 71 |
1 files changed, 71 insertions, 0 deletions
diff --git a/Lib/test/test_threading.py b/Lib/test/test_threading.py index fa666608263..b7688863626 100644 --- a/Lib/test/test_threading.py +++ b/Lib/test/test_threading.py @@ -1171,6 +1171,77 @@ class ThreadTests(BaseTestCase): self.assertEqual(out.strip(), b"OK") self.assertIn(b"can't create new thread at interpreter shutdown", err) + def test_join_daemon_thread_in_finalization(self): + # gh-123940: Py_Finalize() prevents other threads from running Python + # code, so join() can not succeed unless the thread is already done. + # (Non-Python threads, that is `threading._DummyThread`, can't be + # joined at all.) + # We raise an exception rather than hang. + for timeout in (None, 10): + with self.subTest(timeout=timeout): + code = textwrap.dedent(f""" + import threading + + + def loop(): + while True: + pass + + + class Cycle: + def __init__(self): + self.self_ref = self + self.thr = threading.Thread( + target=loop, daemon=True) + self.thr.start() + + def __del__(self): + assert self.thr.is_alive() + try: + self.thr.join(timeout={timeout}) + except PythonFinalizationError: + assert self.thr.is_alive() + print('got the correct exception!') + + # Cycle holds a reference to itself, which ensures it is + # cleaned up during the GC that runs after daemon threads + # have been forced to exit during finalization. + Cycle() + """) + rc, out, err = assert_python_ok("-c", code) + self.assertEqual(err, b"") + self.assertIn(b"got the correct exception", out) + + def test_join_finished_daemon_thread_in_finalization(self): + # (see previous test) + # If the thread is already finished, join() succeeds. + code = textwrap.dedent(""" + import threading + done = threading.Event() + + def loop(): + done.set() + + + class Cycle: + def __init__(self): + self.self_ref = self + self.thr = threading.Thread(target=loop, daemon=True) + self.thr.start() + done.wait() + + def __del__(self): + assert not self.thr.is_alive() + self.thr.join() + assert not self.thr.is_alive() + print('all clear!') + + Cycle() + """) + rc, out, err = assert_python_ok("-c", code) + self.assertEqual(err, b"") + self.assertIn(b"all clear", out) + def test_start_new_thread_failed(self): # gh-109746: if Python fails to start newly created thread # due to failure of underlying PyThread_start_new_thread() call, |