gh-114271: Make `_thread.ThreadHandle` thread-safe in free-threaded builds (GH-115190)

Make `_thread.ThreadHandle` thread-safe in free-threaded builds

We protect the mutable state of `ThreadHandle` using a `_PyOnceFlag`.
Concurrent operations (i.e. `join` or `detach`) on `ThreadHandle` block
until it is their turn to execute or an earlier operation succeeds.
Once an operation has been applied successfully all future operations
complete immediately.

The `join()` method is now idempotent. It may be called multiple times
but the underlying OS thread will only be joined once. After `join()`
succeeds, any future calls to `join()` will succeed immediately.

The internal thread handle `detach()` method has been removed.

1 files changed, 47 insertions, 44 deletions

diff --git a/Lib/test/test_thread.py b/Lib/test/test_thread.py
index 931cb4b..8323523 100644
--- a/Lib/test/test_thread.py
+++ b/Lib/test/test_thread.py
@@ -189,8 +189,8 @@ class ThreadRunningTests(BasicThreadTest):
         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,61 @@ class ThreadRunningTests(BasicThreadTest):
         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 (see comment in `ThreadHandle_join()` for more
+        # details):
+        #
+        # - 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: