gh-157838: Merge biased refcounts on behalf of detached threads (#157839)

Author: kumaraditya303

Include/internal/pycore_pystate.h

@@ -150,6 +150,23 @@ extern void _PyThreadState_Detach(PyThreadState *tstate);
 // to the "detached" state.
 extern void _PyThreadState_Suspend(PyThreadState *tstate);
 
+#ifdef Py_GIL_DISABLED
+// Try to atomically transition a *different* thread's state from "detached"
+// to "suspended". On success, the target thread cannot attach until
+// _PyThreadState_ResumeDetached() is called, and the caller may safely
+// perform operations that are normally only permitted for the owning thread
+// (such as merging the biased reference counts of objects it owns).
+//
+// The caller must not run arbitrary Python code, allocate GC objects, or
+// stop the world while holding the thread in the suspended state.
+// Returns 1 on success, 0 if the thread was not in the "detached" state.
+extern int _PyThreadState_TrySuspendDetached(PyThreadState *tstate);
+
+// Undo a successful _PyThreadState_TrySuspendDetached(): switch the thread
+// back to "detached" and wake it if it is waiting to attach.
+extern void _PyThreadState_ResumeDetached(PyThreadState *tstate);
+#endif
+
 // Mark the thread state as "shutting down". This is used during interpreter
 // and runtime finalization. The thread may no longer attach to the
 // interpreter and will instead block via _PyThreadState_HangThread().

Lib/test/test_free_threading/test_gc.py

@@ -5,7 +5,9 @@
 import time
 from unittest import TestCase
 import gc
+import weakref
 
+from test import support
 from test.support import threading_helper
 
 
@@ -95,6 +97,39 @@ def evil():
         thread.start()
         thread.join()
 
+    def test_merge_brc_queue_of_detached_thread(self):
+        # GH-157838: objects queued for merging by a thread that is detached
+        # (blocked in a lock acquire, sleep, etc.) are merged and freed on its
+        # behalf instead of staying alive until it runs Python code again.
+        lock = threading.Lock()
+        lock.acquire()
+        ready = threading.Event()
+        objs = []
+
+        def worker():
+            # Objects owned by this thread; only the list holds a reference.
+            objs.extend(MyObj() for _ in range(100))
+            ready.set()
+            lock.acquire()  # block while detached
+
+        thread = Thread(target=worker)
+        thread.start()
+        try:
+            ready.wait()
+            # The worker may not have detached yet when the first objects
+            # are dropped; keep trying until one is freed immediately.
+            for _ in support.sleeping_retry(support.SHORT_TIMEOUT, error=False):
+                obj = objs.pop()
+                wr = weakref.ref(obj)
+                del obj
+                if wr() is None:
+                    break
+            else:
+                self.fail("object not freed while owning thread was detached")
+        finally:
+            lock.release()
+            thread.join()
+
     def test_gc_callbacks_race_with_mutation(self):
         def collect():
             b.wait()

Misc/NEWS.d/next/Core_and_Builtins/2026-09-22-12-00-00.gh-issue-157838.brcMrg.rst

@@ -0,0 +1 @@
+Merge biased reference counts on behalf of threads that are detached instead of waiting for them to attach again, in the free-threaded build.

Python/brc.c

@@ -48,6 +48,28 @@ find_thread_state(struct _brc_bucket *bucket, uintptr_t thread_id)
     return NULL;
 }
 
+// Merge the refcounts of all objects in `stack`, keeping the queue's reference.
+static void
+merge_queued_refcounts(_PyObjectStack *stack)
+{
+    for (_PyObjectStackChunk *buf = stack->head; buf != NULL; buf = buf->prev) {
+        for (Py_ssize_t i = 0; i < buf->n; i++) {
+            _Py_ExplicitMergeRefcount(buf->objs[i], 0);
+        }
+    }
+}
+
+// Release the queue's reference to each merged object. This may run
+// destructors, so the bucket mutex must not be held.
+static void
+decref_merged_objects(_PyObjectStack *stack)
+{
+    PyObject *ob;
+    while ((ob = _PyObjectStack_Pop(stack)) != NULL) {
+        Py_DECREF(ob);
+    }
+}
+
 // Enqueue an object to be merged by the owning thread. This steals a
 // reference to the object.
 void
@@ -93,6 +115,22 @@ _Py_brc_queue_object(PyObject *ob)
         return;
     }
 
+    if (_PyThreadState_TrySuspendDetached(&tstate->base)) {
+        // The owning thread is detached (e.g. blocked on a lock or in a
+        // system call) and may not run Python code again for a long time,
+        // so merge its queue on its behalf instead of waiting for it. While
+        // it is held in the "suspended" state it cannot attach and therefore
+        // cannot touch ob_ref_local or ob_tid.
+        _PyObjectStack merged = {0};
+        _PyObjectStack_Merge(&merged, &tstate->brc.objects_to_merge);
+        merge_queued_refcounts(&merged);
+        _PyThreadState_ResumeDetached(&tstate->base);
+        PyMutex_Unlock(&bucket->mutex);
+
+        decref_merged_objects(&merged);
+        return;
+    }
+
     // Notify owning thread
     _Py_set_eval_breaker_bit(&tstate->base, _PY_EVAL_EXPLICIT_MERGE_BIT);
 

Python/pystate.c

@@ -2380,6 +2380,27 @@ _PyThreadState_SetShuttingDown(PyThreadState *tstate)
 #endif
 }
 
+#ifdef Py_GIL_DISABLED
+int
+_PyThreadState_TrySuspendDetached(PyThreadState *tstate)
+{
+    assert(tstate != _PyThreadState_GET());
+    int expected = _Py_THREAD_DETACHED;
+    return _Py_atomic_compare_exchange_int(&tstate->state, &expected,
+                                           _Py_THREAD_SUSPENDED);
+}
+
+void
+_PyThreadState_ResumeDetached(PyThreadState *tstate)
+{
+    assert(tstate != _PyThreadState_GET());
+    assert(_Py_atomic_load_int_relaxed(&tstate->state) == _Py_THREAD_SUSPENDED);
+    _Py_atomic_store_int(&tstate->state, _Py_THREAD_DETACHED);
+    // Wake the thread if it is parked in tstate_wait_attach().
+    _PyParkingLot_UnparkAll(&tstate->state);
+}
+#endif
+
 // Decrease stop-the-world counter of remaining number of threads that need to
 // pause. If we are the final thread to pause, notify the requesting thread.
 static void