PYTHON-5418 Support subinterpreters and InterpreterPoolExecutor

doc/changelog.rst

@@ -7,6 +7,15 @@ 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+). Only in
+  subinterpreters, which 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. Monitor threads in the main
+  interpreter remain daemon threads and shutdown behavior is unchanged.
+  Note that because these threads are non-daemon, a subinterpreter 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,8 @@ 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. Checking daemon
+    # support preserves the normal atexit ordering in the main interpreter.
+    if not periodic_executor._daemon_threads_allowed() and hasattr(threading, "_register_atexit"):
+        threading._register_atexit(_shutdown_resources)  # type: ignore[attr-defined]

pymongo/periodic_executor.py

@@ -121,6 +121,23 @@ async def _run(self) -> None:
             self._event = False
 
 
+def _daemon_threads_allowed() -> bool:
+    """Whether daemon threads are allowed in the current interpreter.
+
+    Subinterpreters do not allow daemon threads, so a PeriodicExecutor
+    there runs on a non-daemon thread instead: _shutdown_executors stops
+    and joins it during interpreter shutdown.
+    """
+    # Python 3.12+ exposes the same check CPython itself uses to enforce
+    # the restriction (in the Thread daemon setter), which is reliable in
+    # the current interpreter. On older versions daemon threads are
+    # always allowed.
+    check = getattr(threading, "_daemon_threads_allowed", None)
+    if check is not None:
+        return check()
+    return True
+
+
 class PeriodicExecutor:
     def __init__(
         self,
@@ -185,7 +202,15 @@ 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 = _daemon_threads_allowed()
+            except RuntimeError:
+                # The restriction is enforced when the thread is started,
+                # handled below.
+                pass
             self._thread = weakref.proxy(thread)
             _register_executor(self)
             # Mitigation to RuntimeError firing when thread starts on shutdown
@@ -196,7 +221,13 @@ def open(self) -> None:
                 if "interpreter shutdown" in str(e) or sys.is_finalizing():
                     self._thread = None
                     return
-                raise
+                if not thread.daemon:
+                    raise
+                # Some Python versions only enforce the subinterpreter daemon
+                # restriction when the thread is started, so fall back to a
+                # non-daemon thread as above.
+                thread.daemon = False
+                thread.start()
 
     def close(self, dummy: Any = None) -> None:
         """Stop. To restart, call open().

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,8 @@ 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. Checking daemon
+    # support preserves the normal atexit ordering in the main interpreter.
+    if not periodic_executor._daemon_threads_allowed() and hasattr(threading, "_register_atexit"):
+        threading._register_atexit(_shutdown_resources)  # type: ignore[attr-defined]

pyproject.toml

@@ -278,7 +278,9 @@ exclude_lines = [
     "return NotImplemented",
     "_use_c = true",
     "if __name__ == '__main__':",
-    "if TYPE_CHECKING:"
+    "if TYPE_CHECKING:",
+    # Sync-only registration glue, generated by unasync into the sync twin.
+    "^if _IS_SYNC"
     ]
 partial_branches = ["if (.*and +)*not _use_c( and.*)*:"]
 

test/asynchronous/test_periodic_executor.py

@@ -17,17 +17,29 @@
 from __future__ import annotations
 
 import asyncio
+import importlib
+import os
 import sys
+import textwrap
 import threading
 import time
+from typing import Any
+from unittest import mock
+
+_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 +190,76 @@ async def target():
             executor._task.exception()
         self.assertEqual(call_count, 2, "executor should run again after re-open")
 
+    async def test_open_non_daemon_thread(self):
+        # Subinterpreters disallow daemon threads; PeriodicExecutor.open()
+        # must fall back to starting the monitor thread as a non-daemon thread.
+        from pymongo.periodic_executor import PeriodicExecutor
+
+        def set_daemon(self, value):
+            raise RuntimeError("daemon threads are disallowed in subinterpreters")
+
+        daemon = threading.Thread.daemon
+        executor = PeriodicExecutor(
+            interval=30.0, min_interval=0.01, target=lambda: True, name="non-daemon"
+        )
+        self.addCleanup(executor.join, 2)
+        self.addCleanup(executor.close)
+        with mock.patch.object(threading.Thread, "daemon", property(daemon.fget, set_daemon)):
+            executor.open()
+            thread = executor._thread
+            assert thread is not None
+            self.assertFalse(thread.daemon, "thread must not be a daemon thread")
+
+    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.
+        # Windows proactor loops cannot start in subinterpreters (set_wakeup_fd
+        # is main-interpreter only), so use a selector loop there.
+        run_stmt = (
+            "main()"
+            if _IS_SYNC
+            else 'asyncio.run(main(), loop_factory=asyncio.SelectorEventLoop if sys.platform == "win32" else None)'
+        )
+        # 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()