From e0785d8d550bb574c2ec97657d4e0838af87975d Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Sun, 20 Sep 2026 12:04:44 +0000 Subject: [PATCH 01/12] 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 6c7751071c3756d51b78250e2fca31e145f7bb4e Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Mon, 21 Sep 2026 13:41:15 -0500 Subject: [PATCH 02/12] 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 6a2f343971624e31cea5132309e64d7bafda4aa1 Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Mon, 21 Sep 2026 14:41:42 -0500 Subject: [PATCH 03/12] 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 c6b1d574528c86e5df27da4a6e2556369daa2a21 Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Mon, 21 Sep 2026 16:37:36 -0500 Subject: [PATCH 04/12] 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 5e1109ccea94388d240afc1cda5df2118cc1f65f Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Mon, 21 Sep 2026 16:39:34 -0500 Subject: [PATCH 05/12] 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 f1ab042f259578d3ce5a420f6415cff947ecdd5b Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Mon, 21 Sep 2026 16:46:36 -0500 Subject: [PATCH 06/12] 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 79611c205d914adf7a6859353f03fc800cbfee3e Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Mon, 21 Sep 2026 16:53:18 -0500 Subject: [PATCH 07/12] 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 623ecb07d5dcd6e1acaf62ad1842dca627679b9a Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Mon, 21 Sep 2026 17:22:40 -0500 Subject: [PATCH 08/12] 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 c8e6c92d64e71fc8618d38f512cd04db62dd8683 Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Tue, 22 Sep 2026 05:14:59 -0500 Subject: [PATCH 09/12] 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 8c058dc22a98884a8db5f6f2eb378a27b4d6081c Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Tue, 22 Sep 2026 05:16:01 -0500 Subject: [PATCH 10/12] 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 089cfab2ee6933472addfa1e6f7c8aff3302a7b1 Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Tue, 22 Sep 2026 05:30:39 -0500 Subject: [PATCH 11/12] 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 3fcdda04c8fde5ebf49611fc13cadb99904052ec Mon Sep 17 00:00:00 2001 From: Steven Silvester Date: Tue, 22 Sep 2026 05:47:33 -0500 Subject: [PATCH 12/12] 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