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

Include/cpython/pystate.h

@@ -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 */

Lib/test/test_atexit.py

@@ -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):
+ 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):

Lib/test/test_capi/test_misc.py

@@ -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
@@ -1641,6 +1642,36 @@ def subthread():
 
  self.assertEqual(actual, int(interpid))
 
+ @threading_helper.requires_working_threading()
+ def test_pending_call_creates_thread(self):
+ source = """
+ import _testinternalcapi
+ 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)
+ _testinternalcapi.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):
 
@@ -1949,6 +1980,41 @@ def test_module_state_shared_in_global(self):
  subinterp_attr_id = os.read(r, 100)
  self.assertEqual(main_attr_id, subinterp_attr_id)
 
+ @threading_helper.requires_working_threading()
+ @unittest.skipUnless(hasattr(os, "pipe"), "requires os.pipe()")
+ @requires_subinterpreters
+ def test_pending_call_creates_thread_subinterpreter(self):
+ interpreters = import_helper.import_module("concurrent.interpreters")
+ r, w = os.pipe()
+ source = f"""if True:
+ import _testinternalcapi
+ import threading
+ import time
+ import os
+
+
+ def output():
+ time.sleep(1)
+ os.write({w}, b"x")
+
+
+ def callback():
+ threading.Thread(target=output).start()
+
+
+ def create_pending_call():
+ time.sleep(1)
+ _testinternalcapi.simple_pending_call(callback)
+
+
+ threading.Thread(target=create_pending_call).start()
+ """
+ interp = interpreters.create()
+ interp.exec(source)
+ interp.close()
+ data = os.read(r, 1)
+ self.assertEqual(data, b"x")
+
 
 @requires_subinterpreters
 class InterpreterConfigTests(unittest.TestCase):

Lib/threading.py

@@ -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

Misc/NEWS.d/next/Core_and_Builtins/2025-06-26-18-44-34.gh-issue-136003.sln51d.rst

@@ -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`).

Modules/_testinternalcapi.c

@@ -2376,6 +2376,16 @@ emscripten_set_up_async_input_device(PyObject *self, PyObject *Py_UNUSED(ignored
 }
 #endif
 
+static PyObject *
+simple_pending_call(PyObject *self, PyObject *callable)
+{
+ if (_PyEval_AddPendingCall(_PyInterpreterState_GET(), _pending_callback, Py_NewRef(callable), 0) < 0){
+ return NULL;
+ }
+
+ Py_RETURN_NONE;
+}
+
 static PyMethodDef module_functions[] ={
 {"get_configs", get_configs, METH_NOARGS},
 {"get_recursion_depth", get_recursion_depth, METH_NOARGS},
@@ -2481,6 +2491,7 @@ static PyMethodDef module_functions[] ={
 #ifdef __EMSCRIPTEN__
 {"emscripten_set_up_async_input_device", emscripten_set_up_async_input_device, METH_NOARGS},
 #endif
+{"simple_pending_call", simple_pending_call, METH_O},
 {NULL, NULL} /* sentinel */
 };
 

Modules/_threadmodule.c

@@ -429,7 +429,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);
@@ -453,7 +453,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()){
@@ -1916,7 +1917,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);
  }