[3.12] gh-109860: Use a New Thread State When Switching Interpreters, When Necessary (gh-110245)

Include/internal/pycore_ceval.h

@@ -26,7 +26,7 @@ extern void _PyEval_FiniState(struct _ceval_state *ceval);
 PyAPI_FUNC(void) _PyEval_SignalReceived(PyInterpreterState *interp);
 PyAPI_FUNC(int) _PyEval_AddPendingCall(
     PyInterpreterState *interp,
-    int (*func)(void *),
+    _Py_pending_call_func func,
     void *arg,
     int mainthreadonly);
 PyAPI_FUNC(void) _PyEval_SignalAsyncExc(PyInterpreterState *interp);

Include/internal/pycore_ceval_state.h

@@ -13,6 +13,8 @@ extern "C" {
 #include "pycore_gil.h"             // struct _gil_runtime_state
 
 
+typedef int (*_Py_pending_call_func)(void *);
+
 struct _pending_calls {
     int busy;
     PyThread_type_lock lock;
@@ -24,7 +26,7 @@ struct _pending_calls {
     int async_exc;
 #define NPENDINGCALLS 32
     struct _pending_call {
-        int (*func)(void *);
+        _Py_pending_call_func func;
         void *arg;
     } calls[NPENDINGCALLS];
     int first;

Include/internal/pycore_interp.h

@@ -68,6 +68,8 @@ struct _is {
         uint64_t next_unique_id;
         /* The linked list of threads, newest first. */
         PyThreadState *head;
+        /* The thread currently executing in the __main__ module, if any. */
+        PyThreadState *main;
         /* Used in Modules/_threadmodule.c. */
         long count;
         /* Support for runtime thread stack size tuning.

Include/internal/pycore_pystate.h

@@ -40,6 +40,12 @@ _Py_IsMainInterpreterFinalizing(PyInterpreterState *interp)
             interp == &interp->runtime->_main_interpreter);
 }
 
+// Export for _xxsubinterpreters module.
+PyAPI_FUNC(int) _PyInterpreterState_SetRunningMain(PyInterpreterState *);
+PyAPI_FUNC(void) _PyInterpreterState_SetNotRunningMain(PyInterpreterState *);
+PyAPI_FUNC(int) _PyInterpreterState_IsRunningMain(PyInterpreterState *);
+PyAPI_FUNC(int) _PyInterpreterState_FailIfRunningMain(PyInterpreterState *);
+
 
 static inline const PyConfig *
 _Py_GetMainConfig(void)
@@ -148,6 +154,8 @@ extern PyStatus _PyInterpreterState_DeleteExceptMain(_PyRuntimeState *runtime);
 extern void _PySignal_AfterFork(void);
 #endif
 
+PyAPI_FUNC(int) _PyCrossInterpreterData_ReleaseAndRawFree(_PyCrossInterpreterData *);
+
 
 PyAPI_FUNC(int) _PyState_AddModule(
     PyThreadState *tstate,

Include/internal/pycore_runtime.h

@@ -105,6 +105,11 @@ typedef struct pyruntimestate {
 
     unsigned long main_thread;
 
+    /* The value to use for sys.path[0] in new subinterpreters.
+       Normally this would be part of the PyConfig struct.  However,
+       we cannot add it there in 3.12 since that's an ABI change. */
+    wchar_t *sys_path_0;
+
     /* ---------- IMPORTANT ---------------------------
      The fields above this line are declared as early as
      possible to facilitate out-of-process observability

Lib/test/test_interpreters.py

@@ -1,5 +1,8 @@
 import contextlib
+import json
 import os
+import os.path
+import sys
 import threading
 from textwrap import dedent
 import unittest
@@ -8,6 +11,7 @@
 from test import support
 from test.support import import_helper
 from test.support import threading_helper
+from test.support import os_helper
 _interpreters = import_helper.import_module('_xxsubinterpreters')
 _channels = import_helper.import_module('_xxinterpchannels')
 from test.support import interpreters
@@ -257,6 +261,16 @@ def test_subinterpreter(self):
             self.assertTrue(interp.is_running())
         self.assertFalse(interp.is_running())
 
+    def test_finished(self):
+        r, w = os.pipe()
+        interp = interpreters.create()
+        interp.run(f"""if True:
+            import os
+            os.write({w}, b'x')
+            """)
+        self.assertFalse(interp.is_running())
+        self.assertEqual(os.read(r, 1), b'x')
+
     def test_from_subinterpreter(self):
         interp = interpreters.create()
         out = _run_output(interp, dedent(f"""
@@ -284,6 +298,31 @@ def test_bad_id(self):
         with self.assertRaises(ValueError):
             interp.is_running()
 
+    def test_with_only_background_threads(self):
+        r_interp, w_interp = os.pipe()
+        r_thread, w_thread = os.pipe()
+
+        DONE = b'D'
+        FINISHED = b'F'
+
+        interp = interpreters.create()
+        interp.run(f"""if True:
+            import os
+            import threading
+
+            def task():
+                v = os.read({r_thread}, 1)
+                assert v == {DONE!r}
+                os.write({w_interp}, {FINISHED!r})
+            t = threading.Thread(target=task)
+            t.start()
+            """)
+        self.assertFalse(interp.is_running())
+
+        os.write(w_thread, DONE)
+        interp.run('t.join()')
+        self.assertEqual(os.read(r_interp, 1), FINISHED)
+
 
 class TestInterpreterClose(TestBase):
 
@@ -385,6 +424,37 @@ def test_still_running(self):
                 interp.close()
             self.assertTrue(interp.is_running())
 
+    def test_subthreads_still_running(self):
+        r_interp, w_interp = os.pipe()
+        r_thread, w_thread = os.pipe()
+
+        FINISHED = b'F'
+
+        interp = interpreters.create()
+        interp.run(f"""if True:
+            import os
+            import threading
+            import time
+
+            done = False
+
+            def notify_fini():
+                global done
+                done = True
+                t.join()
+            threading._register_atexit(notify_fini)
+
+            def task():
+                while not done:
+                    time.sleep(0.1)
+                os.write({w_interp}, {FINISHED!r})
+            t = threading.Thread(target=task)
+            t.start()
+            """)
+        interp.close()
+
+        self.assertEqual(os.read(r_interp, 1), FINISHED)
+
 
 class TestInterpreterRun(TestBase):
 
@@ -461,6 +531,37 @@ def test_bytes_for_script(self):
         with self.assertRaises(TypeError):
             interp.run(b'print("spam")')
 
+    def test_with_background_threads_still_running(self):
+        r_interp, w_interp = os.pipe()
+        r_thread, w_thread = os.pipe()
+
+        RAN = b'R'
+        DONE = b'D'
+        FINISHED = b'F'
+
+        interp = interpreters.create()
+        interp.run(f"""if True:
+            import os
+            import threading
+
+            def task():
+                v = os.read({r_thread}, 1)
+                assert v == {DONE!r}
+                os.write({w_interp}, {FINISHED!r})
+            t = threading.Thread(target=task)
+            t.start()
+            os.write({w_interp}, {RAN!r})
+            """)
+        interp.run(f"""if True:
+            os.write({w_interp}, {RAN!r})
+            """)
+
+        os.write(w_thread, DONE)
+        interp.run('t.join()')
+        self.assertEqual(os.read(r_interp, 1), RAN)
+        self.assertEqual(os.read(r_interp, 1), RAN)
+        self.assertEqual(os.read(r_interp, 1), FINISHED)
+
     # test_xxsubinterpreters covers the remaining Interpreter.run() behavior.
 
 
@@ -488,6 +589,154 @@ def task():
             pass
 
 
+class StartupTests(TestBase):
+
+    # We want to ensure the initial state of subinterpreters
+    # matches expectations.
+
+    _subtest_count = 0
+
+    @contextlib.contextmanager
+    def subTest(self, *args):
+        with super().subTest(*args) as ctx:
+            self._subtest_count += 1
+            try:
+                yield ctx
+            finally:
+                if self._debugged_in_subtest:
+                    if self._subtest_count == 1:
+                        # The first subtest adds a leading newline, so we
+                        # compensate here by not printing a trailing newline.
+                        print('### end subtest debug ###', end='')
+                    else:
+                        print('### end subtest debug ###')
+                self._debugged_in_subtest = False
+
+    def debug(self, msg, *, header=None):
+        if header:
+            self._debug(f'--- {header} ---')
+            if msg:
+                if msg.endswith(os.linesep):
+                    self._debug(msg[:-len(os.linesep)])
+                else:
+                    self._debug(msg)
+                    self._debug('<no newline>')
+            self._debug('------')
+        else:
+            self._debug(msg)
+
+    _debugged = False
+    _debugged_in_subtest = False
+    def _debug(self, msg):
+        if not self._debugged:
+            print()
+            self._debugged = True
+        if self._subtest is not None:
+            if True:
+                if not self._debugged_in_subtest:
+                    self._debugged_in_subtest = True
+                    print('### start subtest debug ###')
+                print(msg)
+        else:
+            print(msg)
+
+    def create_temp_dir(self):
+        import tempfile
+        tmp = tempfile.mkdtemp(prefix='test_interpreters_')
+        tmp = os.path.realpath(tmp)
+        self.addCleanup(os_helper.rmtree, tmp)
+        return tmp
+
+    def write_script(self, *path, text):
+        filename = os.path.join(*path)
+        dirname = os.path.dirname(filename)
+        if dirname:
+            os.makedirs(dirname, exist_ok=True)
+        with open(filename, 'w', encoding='utf-8') as outfile:
+            outfile.write(dedent(text))
+        return filename
+
+    @support.requires_subprocess()
+    def run_python(self, argv, *, cwd=None):
+        # This method is inspired by
+        # EmbeddingTestsMixin.run_embedded_interpreter() in test_embed.py.
+        import shlex
+        import subprocess
+        if isinstance(argv, str):
+            argv = shlex.split(argv)
+        argv = [sys.executable, *argv]
+        try:
+            proc = subprocess.run(
+                argv,
+                cwd=cwd,
+                capture_output=True,
+                text=True,
+            )
+        except Exception as exc:
+            self.debug(f'# cmd: {shlex.join(argv)}')
+            if isinstance(exc, FileNotFoundError) and not exc.filename:
+                if os.path.exists(argv[0]):
+                    exists = 'exists'
+                else:
+                    exists = 'does not exist'
+                self.debug(f'{argv[0]} {exists}')
+            raise  # re-raise
+        assert proc.stderr == '' or proc.returncode != 0, proc.stderr
+        if proc.returncode != 0 and support.verbose:
+            self.debug(f'# python3 {shlex.join(argv[1:])} failed:')
+            self.debug(proc.stdout, header='stdout')
+            self.debug(proc.stderr, header='stderr')
+        self.assertEqual(proc.returncode, 0)
+        self.assertEqual(proc.stderr, '')
+        return proc.stdout
+
+    def test_sys_path_0(self):
+        # The main interpreter's sys.path[0] should be used by subinterpreters.
+        script = '''
+            import sys
+            from test.support import interpreters
+
+            orig = sys.path[0]
+
+            interp = interpreters.create()
+            interp.run(f"""if True:
+                import json
+                import sys
+                print(json.dumps({{
+                    'main': {orig!r},
+                    'sub': sys.path[0],
+                }}, indent=4), flush=True)
+                """)
+            '''
+        # <tmp>/
+        #   pkg/
+        #     __init__.py
+        #     __main__.py
+        #     script.py
+        #   script.py
+        cwd = self.create_temp_dir()
+        self.write_script(cwd, 'pkg', '__init__.py', text='')
+        self.write_script(cwd, 'pkg', '__main__.py', text=script)
+        self.write_script(cwd, 'pkg', 'script.py', text=script)
+        self.write_script(cwd, 'script.py', text=script)
+
+        cases = [
+            ('script.py', cwd),
+            ('-m script', cwd),
+            ('-m pkg', cwd),
+            ('-m pkg.script', cwd),
+            ('-c "import script"', ''),
+        ]
+        for argv, expected in cases:
+            with self.subTest(f'python3 {argv}'):
+                out = self.run_python(argv, cwd=cwd)
+                data = json.loads(out)
+                sp0_main, sp0_sub = data['main'], data['sub']
+                self.assertEqual(sp0_sub, sp0_main)
+                self.assertEqual(sp0_sub, expected)
+        # XXX Also check them all with the -P cmdline flag?
+
+
 class TestIsShareable(TestBase):
 
     def test_default_shareables(self):