Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 17 additions & 0 deletions Include/internal/pycore_pystate.h
Original file line number Diff line number Diff line change
Expand Up @@ -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().
Expand Down
35 changes: 35 additions & 0 deletions Lib/test/test_free_threading/test_gc.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,9 @@
import time
from unittest import TestCase
import gc
import weakref

from test import support
from test.support import threading_helper


Expand Down Expand Up @@ -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()
Expand Down
Original file line number Diff line number Diff line change
@@ -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.
38 changes: 38 additions & 0 deletions Python/brc.c
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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);

Expand Down
21 changes: 21 additions & 0 deletions Python/pystate.c
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
Loading