Skip to content
Open
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
3 changes: 1 addition & 2 deletions Include/cpython/pystate.h
Original file line number Diff line number Diff line change
Expand Up @@ -118,8 +118,7 @@ struct _ts {

int _whence;

/* Thread state (_Py_THREAD_ATTACHED, _Py_THREAD_DETACHED, _Py_THREAD_SUSPENDED).
See Include/internal/pycore_pystate.h for more details. */
/* Thread state. See Include/internal/pycore_pystate.h for details. */
int state;

int py_recursion_remaining;
Expand Down
47 changes: 24 additions & 23 deletions Include/internal/pycore_pystate.h
Original file line number Diff line number Diff line change
Expand Up @@ -21,32 +21,32 @@ extern "C" {
// interpreter at the same time. Only the "bound" thread may perform the
// transitions between "attached" and "detached" on its own PyThreadState.
//
// The "suspended" state is used to implement stop-the-world pauses, such as
// for cyclic garbage collection. It is only used in `--disable-gil` builds.
// The "suspended" state is similar to the "detached" state in that in both
// states the thread is not allowed to call most Python APIs. However, unlike
// the "detached" state, a thread may not transition itself out from the
// "suspended" state. Only the thread performing a stop-the-world pause may
// transition a thread from the "suspended" state back to the "detached" state.
// The "suspended" states are used to implement stop-the-world pauses and to
// merge biased reference counts on behalf of detached threads. They are only
// used in `--disable-gil` builds.
// They are similar to the "detached" state in that the thread is not allowed
// to call most Python APIs. A suspended thread trying to attach marks itself
// as "suspended-waiting". Only the thread responsible for suspending it may
// resume it, moving it to "detached" or "detached-waiting".
// A "detached-waiting" thread must attach before it can be suspended again.
//
// The "shutting down" state is used when the interpreter is being finalized.
// Threads in this state can't do anything other than block the OS thread.
// (See _PyThreadState_HangThread).
//
// State transition diagram:
//
// (bound thread) (stop-the-world thread)
// [attached] <-> [detached] <-> [suspended]
// | ^
// +---------------------------->---------------------------+
// (bound thread)
//
// The (bound thread) and (stop-the-world thread) labels indicate which thread
// is allowed to perform the transition.
#define _Py_THREAD_DETACHED 0
#define _Py_THREAD_ATTACHED 1
#define _Py_THREAD_SUSPENDED 2
#define _Py_THREAD_SHUTTING_DOWN 3
// State transitions:
// Bound thread: attached <-> detached
// attached -> suspended
// suspended -> suspended-waiting
// detached-waiting -> attached
// Suspending thread: detached <-> suspended
// suspended-waiting -> detached-waiting
#define _Py_THREAD_DETACHED 0
#define _Py_THREAD_ATTACHED 1
#define _Py_THREAD_SUSPENDED 2
#define _Py_THREAD_SHUTTING_DOWN 3
#define _Py_THREAD_SUSPENDED_WAITING 4
#define _Py_THREAD_DETACHED_WAITING 5


/* Check if the current thread is the main thread.
Expand Down Expand Up @@ -162,8 +162,9 @@ extern void _PyThreadState_Suspend(PyThreadState *tstate);
// 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.
// Resume a thread suspended by _PyThreadState_TrySuspendDetached() or a
// stop-the-world pause: switch it back to "detached" or "detached-waiting"
// and wake it if it is waiting to attach.
extern void _PyThreadState_ResumeDetached(PyThreadState *tstate);
#endif

Expand Down
39 changes: 38 additions & 1 deletion Lib/test/test_free_threading/test_threading.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,8 @@
import unittest
from test.support import threading_helper
import textwrap

from test import support
from test.support import script_helper, threading_helper

threading_helper.requires_working_threading(module=True)

Expand All @@ -22,5 +25,39 @@ def mutate_thread():
threading_helper.run_concurrently([repr_thread, mutate_thread])


class TestThreadState(unittest.TestCase):
@support.requires_subprocess()
def test_tight_stw_loop_does_not_starve_attach(self):
script = textwrap.dedent(f"""
import faulthandler

faulthandler.dump_traceback_later({support.SHORT_TIMEOUT}, exit=True)

import _testinternalcapi
import threading
import time

started = threading.Event()
stop = threading.Event()

def stop_the_world():
_testinternalcapi.test_stop_the_world()
started.set()
while not stop.is_set():
_testinternalcapi.test_stop_the_world()

thread = threading.Thread(target=stop_the_world)
thread.start()
started.wait()
# Each reattachment must make progress between consecutive pauses.
for _ in range(50):
time.sleep(0.02)
stop.set()
thread.join()
faulthandler.cancel_dump_traceback_later()
""")
script_helper.assert_python_ok("-X", "gil=0", "-c", script)


if __name__ == "__main__":
unittest.main()
Original file line number Diff line number Diff line change
@@ -0,0 +1,2 @@
Fix a free-threaded stop-the-world fairness issue that could starve a thread
reattaching after being suspended while detached.
19 changes: 19 additions & 0 deletions Modules/_testinternalcapi.c
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@
#include "pycore_instruction_sequence.h" // _PyInstructionSequence_New()
#include "pycore_interpframe.h" // _PyFrame_GetFunction()
#include "pycore_jit.h" // _PyJIT_AddressInJitCode()
#include "pycore_lock.h" // PyEvent_WaitTimed()
#include "pycore_object.h" // _PyObject_IsFreed()
#include "pycore_optimizer.h" // _Py_Executor_DependsOn
#include "pycore_pathconfig.h" // _PyPathConfig_ClearGlobal()
Expand Down Expand Up @@ -208,6 +209,23 @@ get_stack_margin(PyObject *self, PyObject *Py_UNUSED(args))
return PyLong_FromSize_t(_PyOS_STACK_MARGIN_BYTES);
}

static PyObject *
test_stop_the_world(PyObject *self, PyObject *Py_UNUSED(args))
{
#ifdef Py_GIL_DISABLED
PyInterpreterState *interp = _PyInterpreterState_GET();
// Request consecutive pauses without running Python code between them.
for (int i = 0; i < 100; i++) {
_PyEval_StopTheWorld(interp);
// Give detached threads time to try to reattach during the pause.
PyEvent event = {0};
PyEvent_WaitTimed(&event, 10 * 1000 * 1000, /*detach=*/0);
_PyEval_StartTheWorld(interp);
}
#endif
Py_RETURN_NONE;
}

#ifdef MS_WINDOWS
static const char *
classify_address(uintptr_t addr, int jit_enabled, PyInterpreterState *interp)
Expand Down Expand Up @@ -3298,6 +3316,7 @@ static PyMethodDef module_functions[] = {
{"get_c_recursion_remaining", get_c_recursion_remaining, METH_NOARGS},
{"get_stack_pointer", get_stack_pointer, METH_NOARGS},
{"get_stack_margin", get_stack_margin, METH_NOARGS},
{"test_stop_the_world", test_stop_the_world, METH_NOARGS},
{"classify_stack_addresses", classify_stack_addresses, METH_VARARGS},
{"get_jit_code_ranges", get_jit_code_ranges, METH_NOARGS},
{"get_jit_backend", get_jit_backend, METH_NOARGS},
Expand Down
67 changes: 55 additions & 12 deletions Python/pystate.c
Original file line number Diff line number Diff line change
Expand Up @@ -1939,7 +1939,10 @@ tstate_delete_common(PyThreadState *tstate, int release_gil)
if (tstate->next) {
tstate->next->prev = tstate->prev;
}
if (tstate->state != _Py_THREAD_SUSPENDED) {
int state = _Py_atomic_load_int_relaxed(&tstate->state);
if (state != _Py_THREAD_SUSPENDED &&
state != _Py_THREAD_SUSPENDED_WAITING)
{
// Any ongoing stop-the-world request should not wait for us because
// our thread is getting deleted.
if (interp->stoptheworld.requested) {
Expand Down Expand Up @@ -2223,6 +2226,22 @@ tstate_try_attach(PyThreadState *tstate)
#endif
}

static int
tstate_try_attach_detached(PyThreadState *tstate, int *state)
{
#ifdef Py_GIL_DISABLED
assert(*state == _Py_THREAD_DETACHED ||
*state == _Py_THREAD_DETACHED_WAITING);
return _Py_atomic_compare_exchange_int(&tstate->state,
state,
_Py_THREAD_ATTACHED);
#else
assert(tstate->state == _Py_THREAD_DETACHED);
tstate->state = _Py_THREAD_ATTACHED;
return 1;
#endif
}

static void
tstate_set_detached(PyThreadState *tstate, int detached_state)
{
Expand All @@ -2237,10 +2256,20 @@ tstate_set_detached(PyThreadState *tstate, int detached_state)
static void
tstate_wait_attach(PyThreadState *tstate)
{
do {
for (;;) {
int state = _Py_atomic_load_int_relaxed(&tstate->state);
if (state == _Py_THREAD_SUSPENDED) {
// Wait until we're switched out of SUSPENDED to DETACHED.
// Register an active attach waiter. The next stop-the-world
// request must let this thread attach before suspending it again.
if (!_Py_atomic_compare_exchange_int(
&tstate->state, &state, _Py_THREAD_SUSPENDED_WAITING))
{
continue;
}
state = _Py_THREAD_SUSPENDED_WAITING;
Comment thread
tpn marked this conversation as resolved.
}
if (state == _Py_THREAD_SUSPENDED_WAITING) {
// Park rechecks the state before sleeping, in case we were resumed.
_PyParkingLot_Park(&tstate->state, &state, sizeof(tstate->state),
/*timeout=*/-1, NULL, /*detach=*/0);
}
Expand All @@ -2249,10 +2278,13 @@ tstate_wait_attach(PyThreadState *tstate)
_PyThreadState_HangThread(tstate);
}
else {
assert(state == _Py_THREAD_DETACHED);
assert(state == _Py_THREAD_DETACHED ||
state == _Py_THREAD_DETACHED_WAITING);
if (tstate_try_attach_detached(tstate, &state)) {
return;
}
Comment thread
tpn marked this conversation as resolved.
}
// Once we're back in DETACHED we can re-attach
} while (!tstate_try_attach(tstate));
}
}

void
Expand Down Expand Up @@ -2394,8 +2426,20 @@ 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);
int state = _Py_atomic_load_int_relaxed(&tstate->state);
int next_state;
do {
assert(state == _Py_THREAD_SUSPENDED ||
state == _Py_THREAD_SUSPENDED_WAITING);
if (state == _Py_THREAD_SUSPENDED_WAITING) {
next_state = _Py_THREAD_DETACHED_WAITING;
}
else {
next_state = _Py_THREAD_DETACHED;
}
// Retry if an attach waiter registered concurrently.
} while (!_Py_atomic_compare_exchange_int(
&tstate->state, &state, next_state));
// Wake the thread if it is parked in tstate_wait_attach().
_PyParkingLot_UnparkAll(&tstate->state);
}
Expand Down Expand Up @@ -2442,6 +2486,8 @@ park_detached_threads(struct _stoptheworld_state *stw)
_Py_FOR_EACH_STW_INTERP(stw, i) {
_Py_FOR_EACH_TSTATE_UNLOCKED(i, t) {
int state = _Py_atomic_load_int_relaxed(&t->state);
// DETACHED_WAITING threads remain counted until they attach and
// stop, so repeated pauses cannot prevent them from attaching.
if (state == _Py_THREAD_DETACHED) {
// Atomically transition to "suspended" if in "detached" state.
if (_Py_atomic_compare_exchange_int(
Expand Down Expand Up @@ -2530,10 +2576,7 @@ start_the_world(struct _stoptheworld_state *stw)
_Py_FOR_EACH_STW_INTERP(stw, i) {
_Py_FOR_EACH_TSTATE_UNLOCKED(i, t) {
if (t != stw->requester) {
assert(_Py_atomic_load_int_relaxed(&t->state) ==
_Py_THREAD_SUSPENDED);
_Py_atomic_store_int(&t->state, _Py_THREAD_DETACHED);
_PyParkingLot_UnparkAll(&t->state);
_PyThreadState_ResumeDetached(t);
}
}
}
Expand Down
Loading