gh-128639: Don't assume one thread in subinterpreter finalization with fixed daemon thread support

Lib/test/test_interpreters/test_api.py

@@ -11,6 +11,7 @@
 from test.support import os_helper
 from test.support import script_helper
 from test.support import import_helper
+from test.support.script_helper import assert_python_ok
 # Raise SkipTest if subinterpreters not supported.
 _interpreters = import_helper.import_module('_interpreters')
 from concurrent import interpreters
@@ -707,6 +708,68 @@ def test_created_with_capi(self):
  self.interp_exists(interpid))
 
 
+ def test_remaining_threads(self):
+ r_interp, w_interp = self.pipe()
+
+ FINISHED = b'F'
+
+ # It's unlikely, but technically speaking, it's possible
+ # that the thread could've finished before interp.close() is
+ # reached, so this test might not properly exercise the case.
+ # However, it's quite unlikely and probably not worth bothering about.
+ interp = interpreters.create()
+ interp.exec(f"""if True:
+ import os
+ import threading
+ import time
+
+ def task():
+ time.sleep(1)
+ os.write({w_interp},{FINISHED!r})
+
+ threads = (threading.Thread(target=task) for _ in range(3))
+ for t in threads:
+ t.start()
+ """)
+ interp.close()
+
+ self.assertEqual(os.read(r_interp, 1), FINISHED)
+
+ def test_remaining_daemon_threads(self):
+ # Daemon threads leak reference by nature, because they hang threads
+ # without allowing them to do cleanup (i.e., release refs).
+ # To prevent that from messing up the refleak hunter and whatnot, we
+ # run this in a subprocess.
+ code = '''if True:
+ import _interpreters
+ import types
+ interp = _interpreters.create(
+ types.SimpleNamespace(
+ use_main_obmalloc=False,
+ allow_fork=False,
+ allow_exec=False,
+ allow_threads=True,
+ allow_daemon_threads=True,
+ check_multi_interp_extensions=True,
+ gil='own',
+ )
+ )
+ _interpreters.exec(interp, f"""if True:
+ import threading
+ import time
+
+ def task():
+ time.sleep(3)
+
+ threads = (threading.Thread(target=task, daemon=True) for _ in range(3))
+ for t in threads:
+ t.start()
+ """)
+ _interpreters.destroy(interp)
+ '''
+ assert_python_ok('-c', code)
+
+
 class TestInterpreterPrepareMain(TestBase):
 
  def test_empty(self):
@@ -815,7 +878,10 @@ def script():
  spam.eggs()
 
  interp = interpreters.create()
- interp.exec(script)
+ try:
+ interp.exec(script)
+ finally:
+ interp.close()
  """)
 
  stdout, stderr = self.assert_python_failure(scriptfile)
@@ -824,7 +890,7 @@ def script():
  # File "{interpreters.__file__}", line 179, in exec
  self.assertEqual(stderr, dedent(f"""\
  Traceback (most recent call last):
- File "{scriptfile}", line 9, in <module>
+ File "{scriptfile}", line 10, in <module>
  interp.exec(script)
  ~~~~~~~~~~~^^^^^^^^
 {interpmod_line.strip()}

Lib/test/test_interpreters/test_lifecycle.py

@@ -132,6 +132,7 @@ def test_sys_path_0(self):
  'sub': sys.path[0],
  }}, indent=4), flush=True)
  """)
+ interp.close()
  '''
  # <tmp>/
  # pkg/
@@ -172,7 +173,10 @@ def test_gh_109793(self):
  argv = [sys.executable, '-c', '''if True:
  from concurrent import interpreters
  interp = interpreters.create()
- raise Exception
+ try:
+ raise Exception
+ finally:
+ interp.close()
  ''']
  proc = subprocess.run(argv, capture_output=True, text=True)
  self.assertIn('Traceback', proc.stderr)

Lib/test/test_threading.py

@@ -1794,10 +1794,7 @@ def f():
 
  _testcapi.run_in_subinterp(%r)
  """ % (subinterp_code,)
- with test.support.SuppressCrashReport():
- rc, out, err = assert_python_failure("-c", script)
- self.assertIn("Fatal Python error: Py_EndInterpreter: "
- "not the last thread", err.decode())
+ assert_python_ok("-c", script)
 
  def _check_allowed(self, before_start='', *,
  allowed=True,

Misc/NEWS.d/next/Core_and_Builtins/2025-01-08-12-52-47.gh-issue-128640.9nbh9z.rst

@@ -0,0 +1 @@
+Fix a crash when using threads inside of a subinterpreter.

Programs/_testembed.c

@@ -1388,9 +1388,12 @@ static int test_audit_subinterpreter(void)
  PySys_AddAuditHook(_audit_subinterpreter_hook, NULL);
  _testembed_initialize();
 
- Py_NewInterpreter();
- Py_NewInterpreter();
- Py_NewInterpreter();
+ PyThreadState *tstate = PyThreadState_Get();
+ for (int i = 0; i < 3; ++i)
+{
+ Py_EndInterpreter(Py_NewInterpreter());
+ PyThreadState_Swap(tstate);
+ }
 
  Py_Finalize();
 

Python/pylifecycle.c

@@ -2000,6 +2000,7 @@ resolve_final_tstate(_PyRuntimeState *runtime)
  }
  else{
  /* Fall back to the current tstate. It's better than nothing. */
+ // XXX No it's not
  main_tstate = tstate;
  }
  }
@@ -2045,6 +2046,16 @@ _Py_Finalize(_PyRuntimeState *runtime)
 
  _PyAtExit_Call(tstate->interp);
 
+ /* Clean up any lingering subinterpreters.
+
+ Two preconditions need to be met here:
+
+ - This has to happen before _PyRuntimeState_SetFinalizing is
+ called, or else threads might get prematurely blocked.
+ - The world must not be stopped, as finalizers can run.
+ */
+ finalize_subinterpreters();
+
  assert(_PyThreadState_GET() == tstate);
 
  /* Copy the core config, PyInterpreterState_Delete() free
@@ -2132,9 +2143,6 @@ _Py_Finalize(_PyRuntimeState *runtime)
  _PyImport_FiniExternal(tstate->interp);
  finalize_modules(tstate);
 
- /* Clean up any lingering subinterpreters. */
- finalize_subinterpreters();
-
  /* Print debug stats if any */
  _PyEval_Fini();
 
@@ -2416,9 +2424,8 @@ Py_NewInterpreter(void)
  return tstate;
 }
 
-/* Delete an interpreter and its last thread. This requires that the
- given thread state is current, that the thread has no remaining
- frames, and that it is its interpreter's only remaining thread.
+/* Delete an interpreter. This requires that the given thread state
+ is current, and that the thread has no remaining frames.
  It is a fatal error to violate these constraints.
 
  (Py_FinalizeEx() doesn't have these constraints -- it zaps
@@ -2448,15 +2455,20 @@ Py_EndInterpreter(PyThreadState *tstate)
  _Py_FinishPendingCalls(tstate);
 
  _PyAtExit_Call(tstate->interp);
-
- if (tstate != interp->threads.head || tstate->next != NULL){
- Py_FatalError("not the last thread");
- }
-
+ _PyRuntimeState *runtime = interp->runtime;
+ _PyEval_StopTheWorldAll(runtime);
  /* Remaining daemon threads will automatically exit
  when they attempt to take the GIL (ex: PyEval_RestoreThread()). */
  _PyInterpreterState_SetFinalizing(interp, tstate);
 
+ PyThreadState *list = _PyThreadState_RemoveExcept(tstate);
+ for (PyThreadState *p = list; p != NULL; p = p->next){
+ _PyThreadState_SetShuttingDown(p);
+ }
+
+ _PyEval_StartTheWorldAll(runtime);
+ _PyThreadState_DeleteList(list, /*is_after_fork=*/0);
+
  // XXX Call something like _PyImport_Disable() here?
 
  _PyImport_FiniExternal(tstate->interp);
@@ -2486,6 +2498,8 @@ finalize_subinterpreters(void)
  PyInterpreterState *main_interp = _PyInterpreterState_Main();
  assert(final_tstate->interp == main_interp);
  _PyRuntimeState *runtime = main_interp->runtime;
+ assert(!runtime->stoptheworld.world_stopped);
+ assert(_PyRuntimeState_GetFinalizing(runtime) == NULL);
  struct pyinterpreters *interpreters = &runtime->interpreters;
 
  /* Get the first interpreter in the list. */
@@ -2514,27 +2528,17 @@ finalize_subinterpreters(void)
 
  /* Clean up all remaining subinterpreters. */
  while (interp != NULL){
- assert(!_PyInterpreterState_IsRunningMain(interp));
-
- /* Find the tstate to use for fini. We assume the interpreter
- will have at most one tstate at this point. */
- PyThreadState *tstate = interp->threads.head;
- if (tstate != NULL){
- /* Ideally we would be able to use tstate as-is, and rely
- on it being in a ready state: no exception set, not
- running anything (tstate->current_frame), matching the
- current thread ID (tstate->thread_id). To play it safe,
- we always delete it and use a fresh tstate instead. */
- assert(tstate != final_tstate);
- _PyThreadState_Attach(tstate);
- PyThreadState_Clear(tstate);
- _PyThreadState_Detach(tstate);
- PyThreadState_Delete(tstate);
+ /* Make a tstate for finalization. */
+ PyThreadState *tstate = _PyThreadState_NewBound(interp, _PyThreadState_WHENCE_FINI);
+ if (tstate == NULL){
+ // XXX Some graceful way to always get a thread state?
+ Py_FatalError("thread state allocation failed");
  }
- tstate = _PyThreadState_NewBound(interp, _PyThreadState_WHENCE_FINI);
 
- /* Destroy the subinterpreter. */
+ /* Enter the subinterpreter. */
  _PyThreadState_Attach(tstate);
+
+ /* Destroy the subinterpreter. */
  Py_EndInterpreter(tstate);
  assert(_PyThreadState_GET() == NULL);