aboutsummaryrefslogtreecommitdiffstatshomepage
path: root/Lib/test/test_threading.py
diff options
context:
space:
mode:
Diffstat (limited to 'Lib/test/test_threading.py')
-rw-r--r--Lib/test/test_threading.py71
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,