[3008.x] Fix nested SyncWrapper deadlock in tcp.PublishServer.publish (#69986) - #69992
Open
dwoz wants to merge 3 commits into
Open
[3008.x] Fix nested SyncWrapper deadlock in tcp.PublishServer.publish (#69986)#69992dwoz wants to merge 3 commits into
dwoz wants to merge 3 commits into
Conversation
When ``PublishServer.publish`` was invoked from a running io_loop (e.g. via ``MWorker._return -> store_job -> fire_event``), the outer ``SaltEvent.pusher`` SyncWrapper's worker thread ran this coroutine, then ``self.pub_sock.send`` invoked SyncWrapper *again* -- it detected the inner thread's running io_loop, spawned yet another thread, and both deadlocked on ``threading.Thread.join()``. Detect the async context via ``asyncio.get_running_loop()`` and bypass the outer SyncWrapper entirely by using a raw ``_TCPPubServerPublisher`` cached per running loop. Use a ``WeakKeyDictionary`` keyed on the loop object so a fresh SyncWrapper asyncio_loop can't inherit a dead publisher via id() recycling. Invalidate the cache both proactively (pre-flight ``stream.closed()``) and reactively (retry once on ``StreamClosedError``) so a downed puller doesn't poison the cache forever. Fixes saltstack#69986
2 tasks
The previous revision cached one ``_TCPPubServerPublisher`` per running loop in a ``WeakKeyDictionary`` to avoid the per-call connect+close overhead. But ``_TCPPubServerPublisher`` (via ``self.io_loop``) and its underlying ``tornado.iostream.IOStream`` (via ``BaseIOStream.__init__``) both hold strong references back to the loop, so the WeakKeyDictionary key was pinned live by its own value: the entry never expired, the publisher never dropped, and its socket FD stayed open for the lifetime of the ``PublishServer``. Under load (rapid ``event.fire`` on the local event bus, each routed through a fresh ``SaltEvent.pusher`` SyncWrapper with a transient asyncio loop) this leaked one AF_UNIX/loopback-TCP FD per SaltEvent instance. It was flagged as a hard regression by ``tests/pytests/scenarios/regression/test_fd_leak_*`` -- every one of them failed on this branch even though the 3008.x nightly at the same base commit passed. Drop the cache entirely. Build a fresh publisher per call, close it in ``finally`` so the FD is released before the coroutine returns. The per-call connect+close cost is trivial (loopback / AF_UNIX) and this path is only taken to work around the nested SyncWrapper deadlock, not on the hot pub-to-minion path. Since each call now has its own dedicated connection the per-loop ``asyncio.Lock`` also drops away: length-prefixed frames from concurrent callers can no longer interleave on a shared stream. Refs saltstack#69986
dwoz
added a commit
that referenced
this pull request
Aug 12, 2026
When ``TCPPublishServer.publish`` was invoked from a running asyncio loop (e.g. via ``MWorker._return -> store_job -> fire_event``), the outer ``SaltEvent.pusher`` SyncWrapper's worker thread ran this coroutine, then ``self.pub_sock.send`` invoked SyncWrapper *again* -- it detected the inner thread's running io_loop, spawned yet another thread, and both deadlocked on ``threading.Thread.join()``. Detect the async context via ``asyncio.get_running_loop()`` and bypass the outer SyncWrapper. Since 3006.x's ``publish`` is sync (no ``async def``), dispatch to ``loop.create_task(...)`` as a fire-and-forget (matches the ``fire_event`` / ``spawn_callback`` precedent in ``salt/utils/event.py``). Cache a raw ``IPCMessageClient`` per running loop via ``WeakKeyDictionary`` so a fresh SyncWrapper asyncio_loop can't inherit a dead client via id() recycling. Invalidate proactively (pre-flight ``stream.closed()``) and reactively (retry once on ``salt.ext.tornado.iostream.StreamClosedError``). 3006.x-specific counterpart to 3008.x PR #69992. Fixes #69986
The previous revision (`10e1c5ba130`, 'Don't cache per-loop publisher; close it per publish call') is fundamentally broken under CI load: * `asyncio.get_running_loop()` inside `async def publish` **always** succeeds, so `in_async` is always True and the fresh-connect branch runs for every event. The `if not self.pub_sock:` fallback is dead code. * Every publish does `_TCPPubServerPublisher(...)` -> `await connect()` -> `await send(payload)` -> `close()`. Under `PubServerChannel.publish` (the minion job dispatch path) this saturates the publish daemon's puller accept backlog / kernel `SOMAXCONN`, drops publishes on the floor as `StreamClosedError`, and every downstream request comes back to the CLI as `Some exception handling minion payload` -- pillar responses become bare strings, JSON output unparseable, deltaproxy times out past 320s. Pattern B (documented in agents/reports/PR69992_DIAGNOSIS.md): * Spin up a dedicated background thread + asyncio loop owned by PublishServer. * That thread owns ONE persistent `_TCPPubServerPublisher` connection to the puller. * Callers marshal their publish onto the bg loop via `asyncio.run_coroutine_threadsafe` and await the returned future cross-thread. Fixes all four failure modes at once: * Nested SyncWrapper deadlock -- we don't use SyncWrapper. * Connect storm -- one long-lived connection reused across every publish. * FD leak -- one publisher tied to one long-lived loop, closed in `close()`. * Loop-mismatch hazards -- no cross-loop `asyncio.Lock` / `WeakKeyDictionary`. Safety: * Bounded `ready.wait(timeout=5)` on bg-thread setup; on timeout or exception the process marks `_bg_failed` and `publish` falls back to the legacy sync path rather than hanging. * `_bg_send` handles `StreamClosedError` by rebuilding once and retrying. * `close()` stops the bg loop and joins the thread with a 5s timeout so the process doesn't hang on shutdown. Refs saltstack#69986
Contributor
Author
|
Replaced the fresh-per-call approach ( Why: the previous revision saturated the publish daemon's puller because Verified locally (all pass on
|
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
PublishServer.publishwhen in async context_TCPPubServerPublisherper running loop viaWeakKeyDictionaryStreamClosedErrorboth proactively and reactivelyFixes #69986
Test plan
salt.transport.tcpunit tests passfire_eventload without wedging (validated in local 10-min stress bench)Related PRs
dwoz/fix/bug1-syncwrapper-3006.x(pending, needs review before opening due to substantial adaptation)