Switch to standalone start_joinable_thread() function and rich thread handle object

Author: pitrou

Lib/test/audit-tests.py

@@ -454,6 +454,9 @@ def __call__(self):
     i = _thread.start_new_thread(test_func(), ())
     lock.acquire()
 
+    handle = _thread.start_joinable_thread(test_func())
+    handle.join()
+
 
 def test_threading_abort():
     # Ensures that aborting PyThreadState_New raises the correct exception

Lib/test/test_audit.py

@@ -209,6 +209,8 @@ def test_threading(self):
         expected = [
             ("_thread.start_new_thread", "(<test_func>, (), None)"),
             ("test.test_func", "()"),
+            ("_thread.start_joinable_thread", "(<test_func>,)"),
+            ("test.test_func", "()"),
         ]
 
         self.assertEqual(actual, expected)

Lib/test/test_concurrent_futures/test_process_pool.py

@@ -194,11 +194,11 @@ def test_python_finalization_error(self):
 
         context = self.get_context()
 
-        # gh-109047: Mock the threading.start_new_thread() function to inject
+        # gh-109047: Mock the threading.start_joinable_thread() function to inject
         # RuntimeError: simulate the error raised during Python finalization.
         # Block the second creation: create _ExecutorManagerThread, but block
         # QueueFeederThread.
-        orig_start_new_thread = threading._start_new_thread
+        orig_start_new_thread = threading._start_joinable_thread
         nthread = 0
         def mock_start_new_thread(func, *args):
             nonlocal nthread
@@ -208,7 +208,7 @@ def mock_start_new_thread(func, *args):
             nthread += 1
             return orig_start_new_thread(func, *args)
 
-        with support.swap_attr(threading, '_start_new_thread',
+        with support.swap_attr(threading, '_start_joinable_thread',
                                mock_start_new_thread):
             executor = self.executor_type(max_workers=2, mp_context=context)
             with executor:

Lib/test/test_thread.py

@@ -165,100 +165,98 @@ def test_join_thread(self):
 
         def task():
             time.sleep(0.05)
-            finished.append(None)
+            finished.append(thread.get_ident())
 
         with threading_helper.wait_threads_exit():
-            joinable = True
-            ident = thread.start_new_thread(task, (), {}, joinable)
-            thread.join_thread(ident)
+            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():
-            joinable = True
-            ident = thread.start_new_thread(task, (), {}, joinable)
+            handle = thread.start_joinable_thread(task)
             time.sleep(0.05)
-            thread.join_thread(ident)
+            handle.join()
 
-    def test_join_non_joinable(self):
+    def test_join_several_times(self):
         def task():
             pass
 
         with threading_helper.wait_threads_exit():
-            ident = thread.start_new_thread(task, ())
+            handle = thread.start_joinable_thread(task)
+            handle.join()
             with self.assertRaisesRegex(ValueError, "not joinable"):
-                thread.join_thread(ident)
+                handle.join()
+
+    def test_joinable_not_joined(self):
+        handle_destroyed = thread.allocate_lock()
+        handle_destroyed.acquire()
 
-    def test_join_several_times(self):
         def task():
-            pass
+            handle_destroyed.acquire()
 
         with threading_helper.wait_threads_exit():
-            joinable = True
-            ident = thread.start_new_thread(task, (), {}, joinable)
-            thread.join_thread(ident)
-            with self.assertRaisesRegex(ValueError, "not joinable"):
-                thread.join_thread(ident)
+            handle = thread.start_joinable_thread(task)
+            del handle
+            handle_destroyed.release()
 
     def test_join_from_self(self):
         errors = []
-        start_new_thread_returned = thread.allocate_lock()
-        start_new_thread_returned.acquire()
+        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():
-            ident = thread.get_ident()
-            # Wait for start_new_thread() to return so that the joinable threads
-            # are populated with the ident, otherwise ValueError would be raised
-            # instead.
-            start_new_thread_returned.acquire()
+            start_joinable_thread_returned.acquire()
             try:
-                thread.join_thread(ident)
+                handles[0].join()
             except Exception as e:
                 errors.append(e)
             finally:
                 task_tried_to_join.release()
 
         with threading_helper.wait_threads_exit():
-            joinable = True
-            ident = thread.start_new_thread(task, (), {}, joinable)
-            start_new_thread_returned.release()
-            # Can still join after join_thread() failed in other thread
+            handle = thread.start_joinable_thread(task)
+            handles.append(handle)
+            start_joinable_thread_returned.release()
+            # Can still join after joining failed in other thread
             task_tried_to_join.acquire()
-            thread.join_thread(ident)
+            handle.join()
 
         assert len(errors) == 1
         with self.assertRaisesRegex(RuntimeError, "Cannot join current thread"):
             raise errors[0]
 
     def test_detach_from_self(self):
         errors = []
-        start_new_thread_returned = thread.allocate_lock()
-        start_new_thread_returned.acquire()
+        handles = []
+        start_joinable_thread_returned = thread.allocate_lock()
+        start_joinable_thread_returned.acquire()
         thread_detached = thread.allocate_lock()
         thread_detached.acquire()
 
         def task():
-            ident = thread.get_ident()
-            start_new_thread_returned.acquire()
+            start_joinable_thread_returned.acquire()
             try:
-                thread.detach_thread(ident)
+                handles[0].detach()
             except Exception as e:
                 errors.append(e)
             finally:
                 thread_detached.release()
 
         with threading_helper.wait_threads_exit():
-            joinable = True
-            ident = thread.start_new_thread(task, (), {}, joinable)
-            start_new_thread_returned.release()
+            handle = thread.start_joinable_thread(task)
+            handles.append(handle)
+            start_joinable_thread_returned.release()
             thread_detached.acquire()
             with self.assertRaisesRegex(ValueError, "not joinable"):
-                thread.join_thread(ident)
+                handle.join()
 
         assert len(errors) == 0
 
@@ -270,25 +268,23 @@ def task():
             lock.acquire()
 
         with threading_helper.wait_threads_exit():
-            joinable = True
-            ident = thread.start_new_thread(task, (), {}, joinable)
-            # detach_thread() returns even though the thread is blocked on lock
-            thread.detach_thread(ident)
-            # join_thread() then cannot be called anymore
+            handle = thread.start_joinable_thread(task)
+            # detach() returns even though the thread is blocked on lock
+            handle.detach()
+            # join() then cannot be called anymore
             with self.assertRaisesRegex(ValueError, "not joinable"):
-                thread.join_thread(ident)
+                handle.join()
             lock.release()
 
     def test_join_then_detach(self):
         def task():
             pass
 
         with threading_helper.wait_threads_exit():
-            joinable = True
-            ident = thread.start_new_thread(task, (), {}, joinable)
-            thread.join_thread(ident)
+            handle = thread.start_joinable_thread(task)
+            handle.join()
             with self.assertRaisesRegex(ValueError, "not joinable"):
-                thread.detach_thread(ident)
+                handle.detach()
 
 
 class Barrier:

Lib/test/test_threading.py

@@ -376,16 +376,16 @@ def test_limbo_cleanup(self):
         # Issue 7481: Failure to start thread should cleanup the limbo map.
         def fail_new_thread(*args):
             raise threading.ThreadError()
-        _start_new_thread = threading._start_new_thread
-        threading._start_new_thread = fail_new_thread
+        _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_new_thread = _start_new_thread
+            threading._start_joinable_thread = _start_joinable_thread
 
     def test_finalize_running_thread(self):
         # Issue 1402: the PyGILState_Ensure / _Release functions may be called

Lib/threading.py

@@ -34,9 +34,7 @@
            'setprofile_all_threads','settrace_all_threads']
 
 # Rename some stuff so "from threading import *" is safe
-_start_new_thread = _thread.start_new_thread
-_join_thread = _thread.join_thread
-_detach_thread = _thread.detach_thread
+_start_joinable_thread = _thread.start_joinable_thread
 _daemon_threads_allowed = _thread.daemon_threads_allowed
 _allocate_lock = _thread.allocate_lock
 _set_sentinel = _thread._set_sentinel
@@ -928,6 +926,7 @@ class is implemented.
             self._native_id = None
         self._tstate_lock = None
         self._join_lock = None
+        self._handle = None
         self._started = Event()
         self._is_stopped = False
         self._initialized = True
@@ -993,19 +992,13 @@ def start(self):
             _limbo[self] = self
         try:
             # Start joinable thread
-            _start_new_thread(self._bootstrap, (), {}, True)
+            self._handle = _start_joinable_thread(self._bootstrap)
         except Exception:
             with _active_limbo_lock:
                 del _limbo[self]
             raise
         self._started.wait()  # Will set ident and native_id
 
-        # We need to make sure the OS thread is either explicitly joined or
-        # detached at some point, otherwise system resources can be leaked.
-        def _finalizer(wr, _detach_thread=_detach_thread, ident=self._ident):
-            _detach_thread(ident)
-        self._non_joined_finalizer = _weakref.ref(self, _finalizer)
-
     def run(self):
         """Method representing the thread's activity.
 
@@ -1168,12 +1161,13 @@ def _join_os_thread(self):
         if join_lock is None:
             return
         with join_lock:
-            # Calling join() multiple times simultaneously would raise
-            # an exception in one of the callers.
-            if self._join_lock is not None:
-                _join_thread(self._ident)
+            # Calling join() multiple times would raise an exception
+            # in one of the callers.
+            if self._handle is not None:
+                self._handle.join()
+                self._handle = None
+                # No need to keep this around
                 self._join_lock = None
-                self._non_joined_finalizer = None
 
     def _wait_for_tstate_lock(self, block=True, timeout=-1):
         # Issue #18808: wait for the thread state to be gone.