Skip to content

gh-136003: Execute pre-finalization callbacks in a loop #136004

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 22 commits into
base: main
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
22 commits
Select commit Hold shift + click to select a range
354e6b5
Execute early finalization handlers in a loop
ZeroIntensity Jun 26, 2025
3090524
Store Py_MAX result in a variable.
ZeroIntensity Jun 26, 2025
43038b8
Add a test in test_atexit.
ZeroIntensity Jun 26, 2025
8d4151c
Deal with it in Py_EndInterpreter() as well.
ZeroIntensity Jun 26, 2025
add4d33
Add a blurb entry.
ZeroIntensity Jun 26, 2025
3edc3a8
Check the return code in the test.
ZeroIntensity Jun 26, 2025
ec57918
Add a test for pending calls.
ZeroIntensity Jun 26, 2025
970153b
Fix tests on Windows.
ZeroIntensity Jun 26, 2025
cfd62b8
Use os.linesep.
ZeroIntensity Jul 3, 2025
859070f
Use an RW lock instead of a counter.
ZeroIntensity Jul 3, 2025
37098a0
Merge branch 'main' of https://github.com/python/cpython into fix-cir…
ZeroIntensity Jul 3, 2025
8a1aa13
Remove more counters.
ZeroIntensity Jul 3, 2025
cbcd552
Remove old artifacts (again).
ZeroIntensity Jul 9, 2025
54613a1
Final time removing artifacts.
ZeroIntensity Jul 9, 2025
9ccdb5f
(I lied)
ZeroIntensity Jul 9, 2025
9cd75b7
Atomically check if there are threads, atexit callbacks, or pending c…
ZeroIntensity Jul 9, 2025
1360059
Remove stray newline change.
ZeroIntensity Jul 9, 2025
475538a
Add a test for atexit with subinterpreters.
ZeroIntensity Jul 10, 2025
a794188
Check for os.pipe() in the test.
ZeroIntensity Jul 10, 2025
2dda7a4
Rely on stop-the-world and the GIL instead of a dedicated RW mutex.
ZeroIntensity Jul 10, 2025
f1460af
Serialize pending calls via the ceval mutex.
ZeroIntensity Jul 10, 2025
1e1301d
Only check for non-daemon threads at finalization.
ZeroIntensity Jul 10, 2025
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
1 change: 1 addition & 0 deletions Include/cpython/pystate.h
Original file line number Diff line number Diff line change
Expand Up @@ -107,6 +107,7 @@ struct _ts {
# define _PyThreadState_WHENCE_THREADING 3
# define _PyThreadState_WHENCE_GILSTATE 4
# define _PyThreadState_WHENCE_EXEC 5
# define _PyThreadState_WHENCE_THREADING_DAEMON 6
#endif

/* Currently holds the GIL. Must be its own field to avoid data races */
Expand Down
56 changes: 56 additions & 0 deletions Lib/test/test_atexit.py
Original file line number Diff line number Diff line change
Expand Up @@ -79,6 +79,62 @@ def thready():
# want them to affect the rest of the tests.
script_helper.assert_python_ok("-c", textwrap.dedent(source))

@threading_helper.requires_working_threading()
def test_thread_created_in_atexit(self):
Comment on lines +82 to +83
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It would probably make sense to extend this with a subtest to check it in a subinterpreter.

source = """if True:
import atexit
import threading
import time


def run():
print(24)
time.sleep(1)
print(42)

@atexit.register
def start_thread():
threading.Thread(target=run).start()
"""
return_code, stdout, stderr = script_helper.assert_python_ok("-c", source)
self.assertEqual(return_code, 0)
self.assertEqual(stdout, f"24{os.linesep}42{os.linesep}".encode("utf-8"))
self.assertEqual(stderr, b"")

@threading_helper.requires_working_threading()
@unittest.skipUnless(hasattr(os, "pipe"), "requires os.pipe()")
def test_thread_created_in_atexit_subinterpreter(self):
try:
from concurrent import interpreters
except ImportError:
self.skipTest("subinterpreters are not available")

read, write = os.pipe()
source = f"""if True:
import atexit
import threading
import time
import os

def run():
os.write({write}, b'spanish')
time.sleep(1)
os.write({write}, b'inquisition')

@atexit.register
def start_thread():
threading.Thread(target=run).start()
"""
interp = interpreters.create()
try:
interp.exec(source)

# Close the interpreter to invoke atexit callbacks
interp.close()
self.assertEqual(os.read(read, 100), b"spanishinquisition")
finally:
os.close(read)
os.close(write)

@support.cpython_only
class SubinterpreterTest(unittest.TestCase):
Expand Down
31 changes: 31 additions & 0 deletions Lib/test/test_capi/test_misc.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
from test import support
from test.support import MISSING_C_DOCSTRINGS
from test.support import import_helper
from test.support import script_helper
from test.support import threading_helper
from test.support import warnings_helper
from test.support import requires_limited_api
Expand Down Expand Up @@ -1641,6 +1642,36 @@ def subthread():

self.assertEqual(actual, int(interpid))

@threading_helper.requires_working_threading()
def test_pending_call_creates_thread(self):
Comment on lines +1645 to +1646
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Likewise, it would probably make sense to extend this with a subtest to check it in a subinterpreter.

Also, it would be good to test adding a pending call to a different interpreter during its finalization.

source = """
import _testcapi
import threading
import time


def output():
print(24)
time.sleep(1)
print(42)


def callback():
threading.Thread(target=output).start()


def create_pending_call():
time.sleep(1)
_testcapi.simple_pending_call(callback)


threading.Thread(target=create_pending_call).start()
"""
return_code, stdout, stderr = script_helper.assert_python_ok('-c', textwrap.dedent(source))
self.assertEqual(return_code, 0)
self.assertEqual(stdout, f"24{os.linesep}42{os.linesep}".encode("utf-8"))
self.assertEqual(stderr, b"")


class SubinterpreterTest(unittest.TestCase):

Expand Down
5 changes: 3 additions & 2 deletions Lib/threading.py
Original file line number Diff line number Diff line change
Expand Up @@ -1557,8 +1557,9 @@ def _shutdown():
# normally - that won't happen until the interpreter is nearly dead. So
# mark it done here.
if _main_thread._os_thread_handle.is_done() and _is_main_interpreter():
# _shutdown() was already called
return
# _shutdown() was already called, but threads might have started
# in the meantime.
return _thread_shutdown()

global _SHUTTING_DOWN
_SHUTTING_DOWN = True
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,3 @@
Fix :class:`threading.Thread` objects becoming incorrectly daemon when
created from an :mod:`atexit` callback or a pending call
(:c:func:`Py_AddPendingCall`).
11 changes: 11 additions & 0 deletions Modules/_testcapimodule.c
Original file line number Diff line number Diff line change
Expand Up @@ -2546,6 +2546,16 @@ toggle_reftrace_printer(PyObject *ob, PyObject *arg)
Py_RETURN_NONE;
}

static PyObject *
simple_pending_call(PyObject *self, PyObject *callable)
{
if (Py_AddPendingCall(_pending_callback, Py_NewRef(callable)) < 0) {
return NULL;
}

Py_RETURN_NONE;
}

static PyMethodDef TestMethods[] = {
{"set_errno", set_errno, METH_VARARGS},
{"test_config", test_config, METH_NOARGS},
Expand Down Expand Up @@ -2640,6 +2650,7 @@ static PyMethodDef TestMethods[] = {
{"test_atexit", test_atexit, METH_NOARGS},
{"code_offset_to_line", _PyCFunction_CAST(code_offset_to_line), METH_FASTCALL},
{"toggle_reftrace_printer", toggle_reftrace_printer, METH_O},
{"simple_pending_call", simple_pending_call, METH_O},
{NULL, NULL} /* sentinel */
};

Expand Down
7 changes: 4 additions & 3 deletions Modules/_threadmodule.c
Original file line number Diff line number Diff line change
Expand Up @@ -415,7 +415,7 @@ force_done(void *arg)

static int
ThreadHandle_start(ThreadHandle *self, PyObject *func, PyObject *args,
PyObject *kwargs)
PyObject *kwargs, int daemon)
{
// Mark the handle as starting to prevent any other threads from doing so
PyMutex_Lock(&self->mutex);
Expand All @@ -439,7 +439,8 @@ ThreadHandle_start(ThreadHandle *self, PyObject *func, PyObject *args,
goto start_failed;
}
PyInterpreterState *interp = _PyInterpreterState_GET();
boot->tstate = _PyThreadState_New(interp, _PyThreadState_WHENCE_THREADING);
uint8_t whence = daemon ? _PyThreadState_WHENCE_THREADING_DAEMON : _PyThreadState_WHENCE_THREADING;
boot->tstate = _PyThreadState_New(interp, whence);
if (boot->tstate == NULL) {
PyMem_RawFree(boot);
if (!PyErr_Occurred()) {
Expand Down Expand Up @@ -1874,7 +1875,7 @@ do_start_new_thread(thread_module_state *state, PyObject *func, PyObject *args,
add_to_shutdown_handles(state, handle);
}

if (ThreadHandle_start(handle, func, args, kwargs) < 0) {
if (ThreadHandle_start(handle, func, args, kwargs, daemon) < 0) {
if (!daemon) {
remove_from_shutdown_handles(handle);
}
Expand Down
124 changes: 99 additions & 25 deletions Python/pylifecycle.c
Original file line number Diff line number Diff line change
Expand Up @@ -2003,6 +2003,99 @@ resolve_final_tstate(_PyRuntimeState *runtime)
return main_tstate;
}

static int
interp_has_threads(PyInterpreterState *interp)
{
/* This needs to check for non-daemon threads only, otherwise we get stuck
* in an infinite loop. */
assert(interp != NULL);
assert(interp->runtime->stoptheworld.world_stopped);
assert(interp->threads.head != NULL);
if (interp->threads.head->next == NULL) {
// No other threads active, easy way out.
return 0;
}

// We don't have to worry about locking this because the
// world is stopped.
_Py_FOR_EACH_TSTATE_UNLOCKED(interp, tstate) {
if (tstate->_whence == _PyThreadState_WHENCE_THREADING) {
return 1;
}
}

return 0;
}

static int
interp_has_pending_calls(PyInterpreterState *interp)
{
assert(interp != NULL);
assert(interp->runtime->stoptheworld.world_stopped);
return interp->ceval.pending.npending != 0;
}

static int
interp_has_atexit_callbacks(PyInterpreterState *interp)
{
assert(interp != NULL);
assert(interp->atexit.callbacks != NULL);
assert(interp->runtime->stoptheworld.world_stopped);
assert(PyList_CheckExact(interp->atexit.callbacks));
return PyList_GET_SIZE(interp->atexit.callbacks) != 0;
}

static void
make_pre_finalization_calls(PyThreadState *tstate)
{
assert(tstate != NULL);
PyInterpreterState *interp = tstate->interp;
/* Each of these functions can start one another, e.g. a pending call
* could start a thread or vice versa. To ensure that we properly clean
* call everything, we run these in a loop until none of them run anything. */
for (;;) {
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Would it make sense to add an arbitrary limit to detect infinite loop? For example, log an error after 16 attemps.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Well, it would prevent deadlocks in rare cases, but it would cause crashes in other equally rare cases. Maybe it would be better to emit a fatal error when there are too many iterations?

assert(!interp->runtime->stoptheworld.world_stopped);

// Wrap up existing "threading"-module-created, non-daemon threads.
wait_for_thread_shutdown(tstate);

// Make any remaining pending calls.
_Py_FinishPendingCalls(tstate);

/* The interpreter is still entirely intact at this point, and the
* exit funcs may be relying on that. In particular, if some thread
* or exit func is still waiting to do an import, the import machinery
* expects Py_IsInitialized() to return true. So don't say the
* runtime is uninitialized until after the exit funcs have run.
* Note that Threading.py uses an exit func to do a join on all the
* threads created thru it, so this also protects pending imports in
* the threads created via Threading.
*/

_PyAtExit_Call(tstate->interp);

/* Stop the world to prevent other threads from creating threads or
* atexit callbacks. On the default build, this is simply locked by
* the GIL. For pending calls, we acquire the dedicated mutex, because
* Py_AddPendingCall() can be called without an attached thread state.
*/

// XXX Why does _PyThreadState_DeleteList() rely on all interpreters
// being stopped?
PyMutex_Lock(&interp->ceval.pending.mutex);
_PyEval_StopTheWorldAll(interp->runtime);
int should_continue = (interp_has_threads(interp)
|| interp_has_atexit_callbacks(interp)
|| interp_has_pending_calls(interp));
if (!should_continue) {
break;
}
_PyEval_StartTheWorldAll(interp->runtime);
PyMutex_Unlock(&interp->ceval.pending.mutex);
}
assert(interp->runtime->stoptheworld.world_stopped);
}

static int
_Py_Finalize(_PyRuntimeState *runtime)
{
Expand All @@ -2019,23 +2112,7 @@ _Py_Finalize(_PyRuntimeState *runtime)
// Block some operations.
tstate->interp->finalizing = 1;

// Wrap up existing "threading"-module-created, non-daemon threads.
wait_for_thread_shutdown(tstate);

// Make any remaining pending calls.
_Py_FinishPendingCalls(tstate);

/* The interpreter is still entirely intact at this point, and the
* exit funcs may be relying on that. In particular, if some thread
* or exit func is still waiting to do an import, the import machinery
* expects Py_IsInitialized() to return true. So don't say the
* runtime is uninitialized until after the exit funcs have run.
* Note that Threading.py uses an exit func to do a join on all the
* threads created thru it, so this also protects pending imports in
* the threads created via Threading.
*/

_PyAtExit_Call(tstate->interp);
make_pre_finalization_calls(tstate);

assert(_PyThreadState_GET() == tstate);

Expand All @@ -2053,7 +2130,7 @@ _Py_Finalize(_PyRuntimeState *runtime)
#endif

/* Ensure that remaining threads are detached */
_PyEval_StopTheWorldAll(runtime);
assert(tstate->interp->runtime->stoptheworld.world_stopped);

/* Remaining daemon threads will be trapped in PyThread_hang_thread
when they attempt to take the GIL (ex: PyEval_RestoreThread()). */
Expand All @@ -2074,6 +2151,7 @@ _Py_Finalize(_PyRuntimeState *runtime)
_PyThreadState_SetShuttingDown(p);
}
_PyEval_StartTheWorldAll(runtime);
PyMutex_Unlock(&tstate->interp->ceval.pending.mutex);

/* Clear frames of other threads to call objects destructors. Destructors
will be called in the current Python thread. Since
Expand Down Expand Up @@ -2435,13 +2513,7 @@ Py_EndInterpreter(PyThreadState *tstate)
}
interp->finalizing = 1;

// Wrap up existing "threading"-module-created, non-daemon threads.
wait_for_thread_shutdown(tstate);

// Make any remaining pending calls.
_Py_FinishPendingCalls(tstate);

_PyAtExit_Call(tstate->interp);
make_pre_finalization_calls(tstate);

if (tstate != interp->threads.head || tstate->next != NULL) {
Py_FatalError("not the last thread");
Expand All @@ -2450,6 +2522,8 @@ Py_EndInterpreter(PyThreadState *tstate)
/* Remaining daemon threads will automatically exit
when they attempt to take the GIL (ex: PyEval_RestoreThread()). */
_PyInterpreterState_SetFinalizing(interp, tstate);
_PyEval_StartTheWorldAll(interp->runtime);
PyMutex_Unlock(&interp->ceval.pending.mutex);

// XXX Call something like _PyImport_Disable() here?

Expand Down
2 changes: 1 addition & 1 deletion Python/pystate.c
Original file line number Diff line number Diff line change
Expand Up @@ -1447,7 +1447,7 @@ init_threadstate(_PyThreadStateImpl *_tstate,
assert(tstate->prev == NULL);

assert(tstate->_whence == _PyThreadState_WHENCE_NOTSET);
assert(whence >= 0 && whence <= _PyThreadState_WHENCE_EXEC);
assert(whence >= 0 && whence <= _PyThreadState_WHENCE_THREADING_DAEMON);
tstate->_whence = whence;

assert(id > 0);
Expand Down
Loading