From 0425e28414c63132eef3f69f5b008291ee726b31 Mon Sep 17 00:00:00 2001 From: David Dotson Date: Wed, 3 Jun 2026 22:36:02 -0600 Subject: [PATCH] Replace nest-asyncio with a dedicated background event loop `BaseClient._run_async` relied on `nest_asyncio` to allow re-entrant `run_until_complete` so the synchronous client could drive its async, concurrency-bearing methods (batched `asyncio.gather` fan-out, concurrent ProtocolDAGResult retrieval) even from environments with an already-running loop (e.g. Jupyter). `nest_asyncio` does this by monkeypatching asyncio's private internals, which is fragile across CPython releases and is widely considered bad practice. Instead, run a single persistent event loop in a dedicated daemon thread and submit coroutines to it with `asyncio.run_coroutine_threadsafe(...).result()`. The calling thread blocks on the result while the loop thread does the work, so: - all existing `gather`/`as_completed` concurrency is preserved (it runs inside the loop thread); - re-entrancy "just works" without monkeypatching --- a caller already inside a running loop simply waits on another thread; - `alru_cache` stays bound to one stable, process-wide loop, which is the invariant the previous shared-loop dance was straining to maintain. The loop is created lazily under a lock so concurrent callers cannot start competing loops. `nest-asyncio` is dropped from the client environment, as it is no longer used anywhere. Co-Authored-By: Claude Opus 4.8 (1M context) --- alchemiscale/base/client.py | 67 ++++++++++++++++----- devtools/conda-envs/alchemiscale-client.yml | 1 - 2 files changed, 51 insertions(+), 17 deletions(-) diff --git a/alchemiscale/base/client.py b/alchemiscale/base/client.py index cec53b44..f7dd1339 100644 --- a/alchemiscale/base/client.py +++ b/alchemiscale/base/client.py @@ -5,6 +5,7 @@ """ import asyncio +import threading import time import random from itertools import islice @@ -116,7 +117,15 @@ class AlchemiscaleBaseClient: _exception = AlchemiscaleBaseClientError _retry_status_codes = [404, 502, 503, 504] + + # A single event loop, running in a dedicated daemon thread, is shared + # across all client instances. Coroutines are submitted to it from the + # calling thread (see ``_run_async``). Sharing one loop process-wide keeps + # ``alru_cache`` --- which binds to the loop it is first used with --- bound + # to a stable loop regardless of which client invoked it. _shared_event_loop = None + _shared_event_loop_thread = None + _event_loop_lock = threading.Lock() _PARAMS = { "api_url": AlchemiscaleBaseClientParam( @@ -263,27 +272,53 @@ def _settings(self): def __repr__(self): return f"{self.__class__.__name__}('{self.api_url}')" - def _run_async(self, coro): - """Run an async coroutine, reusing event loop for alru_cache compatibility. - - ``alru_cache`` binds to the event loop it is first used with and rejects - calls from different loops. ``asyncio.run()`` creates a new loop each time, - which is incompatible. This method maintains a single shared event loop - per class to match ``alru_cache``'s class-level loop tracking. + @classmethod + def _get_event_loop(cls) -> asyncio.AbstractEventLoop: + """Return the shared event loop, starting its daemon thread if needed. - We apply ``nest_asyncio`` proactively so the loop also works in - environments where an event loop is already running (e.g. Jupyter). + The loop runs in a dedicated background thread via ``run_forever`` and + persists for the lifetime of the process. A single loop is shared across + all client instances so that ``alru_cache`` (which binds to the loop it + is first used with) always sees the same loop. Lazy initialization is + guarded by a lock so concurrent callers cannot start competing loops. """ - import nest_asyncio + loop = AlchemiscaleBaseClient._shared_event_loop + if loop is not None and not loop.is_closed(): + return loop + + with AlchemiscaleBaseClient._event_loop_lock: + # re-check inside the lock; another thread may have started it + loop = AlchemiscaleBaseClient._shared_event_loop + if loop is not None and not loop.is_closed(): + return loop + + loop = asyncio.new_event_loop() + thread = threading.Thread( + target=loop.run_forever, + name="alchemiscale-client-asyncio", + daemon=True, + ) + thread.start() - cls = type(self) - if cls._shared_event_loop is None or cls._shared_event_loop.is_closed(): - cls._shared_event_loop = asyncio.new_event_loop() - asyncio.set_event_loop(cls._shared_event_loop) - nest_asyncio.apply(cls._shared_event_loop) + AlchemiscaleBaseClient._shared_event_loop = loop + AlchemiscaleBaseClient._shared_event_loop_thread = thread + return loop - return cls._shared_event_loop.run_until_complete(coro) + def _run_async(self, coro): + """Run an async coroutine on the shared background event loop. + + The coroutine is scheduled on a persistent event loop running in a + dedicated daemon thread, and the calling thread blocks until it + completes. Because the loop lives in its own thread, this works even + when the caller is itself inside a running event loop (e.g. Jupyter): + the calling thread simply waits on the result while the loop thread does + the work. This avoids ``nest_asyncio``'s monkeypatching of asyncio + internals, and keeps ``alru_cache`` bound to a single stable loop. + + """ + loop = self._get_event_loop() + return asyncio.run_coroutine_threadsafe(coro, loop).result() def _retry(f): """Automatically retry with exponential backoff if API service is diff --git a/devtools/conda-envs/alchemiscale-client.yml b/devtools/conda-envs/alchemiscale-client.yml index 6ef6d801..e8e500cd 100644 --- a/devtools/conda-envs/alchemiscale-client.yml +++ b/devtools/conda-envs/alchemiscale-client.yml @@ -23,7 +23,6 @@ dependencies: ## user client - rich - - nest-asyncio # openmm protocols - feflow=0.1.4