diff --git a/doc/changelog.rst b/doc/changelog.rst index 5fb2fb4731..33a7c2acab 100644 --- a/doc/changelog.rst +++ b/doc/changelog.rst @@ -7,6 +7,14 @@ Changes in Version 4.19.0 (2026/XX/XX) PyMongo 4.19 brings a number of changes including: - Added support for Python 3.15. +- Added support for running the synchronous and asynchronous clients in + subinterpreters, including inside + ``concurrent.futures.InterpreterPoolExecutor`` (Python 3.14+). In + interpreters that do not allow daemon threads, the synchronous client's + monitor threads now start as non-daemon threads and are stopped and joined + when the interpreter is torn down. Note that because these threads are + non-daemon, an interpreter may block on teardown until any in-flight + monitor work completes. Bug fixes ......... diff --git a/pymongo/asynchronous/monitor.py b/pymongo/asynchronous/monitor.py index ba0e3804a6..eca692bf96 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,11 @@ def _shutdown_resources() -> None: if _IS_SYNC: atexit.register(_shutdown_resources) + # In subinterpreters, the executors' threads are non-daemon and are + # joined at shutdown, so they must be stopped first. The probe preserves + # the normal atexit ordering in the main interpreter. + try: + threading.Thread().daemon = True + except RuntimeError: + if hasattr(threading, "_register_atexit"): + threading._register_atexit(_shutdown_resources) # type: ignore[attr-defined] diff --git a/pymongo/periodic_executor.py b/pymongo/periodic_executor.py index 4b979ca9f9..4338f5dd49 100644 --- a/pymongo/periodic_executor.py +++ b/pymongo/periodic_executor.py @@ -185,7 +185,13 @@ def open(self) -> None: if not started: thread = threading.Thread(target=self._run, name=self._name) - thread.daemon = True + try: + # Subinterpreters do not allow daemon threads, so fall back + # to a non-daemon thread: _shutdown_executors stops and + # joins it during interpreter shutdown. + thread.daemon = True + except RuntimeError: + pass self._thread = weakref.proxy(thread) _register_executor(self) # Mitigation to RuntimeError firing when thread starts on shutdown diff --git a/pymongo/synchronous/monitor.py b/pymongo/synchronous/monitor.py index 9a25757f03..aea1b9fc12 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,11 @@ def _shutdown_resources() -> None: if _IS_SYNC: atexit.register(_shutdown_resources) + # In subinterpreters, the executors' threads are non-daemon and are + # joined at shutdown, so they must be stopped first. The probe preserves + # the normal atexit ordering in the main interpreter. + try: + threading.Thread().daemon = True + except RuntimeError: + if hasattr(threading, "_register_atexit"): + threading._register_atexit(_shutdown_resources) # type: ignore[attr-defined] diff --git a/test/asynchronous/test_periodic_executor.py b/test/asynchronous/test_periodic_executor.py index 15186c4f88..8d2ecd4af4 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,50 @@ async def target(): executor._task.exception() self.assertEqual(call_count, 2, "executor should run again after re-open") + async def test_subinterpreter_shutdown(self): + if _interpreters is None: + self.skipTest("concurrent.interpreters requires Python 3.14+") + return + + root = _PYMONGO_ROOT + # The async executor's open() starts a task, which requires a running + # event loop; synchro translates the rest of the block for the sync suite. + run_stmt = "asyncio.run(main())" if not _IS_SYNC else "main()" + # The sync monitor signals its thread through the event; the async + # monitor yields to its loop instead. Synchro drops the async yield. + wait_stmt = "assert started.wait(10)" if _IS_SYNC else "" + code = textwrap.dedent( + f""" + import asyncio + import sys + import threading + sys.path.insert(0, {root!r}) + from pymongo.periodic_executor import AsyncPeriodicExecutor + + started = threading.Event() + + async def target(): + started.set() + return True + + async def main(): + executor = AsyncPeriodicExecutor( + interval=30.0, min_interval=0.05, target=target, name="subinterp" + ) + executor.open() + {wait_stmt} + await asyncio.sleep(0) + + {run_stmt} + """ + ) + interp = _interpreters.create() + try: + interp.exec(code) + finally: + # Destroying the subinterpreter must not crash or hang. + interp.close() + if __name__ == "__main__": unittest.main() diff --git a/test/asynchronous/test_subinterpreters.py b/test/asynchronous/test_subinterpreters.py new file mode 100644 index 0000000000..a31cba1b6a --- /dev/null +++ b/test/asynchronous/test_subinterpreters.py @@ -0,0 +1,210 @@ +# Copyright 2026-present MongoDB, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Test running pymongo in subinterpreters.""" + +from __future__ import annotations + +import sys +import textwrap +import threading +import uuid + +sys.path[0:0] = [""] + +try: + from concurrent import interpreters +except ImportError: # pragma: no cover - Python < 3.14 + interpreters = None # type: ignore[assignment] + +try: + from concurrent.futures import InterpreterPoolExecutor +except ImportError: # pragma: no cover - Python < 3.14 + InterpreterPoolExecutor = None # type: ignore[assignment,misc] + +from test.asynchronous import AsyncIntegrationTest, async_client_context, unittest + +_IS_SYNC = False + + +class TestSubinterpreters(AsyncIntegrationTest): + @staticmethod + def _get_n(queue, n, errors): + try: + return sorted(queue.get(timeout=30) for _ in range(n)) + except interpreters.QueueEmpty: + raise AssertionError(f"subinterpreters failed to run: {errors!r}") from None + + @unittest.skipUnless( + sys.version_info >= (3, 14), "concurrent.interpreters requires Python 3.14+" + ) + async def test_subinterpreters(self): + if interpreters is None: + self.skipTest("concurrent.interpreters is not available") + + # Run live clients in multiple subinterpreters at once, mirroring the + # mod_wsgi test, which mounts the same app in two interpreters. + n_interpreters = 2 + coll_name = f"subinterp-{uuid.uuid4().hex}" + self.addCleanup(self.db.drop_collection, coll_name) + + ready = interpreters.create_queue() + release = interpreters.create_queue() + done = interpreters.create_queue() + # The async client's constructor starts a task, which requires a + # running event loop; synchro translates the rest of the block. + run_stmt = "asyncio.run(main())" if not _IS_SYNC else "main()" + code = textwrap.dedent( + f""" + import asyncio + import sys + sys.path[:0] = path + + from pymongo import AsyncMongoClient + + async def main(): + client = AsyncMongoClient(uri, serverSelectionTimeoutMS=30000) + collection = client.get_database(db_name).get_collection(coll_name) + await collection.insert_one({{"subinterp": i}}) + assert await collection.find_one({{"subinterp": i}}) is not None + ready.put(i) + # Hold the client open until every interpreter has connected. + release.get(timeout=60) + assert await collection.find_one({{"subinterp": i}}) is not None + done.put(i) + + {run_stmt} + """ + ) + + errors: list[BaseException] = [] + + def run(interp): + try: + interp.exec(code) + except BaseException as exc: + errors.append(exc) + + uri = await async_client_context.uri + interps = [] + threads = [] + try: + for i in range(n_interpreters): + interp = interpreters.create() + interp.prepare_main( + uri=uri, + db_name=self.db.name, + coll_name=coll_name, + i=i, + path=tuple(sys.path), + ready=ready, + release=release, + done=done, + ) + interps.append(interp) + thread = threading.Thread(target=run, args=(interp,), name=f"subinterp-{i}") + threads.append(thread) + thread.start() + + started = self._get_n(ready, n_interpreters, errors) + self.assertEqual(started, list(range(n_interpreters))) + for _ in range(n_interpreters): + release.put(True) + + for thread in threads: + thread.join(60) + self.assertFalse(thread.is_alive(), f"{thread.name} did not exit") + + finished = self._get_n(done, n_interpreters, errors) + self.assertEqual(finished, list(range(n_interpreters))) + if errors: + self.fail(f"subinterpreter errors: {errors!r}") + finally: + # Unblock any interpreter still waiting, then destroy them all. + for _ in range(n_interpreters): + release.put(True) + for interp in interps: + # Closing an idle interpreter runs threading._shutdown, which + # stops and joins pymongo's monitor threads; skip running ones + # to avoid masking errors. + if not interp.is_running(): + interp.close() + + docs = await self.db[coll_name].find({}, {"subinterp": 1}).to_list() + found = sorted(doc["subinterp"] for doc in docs) + self.assertEqual(found, list(range(n_interpreters))) + + @unittest.skipUnless( + sys.version_info >= (3, 14), "InterpreterPoolExecutor requires Python 3.14+" + ) + async def test_interpreter_pool_executor(self): + if InterpreterPoolExecutor is None: + self.skipTest("InterpreterPoolExecutor is not available") + + # The interpreters disallow daemon threads, so the sync client starts + # non-daemon monitor threads; the async client runs tasks on the + # interpreter's own event loop. + n_interpreters = 2 + coll_name = f"interp-pool-{uuid.uuid4().hex}" + self.addCleanup(self.db.drop_collection, coll_name) + + # The async client's constructor starts a task, which requires a + # running event loop; synchro translates the rest of the block. + run_stmt = "asyncio.run(main())" if not _IS_SYNC else "main()" + + # The worker's sys.path lacks the repo root, so submit the builtin exec + # and fix sys.path in the code. + code = textwrap.dedent( + f""" + import asyncio + import sys + sys.path[:0] = path + + from pymongo import AsyncMongoClient + + async def main(): + client = AsyncMongoClient(uri, serverSelectionTimeoutMS=30000) + collection = client.get_database(db_name).get_collection(coll_name) + await collection.insert_one({{"interp-pool": i}}) + assert await collection.find_one({{"interp-pool": i}}) is not None + + {run_stmt} + """ + ) + with InterpreterPoolExecutor(max_workers=n_interpreters) as executor: + uri = await async_client_context.uri + futures = [ + executor.submit( + exec, + code, + { + "i": i, + "path": tuple(sys.path), + "uri": uri, + "db_name": self.db.name, + "coll_name": coll_name, + }, + ) + for i in range(n_interpreters) + ] + for future in futures: + future.result(timeout=120) + + docs = await self.db[coll_name].find({}, {"interp-pool": 1}).to_list() + found = sorted(doc["interp-pool"] for doc in docs) + self.assertEqual(found, list(range(n_interpreters))) + + +if __name__ == "__main__": + unittest.main() diff --git a/test/test_periodic_executor.py b/test/test_periodic_executor.py index e6bbb31a72..66101e89b6 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,49 @@ def target(): executor._task.exception() self.assertEqual(call_count, 2, "executor should run again after re-open") + def test_subinterpreter_shutdown(self): + if _interpreters is None: + self.skipTest("concurrent.interpreters requires Python 3.14+") + return + + root = _PYMONGO_ROOT + # The async executor's open() starts a task, which requires a running + # event loop; synchro translates the rest of the block for the sync suite. + run_stmt = "asyncio.run(main())" if not _IS_SYNC else "main()" + # The sync monitor signals its thread through the event; the async + # monitor yields to its loop instead. Synchro drops the async yield. + wait_stmt = "assert started.wait(10)" if _IS_SYNC else "" + code = textwrap.dedent( + f""" + import asyncio + import sys + import threading + sys.path.insert(0, {root!r}) + from pymongo.periodic_executor import PeriodicExecutor + + started = threading.Event() + + def target(): + started.set() + return True + + def main(): + executor = PeriodicExecutor( + interval=30.0, min_interval=0.05, target=target, name="subinterp" + ) + executor.open() + {wait_stmt} + + {run_stmt} + """ + ) + interp = _interpreters.create() + try: + interp.exec(code) + finally: + # Destroying the subinterpreter must not crash or hang. + interp.close() + if __name__ == "__main__": unittest.main() diff --git a/test/test_subinterpreters.py b/test/test_subinterpreters.py new file mode 100644 index 0000000000..68e93fe575 --- /dev/null +++ b/test/test_subinterpreters.py @@ -0,0 +1,210 @@ +# Copyright 2026-present MongoDB, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Test running pymongo in subinterpreters.""" + +from __future__ import annotations + +import sys +import textwrap +import threading +import uuid + +sys.path[0:0] = [""] + +try: + from concurrent import interpreters +except ImportError: # pragma: no cover - Python < 3.14 + interpreters = None # type: ignore[assignment] + +try: + from concurrent.futures import InterpreterPoolExecutor +except ImportError: # pragma: no cover - Python < 3.14 + InterpreterPoolExecutor = None # type: ignore[assignment,misc] + +from test 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 multiple subinterpreters at once, mirroring the + # mod_wsgi test, which mounts the same app in two interpreters. + n_interpreters = 2 + coll_name = f"subinterp-{uuid.uuid4().hex}" + self.addCleanup(self.db.drop_collection, coll_name) + + ready = interpreters.create_queue() + release = interpreters.create_queue() + done = interpreters.create_queue() + # The async client's constructor starts a task, which requires a + # running event loop; synchro translates the rest of the block. + run_stmt = "asyncio.run(main())" if not _IS_SYNC else "main()" + code = textwrap.dedent( + f""" + import asyncio + import sys + sys.path[:0] = path + + from pymongo import 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. + 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 errors. + 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") + + # The interpreters disallow daemon threads, so the sync client starts + # non-daemon monitor threads; the async client runs tasks on the + # interpreter's own event loop. + n_interpreters = 2 + coll_name = f"interp-pool-{uuid.uuid4().hex}" + self.addCleanup(self.db.drop_collection, coll_name) + + # The async client's constructor starts a task, which requires a + # running event loop; synchro translates the rest of the block. + run_stmt = "asyncio.run(main())" if not _IS_SYNC else "main()" + + # The worker's sys.path lacks the repo root, so submit the builtin exec + # and fix sys.path in the code. + code = textwrap.dedent( + f""" + import asyncio + import sys + sys.path[:0] = path + + from pymongo import 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()