Skip to content

Commit afbb9f2

Browse files
gh-157838: Merge biased refcounts on behalf of detached threads (#157839)
(cherry picked from commit 030e913)
1 parent ffbbdad commit afbb9f2

5 files changed

Lines changed: 112 additions & 0 deletions

File tree

‎Include/internal/pycore_pystate.h‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -150,6 +150,23 @@ extern void _PyThreadState_Detach(PyThreadState *tstate);
150150
// to the "detached" state.
151151
extern void _PyThreadState_Suspend(PyThreadState *tstate);
152152

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

‎Lib/test/test_free_threading/test_gc.py‎

Lines changed: 35 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,9 @@
44
from threading import Thread
55
from unittest import TestCase
66
import gc
7+
import weakref
78

9+
from test import support
810
from test.support import threading_helper
911

1012

@@ -94,6 +96,39 @@ def evil():
9496
thread.start()
9597
thread.join()
9698

99+
def test_merge_brc_queue_of_detached_thread(self):
100+
# GH-157838: objects queued for merging by a thread that is detached
101+
# (blocked in a lock acquire, sleep, etc.) are merged and freed on its
102+
# behalf instead of staying alive until it runs Python code again.
103+
lock = threading.Lock()
104+
lock.acquire()
105+
ready = threading.Event()
106+
objs = []
107+
108+
def worker():
109+
# Objects owned by this thread; only the list holds a reference.
110+
objs.extend(MyObj() for _ in range(100))
111+
ready.set()
112+
lock.acquire() # block while detached
113+
114+
thread = Thread(target=worker)
115+
thread.start()
116+
try:
117+
ready.wait()
118+
# The worker may not have detached yet when the first objects
119+
# are dropped; keep trying until one is freed immediately.
120+
for _ in support.sleeping_retry(support.SHORT_TIMEOUT, error=False):
121+
obj = objs.pop()
122+
wr = weakref.ref(obj)
123+
del obj
124+
if wr() is None:
125+
break
126+
else:
127+
self.fail("object not freed while owning thread was detached")
128+
finally:
129+
lock.release()
130+
thread.join()
131+
97132
def test_set_threshold(self):
98133
# GH-148613: Setting the GC threshold from another thread could cause a
99134
# race between the `gc_should_collect` and `gc_set_threshold` functions.
Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
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‎

Lines changed: 38 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -48,6 +48,28 @@ find_thread_state(struct _brc_bucket *bucket, uintptr_t thread_id)
4848
return NULL;
4949
}
5050

51+
// Merge the refcounts of all objects in `stack`, keeping the queue's reference.
52+
static void
53+
merge_queued_refcounts(_PyObjectStack *stack)
54+
{
55+
for (_PyObjectStackChunk *buf = stack->head; buf != NULL; buf = buf->prev) {
56+
for (Py_ssize_t i = 0; i < buf->n; i++) {
57+
_Py_ExplicitMergeRefcount(buf->objs[i], 0);
58+
}
59+
}
60+
}
61+
62+
// Release the queue's reference to each merged object. This may run
63+
// destructors, so the bucket mutex must not be held.
64+
static void
65+
decref_merged_objects(_PyObjectStack *stack)
66+
{
67+
PyObject *ob;
68+
while ((ob = _PyObjectStack_Pop(stack)) != NULL) {
69+
Py_DECREF(ob);
70+
}
71+
}
72+
5173
// Enqueue an object to be merged by the owning thread. This steals a
5274
// reference to the object.
5375
void
@@ -93,6 +115,22 @@ _Py_brc_queue_object(PyObject *ob)
93115
return;
94116
}
95117

118+
if (_PyThreadState_TrySuspendDetached(&tstate->base)) {
119+
// The owning thread is detached (e.g. blocked on a lock or in a
120+
// system call) and may not run Python code again for a long time,
121+
// so merge its queue on its behalf instead of waiting for it. While
122+
// it is held in the "suspended" state it cannot attach and therefore
123+
// cannot touch ob_ref_local or ob_tid.
124+
_PyObjectStack merged = {0};
125+
_PyObjectStack_Merge(&merged, &tstate->brc.objects_to_merge);
126+
merge_queued_refcounts(&merged);
127+
_PyThreadState_ResumeDetached(&tstate->base);
128+
PyMutex_Unlock(&bucket->mutex);
129+
130+
decref_merged_objects(&merged);
131+
return;
132+
}
133+
96134
// Notify owning thread
97135
_Py_set_eval_breaker_bit(&tstate->base, _PY_EVAL_EXPLICIT_MERGE_BIT);
98136

‎Python/pystate.c‎

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2349,6 +2349,27 @@ _PyThreadState_SetShuttingDown(PyThreadState *tstate)
23492349
#endif
23502350
}
23512351

2352+
#ifdef Py_GIL_DISABLED
2353+
int
2354+
_PyThreadState_TrySuspendDetached(PyThreadState *tstate)
2355+
{
2356+
assert(tstate != _PyThreadState_GET());
2357+
int expected = _Py_THREAD_DETACHED;
2358+
return _Py_atomic_compare_exchange_int(&tstate->state, &expected,
2359+
_Py_THREAD_SUSPENDED);
2360+
}
2361+
2362+
void
2363+
_PyThreadState_ResumeDetached(PyThreadState *tstate)
2364+
{
2365+
assert(tstate != _PyThreadState_GET());
2366+
assert(_Py_atomic_load_int_relaxed(&tstate->state) == _Py_THREAD_SUSPENDED);
2367+
_Py_atomic_store_int(&tstate->state, _Py_THREAD_DETACHED);
2368+
// Wake the thread if it is parked in tstate_wait_attach().
2369+
_PyParkingLot_UnparkAll(&tstate->state);
2370+
}
2371+
#endif
2372+
23522373
// Decrease stop-the-world counter of remaining number of threads that need to
23532374
// pause. If we are the final thread to pause, notify the requesting thread.
23542375
static void

0 commit comments

Comments
 (0)