PYTHON-5418 Support subinterpreters and InterpreterPoolExecutor

doc/changelog.rst

@@ -7,6 +7,14 @@ Changes in Version 4.19.0 (2026/XX/XX)
 PyMongo 4.19 brings a number of changes including:
 
 - Added support for Python 3.15.
+- Added support for running the synchronous and asynchronous clients in
+  subinterpreters, including inside
+  ``concurrent.futures.InterpreterPoolExecutor`` (Python 3.14+). In
+  interpreters that do not allow daemon threads, the synchronous client's
+  monitor threads now start as non-daemon threads and are stopped and joined
+  when the interpreter is torn down. Note that because these threads are
+  non-daemon, an interpreter may block on teardown until any in-flight
+  monitor work completes.
 
 Bug fixes
 .........

pymongo/asynchronous/monitor.py

@@ -18,6 +18,7 @@
 
 import asyncio
 import atexit
+import threading
 import time
 import weakref
 from typing import TYPE_CHECKING, Any, Optional
@@ -492,3 +493,11 @@ def _shutdown_resources() -> None:
 
 if _IS_SYNC:
     atexit.register(_shutdown_resources)
+    # In subinterpreters, the executors' threads are non-daemon and are
+    # joined at shutdown, so they must be stopped first. The probe preserves
+    # the normal atexit ordering in the main interpreter.
+    try:
+        threading.Thread().daemon = True
+    except RuntimeError:
+        if hasattr(threading, "_register_atexit"):
+            threading._register_atexit(_shutdown_resources)  # type: ignore[attr-defined]

pymongo/periodic_executor.py

@@ -185,7 +185,13 @@ def open(self) -> None:
 
         if not started:
             thread = threading.Thread(target=self._run, name=self._name)
-            thread.daemon = True
+            try:
+                # Subinterpreters do not allow daemon threads, so fall back
+                # to a non-daemon thread: _shutdown_executors stops and
+                # joins it during interpreter shutdown.
+                thread.daemon = True
+            except RuntimeError:
+                pass
             self._thread = weakref.proxy(thread)
             _register_executor(self)
             # Mitigation to RuntimeError firing when thread starts on shutdown

pymongo/synchronous/monitor.py

@@ -18,6 +18,7 @@
 
 import asyncio
 import atexit
+import threading
 import time
 import weakref
 from typing import TYPE_CHECKING, Any, Optional
@@ -490,3 +491,11 @@ def _shutdown_resources() -> None:
 
 if _IS_SYNC:
     atexit.register(_shutdown_resources)
+    # In subinterpreters, the executors' threads are non-daemon and are
+    # joined at shutdown, so they must be stopped first. The probe preserves
+    # the normal atexit ordering in the main interpreter.
+    try:
+        threading.Thread().daemon = True
+    except RuntimeError:
+        if hasattr(threading, "_register_atexit"):
+            threading._register_atexit(_shutdown_resources)  # type: ignore[attr-defined]

test/asynchronous/test_periodic_executor.py

@@ -17,17 +17,28 @@
 from __future__ import annotations
 
 import asyncio
+import importlib
+import os
 import sys
+import textwrap
 import threading
 import time
+from typing import Any
+
+_interpreters: Any = None
+if sys.version_info >= (3, 14):
+    _interpreters = importlib.import_module("concurrent.interpreters")
 
 sys.path[0:0] = [""]
 
+import pymongo
 from pymongo.periodic_executor import AsyncPeriodicExecutor
 from test.asynchronous import AsyncUnitTest, unittest
 
 _IS_SYNC = False
 
+_PYMONGO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(pymongo.__file__)))
+
 
 class TestAsyncPeriodicExecutor(AsyncUnitTest):
     def _make_executor(self, interval=30.0, min_interval=0.01, target=None, name="test"):
@@ -178,6 +189,50 @@ async def target():
             executor._task.exception()
         self.assertEqual(call_count, 2, "executor should run again after re-open")
 
+    async def test_subinterpreter_shutdown(self):
+        if _interpreters is None:
+            self.skipTest("concurrent.interpreters requires Python 3.14+")
+            return
+
+        root = _PYMONGO_ROOT
+        # The async executor's open() starts a task, which requires a running
+        # event loop; synchro translates the rest of the block for the sync suite.
+        run_stmt = "asyncio.run(main())" if not _IS_SYNC else "main()"
+        # The sync monitor signals its thread through the event; the async
+        # monitor yields to its loop instead. Synchro drops the async yield.
+        wait_stmt = "assert started.wait(10)" if _IS_SYNC else ""
+        code = textwrap.dedent(
+            f"""
+            import asyncio
+            import sys
+            import threading
+            sys.path.insert(0, {root!r})
+            from pymongo.periodic_executor import AsyncPeriodicExecutor
+
+            started = threading.Event()
+
+            async def target():
+                started.set()
+                return True
+
+            async def main():
+                executor = AsyncPeriodicExecutor(
+                    interval=30.0, min_interval=0.05, target=target, name="subinterp"
+                )
+                executor.open()
+                {wait_stmt}
+                await asyncio.sleep(0)
+
+            {run_stmt}
+            """
+        )
+        interp = _interpreters.create()
+        try:
+            interp.exec(code)
+        finally:
+            # Destroying the subinterpreter must not crash or hang.
+            interp.close()
+
 
 if __name__ == "__main__":
     unittest.main()

test/asynchronous/test_subinterpreters.py

@@ -0,0 +1,210 @@
+# Copyright 2026-present MongoDB, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Test running pymongo in subinterpreters."""
+
+from __future__ import annotations
+
+import sys
+import textwrap
+import threading
+import uuid
+
+sys.path[0:0] = [""]
+
+try:
+    from concurrent import interpreters
+except ImportError:  # pragma: no cover - Python < 3.14
+    interpreters = None  # type: ignore[assignment]
+
+try:
+    from concurrent.futures import InterpreterPoolExecutor
+except ImportError:  # pragma: no cover - Python < 3.14
+    InterpreterPoolExecutor = None  # type: ignore[assignment,misc]
+
+from test.asynchronous import AsyncIntegrationTest, async_client_context, unittest
+
+_IS_SYNC = False
+
+
+class TestSubinterpreters(AsyncIntegrationTest):
+    @staticmethod
+    def _get_n(queue, n, errors):
+        try:
+            return sorted(queue.get(timeout=30) for _ in range(n))
+        except interpreters.QueueEmpty:
+            raise AssertionError(f"subinterpreters failed to run: {errors!r}") from None
+
+    @unittest.skipUnless(
+        sys.version_info >= (3, 14), "concurrent.interpreters requires Python 3.14+"
+    )
+    async def test_subinterpreters(self):
+        if interpreters is None:
+            self.skipTest("concurrent.interpreters is not available")
+
+        # Run live clients in multiple subinterpreters at once, mirroring the
+        # mod_wsgi test, which mounts the same app in two interpreters.
+        n_interpreters = 2
+        coll_name = f"subinterp-{uuid.uuid4().hex}"
+        self.addCleanup(self.db.drop_collection, coll_name)
+
+        ready = interpreters.create_queue()
+        release = interpreters.create_queue()
+        done = interpreters.create_queue()
+        # The async client's constructor starts a task, which requires a
+        # running event loop; synchro translates the rest of the block.
+        run_stmt = "asyncio.run(main())" if not _IS_SYNC else "main()"
+        code = textwrap.dedent(
+            f"""
+            import asyncio
+            import sys
+            sys.path[:0] = path
+
+            from pymongo import AsyncMongoClient
+
+            async def main():
+                client = AsyncMongoClient(uri, serverSelectionTimeoutMS=30000)
+                collection = client.get_database(db_name).get_collection(coll_name)
+                await collection.insert_one({{"subinterp": i}})
+                assert await collection.find_one({{"subinterp": i}}) is not None
+                ready.put(i)
+                # Hold the client open until every interpreter has connected.
+                release.get(timeout=60)
+                assert await collection.find_one({{"subinterp": i}}) is not None
+                done.put(i)
+
+            {run_stmt}
+            """
+        )
+
+        errors: list[BaseException] = []
+
+        def run(interp):
+            try:
+                interp.exec(code)
+            except BaseException as exc:
+                errors.append(exc)
+
+        uri = await async_client_context.uri
+        interps = []
+        threads = []
+        try:
+            for i in range(n_interpreters):
+                interp = interpreters.create()
+                interp.prepare_main(
+                    uri=uri,
+                    db_name=self.db.name,
+                    coll_name=coll_name,
+                    i=i,
+                    path=tuple(sys.path),
+                    ready=ready,
+                    release=release,
+                    done=done,
+                )
+                interps.append(interp)
+                thread = threading.Thread(target=run, args=(interp,), name=f"subinterp-{i}")
+                threads.append(thread)
+                thread.start()
+
+            started = self._get_n(ready, n_interpreters, errors)
+            self.assertEqual(started, list(range(n_interpreters)))
+            for _ in range(n_interpreters):
+                release.put(True)
+
+            for thread in threads:
+                thread.join(60)
+                self.assertFalse(thread.is_alive(), f"{thread.name} did not exit")
+
+            finished = self._get_n(done, n_interpreters, errors)
+            self.assertEqual(finished, list(range(n_interpreters)))
+            if errors:
+                self.fail(f"subinterpreter errors: {errors!r}")
+        finally:
+            # Unblock any interpreter still waiting, then destroy them all.
+            for _ in range(n_interpreters):
+                release.put(True)
+            for interp in interps:
+                # Closing an idle interpreter runs threading._shutdown, which
+                # stops and joins pymongo's monitor threads; skip running ones
+                # to avoid masking errors.
+                if not interp.is_running():
+                    interp.close()
+
+        docs = await self.db[coll_name].find({}, {"subinterp": 1}).to_list()
+        found = sorted(doc["subinterp"] for doc in docs)
+        self.assertEqual(found, list(range(n_interpreters)))
+
+    @unittest.skipUnless(
+        sys.version_info >= (3, 14), "InterpreterPoolExecutor requires Python 3.14+"
+    )
+    async def test_interpreter_pool_executor(self):
+        if InterpreterPoolExecutor is None:
+            self.skipTest("InterpreterPoolExecutor is not available")
+
+        # The interpreters disallow daemon threads, so the sync client starts
+        # non-daemon monitor threads; the async client runs tasks on the
+        # interpreter's own event loop.
+        n_interpreters = 2
+        coll_name = f"interp-pool-{uuid.uuid4().hex}"
+        self.addCleanup(self.db.drop_collection, coll_name)
+
+        # The async client's constructor starts a task, which requires a
+        # running event loop; synchro translates the rest of the block.
+        run_stmt = "asyncio.run(main())" if not _IS_SYNC else "main()"
+
+        # The worker's sys.path lacks the repo root, so submit the builtin exec
+        # and fix sys.path in the code.
+        code = textwrap.dedent(
+            f"""
+            import asyncio
+            import sys
+            sys.path[:0] = path
+
+            from pymongo import AsyncMongoClient
+
+            async def main():
+                client = AsyncMongoClient(uri, serverSelectionTimeoutMS=30000)
+                collection = client.get_database(db_name).get_collection(coll_name)
+                await collection.insert_one({{"interp-pool": i}})
+                assert await collection.find_one({{"interp-pool": i}}) is not None
+
+            {run_stmt}
+            """
+        )
+        with InterpreterPoolExecutor(max_workers=n_interpreters) as executor:
+            uri = await async_client_context.uri
+            futures = [
+                executor.submit(
+                    exec,
+                    code,
+                    {
+                        "i": i,
+                        "path": tuple(sys.path),
+                        "uri": uri,
+                        "db_name": self.db.name,
+                        "coll_name": coll_name,
+                    },
+                )
+                for i in range(n_interpreters)
+            ]
+            for future in futures:
+                future.result(timeout=120)
+
+        docs = await self.db[coll_name].find({}, {"interp-pool": 1}).to_list()
+        found = sorted(doc["interp-pool"] for doc in docs)
+        self.assertEqual(found, list(range(n_interpreters)))
+
+
+if __name__ == "__main__":
+    unittest.main()