fix(sdk/python): resolve ResultCache cross-loop deadlock (#623) - #799
fix(sdk/python): resolve ResultCache cross-loop deadlock (#623)#7997vignesh wants to merge 18 commits into
Conversation
…#623) ResultCache mixed a threading.RLock (for its data) with loop-bound asyncio primitives (asyncio.Event for shutdown, asyncio.Task for the cleanup loop). When start() and stop() ran on different event loops — which happens when the AgentFieldClient's sync and async execution paths are mixed (Agent-Field#620) — stop() would raise 'got Future attached to a different loop' and could wedge the process waiting on a task it can never await. Fix: make the cache lifecycle loop-aware. - Record the event loop the cleanup task/shutdown event are bound to. - start() is now idempotent on the same loop and rebinds cleanly when called on a new loop, discarding the stale task via call_soon_threadsafe(task.cancel) on its owning loop — never a cross-loop await. - stop() only awaits the cleanup task when on its owning loop; from a different loop it cancels without awaiting. The cache is always cleared regardless (that path only needs the thread lock). - Shrink the cleanup loop's critical section so stats logging no longer runs while holding the lock, reducing contention with sync callers. Adds tests/test_result_cache_deadlock.py covering cross-loop stop, idempotent/rebinding start, concurrent sync access during cleanup, and the disabled-cache no-op path. result_cache.py coverage: 88% -> 95%.
Performance
⚠ Regression detected:
|
📊 Coverage gateThresholds from
✅ Gate passedNo surface regressed past the allowed threshold and the aggregate stayed above the floor. |
📐 Patch coverage gateThreshold: 80% on lines this PR touches vs
✅ Patch gate passedEvery surface whose lines were touched by this PR has patch coverage at or above the threshold. |
AbirAbbas
left a comment
AbirAbbas
left a comment
There was a problem hiding this comment.
I took this for a spin rather than just reading the diff. On the merge-base I reproduced the failure with start() on a background thread's loop and stop() via asyncio.run() on the main thread: RuntimeError: got Future attached to a different loop, and since stop() aborts mid-way the cache is left populated. On this branch the same scenario (and the owning-loop-closed variant) completes cleanly, and I checked that the stale cleanup task really does get cancelled on its owning loop rather than leaked. I also ran the new test file against the merge-base — 3 of the 8 genuinely fail pre-fix, so these are real regression tests rather than mirrors of the implementation. Full suite on the branch is green for me locally (1767 passed, 4 skipped) and ruff is clean. Nice touch that stop() now clears the cache even on the cross-loop path — before, the failed stop left the contents behind.
The one thing I'd like sorted before this closes #623: the same foreign-loop task.cancel(); await task pattern still lives in AsyncExecutionManager.stop() (async_execution_manager.py:337-340) and ConnectionManager.close() (connection_manager.py:99-102). I verified on this branch that starting the manager on one loop and stopping it from another still raises the identical RuntimeError at async_execution_manager.py:339 — before result_cache.stop() is ever reached. Since client.aclose() goes through manager.stop(), the end-to-end sync/async mixing case from #620/#623 still fails one level up. Happy with either extending the same loop-aware treatment to those two sites in this PR or a follow-up, but either way I'd change "Fixes #623" to "Partially addresses #623" so the issue doesn't auto-close on merge.
| # Cross-loop stop: cancel on the owning loop, never await here. | ||
| self._discard_stale_cleanup_task() | ||
|
|
||
| self._cleanup_task = None |
There was a problem hiding this comment.
One residual race here: stop() snapshots these fields at the top, awaits the task in between, and then unconditionally nulls all three — with no lock on either side. A start() racing in from another thread (or even on the same loop during the await task) can install a fresh task that gets clobbered here and leaks, still running on its loop. The serialized cross-loop case — the #623 scenario — is handled, and the old code was just as unguarded, so not blocking. But since the instance already has self._lock, it'd be cheap to take it around the snapshot and around this mutation block (never across the await).
…ool (Agent-Field#623) Addresses review feedback on Agent-Field#799: the same foreign-loop 'task.cancel(); await task' hazard that affected ResultCache also lived in AsyncExecutionManager.stop() and http_connection_manager ConnectionManager.close(). Since client.aclose() flows through manager.stop() -> connection_manager.close(), the end-to-end sync/async mixing case (Agent-Field#620/Agent-Field#623) still raised 'got Future attached to a different loop' one level up, before result_cache.stop() was reached. Changes: - New agentfield/async_lifecycle.py with shared loop-aware teardown helpers: current_running_loop(), cancel_task_cross_loop(), cancel_and_await_if_same_loop(). Single source of truth for the safe pattern. - ResultCache refactored to use the shared helpers (behaviour unchanged). - AsyncExecutionManager records its owning loop at start(); stop() only awaits background tasks / sets the shutdown Event on that loop, and cancels cross-loop without awaiting otherwise. - http_connection_manager ConnectionManager records its owning loop at start(); close() takes the async lock + awaits the session only on the owning loop, and on a foreign loop schedules session/connector close on the owning loop via call_soon_threadsafe without awaiting. Tests: new tests/test_async_lifecycle_deadlock.py covers cross-loop teardown of both components (3 of its 4 checks fail pre-fix on the merge-base). Full result_cache + manager + connection suite green (110 passed); result_cache.py coverage 98%.
|
Good catch you're right that the same foreign-loop What changed in the latest push:
I verified the end-to-end path: starting the manager on a background thread's loop and stopping it from another no longer raises at async_execution_manager.py:339, and New Since the whole |
|
@santoshkumarradha pls review |
ruff 0.16.0 was released between CI runs and changed default-enabled rules, causing 2730 new lint findings across the entire sdk/python codebase. Pin to the last known-good version (0.15.22) that all existing code passes clean against. The upgrade to 0.16.0 can be done deliberately in a separate PR that addresses its new defaults.
|
@AbirAbbas review |
AbirAbbas
left a comment
AbirAbbas
left a comment
There was a problem hiding this comment.
Went through the latest revision and hand-verified everything against the diff (review tooling first, then manually). The core loop-aware teardown is right — I couldn't break the main paths, and the earlier repro stays fixed. What's left is edge-case hardening: two items are squarely in scope for this PR and cheap (the execution-lock note and the ResultCache stop/start race, inline below); the other two are fine as follow-ups if you'd rather keep this focused.
| await task | ||
| except asyncio.CancelledError: | ||
| pass | ||
| except RuntimeError: |
There was a problem hiding this comment.
This catches more than the loop-association error: a RuntimeError raised by the task's own cancellation cleanup (an __aexit__, say) is swallowed at debug level, where the old code only suppressed CancelledError. Worth matching on the message (attached to a different loop) or at least logging at warning so a failed teardown isn't invisible.
| self._loop = None | ||
|
|
||
| # Cancel all active executions | ||
| async with self._execution_lock: |
There was a problem hiding this comment.
The task cancellation above is loop-aware now, but this async with self._execution_lock still runs on whatever loop called stop(). Uncontended it happens to work, but if the owner loop is holding the lock the waiter future binds to the foreign loop and we're back to got Future attached to a different loop — one line below the fix. The cross-loop path needs to skip or marshal this section, like the connection manager does.
| self._loop = None | ||
| logger.info("ConnectionManager closed") | ||
|
|
||
| def _close_cross_loop(self, owning_loop: asyncio.AbstractEventLoop) -> None: |
There was a problem hiding this comment.
Two notes on this path, both fine as follow-ups: (1) it mutates _closed/_session/_connector from the foreign thread without the lock, so an in-flight get_session() on the owner loop can read _session as None mid-request and die with AttributeError — marshaling the whole teardown to the owning loop would close that window; (2) when the owning loop is already closed (the common sequential asyncio.run() case) the session/connector are silently dropped and aiohttp's unclosed-session warnings fire anyway, while this logs closed (cross-loop) at info and buries the detail at debug. A warning there would make the leak visible.
| shutdown_event.set() | ||
| await cancel_and_await_if_same_loop(task, owning_loop) | ||
|
|
||
| self._cleanup_task = None |
There was a problem hiding this comment.
These clears run unconditionally after an await, so a concurrent start() on another loop that installs a fresh cleanup task in that window gets its references nulled without the new task being cancelled — orphan cleanup loop, cache looks stopped. Compare-before-clear (if self._cleanup_task is task: ..., same for the event and loop) closes it.
|
Heads up: main just picked up a ruff pin in |
…op lock (Agent-Field#623) 1. ResultCache.stop(): compare-before-clear so a concurrent start() that installs fresh references between the await and the cleanup doesn't get its new task orphaned. 2. AsyncExecutionManager.stop(): skip the _execution_lock section on cross-loop stop — the lock is bound to the owning loop, taking it from a foreign loop raises the same 'got Future attached to a different loop' error one line below the fix.
|
Thanks for the thorough review @AbirAbbas! Just pushed the two in-scope fixes:
For the other two (RuntimeError catch breadth in async_lifecycle.py and the connection manager's lockless mutation on cross-loop close), I'll address those in a follow-up once this lands. They're on my radar. Also already resolved the merge conflict from the ruff pin in the previous push. |
Summary
ResultCachemixed athreading.RLockwith loop-bound asyncio primitives (anasyncio.Eventfor shutdown and anasyncio.Taskfor the cleanup loop). Whenstart()andstop()ran on different event loops which happens when the client's sync and async execution paths are mixed (#620)stop()raisedgot Future attached to a different loopand could wedge the process. This makes the cache lifecycle loop-aware so it never awaits a task from the wrong loop.Type of change
Test plan
cd sdk/python && python -m pytest tests/test_result_cache_deadlock.py -vcd sdk/python && python -m pytest tests/test_result_cache.py tests/test_result_cache_bigfiles_coverage.pyRuntimeError: got Future attached to a different loop) and confirmed cross-loopstop()now completes cleanly with no lingering pending-task warnings.Test coverage
coverage-baseline.jsonin this PR only if the removal caused a legitimate regression and I called it out in the summary above.result_cache.pycoverage went from 88% to 95%; the sdk-python aggregate stays at 94% (baseline 93.73%). Newtests/test_result_cache_deadlock.pycovers cross-loop stop, idempotent/rebinding start, concurrent sync access during cleanup, stop-without-start, and the disabled-cache no-op path.Checklist
Related issues / PRs
Fixes #623
Related to #620 (mixing threads and async code)