From e7180a7c5aa5dc1ae1e351f75ab959eb45966b66 Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Sun, 20 Sep 2026 12:04:44 +0000 Subject: [PATCH 01/21] PYTHON-5418 Stop executor threads on subinterpreter shutdown PeriodicExecutor now falls back to a non-daemon thread when daemon threads are disallowed (subinterpreters). The monitor shutdown handler is also registered with threading._register_atexit so executors are stopped and joined before interpreter teardown. Adds regression tests for PeriodicExecutor in a real subinterpreter and for concurrent MongoClients across subinterpreters (Python 3.14+). --- pymongo/asynchronous/monitor.py | 7 ++ pymongo/periodic_executor.py | 9 +- pymongo/synchronous/monitor.py | 7 ++ test/asynchronous/test_periodic_executor.py | 43 ++++++++ test/test_periodic_executor.py | 43 ++++++++ test/test_threads.py | 104 ++++++++++++++++++++ 6 files changed, 212 insertions(+), 1 deletion(-) diff --git a/pymongo/asynchronous/monitor.py b/pymongo/asynchronous/monitor.py index ba0e3804a6..179f459837 100644 --- a/pymongo/asynchronous/monitor.py +++ b/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,9 @@ def _shutdown_resources() -> None: if _IS_SYNC: atexit.register(_shutdown_resources) + # Also register with threading so executors are stopped before the + # interpreter tries to join their (possibly non-daemon) threads. Unlike + # atexit, this runs for subinterpreters, where daemon threads are + # generally not allowed and CPython is strict about background threads. + if hasattr(threading, "_register_atexit"): + threading._register_atexit(_shutdown_resources) # type: ignore[attr-defined] diff --git a/pymongo/periodic_executor.py b/pymongo/periodic_executor.py index 4b979ca9f9..c6f444480c 100644 --- a/pymongo/periodic_executor.py +++ b/pymongo/periodic_executor.py @@ -185,7 +185,14 @@ def open(self) -> None: if not started: thread = threading.Thread(target=self._run, name=self._name) - thread.daemon = True + try: + # Daemon threads are disabled in subinterpreters unless the + # interpreter was created with allow_daemon_threads=True. + # _shutdown_executors stops and joins the thread during + # interpreter shutdown, so a non-daemon thread is safe. + thread.daemon = True + except RuntimeError: + pass self._thread = weakref.proxy(thread) _register_executor(self) # Mitigation to RuntimeError firing when thread starts on shutdown diff --git a/pymongo/synchronous/monitor.py b/pymongo/synchronous/monitor.py index 9a25757f03..8a445fce5d 100644 --- a/pymongo/synchronous/monitor.py +++ b/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,9 @@ def _shutdown_resources() -> None: if _IS_SYNC: atexit.register(_shutdown_resources) + # Also register with threading so executors are stopped before the + # interpreter tries to join their (possibly non-daemon) threads. Unlike + # atexit, this runs for subinterpreters, where daemon threads are + # generally not allowed and CPython is strict about background threads. + if hasattr(threading, "_register_atexit"): + threading._register_atexit(_shutdown_resources) # type: ignore[attr-defined] diff --git a/test/asynchronous/test_periodic_executor.py b/test/asynchronous/test_periodic_executor.py index 15186c4f88..97a2e1fd13 100644 --- a/test/asynchronous/test_periodic_executor.py +++ b/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,38 @@ 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 not _IS_SYNC: + self.skipTest("subinterpreters are only used with the sync driver") + if _interpreters is None: + self.skipTest("concurrent.interpreters requires Python 3.14+") + return + + root = _PYMONGO_ROOT + code = textwrap.dedent( + f""" + import sys + sys.path.insert(0, {root!r}) + from pymongo.periodic_executor import PeriodicExecutor + + def target(): + return True + + executor = PeriodicExecutor( + interval=30.0, min_interval=0.05, target=target, name="subinterp" + ) + executor.open() + """ + ) + interp = _interpreters.create() + try: + interp.exec(code) + finally: + # Destroying the subinterpreter must stop and join the executor + # thread without crashing or hanging. Regression test for + # PYTHON-6114. + interp.close() + if __name__ == "__main__": unittest.main() diff --git a/test/test_periodic_executor.py b/test/test_periodic_executor.py index e6bbb31a72..622e9ad070 100644 --- a/test/test_periodic_executor.py +++ b/test/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 PeriodicExecutor from test import UnitTest, unittest _IS_SYNC = True +_PYMONGO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(pymongo.__file__))) + class TestPeriodicExecutor(UnitTest): def _make_executor(self, interval=30.0, min_interval=0.01, target=None, name="test"): @@ -178,6 +189,38 @@ def target(): executor._task.exception() self.assertEqual(call_count, 2, "executor should run again after re-open") + def test_subinterpreter_shutdown(self): + if not _IS_SYNC: + self.skipTest("subinterpreters are only used with the sync driver") + if _interpreters is None: + self.skipTest("concurrent.interpreters requires Python 3.14+") + return + + root = _PYMONGO_ROOT + code = textwrap.dedent( + f""" + import sys + sys.path.insert(0, {root!r}) + from pymongo.periodic_executor import PeriodicExecutor + + def target(): + return True + + executor = PeriodicExecutor( + interval=30.0, min_interval=0.05, target=target, name="subinterp" + ) + executor.open() + """ + ) + interp = _interpreters.create() + try: + interp.exec(code) + finally: + # Destroying the subinterpreter must stop and join the executor + # thread without crashing or hanging. Regression test for + # PYTHON-6114. + interp.close() + if __name__ == "__main__": unittest.main() diff --git a/test/test_threads.py b/test/test_threads.py index 1b241751ee..11a1bfe8e6 100644 --- a/test/test_threads.py +++ b/test/test_threads.py @@ -16,7 +16,15 @@ from __future__ import annotations +import sys +import textwrap import threading +import uuid + +try: + from concurrent import interpreters +except ImportError: # pragma: no cover - Python < 3.14 + interpreters = None # type: ignore[assignment] from test import IntegrationTest, client_context, unittest from test.utils import joinall @@ -160,6 +168,102 @@ def test_safe_update(self): error.join() okay.join() + @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+" + ) + def test_subinterpreters(self): + if interpreters is None: + self.skipTest("concurrent.interpreters is not available") + + # Run live MongoClients in more than one subinterpreter at the same + # time. This mirrors the mod_wsgi test, which mounts the same app in + # two interpreters, and covers pymongo shutting down its background + # threads when an interpreter is destroyed (PYTHON-6114). + 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() + code = textwrap.dedent( + """ + import sys + sys.path[:0] = path + + from pymongo import MongoClient + + client = MongoClient(uri, serverSelectionTimeoutMS=30000) + collection = client.get_database(db_name).get_collection(coll_name) + collection.insert_one({"subinterp": i}) + assert collection.find_one({"subinterp": i}) is not None + ready.put(i) + # Hold the client open until every interpreter has connected, so + # that all of the clients are live at the same time. + release.get(timeout=60) + assert collection.find_one({"subinterp": i}) is not None + done.put(i) + """ + ) + + errors: list[BaseException] = [] + + def run(interp): + try: + interp.exec(code) + except BaseException as exc: + errors.append(exc) + + interps = [] + threads = [] + try: + for i in range(n_interpreters): + interp = interpreters.create() + interp.prepare_main( + uri=client_context.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: + interp.close() + + found = sorted(doc["subinterp"] for doc in self.db[coll_name].find({}, {"subinterp": 1})) + self.assertEqual(found, list(range(n_interpreters))) + if __name__ == "__main__": unittest.main() From dff3a0c81057e84972ecf1eedf866eba4709f112 Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Mon, 21 Sep 2026 13:41:15 -0500 Subject: [PATCH 02/21] PYTHON-5418 Add a test using InterpreterPoolExecutor Runs live MongoClients inside interpreters managed by the standard InterpreterPoolExecutor, mirroring the existing multi-threaded checks. The pool's interpreters do not allow daemon threads, so this exercises the non-daemon monitor thread fallback and its shutdown handling. --- test/test_threads.py | 47 ++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 47 insertions(+) diff --git a/test/test_threads.py b/test/test_threads.py index 11a1bfe8e6..edd2dd85f8 100644 --- a/test/test_threads.py +++ b/test/test_threads.py @@ -26,6 +26,11 @@ 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 import IntegrationTest, client_context, unittest from test.utils import joinall @@ -35,6 +40,20 @@ def setUpModule(): pass +def _interpreter_pool_worker(i, uri, db_name, coll_name, path): + import sys + + sys.path[:0] = list(path) + + from pymongo import MongoClient + + client: MongoClient = MongoClient(uri, serverSelectionTimeoutMS=30000) + collection = client.get_database(db_name).get_collection(coll_name) + collection.insert_one({"interp-pool": i}) + assert collection.find_one({"interp-pool": i}) is not None + return i + + class AutoAuthenticateThreads(threading.Thread): def __init__(self, collection, num): threading.Thread.__init__(self) @@ -264,6 +283,34 @@ def run(interp): found = sorted(doc["subinterp"] for doc in self.db[coll_name].find({}, {"subinterp": 1})) self.assertEqual(found, list(range(n_interpreters))) + @unittest.skipUnless( + sys.version_info >= (3, 14), "InterpreterPoolExecutor requires Python 3.14+" + ) + def test_interpreter_pool_executor(self): + if InterpreterPoolExecutor is None: + self.skipTest("InterpreterPoolExecutor is not available") + + # Run live MongoClients inside interpreters managed by the standard + # InterpreterPoolExecutor (PYTHON-5418). The pool's interpreters do + # not allow daemon threads, so pymongo must start non-daemon monitor + # threads and stop them when the interpreter is destroyed. + n_interpreters = 2 + coll_name = f"interp-pool-{uuid.uuid4().hex}" + self.addCleanup(self.db.drop_collection, coll_name) + + args = (client_context.uri, self.db.name, coll_name, tuple(sys.path)) + with InterpreterPoolExecutor(max_workers=n_interpreters) as executor: + futures = [ + executor.submit(_interpreter_pool_worker, i, *args) for i in range(n_interpreters) + ] + for i, future in enumerate(futures): + self.assertEqual(future.result(timeout=120), i) + + found = sorted( + doc["interp-pool"] for doc in self.db[coll_name].find({}, {"interp-pool": 1}) + ) + self.assertEqual(found, list(range(n_interpreters))) + if __name__ == "__main__": unittest.main() From 0f43983bda781cc016606ad0b8b8650050a46da7 Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Mon, 21 Sep 2026 14:41:42 -0500 Subject: [PATCH 03/21] PYTHON-5418 Submit exec to the InterpreterPoolExecutor The worker interpreters may not have the repo root on sys.path (the stdlib test package shadows the repo's), so the pickled worker function failed to unpickle. Submit the builtin exec with a code string that inserts the main interpreter's sys.path before importing pymongo. --- test/test_threads.py | 39 +++++++++++++++++++-------------------- 1 file changed, 19 insertions(+), 20 deletions(-) diff --git a/test/test_threads.py b/test/test_threads.py index edd2dd85f8..fa015856a2 100644 --- a/test/test_threads.py +++ b/test/test_threads.py @@ -40,20 +40,6 @@ def setUpModule(): pass -def _interpreter_pool_worker(i, uri, db_name, coll_name, path): - import sys - - sys.path[:0] = list(path) - - from pymongo import MongoClient - - client: MongoClient = MongoClient(uri, serverSelectionTimeoutMS=30000) - collection = client.get_database(db_name).get_collection(coll_name) - collection.insert_one({"interp-pool": i}) - assert collection.find_one({"interp-pool": i}) is not None - return i - - class AutoAuthenticateThreads(threading.Thread): def __init__(self, collection, num): threading.Thread.__init__(self) @@ -298,13 +284,26 @@ def test_interpreter_pool_executor(self): coll_name = f"interp-pool-{uuid.uuid4().hex}" self.addCleanup(self.db.drop_collection, coll_name) - args = (client_context.uri, self.db.name, coll_name, tuple(sys.path)) + # The worker runs in an interpreter whose sys.path has not picked up + # the repo root, so pass the callable as a builtin (exec) and insert + # the main interpreter's sys.path before importing pymongo. + code = textwrap.dedent( + f""" + import sys + sys.path[:0] = {tuple(sys.path)!r} + + from pymongo import MongoClient + + client = MongoClient({client_context.uri!r}, serverSelectionTimeoutMS=30000) + collection = client.get_database({self.db.name!r}).get_collection({coll_name!r}) + collection.insert_one({{"interp-pool": i}}) + assert collection.find_one({{"interp-pool": i}}) is not None + """ + ) with InterpreterPoolExecutor(max_workers=n_interpreters) as executor: - futures = [ - executor.submit(_interpreter_pool_worker, i, *args) for i in range(n_interpreters) - ] - for i, future in enumerate(futures): - self.assertEqual(future.result(timeout=120), i) + futures = [executor.submit(exec, code, {"i": i}) for i in range(n_interpreters)] + for future in futures: + future.result(timeout=120) found = sorted( doc["interp-pool"] for doc in self.db[coll_name].find({}, {"interp-pool": 1}) From 16e833db118314c9ce31fd1a25064d5817799a75 Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Mon, 21 Sep 2026 16:37:36 -0500 Subject: [PATCH 04/21] PYTHON-5418 Note subinterpreter support in the changelog --- doc/changelog.rst | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/doc/changelog.rst b/doc/changelog.rst index 5fb2fb4731..e9da8c6678 100644 --- a/doc/changelog.rst +++ b/doc/changelog.rst @@ -7,6 +7,13 @@ 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 PyMongo in subinterpreters, including inside + ``concurrent.futures.InterpreterPoolExecutor`` (Python 3.14+). In + interpreters that do not allow daemon threads, 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. Subinterpreter + support is currently only exercised with the synchronous client. Bug fixes ......... From 8a1d9174fee332bf0dd4c5ecaffaea3ddc1af4d0 Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Mon, 21 Sep 2026 16:39:34 -0500 Subject: [PATCH 05/21] PYTHON-5418 Test the async client in subinterpreters Runs live AsyncMongoClients inside interpreters managed by InterpreterPoolExecutor. The async client runs its background tasks on the interpreter's own event loop rather than in threads, covering the other shutdown path. --- test/test_threads.py | 60 ++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 60 insertions(+) diff --git a/test/test_threads.py b/test/test_threads.py index fa015856a2..ff5915d5ae 100644 --- a/test/test_threads.py +++ b/test/test_threads.py @@ -310,6 +310,66 @@ def test_interpreter_pool_executor(self): ) self.assertEqual(found, list(range(n_interpreters))) + @unittest.skipUnless( + sys.version_info >= (3, 14), "InterpreterPoolExecutor requires Python 3.14+" + ) + def test_interpreter_pool_executor_async(self): + if InterpreterPoolExecutor is None: + self.skipTest("InterpreterPoolExecutor is not available") + + # Run live AsyncMongoClients inside interpreters managed by the + # standard InterpreterPoolExecutor. The async client runs its + # background tasks on the interpreter's own event loop instead of in + # threads. + n_interpreters = 2 + coll_name = f"interp-pool-async-{uuid.uuid4().hex}" + self.addCleanup(self.db.drop_collection, coll_name) + + # The worker runs in an interpreter whose sys.path has not picked up + # the repo root, so pass the callable as a builtin (exec) and insert + # the main interpreter's sys.path before importing pymongo. + code = textwrap.dedent( + """ + 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-async": i}) + assert await collection.find_one({"interp-pool-async": i}) is not None + await client.close() + + asyncio.run(main()) + """ + ) + with InterpreterPoolExecutor(max_workers=n_interpreters) as executor: + futures = [ + executor.submit( + exec, + code, + { + "i": i, + "path": tuple(sys.path), + "uri": client_context.uri, + "db_name": self.db.name, + "coll_name": coll_name, + }, + ) + for i in range(n_interpreters) + ] + for future in futures: + future.result(timeout=120) + + found = sorted( + doc["interp-pool-async"] + for doc in self.db[coll_name].find({}, {"interp-pool-async": 1}) + ) + self.assertEqual(found, list(range(n_interpreters))) + if __name__ == "__main__": unittest.main() From 34bb47d2a444ea98ad45c8738bccf3bed100fb4e Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Mon, 21 Sep 2026 16:46:36 -0500 Subject: [PATCH 06/21] PYTHON-5418 Finalize idle interpreters and update the changelog Invert the is_running guard in test_subinterpreters: closing an idle interpreter runs threading._shutdown, which stops and joins pymongo's monitor threads, so the successful path must close them. Note in the changelog that both clients are covered by the subinterpreter tests. --- doc/changelog.rst | 11 ++++++----- test/test_threads.py | 7 ++++++- 2 files changed, 12 insertions(+), 6 deletions(-) diff --git a/doc/changelog.rst b/doc/changelog.rst index e9da8c6678..ddf56b3a9d 100644 --- a/doc/changelog.rst +++ b/doc/changelog.rst @@ -9,11 +9,12 @@ PyMongo 4.19 brings a number of changes including: - Added support for Python 3.15. - Added support for running PyMongo in subinterpreters, including inside ``concurrent.futures.InterpreterPoolExecutor`` (Python 3.14+). In - interpreters that do not allow daemon threads, 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. Subinterpreter - support is currently only exercised with the synchronous client. + 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. Tests cover both the synchronous and asynchronous + clients running in subinterpreters. Bug fixes ......... diff --git a/test/test_threads.py b/test/test_threads.py index ff5915d5ae..9b4585de7e 100644 --- a/test/test_threads.py +++ b/test/test_threads.py @@ -264,7 +264,12 @@ def run(interp): for _ in range(n_interpreters): release.put(True) for interp in interps: - interp.close() + # Finalize the idle interpreters: closing one runs + # threading._shutdown, which stops and joins pymongo's + # monitor threads. Skip any that are still executing to + # avoid masking the original error. + if not interp.is_running(): + interp.close() found = sorted(doc["subinterp"] for doc in self.db[coll_name].find({}, {"subinterp": 1})) self.assertEqual(found, list(range(n_interpreters))) From 74dbcd36167e65e3cfa44edeb8612916309a891d Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Mon, 21 Sep 2026 16:53:18 -0500 Subject: [PATCH 07/21] PYTHON-5418 Limit the early shutdown registration to subinterpreters Only register the monitor shutdown handler with threading._register_atexit when the interpreter disallows daemon threads, preserving normal main interpreter shutdown ordering. --- pymongo/asynchronous/monitor.py | 17 +++++++++++------ pymongo/periodic_executor.py | 5 ++++- pymongo/synchronous/monitor.py | 17 +++++++++++------ 3 files changed, 26 insertions(+), 13 deletions(-) diff --git a/pymongo/asynchronous/monitor.py b/pymongo/asynchronous/monitor.py index 179f459837..406bf11249 100644 --- a/pymongo/asynchronous/monitor.py +++ b/pymongo/asynchronous/monitor.py @@ -493,9 +493,14 @@ def _shutdown_resources() -> None: if _IS_SYNC: atexit.register(_shutdown_resources) - # Also register with threading so executors are stopped before the - # interpreter tries to join their (possibly non-daemon) threads. Unlike - # atexit, this runs for subinterpreters, where daemon threads are - # generally not allowed and CPython is strict about background threads. - if hasattr(threading, "_register_atexit"): - threading._register_atexit(_shutdown_resources) # type: ignore[attr-defined] + # In subinterpreters, daemon threads are not allowed and the executors' + # threads are joined (unlike atexit, threading._register_atexit runs for + # subinterpreters), so the executors must be stopped before the + # interpreter tries to join them. Probe for that restriction: in the + # main interpreter the assignment always succeeds and the normal atexit + # ordering is preserved. + try: + threading.Thread().daemon = True + except RuntimeError: + if hasattr(threading, "_register_atexit"): + threading._register_atexit(_shutdown_resources) # type: ignore[attr-defined] diff --git a/pymongo/periodic_executor.py b/pymongo/periodic_executor.py index c6f444480c..bf99469b81 100644 --- a/pymongo/periodic_executor.py +++ b/pymongo/periodic_executor.py @@ -189,7 +189,10 @@ def open(self) -> None: # Daemon threads are disabled in subinterpreters unless the # interpreter was created with allow_daemon_threads=True. # _shutdown_executors stops and joins the thread during - # interpreter shutdown, so a non-daemon thread is safe. + # interpreter shutdown, so a non-daemon thread is safe. Note + # that if the join times out, interpreter shutdown still + # joins the thread without a timeout, so a stuck target can + # block interpreter teardown. thread.daemon = True except RuntimeError: pass diff --git a/pymongo/synchronous/monitor.py b/pymongo/synchronous/monitor.py index 8a445fce5d..ab86559a1d 100644 --- a/pymongo/synchronous/monitor.py +++ b/pymongo/synchronous/monitor.py @@ -491,9 +491,14 @@ def _shutdown_resources() -> None: if _IS_SYNC: atexit.register(_shutdown_resources) - # Also register with threading so executors are stopped before the - # interpreter tries to join their (possibly non-daemon) threads. Unlike - # atexit, this runs for subinterpreters, where daemon threads are - # generally not allowed and CPython is strict about background threads. - if hasattr(threading, "_register_atexit"): - threading._register_atexit(_shutdown_resources) # type: ignore[attr-defined] + # In subinterpreters, daemon threads are not allowed and the executors' + # threads are joined (unlike atexit, threading._register_atexit runs for + # subinterpreters), so the executors must be stopped before the + # interpreter tries to join them. Probe for that restriction: in the + # main interpreter the assignment always succeeds and the normal atexit + # ordering is preserved. + try: + threading.Thread().daemon = True + except RuntimeError: + if hasattr(threading, "_register_atexit"): + threading._register_atexit(_shutdown_resources) # type: ignore[attr-defined] From e20ed843f48afd1a1a356d3f0b7850e83f56f299 Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Mon, 21 Sep 2026 17:22:40 -0500 Subject: [PATCH 08/21] PYTHON-5418 Address review feedback Test AsyncPeriodicExecutor in a subinterpreter in the async suite, tighten the code comments, and note that both clients are supported in the changelog. --- doc/changelog.rst | 6 +-- pymongo/asynchronous/monitor.py | 9 ++-- pymongo/periodic_executor.py | 10 ++-- test/asynchronous/test_periodic_executor.py | 55 ++++++++++++++------- test/test_periodic_executor.py | 55 ++++++++++++++------- test/test_threads.py | 36 ++++++-------- 6 files changed, 96 insertions(+), 75 deletions(-) diff --git a/doc/changelog.rst b/doc/changelog.rst index ddf56b3a9d..33a7c2acab 100644 --- a/doc/changelog.rst +++ b/doc/changelog.rst @@ -7,14 +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 PyMongo in subinterpreters, including inside +- 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. Tests cover both the synchronous and asynchronous - clients running in subinterpreters. + monitor work completes. Bug fixes ......... diff --git a/pymongo/asynchronous/monitor.py b/pymongo/asynchronous/monitor.py index 406bf11249..eca692bf96 100644 --- a/pymongo/asynchronous/monitor.py +++ b/pymongo/asynchronous/monitor.py @@ -493,12 +493,9 @@ def _shutdown_resources() -> None: if _IS_SYNC: atexit.register(_shutdown_resources) - # In subinterpreters, daemon threads are not allowed and the executors' - # threads are joined (unlike atexit, threading._register_atexit runs for - # subinterpreters), so the executors must be stopped before the - # interpreter tries to join them. Probe for that restriction: in the - # main interpreter the assignment always succeeds and the normal atexit - # ordering is preserved. + # 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: diff --git a/pymongo/periodic_executor.py b/pymongo/periodic_executor.py index bf99469b81..4338f5dd49 100644 --- a/pymongo/periodic_executor.py +++ b/pymongo/periodic_executor.py @@ -186,13 +186,9 @@ def open(self) -> None: if not started: thread = threading.Thread(target=self._run, name=self._name) try: - # Daemon threads are disabled in subinterpreters unless the - # interpreter was created with allow_daemon_threads=True. - # _shutdown_executors stops and joins the thread during - # interpreter shutdown, so a non-daemon thread is safe. Note - # that if the join times out, interpreter shutdown still - # joins the thread without a timeout, so a stuck target can - # block interpreter teardown. + # 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 diff --git a/test/asynchronous/test_periodic_executor.py b/test/asynchronous/test_periodic_executor.py index 97a2e1fd13..c1b39b3f81 100644 --- a/test/asynchronous/test_periodic_executor.py +++ b/test/asynchronous/test_periodic_executor.py @@ -190,35 +190,52 @@ async def target(): self.assertEqual(call_count, 2, "executor should run again after re-open") async def test_subinterpreter_shutdown(self): - if not _IS_SYNC: - self.skipTest("subinterpreters are only used with the sync driver") if _interpreters is None: self.skipTest("concurrent.interpreters requires Python 3.14+") return root = _PYMONGO_ROOT - code = textwrap.dedent( - f""" - import sys - sys.path.insert(0, {root!r}) - from pymongo.periodic_executor import PeriodicExecutor - - def target(): - return True - - executor = PeriodicExecutor( - interval=30.0, min_interval=0.05, target=target, name="subinterp" + if _IS_SYNC: + code = textwrap.dedent( + f""" + import sys + sys.path.insert(0, {root!r}) + from pymongo.periodic_executor import PeriodicExecutor + + def target(): + return True + + executor = PeriodicExecutor( + interval=30.0, min_interval=0.05, target=target, name="subinterp" + ) + executor.open() + """ + ) + else: + code = textwrap.dedent( + f""" + import asyncio + import sys + sys.path.insert(0, {root!r}) + from pymongo.periodic_executor import AsyncPeriodicExecutor + + def target(): + return True + + async def main(): + executor = AsyncPeriodicExecutor( + interval=30.0, min_interval=0.05, target=target, name="subinterp" + ) + executor.open() + + asyncio.run(main()) + """ ) - executor.open() - """ - ) interp = _interpreters.create() try: interp.exec(code) finally: - # Destroying the subinterpreter must stop and join the executor - # thread without crashing or hanging. Regression test for - # PYTHON-6114. + # Destroying the subinterpreter must not crash or hang. interp.close() diff --git a/test/test_periodic_executor.py b/test/test_periodic_executor.py index 622e9ad070..70bfa5d2a5 100644 --- a/test/test_periodic_executor.py +++ b/test/test_periodic_executor.py @@ -190,35 +190,52 @@ def target(): self.assertEqual(call_count, 2, "executor should run again after re-open") def test_subinterpreter_shutdown(self): - if not _IS_SYNC: - self.skipTest("subinterpreters are only used with the sync driver") if _interpreters is None: self.skipTest("concurrent.interpreters requires Python 3.14+") return root = _PYMONGO_ROOT - code = textwrap.dedent( - f""" - import sys - sys.path.insert(0, {root!r}) - from pymongo.periodic_executor import PeriodicExecutor - - def target(): - return True - - executor = PeriodicExecutor( - interval=30.0, min_interval=0.05, target=target, name="subinterp" + if _IS_SYNC: + code = textwrap.dedent( + f""" + import sys + sys.path.insert(0, {root!r}) + from pymongo.periodic_executor import PeriodicExecutor + + def target(): + return True + + executor = PeriodicExecutor( + interval=30.0, min_interval=0.05, target=target, name="subinterp" + ) + executor.open() + """ + ) + else: + code = textwrap.dedent( + f""" + import asyncio + import sys + sys.path.insert(0, {root!r}) + from pymongo.periodic_executor import PeriodicExecutor + + def target(): + return True + + def main(): + executor = PeriodicExecutor( + interval=30.0, min_interval=0.05, target=target, name="subinterp" + ) + executor.open() + + asyncio.run(main()) + """ ) - executor.open() - """ - ) interp = _interpreters.create() try: interp.exec(code) finally: - # Destroying the subinterpreter must stop and join the executor - # thread without crashing or hanging. Regression test for - # PYTHON-6114. + # Destroying the subinterpreter must not crash or hang. interp.close() diff --git a/test/test_threads.py b/test/test_threads.py index 9b4585de7e..b141f5aa11 100644 --- a/test/test_threads.py +++ b/test/test_threads.py @@ -188,9 +188,8 @@ def test_subinterpreters(self): self.skipTest("concurrent.interpreters is not available") # Run live MongoClients in more than one subinterpreter at the same - # time. This mirrors the mod_wsgi test, which mounts the same app in - # two interpreters, and covers pymongo shutting down its background - # threads when an interpreter is destroyed (PYTHON-6114). + # time, 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) @@ -264,10 +263,9 @@ def run(interp): for _ in range(n_interpreters): release.put(True) for interp in interps: - # Finalize the idle interpreters: closing one runs - # threading._shutdown, which stops and joins pymongo's - # monitor threads. Skip any that are still executing to - # avoid masking the original error. + # Closing an idle interpreter runs threading._shutdown, which + # stops and joins pymongo's monitor threads. Skip running + # ones to avoid masking the original error. if not interp.is_running(): interp.close() @@ -281,17 +279,15 @@ def test_interpreter_pool_executor(self): if InterpreterPoolExecutor is None: self.skipTest("InterpreterPoolExecutor is not available") - # Run live MongoClients inside interpreters managed by the standard - # InterpreterPoolExecutor (PYTHON-5418). The pool's interpreters do - # not allow daemon threads, so pymongo must start non-daemon monitor - # threads and stop them when the interpreter is destroyed. + # Run live MongoClients in InterpreterPoolExecutor workers. The + # interpreters disallow daemon threads, so pymongo starts non-daemon + # monitor threads. n_interpreters = 2 coll_name = f"interp-pool-{uuid.uuid4().hex}" self.addCleanup(self.db.drop_collection, coll_name) - # The worker runs in an interpreter whose sys.path has not picked up - # the repo root, so pass the callable as a builtin (exec) and insert - # the main interpreter's sys.path before importing pymongo. + # 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 sys @@ -322,17 +318,15 @@ def test_interpreter_pool_executor_async(self): if InterpreterPoolExecutor is None: self.skipTest("InterpreterPoolExecutor is not available") - # Run live AsyncMongoClients inside interpreters managed by the - # standard InterpreterPoolExecutor. The async client runs its - # background tasks on the interpreter's own event loop instead of in - # threads. + # Run live AsyncMongoClients in InterpreterPoolExecutor workers; the + # async client runs its background tasks on the interpreter's own + # event loop. n_interpreters = 2 coll_name = f"interp-pool-async-{uuid.uuid4().hex}" self.addCleanup(self.db.drop_collection, coll_name) - # The worker runs in an interpreter whose sys.path has not picked up - # the repo root, so pass the callable as a builtin (exec) and insert - # the main interpreter's sys.path before importing pymongo. + # The worker's sys.path lacks the repo root, so submit the builtin + # exec and fix sys.path in the code. code = textwrap.dedent( """ import asyncio From 65dae401cb3b13debb1a9c122d69dccc0d3e709c Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Tue, 22 Sep 2026 05:14:59 -0500 Subject: [PATCH 09/21] PYTHON-5418 Move the subinterpreter client tests to a mirrored file Add test/asynchronous/test_subinterpreters.py with the subinterpreter and InterpreterPoolExecutor client tests, written once in async style and mirrored by synchro. Remove them from test_threads.py, which is not mirrored. Collapse the periodic executor test's duplicated worker blocks into one block and make its async target a coroutine. --- test/asynchronous/test_periodic_executor.py | 48 ++--- test/asynchronous/test_subinterpreters.py | 215 ++++++++++++++++++++ test/test_periodic_executor.py | 46 ++--- test/test_subinterpreters.py | 215 ++++++++++++++++++++ test/test_threads.py | 209 ------------------- 5 files changed, 463 insertions(+), 270 deletions(-) create mode 100644 test/asynchronous/test_subinterpreters.py create mode 100644 test/test_subinterpreters.py diff --git a/test/asynchronous/test_periodic_executor.py b/test/asynchronous/test_periodic_executor.py index c1b39b3f81..f81b736152 100644 --- a/test/asynchronous/test_periodic_executor.py +++ b/test/asynchronous/test_periodic_executor.py @@ -195,42 +195,28 @@ async def test_subinterpreter_shutdown(self): return root = _PYMONGO_ROOT - if _IS_SYNC: - code = textwrap.dedent( - f""" - import sys - sys.path.insert(0, {root!r}) - from pymongo.periodic_executor import PeriodicExecutor + # 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()" + code = textwrap.dedent( + f""" + import asyncio + import sys + sys.path.insert(0, {root!r}) + from pymongo.periodic_executor import AsyncPeriodicExecutor - def target(): - return True + async def target(): + return True - executor = PeriodicExecutor( + async def main(): + executor = AsyncPeriodicExecutor( interval=30.0, min_interval=0.05, target=target, name="subinterp" ) executor.open() - """ - ) - else: - code = textwrap.dedent( - f""" - import asyncio - import sys - sys.path.insert(0, {root!r}) - from pymongo.periodic_executor import AsyncPeriodicExecutor - - def target(): - return True - - async def main(): - executor = AsyncPeriodicExecutor( - interval=30.0, min_interval=0.05, target=target, name="subinterp" - ) - executor.open() - - asyncio.run(main()) - """ - ) + + {run_stmt} + """ + ) interp = _interpreters.create() try: interp.exec(code) diff --git a/test/asynchronous/test_subinterpreters.py b/test/asynchronous/test_subinterpreters.py new file mode 100644 index 0000000000..7c16d789ed --- /dev/null +++ b/test/asynchronous/test_subinterpreters.py @@ -0,0 +1,215 @@ +# 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 more than one subinterpreter at the same time, + # 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 for the + # sync suite. + 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, so + # that all of the clients are live at the same time. + 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 the original error. + 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") + + # Run live clients in InterpreterPoolExecutor workers. The + # interpreters disallow daemon threads, so the sync client starts + # non-daemon monitor threads, while the async client runs its + # background 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 for the + # sync suite. + 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() diff --git a/test/test_periodic_executor.py b/test/test_periodic_executor.py index 70bfa5d2a5..5c4da97b4b 100644 --- a/test/test_periodic_executor.py +++ b/test/test_periodic_executor.py @@ -195,42 +195,28 @@ def test_subinterpreter_shutdown(self): return root = _PYMONGO_ROOT - if _IS_SYNC: - code = textwrap.dedent( - f""" - import sys - sys.path.insert(0, {root!r}) - from pymongo.periodic_executor import PeriodicExecutor + # 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()" + code = textwrap.dedent( + f""" + import asyncio + import sys + sys.path.insert(0, {root!r}) + from pymongo.periodic_executor import PeriodicExecutor - def target(): - return True + def target(): + return True + def main(): executor = PeriodicExecutor( interval=30.0, min_interval=0.05, target=target, name="subinterp" ) executor.open() - """ - ) - else: - code = textwrap.dedent( - f""" - import asyncio - import sys - sys.path.insert(0, {root!r}) - from pymongo.periodic_executor import PeriodicExecutor - - def target(): - return True - - def main(): - executor = PeriodicExecutor( - interval=30.0, min_interval=0.05, target=target, name="subinterp" - ) - executor.open() - - asyncio.run(main()) - """ - ) + + {run_stmt} + """ + ) interp = _interpreters.create() try: interp.exec(code) diff --git a/test/test_subinterpreters.py b/test/test_subinterpreters.py new file mode 100644 index 0000000000..7010e2b2e1 --- /dev/null +++ b/test/test_subinterpreters.py @@ -0,0 +1,215 @@ +# 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 import IntegrationTest, client_context, unittest + +_IS_SYNC = True + + +class TestSubinterpreters(IntegrationTest): + @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+" + ) + def test_subinterpreters(self): + if interpreters is None: + self.skipTest("concurrent.interpreters is not available") + + # Run live clients in more than one subinterpreter at the same time, + # 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 for the + # sync suite. + 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 MongoClient + + def main(): + client = MongoClient(uri, serverSelectionTimeoutMS=30000) + collection = client.get_database(db_name).get_collection(coll_name) + collection.insert_one({{"subinterp": i}}) + assert collection.find_one({{"subinterp": i}}) is not None + ready.put(i) + # Hold the client open until every interpreter has connected, so + # that all of the clients are live at the same time. + release.get(timeout=60) + assert 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 = 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 the original error. + if not interp.is_running(): + interp.close() + + docs = 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+" + ) + def test_interpreter_pool_executor(self): + if InterpreterPoolExecutor is None: + self.skipTest("InterpreterPoolExecutor is not available") + + # Run live clients in InterpreterPoolExecutor workers. The + # interpreters disallow daemon threads, so the sync client starts + # non-daemon monitor threads, while the async client runs its + # background 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 for the + # sync suite. + 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 MongoClient + + def main(): + client = MongoClient(uri, serverSelectionTimeoutMS=30000) + collection = client.get_database(db_name).get_collection(coll_name) + collection.insert_one({{"interp-pool": i}}) + assert collection.find_one({{"interp-pool": i}}) is not None + + {run_stmt} + """ + ) + with InterpreterPoolExecutor(max_workers=n_interpreters) as executor: + uri = 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 = 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() diff --git a/test/test_threads.py b/test/test_threads.py index b141f5aa11..1b241751ee 100644 --- a/test/test_threads.py +++ b/test/test_threads.py @@ -16,20 +16,7 @@ from __future__ import annotations -import sys -import textwrap import threading -import uuid - -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 import IntegrationTest, client_context, unittest from test.utils import joinall @@ -173,202 +160,6 @@ def test_safe_update(self): error.join() okay.join() - @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+" - ) - def test_subinterpreters(self): - if interpreters is None: - self.skipTest("concurrent.interpreters is not available") - - # Run live MongoClients in more than one subinterpreter at the same - # time, 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() - code = textwrap.dedent( - """ - import sys - sys.path[:0] = path - - from pymongo import MongoClient - - client = MongoClient(uri, serverSelectionTimeoutMS=30000) - collection = client.get_database(db_name).get_collection(coll_name) - collection.insert_one({"subinterp": i}) - assert collection.find_one({"subinterp": i}) is not None - ready.put(i) - # Hold the client open until every interpreter has connected, so - # that all of the clients are live at the same time. - release.get(timeout=60) - assert collection.find_one({"subinterp": i}) is not None - done.put(i) - """ - ) - - errors: list[BaseException] = [] - - def run(interp): - try: - interp.exec(code) - except BaseException as exc: - errors.append(exc) - - interps = [] - threads = [] - try: - for i in range(n_interpreters): - interp = interpreters.create() - interp.prepare_main( - uri=client_context.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 the original error. - if not interp.is_running(): - interp.close() - - found = sorted(doc["subinterp"] for doc in self.db[coll_name].find({}, {"subinterp": 1})) - self.assertEqual(found, list(range(n_interpreters))) - - @unittest.skipUnless( - sys.version_info >= (3, 14), "InterpreterPoolExecutor requires Python 3.14+" - ) - def test_interpreter_pool_executor(self): - if InterpreterPoolExecutor is None: - self.skipTest("InterpreterPoolExecutor is not available") - - # Run live MongoClients in InterpreterPoolExecutor workers. The - # interpreters disallow daemon threads, so pymongo starts non-daemon - # monitor threads. - n_interpreters = 2 - coll_name = f"interp-pool-{uuid.uuid4().hex}" - self.addCleanup(self.db.drop_collection, coll_name) - - # 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 sys - sys.path[:0] = {tuple(sys.path)!r} - - from pymongo import MongoClient - - client = MongoClient({client_context.uri!r}, serverSelectionTimeoutMS=30000) - collection = client.get_database({self.db.name!r}).get_collection({coll_name!r}) - collection.insert_one({{"interp-pool": i}}) - assert collection.find_one({{"interp-pool": i}}) is not None - """ - ) - with InterpreterPoolExecutor(max_workers=n_interpreters) as executor: - futures = [executor.submit(exec, code, {"i": i}) for i in range(n_interpreters)] - for future in futures: - future.result(timeout=120) - - found = sorted( - doc["interp-pool"] for doc in self.db[coll_name].find({}, {"interp-pool": 1}) - ) - self.assertEqual(found, list(range(n_interpreters))) - - @unittest.skipUnless( - sys.version_info >= (3, 14), "InterpreterPoolExecutor requires Python 3.14+" - ) - def test_interpreter_pool_executor_async(self): - if InterpreterPoolExecutor is None: - self.skipTest("InterpreterPoolExecutor is not available") - - # Run live AsyncMongoClients in InterpreterPoolExecutor workers; the - # async client runs its background tasks on the interpreter's own - # event loop. - n_interpreters = 2 - coll_name = f"interp-pool-async-{uuid.uuid4().hex}" - self.addCleanup(self.db.drop_collection, coll_name) - - # The worker's sys.path lacks the repo root, so submit the builtin - # exec and fix sys.path in the code. - code = textwrap.dedent( - """ - 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-async": i}) - assert await collection.find_one({"interp-pool-async": i}) is not None - await client.close() - - asyncio.run(main()) - """ - ) - with InterpreterPoolExecutor(max_workers=n_interpreters) as executor: - futures = [ - executor.submit( - exec, - code, - { - "i": i, - "path": tuple(sys.path), - "uri": client_context.uri, - "db_name": self.db.name, - "coll_name": coll_name, - }, - ) - for i in range(n_interpreters) - ] - for future in futures: - future.result(timeout=120) - - found = sorted( - doc["interp-pool-async"] - for doc in self.db[coll_name].find({}, {"interp-pool-async": 1}) - ) - self.assertEqual(found, list(range(n_interpreters))) - if __name__ == "__main__": unittest.main() From 803229e845e3c97a6fc0cbefd99eef92a23ca7e5 Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Tue, 22 Sep 2026 05:16:01 -0500 Subject: [PATCH 10/21] PYTHON-5418 Sync the generated monitor comment with the async source Commit 623ecb07 tightened the comment in the async monitor but did not regenerate the sync mirror. --- pymongo/synchronous/monitor.py | 9 +++------ 1 file changed, 3 insertions(+), 6 deletions(-) diff --git a/pymongo/synchronous/monitor.py b/pymongo/synchronous/monitor.py index ab86559a1d..aea1b9fc12 100644 --- a/pymongo/synchronous/monitor.py +++ b/pymongo/synchronous/monitor.py @@ -491,12 +491,9 @@ def _shutdown_resources() -> None: if _IS_SYNC: atexit.register(_shutdown_resources) - # In subinterpreters, daemon threads are not allowed and the executors' - # threads are joined (unlike atexit, threading._register_atexit runs for - # subinterpreters), so the executors must be stopped before the - # interpreter tries to join them. Probe for that restriction: in the - # main interpreter the assignment always succeeds and the normal atexit - # ordering is preserved. + # 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: From 52b2e2327478be4ed2483dd035170688fe3bd6d9 Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Tue, 22 Sep 2026 05:30:39 -0500 Subject: [PATCH 11/21] PYTHON-5418 Wait for the monitor to start before closing the subinterpreter The sync worker blocks on an event set by the target's first run, so the interpreter is destroyed with the monitor live in its interval loop. The async worker yields to its loop once, which runs the monitor task's first step. --- test/asynchronous/test_periodic_executor.py | 9 +++++++++ test/test_periodic_executor.py | 8 ++++++++ 2 files changed, 17 insertions(+) diff --git a/test/asynchronous/test_periodic_executor.py b/test/asynchronous/test_periodic_executor.py index f81b736152..8d2ecd4af4 100644 --- a/test/asynchronous/test_periodic_executor.py +++ b/test/asynchronous/test_periodic_executor.py @@ -198,14 +198,21 @@ async def test_subinterpreter_shutdown(self): # 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(): @@ -213,6 +220,8 @@ async def main(): interval=30.0, min_interval=0.05, target=target, name="subinterp" ) executor.open() + {wait_stmt} + await asyncio.sleep(0) {run_stmt} """ diff --git a/test/test_periodic_executor.py b/test/test_periodic_executor.py index 5c4da97b4b..66101e89b6 100644 --- a/test/test_periodic_executor.py +++ b/test/test_periodic_executor.py @@ -198,14 +198,21 @@ def test_subinterpreter_shutdown(self): # 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 PeriodicExecutor + started = threading.Event() + def target(): + started.set() return True def main(): @@ -213,6 +220,7 @@ def main(): interval=30.0, min_interval=0.05, target=target, name="subinterp" ) executor.open() + {wait_stmt} {run_stmt} """ From e9a9288aed73777c9352e8ec10d8115b4c53b039 Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Tue, 22 Sep 2026 05:47:33 -0500 Subject: [PATCH 12/21] PYTHON-5418 Trim the subinterpreter test comments --- test/asynchronous/test_subinterpreters.py | 29 ++++++++++------------- test/test_subinterpreters.py | 29 ++++++++++------------- 2 files changed, 24 insertions(+), 34 deletions(-) diff --git a/test/asynchronous/test_subinterpreters.py b/test/asynchronous/test_subinterpreters.py index 7c16d789ed..a31cba1b6a 100644 --- a/test/asynchronous/test_subinterpreters.py +++ b/test/asynchronous/test_subinterpreters.py @@ -53,9 +53,8 @@ async def test_subinterpreters(self): if interpreters is None: self.skipTest("concurrent.interpreters is not available") - # Run live clients in more than one subinterpreter at the same time, - # mirroring the mod_wsgi test, which mounts the same app in two - # interpreters. + # 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) @@ -64,8 +63,7 @@ async def test_subinterpreters(self): 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 for the - # sync suite. + # 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""" @@ -81,8 +79,7 @@ async def main(): 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, so - # that all of the clients are live at the same time. + # 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) @@ -139,8 +136,8 @@ def run(interp): 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 the original error. + # stops and joins pymongo's monitor threads; skip running ones + # to avoid masking errors. if not interp.is_running(): interp.close() @@ -155,21 +152,19 @@ async def test_interpreter_pool_executor(self): if InterpreterPoolExecutor is None: self.skipTest("InterpreterPoolExecutor is not available") - # Run live clients in InterpreterPoolExecutor workers. The - # interpreters disallow daemon threads, so the sync client starts - # non-daemon monitor threads, while the async client runs its - # background tasks on the interpreter's own event loop. + # 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 for the - # sync suite. + # 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. + # 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 diff --git a/test/test_subinterpreters.py b/test/test_subinterpreters.py index 7010e2b2e1..68e93fe575 100644 --- a/test/test_subinterpreters.py +++ b/test/test_subinterpreters.py @@ -53,9 +53,8 @@ def test_subinterpreters(self): if interpreters is None: self.skipTest("concurrent.interpreters is not available") - # Run live clients in more than one subinterpreter at the same time, - # mirroring the mod_wsgi test, which mounts the same app in two - # interpreters. + # 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) @@ -64,8 +63,7 @@ def test_subinterpreters(self): 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 for the - # sync suite. + # 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""" @@ -81,8 +79,7 @@ def main(): collection.insert_one({{"subinterp": i}}) assert collection.find_one({{"subinterp": i}}) is not None ready.put(i) - # Hold the client open until every interpreter has connected, so - # that all of the clients are live at the same time. + # Hold the client open until every interpreter has connected. release.get(timeout=60) assert collection.find_one({{"subinterp": i}}) is not None done.put(i) @@ -139,8 +136,8 @@ def run(interp): 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 the original error. + # stops and joins pymongo's monitor threads; skip running ones + # to avoid masking errors. if not interp.is_running(): interp.close() @@ -155,21 +152,19 @@ def test_interpreter_pool_executor(self): if InterpreterPoolExecutor is None: self.skipTest("InterpreterPoolExecutor is not available") - # Run live clients in InterpreterPoolExecutor workers. The - # interpreters disallow daemon threads, so the sync client starts - # non-daemon monitor threads, while the async client runs its - # background tasks on the interpreter's own event loop. + # 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 for the - # sync suite. + # 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. + # 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 From ecc6bf289c0ae8b061ef6bb77000381fec85e7e4 Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Tue, 22 Sep 2026 06:23:59 -0500 Subject: [PATCH 13/21] PYTHON-5418 Join the subinterpreter workers before closing them A failed assertion left the non-daemon worker threads running, which blocked the test process on exit and leaked the running interpreters. Join the workers in the cleanup path before closing the interpreters. --- test/asynchronous/test_subinterpreters.py | 5 ++++- test/test_subinterpreters.py | 5 ++++- 2 files changed, 8 insertions(+), 2 deletions(-) diff --git a/test/asynchronous/test_subinterpreters.py b/test/asynchronous/test_subinterpreters.py index a31cba1b6a..05aa041d58 100644 --- a/test/asynchronous/test_subinterpreters.py +++ b/test/asynchronous/test_subinterpreters.py @@ -131,9 +131,12 @@ def run(interp): if errors: self.fail(f"subinterpreter errors: {errors!r}") finally: - # Unblock any interpreter still waiting, then destroy them all. + # Unblock any interpreter still waiting, join its worker, then + # destroy them all. for _ in range(n_interpreters): release.put(True) + for thread in threads: + thread.join(60) for interp in interps: # Closing an idle interpreter runs threading._shutdown, which # stops and joins pymongo's monitor threads; skip running ones diff --git a/test/test_subinterpreters.py b/test/test_subinterpreters.py index 68e93fe575..8db24618c2 100644 --- a/test/test_subinterpreters.py +++ b/test/test_subinterpreters.py @@ -131,9 +131,12 @@ def run(interp): if errors: self.fail(f"subinterpreter errors: {errors!r}") finally: - # Unblock any interpreter still waiting, then destroy them all. + # Unblock any interpreter still waiting, join its worker, then + # destroy them all. for _ in range(n_interpreters): release.put(True) + for thread in threads: + thread.join(60) for interp in interps: # Closing an idle interpreter runs threading._shutdown, which # stops and joins pymongo's monitor threads; skip running ones From 4f3005601ec73a2f883573ce7bcf0afcab596aa1 Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Tue, 22 Sep 2026 10:04:50 -0500 Subject: [PATCH 14/21] PYTHON-5418 Wait off-loop in the async subinterpreter tests The queue read in the subinterpreter worker and the pool future waits blocked the event loop, freezing the client tasks under test. The async suite waits through asyncio.to_thread instead, while the sync suite keeps the direct calls. --- test/asynchronous/test_subinterpreters.py | 18 ++++++++++++++++-- test/test_subinterpreters.py | 18 ++++++++++++++++-- 2 files changed, 32 insertions(+), 4 deletions(-) diff --git a/test/asynchronous/test_subinterpreters.py b/test/asynchronous/test_subinterpreters.py index 05aa041d58..a1b98de722 100644 --- a/test/asynchronous/test_subinterpreters.py +++ b/test/asynchronous/test_subinterpreters.py @@ -65,6 +65,13 @@ async def test_subinterpreters(self): # 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 sync monitor reads the queue on its thread; the async monitor + # reads it off-loop so the client's tasks keep running. + release_stmt = ( + "await asyncio.to_thread(release.get, timeout=60)" + if not _IS_SYNC + else "release.get(timeout=60)" + ) code = textwrap.dedent( f""" import asyncio @@ -80,7 +87,7 @@ async def main(): 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) + {release_stmt} assert await collection.find_one({{"subinterp": i}}) is not None done.put(i) @@ -185,6 +192,13 @@ async def main(): {run_stmt} """ ) + # Wait for the workers off-loop in the async suite; synchro keeps the + # direct call for the sync suite. + result_stmt = ( + "await asyncio.to_thread(future.result, 120)" + if not _IS_SYNC + else "future.result(timeout=120)" + ) with InterpreterPoolExecutor(max_workers=n_interpreters) as executor: uri = await async_client_context.uri futures = [ @@ -202,7 +216,7 @@ async def main(): for i in range(n_interpreters) ] for future in futures: - future.result(timeout=120) + {result_stmt} docs = await self.db[coll_name].find({}, {"interp-pool": 1}).to_list() found = sorted(doc["interp-pool"] for doc in docs) diff --git a/test/test_subinterpreters.py b/test/test_subinterpreters.py index 8db24618c2..0b6c9a62aa 100644 --- a/test/test_subinterpreters.py +++ b/test/test_subinterpreters.py @@ -65,6 +65,13 @@ def test_subinterpreters(self): # 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 sync monitor reads the queue on its thread; the async monitor + # reads it off-loop so the client's tasks keep running. + release_stmt = ( + "asyncio.to_thread(release.get, timeout=60)" + if not _IS_SYNC + else "release.get(timeout=60)" + ) code = textwrap.dedent( f""" import asyncio @@ -80,7 +87,7 @@ def main(): assert collection.find_one({{"subinterp": i}}) is not None ready.put(i) # Hold the client open until every interpreter has connected. - release.get(timeout=60) + {release_stmt} assert collection.find_one({{"subinterp": i}}) is not None done.put(i) @@ -185,6 +192,13 @@ def main(): {run_stmt} """ ) + # Wait for the workers off-loop in the async suite; synchro keeps the + # direct call for the sync suite. + result_stmt = ( + "asyncio.to_thread(future.result, 120)" + if not _IS_SYNC + else "future.result(timeout=120)" + ) with InterpreterPoolExecutor(max_workers=n_interpreters) as executor: uri = client_context.uri futures = [ @@ -202,7 +216,7 @@ def main(): for i in range(n_interpreters) ] for future in futures: - future.result(timeout=120) + {result_stmt} docs = self.db[coll_name].find({}, {"interp-pool": 1}).to_list() found = sorted(doc["interp-pool"] for doc in docs) From 41e55ad0f1739b395cd74e6d713340569507acfc Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Tue, 22 Sep 2026 10:35:08 -0500 Subject: [PATCH 15/21] PYTHON-5418 Cover the non-daemon thread fallback and exclude sync-only blocks Test that PeriodicExecutor.open() starts a non-daemon thread when the daemon assignment raises, and exclude the monitor shutdown registration blocks, which only run in the sync twin (the async source's block is dead code). --- pyproject.toml | 4 +++- test/asynchronous/test_periodic_executor.py | 21 +++++++++++++++++++++ test/test_periodic_executor.py | 21 +++++++++++++++++++++ 3 files changed, 45 insertions(+), 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index 2adddcd29b..eb8aa80984 100644 --- a/pyproject.toml +++ b/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.*)*:"] diff --git a/test/asynchronous/test_periodic_executor.py b/test/asynchronous/test_periodic_executor.py index 8d2ecd4af4..a3a36e632f 100644 --- a/test/asynchronous/test_periodic_executor.py +++ b/test/asynchronous/test_periodic_executor.py @@ -24,6 +24,7 @@ import threading import time from typing import Any +from unittest import mock _interpreters: Any = None if sys.version_info >= (3, 14): @@ -189,6 +190,26 @@ 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+") diff --git a/test/test_periodic_executor.py b/test/test_periodic_executor.py index 66101e89b6..1cacefcadc 100644 --- a/test/test_periodic_executor.py +++ b/test/test_periodic_executor.py @@ -24,6 +24,7 @@ import threading import time from typing import Any +from unittest import mock _interpreters: Any = None if sys.version_info >= (3, 14): @@ -189,6 +190,26 @@ def target(): executor._task.exception() self.assertEqual(call_count, 2, "executor should run again after re-open") + 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") + def test_subinterpreter_shutdown(self): if _interpreters is None: self.skipTest("concurrent.interpreters requires Python 3.14+") From ebb584fb389e5efe2b933412db55d6045dec7d77 Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Tue, 22 Sep 2026 12:31:45 -0500 Subject: [PATCH 16/21] PYTHON-5418 Await the pool worker results The result_stmt interpolation landed outside the worker string as a set literal, so the pool's futures were never awaited and worker exceptions were silently ignored. Call future.result() explicitly, through asyncio.to_thread in the async suite. --- test/asynchronous/test_subinterpreters.py | 14 ++++++-------- test/test_subinterpreters.py | 14 ++++++-------- 2 files changed, 12 insertions(+), 16 deletions(-) diff --git a/test/asynchronous/test_subinterpreters.py b/test/asynchronous/test_subinterpreters.py index a1b98de722..71970cd034 100644 --- a/test/asynchronous/test_subinterpreters.py +++ b/test/asynchronous/test_subinterpreters.py @@ -16,6 +16,7 @@ from __future__ import annotations +import asyncio import sys import textwrap import threading @@ -192,13 +193,6 @@ async def main(): {run_stmt} """ ) - # Wait for the workers off-loop in the async suite; synchro keeps the - # direct call for the sync suite. - result_stmt = ( - "await asyncio.to_thread(future.result, 120)" - if not _IS_SYNC - else "future.result(timeout=120)" - ) with InterpreterPoolExecutor(max_workers=n_interpreters) as executor: uri = await async_client_context.uri futures = [ @@ -216,7 +210,11 @@ async def main(): for i in range(n_interpreters) ] for future in futures: - {result_stmt} + if _IS_SYNC: + future.result(timeout=120) + else: + # Keep the event loop free while waiting for the workers. + await asyncio.to_thread(future.result, 120) docs = await self.db[coll_name].find({}, {"interp-pool": 1}).to_list() found = sorted(doc["interp-pool"] for doc in docs) diff --git a/test/test_subinterpreters.py b/test/test_subinterpreters.py index 0b6c9a62aa..2778cc3757 100644 --- a/test/test_subinterpreters.py +++ b/test/test_subinterpreters.py @@ -16,6 +16,7 @@ from __future__ import annotations +import asyncio import sys import textwrap import threading @@ -192,13 +193,6 @@ def main(): {run_stmt} """ ) - # Wait for the workers off-loop in the async suite; synchro keeps the - # direct call for the sync suite. - result_stmt = ( - "asyncio.to_thread(future.result, 120)" - if not _IS_SYNC - else "future.result(timeout=120)" - ) with InterpreterPoolExecutor(max_workers=n_interpreters) as executor: uri = client_context.uri futures = [ @@ -216,7 +210,11 @@ def main(): for i in range(n_interpreters) ] for future in futures: - {result_stmt} + if _IS_SYNC: + future.result(timeout=120) + else: + # Keep the event loop free while waiting for the workers. + asyncio.to_thread(future.result, 120) docs = self.db[coll_name].find({}, {"interp-pool": 1}).to_list() found = sorted(doc["interp-pool"] for doc in docs) From db6cdc500780636af2edb8d945201e101539e3e0 Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Tue, 22 Sep 2026 12:31:45 -0500 Subject: [PATCH 17/21] PYTHON-5418 Await the pool worker results The result_stmt interpolation landed outside the worker string as a set literal, so the pool's futures were never awaited and worker exceptions were silently ignored. Call future.result() explicitly, through asyncio.to_thread in the async suite. --- test/asynchronous/test_subinterpreters.py | 14 ++++++-------- test/test_subinterpreters.py | 14 ++++++-------- 2 files changed, 12 insertions(+), 16 deletions(-) diff --git a/test/asynchronous/test_subinterpreters.py b/test/asynchronous/test_subinterpreters.py index a1b98de722..71970cd034 100644 --- a/test/asynchronous/test_subinterpreters.py +++ b/test/asynchronous/test_subinterpreters.py @@ -16,6 +16,7 @@ from __future__ import annotations +import asyncio import sys import textwrap import threading @@ -192,13 +193,6 @@ async def main(): {run_stmt} """ ) - # Wait for the workers off-loop in the async suite; synchro keeps the - # direct call for the sync suite. - result_stmt = ( - "await asyncio.to_thread(future.result, 120)" - if not _IS_SYNC - else "future.result(timeout=120)" - ) with InterpreterPoolExecutor(max_workers=n_interpreters) as executor: uri = await async_client_context.uri futures = [ @@ -216,7 +210,11 @@ async def main(): for i in range(n_interpreters) ] for future in futures: - {result_stmt} + if _IS_SYNC: + future.result(timeout=120) + else: + # Keep the event loop free while waiting for the workers. + await asyncio.to_thread(future.result, 120) docs = await self.db[coll_name].find({}, {"interp-pool": 1}).to_list() found = sorted(doc["interp-pool"] for doc in docs) diff --git a/test/test_subinterpreters.py b/test/test_subinterpreters.py index 0b6c9a62aa..2778cc3757 100644 --- a/test/test_subinterpreters.py +++ b/test/test_subinterpreters.py @@ -16,6 +16,7 @@ from __future__ import annotations +import asyncio import sys import textwrap import threading @@ -192,13 +193,6 @@ def main(): {run_stmt} """ ) - # Wait for the workers off-loop in the async suite; synchro keeps the - # direct call for the sync suite. - result_stmt = ( - "asyncio.to_thread(future.result, 120)" - if not _IS_SYNC - else "future.result(timeout=120)" - ) with InterpreterPoolExecutor(max_workers=n_interpreters) as executor: uri = client_context.uri futures = [ @@ -216,7 +210,11 @@ def main(): for i in range(n_interpreters) ] for future in futures: - {result_stmt} + if _IS_SYNC: + future.result(timeout=120) + else: + # Keep the event loop free while waiting for the workers. + asyncio.to_thread(future.result, 120) docs = self.db[coll_name].find({}, {"interp-pool": 1}).to_list() found = sorted(doc["interp-pool"] for doc in docs) From 78ffa91d9a21352e7ee88a55f5a199437f7f66be Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Tue, 22 Sep 2026 16:41:20 -0500 Subject: [PATCH 18/21] PYTHON-5418 Keep the async subinterpreter test loop free Offload the queue waits, thread joins, and pool shutdown to worker threads in the async suite, where they would otherwise block the event loop for up to their full timeouts. The sync suite keeps the direct calls. --- test/asynchronous/test_subinterpreters.py | 32 +++++++++++++++++++---- test/test_subinterpreters.py | 32 +++++++++++++++++++---- 2 files changed, 54 insertions(+), 10 deletions(-) diff --git a/test/asynchronous/test_subinterpreters.py b/test/asynchronous/test_subinterpreters.py index 71970cd034..8a9ae311e4 100644 --- a/test/asynchronous/test_subinterpreters.py +++ b/test/asynchronous/test_subinterpreters.py @@ -125,16 +125,27 @@ def run(interp): threads.append(thread) thread.start() - started = self._get_n(ready, n_interpreters, errors) + started = ( + await asyncio.to_thread(self._get_n, ready, n_interpreters, errors) + if not _IS_SYNC + else 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) + if _IS_SYNC: + thread.join(60) + else: + await asyncio.to_thread(thread.join, 60) self.assertFalse(thread.is_alive(), f"{thread.name} did not exit") - finished = self._get_n(done, n_interpreters, errors) + finished = ( + await asyncio.to_thread(self._get_n, done, n_interpreters, errors) + if not _IS_SYNC + else self._get_n(done, n_interpreters, errors) + ) self.assertEqual(finished, list(range(n_interpreters))) if errors: self.fail(f"subinterpreter errors: {errors!r}") @@ -144,7 +155,10 @@ def run(interp): for _ in range(n_interpreters): release.put(True) for thread in threads: - thread.join(60) + if _IS_SYNC: + thread.join(60) + else: + await asyncio.to_thread(thread.join, 60) for interp in interps: # Closing an idle interpreter runs threading._shutdown, which # stops and joins pymongo's monitor threads; skip running ones @@ -193,7 +207,8 @@ async def main(): {run_stmt} """ ) - with InterpreterPoolExecutor(max_workers=n_interpreters) as executor: + executor = InterpreterPoolExecutor(max_workers=n_interpreters) + try: uri = await async_client_context.uri futures = [ executor.submit( @@ -215,6 +230,13 @@ async def main(): else: # Keep the event loop free while waiting for the workers. await asyncio.to_thread(future.result, 120) + finally: + # Worker interpreter teardown joins monitor threads, which can + # take a while; keep the event loop free while waiting. + if _IS_SYNC: + executor.shutdown(wait=True) + else: + await asyncio.to_thread(executor.shutdown) docs = await self.db[coll_name].find({}, {"interp-pool": 1}).to_list() found = sorted(doc["interp-pool"] for doc in docs) diff --git a/test/test_subinterpreters.py b/test/test_subinterpreters.py index 2778cc3757..e65984831b 100644 --- a/test/test_subinterpreters.py +++ b/test/test_subinterpreters.py @@ -125,16 +125,27 @@ def run(interp): threads.append(thread) thread.start() - started = self._get_n(ready, n_interpreters, errors) + started = ( + asyncio.to_thread(self._get_n, ready, n_interpreters, errors) + if not _IS_SYNC + else 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) + if _IS_SYNC: + thread.join(60) + else: + asyncio.to_thread(thread.join, 60) self.assertFalse(thread.is_alive(), f"{thread.name} did not exit") - finished = self._get_n(done, n_interpreters, errors) + finished = ( + asyncio.to_thread(self._get_n, done, n_interpreters, errors) + if not _IS_SYNC + else self._get_n(done, n_interpreters, errors) + ) self.assertEqual(finished, list(range(n_interpreters))) if errors: self.fail(f"subinterpreter errors: {errors!r}") @@ -144,7 +155,10 @@ def run(interp): for _ in range(n_interpreters): release.put(True) for thread in threads: - thread.join(60) + if _IS_SYNC: + thread.join(60) + else: + asyncio.to_thread(thread.join, 60) for interp in interps: # Closing an idle interpreter runs threading._shutdown, which # stops and joins pymongo's monitor threads; skip running ones @@ -193,7 +207,8 @@ def main(): {run_stmt} """ ) - with InterpreterPoolExecutor(max_workers=n_interpreters) as executor: + executor = InterpreterPoolExecutor(max_workers=n_interpreters) + try: uri = client_context.uri futures = [ executor.submit( @@ -215,6 +230,13 @@ def main(): else: # Keep the event loop free while waiting for the workers. asyncio.to_thread(future.result, 120) + finally: + # Worker interpreter teardown joins monitor threads, which can + # take a while; keep the event loop free while waiting. + if _IS_SYNC: + executor.shutdown(wait=True) + else: + asyncio.to_thread(executor.shutdown) docs = self.db[coll_name].find({}, {"interp-pool": 1}).to_list() found = sorted(doc["interp-pool"] for doc in docs) From 1c74dcc33e0634dc786cbf301eefbd802dc27d09 Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Tue, 22 Sep 2026 20:44:30 -0500 Subject: [PATCH 19/21] PYTHON-5418 Use a selector loop in subinterpreters on Windows The Windows proactor loop cannot be created in a subinterpreter: signal.set_wakeup_fd only works in the main interpreter. Run the async workers on a selector loop there instead. --- test/asynchronous/test_periodic_executor.py | 8 +++++++- test/asynchronous/test_subinterpreters.py | 16 ++++++++++++++-- test/test_periodic_executor.py | 8 +++++++- test/test_subinterpreters.py | 16 ++++++++++++++-- 4 files changed, 42 insertions(+), 6 deletions(-) diff --git a/test/asynchronous/test_periodic_executor.py b/test/asynchronous/test_periodic_executor.py index a3a36e632f..42d64367ee 100644 --- a/test/asynchronous/test_periodic_executor.py +++ b/test/asynchronous/test_periodic_executor.py @@ -218,7 +218,13 @@ async def test_subinterpreter_shutdown(self): 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()" + # 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 "" diff --git a/test/asynchronous/test_subinterpreters.py b/test/asynchronous/test_subinterpreters.py index 8a9ae311e4..c75fe47e8e 100644 --- a/test/asynchronous/test_subinterpreters.py +++ b/test/asynchronous/test_subinterpreters.py @@ -65,7 +65,13 @@ async def test_subinterpreters(self): 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()" + # 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 reads the queue on its thread; the async monitor # reads it off-loop so the client's tasks keep running. release_stmt = ( @@ -186,7 +192,13 @@ async def test_interpreter_pool_executor(self): # 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()" + # 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 worker's sys.path lacks the repo root, so submit the builtin exec # and fix sys.path in the code. diff --git a/test/test_periodic_executor.py b/test/test_periodic_executor.py index 1cacefcadc..d564042c30 100644 --- a/test/test_periodic_executor.py +++ b/test/test_periodic_executor.py @@ -218,7 +218,13 @@ def test_subinterpreter_shutdown(self): 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()" + # 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 "" diff --git a/test/test_subinterpreters.py b/test/test_subinterpreters.py index e65984831b..70030c056e 100644 --- a/test/test_subinterpreters.py +++ b/test/test_subinterpreters.py @@ -65,7 +65,13 @@ def test_subinterpreters(self): 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()" + # 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 reads the queue on its thread; the async monitor # reads it off-loop so the client's tasks keep running. release_stmt = ( @@ -186,7 +192,13 @@ def test_interpreter_pool_executor(self): # 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()" + # 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 worker's sys.path lacks the repo root, so submit the builtin exec # and fix sys.path in the code. From 9980e3b916ced6c79722d944bf45328506213591 Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Wed, 23 Sep 2026 05:29:06 -0500 Subject: [PATCH 20/21] PYTHON-5418 Address copilot review and pass server_api in subinterpreter tests --- pymongo/asynchronous/monitor.py | 11 ++++----- pymongo/periodic_executor.py | 29 +++++++++++++++++++++-- pymongo/synchronous/monitor.py | 11 ++++----- test/asynchronous/test_subinterpreters.py | 21 +++++++++++++--- test/test_subinterpreters.py | 21 +++++++++++++--- 5 files changed, 71 insertions(+), 22 deletions(-) diff --git a/pymongo/asynchronous/monitor.py b/pymongo/asynchronous/monitor.py index eca692bf96..da261e37ea 100644 --- a/pymongo/asynchronous/monitor.py +++ b/pymongo/asynchronous/monitor.py @@ -494,10 +494,7 @@ 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] + # 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] diff --git a/pymongo/periodic_executor.py b/pymongo/periodic_executor.py index 4338f5dd49..867fc33e06 100644 --- a/pymongo/periodic_executor.py +++ b/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, @@ -189,8 +206,10 @@ def open(self) -> None: # 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 + 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) @@ -202,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(). diff --git a/pymongo/synchronous/monitor.py b/pymongo/synchronous/monitor.py index aea1b9fc12..3a1e1a49d0 100644 --- a/pymongo/synchronous/monitor.py +++ b/pymongo/synchronous/monitor.py @@ -492,10 +492,7 @@ 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] + # 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] diff --git a/test/asynchronous/test_subinterpreters.py b/test/asynchronous/test_subinterpreters.py index c75fe47e8e..f3e8c2883c 100644 --- a/test/asynchronous/test_subinterpreters.py +++ b/test/asynchronous/test_subinterpreters.py @@ -34,7 +34,12 @@ except ImportError: # pragma: no cover - Python < 3.14 InterpreterPoolExecutor = None # type: ignore[assignment,misc] -from test.asynchronous import AsyncIntegrationTest, async_client_context, unittest +from test.asynchronous import ( + MONGODB_API_VERSION, + AsyncIntegrationTest, + async_client_context, + unittest, +) _IS_SYNC = False @@ -86,9 +91,13 @@ async def test_subinterpreters(self): sys.path[:0] = path from pymongo import AsyncMongoClient + from pymongo.server_api import ServerApi async def main(): - client = AsyncMongoClient(uri, serverSelectionTimeoutMS=30000) + kwargs = {{"serverSelectionTimeoutMS": 30000}} + if api_version: + kwargs["server_api"] = ServerApi(api_version) + client = AsyncMongoClient(uri, **kwargs) 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 @@ -122,6 +131,7 @@ def run(interp): coll_name=coll_name, i=i, path=tuple(sys.path), + api_version=MONGODB_API_VERSION, ready=ready, release=release, done=done, @@ -209,9 +219,13 @@ async def test_interpreter_pool_executor(self): sys.path[:0] = path from pymongo import AsyncMongoClient + from pymongo.server_api import ServerApi async def main(): - client = AsyncMongoClient(uri, serverSelectionTimeoutMS=30000) + kwargs = {{"serverSelectionTimeoutMS": 30000}} + if api_version: + kwargs["server_api"] = ServerApi(api_version) + client = AsyncMongoClient(uri, **kwargs) 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 @@ -232,6 +246,7 @@ async def main(): "uri": uri, "db_name": self.db.name, "coll_name": coll_name, + "api_version": MONGODB_API_VERSION, }, ) for i in range(n_interpreters) diff --git a/test/test_subinterpreters.py b/test/test_subinterpreters.py index 70030c056e..e16d8be642 100644 --- a/test/test_subinterpreters.py +++ b/test/test_subinterpreters.py @@ -34,7 +34,12 @@ except ImportError: # pragma: no cover - Python < 3.14 InterpreterPoolExecutor = None # type: ignore[assignment,misc] -from test import IntegrationTest, client_context, unittest +from test import ( + MONGODB_API_VERSION, + IntegrationTest, + client_context, + unittest, +) _IS_SYNC = True @@ -86,9 +91,13 @@ def test_subinterpreters(self): sys.path[:0] = path from pymongo import MongoClient + from pymongo.server_api import ServerApi def main(): - client = MongoClient(uri, serverSelectionTimeoutMS=30000) + kwargs = {{"serverSelectionTimeoutMS": 30000}} + if api_version: + kwargs["server_api"] = ServerApi(api_version) + client = MongoClient(uri, **kwargs) collection = client.get_database(db_name).get_collection(coll_name) collection.insert_one({{"subinterp": i}}) assert collection.find_one({{"subinterp": i}}) is not None @@ -122,6 +131,7 @@ def run(interp): coll_name=coll_name, i=i, path=tuple(sys.path), + api_version=MONGODB_API_VERSION, ready=ready, release=release, done=done, @@ -209,9 +219,13 @@ def test_interpreter_pool_executor(self): sys.path[:0] = path from pymongo import MongoClient + from pymongo.server_api import ServerApi def main(): - client = MongoClient(uri, serverSelectionTimeoutMS=30000) + kwargs = {{"serverSelectionTimeoutMS": 30000}} + if api_version: + kwargs["server_api"] = ServerApi(api_version) + client = MongoClient(uri, **kwargs) collection = client.get_database(db_name).get_collection(coll_name) collection.insert_one({{"interp-pool": i}}) assert collection.find_one({{"interp-pool": i}}) is not None @@ -232,6 +246,7 @@ def main(): "uri": uri, "db_name": self.db.name, "coll_name": coll_name, + "api_version": MONGODB_API_VERSION, }, ) for i in range(n_interpreters) From 6c3a857fdd1516d3fc0efc118a05ff91dc4c0cbc Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Wed, 23 Sep 2026 05:46:47 -0500 Subject: [PATCH 21/21] PYTHON-5418 Clarify that non-daemon monitor threads only apply to subinterpreters --- doc/changelog.rst | 13 +++++++------ 1 file changed, 7 insertions(+), 6 deletions(-) diff --git a/doc/changelog.rst b/doc/changelog.rst index 33a7c2acab..9f16f0ee99 100644 --- a/doc/changelog.rst +++ b/doc/changelog.rst @@ -9,12 +9,13 @@ 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. + ``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 .........