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 21 commits into
base: main
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
21 commits
Select commit Hold shift + click to select a range
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
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
106 changes: 81 additions & 25 deletions Python/pylifecycle.c
Original file line number Diff line number Diff line change
Expand Up @@ -2003,6 +2003,81 @@ resolve_final_tstate(_PyRuntimeState *runtime)
return main_tstate;
}

static int
interp_has_threads(PyInterpreterState *interp)
{
assert(interp != NULL);
assert(interp->threads.head != NULL);
return interp->threads.head->next != NULL;
}

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

static int
interp_has_atexit_callbacks(PyInterpreterState *interp)
{
assert(interp != NULL);
assert(interp->atexit.callbacks != NULL);
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 +2094,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 +2112,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 +2133,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 +2495,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 +2504,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
Loading