Add shard async loop #25
Conversation
There was a problem hiding this comment.
Pull request overview
This PR introduces a shared async event loop implementation to handle async-only providers (like Google GenAI) and refactors the safe_generation function to use it instead of the synchronous run_sync method. The implementation provides a thread-safe singleton pattern for managing a background event loop that can be accessed from multiple threads.
Key changes:
- Adds
SharedAsyncLoopclass that manages a dedicated background thread with an asyncio event loop - Implements singleton pattern with double-checked locking for thread-safe access
- Replaces
agent.run_sync()withshared_async_loop().run(agent.run())in the safe_generation function
Reviewed changes
Copilot reviewed 2 out of 2 changed files in this pull request and generated 2 comments.
| File | Description |
|---|---|
| src/ale_bench_eval/shared_async_loop.py | New file implementing SharedAsyncLoop class with background event loop management, cleanup handlers, and thread-safe singleton access |
| src/ale_bench_eval/safe_generation.py | Updated to use the shared async loop instead of run_sync for executing agent operations |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| self._loop.run_forever() | ||
|
|
||
| async def _drain_pending(self) -> None: | ||
| tasks = [t for t in asyncio.all_tasks() if t is not asyncio.current_task()] |
There was a problem hiding this comment.
The asyncio.all_tasks() call is missing the loop parameter. When called from a different thread context, asyncio.all_tasks() without arguments returns tasks from the current thread's event loop (which may not exist), not from self._loop. This should be asyncio.all_tasks(self._loop) to ensure it retrieves tasks from the correct event loop that is being shut down.
| tasks = [t for t in asyncio.all_tasks() if t is not asyncio.current_task()] | |
| tasks = [t for t in asyncio.all_tasks(self._loop) if t is not asyncio.current_task(self._loop)] |
| self._loop = asyncio.new_event_loop() | ||
| self._thread = threading.Thread(target=self._run_loop, daemon=True) | ||
| self._thread.start() | ||
| atexit.register(self.shutdown) |
There was a problem hiding this comment.
The atexit registration in the constructor can lead to resource leaks if multiple SharedAsyncLoop instances are created and discarded. Each instance registers its own shutdown handler, but when an instance is replaced (e.g., in the shared_async_loop() function), the old instance's atexit handler remains registered but can no longer properly clean up since SHARED_ASYNC_LOOP has been reassigned. Consider unregistering atexit callbacks when instances are replaced, or restructure to only register atexit once for the global instance.
|
@codex review |
|
Codex Review: Didn't find any major issues. 👍 ℹ️ About Codex in GitHubYour team has set up Codex to review pull requests in this repo. Reviews are triggered when you
If Codex has suggestions, it will comment; otherwise it will react with 👍. Codex can also answer questions or update the PR. Try commenting "@codex address that feedback". |
|
@codex review carefully and report major or minor issues. |
|
Summary
Testing
|
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 2 out of 2 changed files in this pull request and generated 8 comments.
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 2 out of 2 changed files in this pull request and generated 5 comments.
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 2 out of 2 changed files in this pull request and generated 4 comments.
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
Resolve the below error when Google GenAI is used: