diff --git a/Include/internal/pycore_pystate.h b/Include/internal/pycore_pystate.h index 253d26fe3a3cd8b..6caa7a5d30116e9 100644 --- a/Include/internal/pycore_pystate.h +++ b/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(). diff --git a/Python/brc.c b/Python/brc.c index d27687052aec190..aeef74b922b86e6 100644 --- a/Python/brc.c +++ b/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); diff --git a/Python/pystate.c b/Python/pystate.c index 9a2dc9431f8bb24..75bb7520c9ec8cc 100644 --- a/Python/pystate.c +++ b/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