Merge branch 'pythongh-114271-_thread-ThreadHandle' into pythongh-11427-cherry-pick-ThreadHandle

Author: mpage

Include/internal/pycore_lock.h

@@ -135,6 +135,10 @@ typedef struct {
     uint8_t v;
 } PyEvent;
 
+// Check if the event is set without blocking. Returns 1 if the event is set or
+// 0 otherwise.
+PyAPI_FUNC(int) _PyEvent_IsSet(PyEvent *evt);
+
 // Set the event and notify any waiting threads.
 // Export for '_testinternalcapi' shared extension
 PyAPI_FUNC(void) _PyEvent_Notify(PyEvent *evt);
@@ -158,29 +162,9 @@ typedef struct _PyEventRc {
     Py_ssize_t refcount;
 } _PyEventRc;
 
-static inline _PyEventRc *
-_PyEventRc_New(void)
-{
-    _PyEventRc *erc = (_PyEventRc *)PyMem_RawCalloc(1, sizeof(_PyEventRc));
-    if (erc != NULL) {
-        erc->refcount = 1;
-    }
-    return erc;
-}
-
-static inline void
-_PyEventRc_Incref(_PyEventRc *erc)
-{
-    _Py_atomic_add_ssize(&erc->refcount, 1);
-}
-
-static inline void
-_PyEventRc_Decref(_PyEventRc *erc)
-{
-    if (_Py_atomic_add_ssize(&erc->refcount, -1) == 1) {
-        PyMem_RawFree(erc);
-    }
-}
+_PyEventRc *_PyEventRc_New(void);
+void _PyEventRc_Incref(_PyEventRc *erc);
+void _PyEventRc_Decref(_PyEventRc *erc);
 
 // _PyRawMutex implements a word-sized mutex that that does not depend on the
 // parking lot API, and therefore can be used in the parking lot

Lib/test/test_thread.py

@@ -189,8 +189,8 @@ def task():
         with threading_helper.wait_threads_exit():
             handle = thread.start_joinable_thread(task)
             handle.join()
-            with self.assertRaisesRegex(ValueError, "not joinable"):
-                handle.join()
+            # Subsequent join() calls should succeed
+            handle.join()
 
     def test_joinable_not_joined(self):
         handle_destroyed = thread.allocate_lock()
@@ -233,58 +233,60 @@ def task():
         with self.assertRaisesRegex(RuntimeError, "Cannot join current thread"):
             raise errors[0]
 
-    def test_detach_from_self(self):
-        errors = []
-        handles = []
-        start_joinable_thread_returned = thread.allocate_lock()
-        start_joinable_thread_returned.acquire()
-        thread_detached = thread.allocate_lock()
-        thread_detached.acquire()
+    def test_join_then_self_join(self):
+        # make sure we can't deadlock in the following scenario with
+        # threads t0 and t1:
+        #
+        # - t0 joins t1
+        # - t1 self joins
+        def make_lock():
+            lock = thread.allocate_lock()
+            lock.acquire()
+            return lock
+
+        error = None
+        self_joiner_handle = None
+        self_joiner_started = make_lock()
+        self_joiner_barrier = make_lock()
+        def self_joiner():
+            nonlocal error
+
+            self_joiner_started.release()
+            self_joiner_barrier.acquire()
 
-        def task():
-            start_joinable_thread_returned.acquire()
             try:
-                handles[0].detach()
+                self_joiner_handle.join()
             except Exception as e:
-                errors.append(e)
-            finally:
-                thread_detached.release()
+                error = e
+
+        joiner_started = make_lock()
+        def joiner():
+            joiner_started.release()
+            self_joiner_handle.join()
 
         with threading_helper.wait_threads_exit():
-            handle = thread.start_joinable_thread(task)
-            handles.append(handle)
-            start_joinable_thread_returned.release()
-            thread_detached.acquire()
-            with self.assertRaisesRegex(ValueError, "not joinable"):
-                handle.join()
+            self_joiner_handle = thread.start_joinable_thread(self_joiner)
+            # Wait for the self-joining thread to start
+            self_joiner_started.acquire()
 
-        assert len(errors) == 0
+            # Start the thread that joins the self-joiner
+            joiner_handle = thread.start_joinable_thread(joiner)
 
-    def test_detach_then_join(self):
-        lock = thread.allocate_lock()
-        lock.acquire()
+            # Wait for the joiner to start
+            joiner_started.acquire()
 
-        def task():
-            lock.acquire()
+            # Not great, but I don't think there's a deterministic way to make
+            # sure that the self-joining thread has been joined.
+            time.sleep(0.1)
 
-        with threading_helper.wait_threads_exit():
-            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"):
-                handle.join()
-            lock.release()
-
-    def test_join_then_detach(self):
-        def task():
-            pass
+            # Unblock the self-joiner
+            self_joiner_barrier.release()
 
-        with threading_helper.wait_threads_exit():
-            handle = thread.start_joinable_thread(task)
-            handle.join()
-            with self.assertRaisesRegex(ValueError, "not joinable"):
-                handle.detach()
+            self_joiner_handle.join()
+            joiner_handle.join()
+
+            with self.assertRaisesRegex(RuntimeError, "Cannot join current thread"):
+                raise error
 
 
 class Barrier:

Lib/threading.py

@@ -932,13 +932,10 @@ def _after_fork(self, new_ident=None):
             self._ident = new_ident
             if self._handle is not None:
                 assert self._handle.ident == new_ident
-            if self._join_lock is not None:
-                self._join_lock._at_fork_reinit()
         else:
             # This thread isn't alive after fork: it doesn't have a tstate
             # anymore.
             self._done_event.set()
-            self._join_lock = None
             self._handle = None
 
     def __repr__(self):
@@ -970,8 +967,6 @@ def start(self):
         if self._started.is_set():
             raise RuntimeError("threads can only be started once")
 
-        self._join_lock = _allocate_lock()
-
         with _active_limbo_lock:
             _limbo[self] = self
         try:
@@ -1099,17 +1094,9 @@ def join(self, timeout=None):
             self._join_os_thread()
 
     def _join_os_thread(self):
-        join_lock = self._join_lock
-        if join_lock is None:
-            return
-        with join_lock:
-            # 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._handle may be cleared post-fork
+        if self._handle is not None:
+            self._handle.join()
 
     @property
     def name(self):
@@ -1377,6 +1364,10 @@ def __init__(self):
         with _active_limbo_lock:
             _active[self._ident] = self
 
+    def _join_os_thread(self):
+        # No ThreadHandle for main thread
+        pass
+
 
 # Helper thread-local instance to detect when a _DummyThread
 # is collected. Not a part of the public API.