From 468f4d8f10dc029f4196c125706460039d730d45 Mon Sep 17 00:00:00 2001 From: "Daniel A. Wozniak" Date: Sun, 9 Aug 2026 15:28:38 -0700 Subject: [PATCH 1/2] Fix nested SyncWrapper deadlock in tcp.TCPPublishServer.publish (3006.x) 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 --- changelog/69986.fixed.md | 1 + salt/transport/tcp.py | 112 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 113 insertions(+) create mode 100644 changelog/69986.fixed.md diff --git a/changelog/69986.fixed.md b/changelog/69986.fixed.md new file mode 100644 index 000000000000..33aa2cb96a08 --- /dev/null +++ b/changelog/69986.fixed.md @@ -0,0 +1 @@ +Fix MWorker deadlock caused by nested ``SyncWrapper`` recursion in ``tcp.TCPPublishServer.publish``. When ``fire_event`` invoked ``publish`` from inside a running asyncio loop, the outer ``SaltEvent.pusher`` SyncWrapper's thread spawned another SyncWrapper which deadlocked on ``threading.Thread.join()``, wedging all MWorkers. On 3006.x the fix uses a fire-and-forget dispatch via ``loop.create_task`` (``publish`` remains sync on 3006.x) with a per-loop ``IPCMessageClient`` cache. diff --git a/salt/transport/tcp.py b/salt/transport/tcp.py index b5e6e518c0f1..ea10cbc0b584 100644 --- a/salt/transport/tcp.py +++ b/salt/transport/tcp.py @@ -6,6 +6,7 @@ """ +import asyncio import errno import logging import multiprocessing @@ -16,6 +17,7 @@ import urllib import uuid import warnings +import weakref import salt.ext.tornado import salt.ext.tornado.concurrent @@ -1044,6 +1046,116 @@ def publish(self, payload, **kwargs): pull_uri = int(self.opts.get("tcp_master_publish_pull", 4514)) else: pull_uri = os.path.join(self.opts["sock_dir"], "publish_pull.ipc") + + # PATCH: avoid the nested-SyncWrapper deadlock in the + # ``fire_event`` -> ``TCPPublishServer.publish`` -> ``pub_sock.send`` + # chain. ``self.pub_sock`` is a ``SyncWrapper(IPCMessageClient)``. + # When ``publish`` is invoked while an asyncio/Tornado io_loop is + # running on the current thread (as is the case in every + # ``MWorker._handle_payload`` coroutine -> ``_handle_clear`` -> + # ``_send_pub`` -> ``chan.publish`` path), the outer SyncWrapper + # around ``SaltEvent.pusher`` (or the MWorker's own io_loop) has + # already spawned a worker thread that runs this method, then + # ``self.pub_sock.send`` invokes SyncWrapper *again* -- it detects + # the running io_loop, spawns yet another thread, and the two + # threads can deadlock on ``threading.Thread.join()``. All + # MWorkers wedge, MWQ's DEALER send() blocks (queue backlog), + # minions time out and reconnect, dead-peer TCP conns pile up. + # + # Fix: when we're already in an async context, bypass the outer + # SyncWrapper entirely and schedule an ``IPCMessageClient.send`` + # coroutine directly on the running loop (fire-and-forget, which + # matches the ``fire_event`` precedent at + # ``salt/utils/event.py``: ``self.io_loop.spawn_callback( + # self.pusher.send, msg)``). Cache the raw + # ``IPCMessageClient`` per running loop because Tornado + # ``IOStream`` instances (and any locks bound to a specific loop) + # cannot be safely shared across loops -- this + # ``TCPPublishServer`` is used by both the sync-mode SyncWrapper + # thread's io_loop and any coroutine-mode io_loop. A per-loop + # ``asyncio.Lock`` serializes concurrent ``fire_event`` tasks so + # their length-prefixed frames don't interleave on the shared + # stream (framing corruption would otherwise surface as bogus + # ~GB length prefixes on the puller side). + try: + loop = asyncio.get_running_loop() + in_async = True + except RuntimeError: + in_async = False + + if in_async: + per_loop = getattr(self, "_async_pub_by_loop", None) + if per_loop is None: + # PATCH: WeakKeyDictionary so entries drop when the loop + # is GC'd. Keying on ``id(loop)`` would be unsafe -- + # CPython recycles integer ids after GC and a fresh + # ``SyncWrapper.asyncio_loop`` could land on the same id + # as a dead one and inherit that dead loop's cached + # (dead) publisher. + per_loop = self._async_pub_by_loop = weakref.WeakKeyDictionary() + + entry = per_loop.get(loop) + if entry is not None: + pub, _lock = entry + # PATCH: invalidate on a dead stream. If the puller + # side went away (slow-subscriber discard, subscriber + # process restart) the stream is closed but the entry + # is still cached -- next ``send`` raises + # ``StreamClosedError`` forever until we rebuild. A + # closed stream is unrecoverable in tornado's + # ``IOStream``; drop the entry so we reconnect below. + stream = getattr(pub, "stream", None) + if stream is None or stream.closed(): + del per_loop[loop] + entry = None + + if entry is None: + # ``IPCMessageClient`` expects a Tornado ``IOLoop`` for + # ``add_callback``/``add_future`` scheduling. In Tornado + # 6.x ``IOLoop.current()`` is a thin wrapper around the + # currently-running asyncio loop, so constructing it + # here (on the loop's own thread) binds the client to + # the correct loop. + tio_loop = salt.ext.tornado.ioloop.IOLoop.current() + pub = salt.transport.ipc.IPCMessageClient(pull_uri, io_loop=tio_loop) + lock = asyncio.Lock() + entry = (pub, lock) + per_loop[loop] = entry + pub, lock = entry + + async def _send_async(): + async with lock: + try: + if not pub.connected(): + await pub.connect() + await pub.send(payload) + except salt.ext.tornado.iostream.StreamClosedError: + # PATCH: puller closed on us mid-send. Drop the + # cached publisher and rebuild once so the next + # call can succeed. We do a single retry inside + # the lock to preserve message ordering for + # concurrent callers on this loop. + per_loop.pop(loop, None) + tio_loop = salt.ext.tornado.ioloop.IOLoop.current() + new_pub = salt.transport.ipc.IPCMessageClient( + pull_uri, io_loop=tio_loop + ) + per_loop[loop] = (new_pub, lock) + try: + await new_pub.connect() + await new_pub.send(payload) + except Exception: # pylint: disable=broad-except + log.exception("TCPPublishServer async publish retry failed") + except Exception: # pylint: disable=broad-except + log.exception("TCPPublishServer async publish failed") + + # Schedule on the running loop. Fire-and-forget matches the + # sync-branch semantics (SyncWrapper.send returns after the + # local IPC write completes; callers do not observe an ack + # from the puller side). + loop.create_task(_send_async()) + return + if not self.pub_sock: self.pub_sock = salt.utils.asynchronous.SyncWrapper( salt.transport.ipc.IPCMessageClient, From 41022746e0c59d5526a44e2f0aa42d478c5b09cd Mon Sep 17 00:00:00 2001 From: "Daniel A. Wozniak" Date: Tue, 11 Aug 2026 17:11:22 -0700 Subject: [PATCH 2/2] Address twangboy review: re-resolve active publisher inside lock Before: concurrent tasks captured pub from the outer scope before taking the per-loop lock. If task A hit StreamClosedError, replaced per_loop[loop] with a healthy pub2, and released the lock, task B then acquired the lock still holding its captured pub1, tried to send on that already-closed publisher, and evicted the healthy pub2 from the cache -- cascading unnecessary reconnects under sustained concurrent publish. Re-resolve the active publisher from per_loop inside the lock so waiting tasks pick up the newly reconnected instance. Also guard the eviction path so we only pop the cache entry when it still points at the publisher we tried; a peer task's successful replacement must not be dropped. Refs review comment on PR #69998. --- salt/transport/tcp.py | 29 ++++++++++++++++++++--------- 1 file changed, 20 insertions(+), 9 deletions(-) diff --git a/salt/transport/tcp.py b/salt/transport/tcp.py index ea10cbc0b584..d037b8d2c181 100644 --- a/salt/transport/tcp.py +++ b/salt/transport/tcp.py @@ -1125,17 +1125,28 @@ def publish(self, payload, **kwargs): async def _send_async(): async with lock: + # Re-resolve the active publisher inside the lock so + # tasks that captured a stale ``pub`` from the outer + # scope before a concurrent StreamClosedError retry + # replaced ``per_loop[loop]`` don't cascade evict + # the healthy replacement. Falls back to the + # outer-scope ``pub`` if the cache is empty (e.g. + # close() ran between capture and lock acquisition). + current_entry = per_loop.get(loop) + active_pub = current_entry[0] if current_entry else pub try: - if not pub.connected(): - await pub.connect() - await pub.send(payload) + if not active_pub.connected(): + await active_pub.connect() + await active_pub.send(payload) except salt.ext.tornado.iostream.StreamClosedError: - # PATCH: puller closed on us mid-send. Drop the - # cached publisher and rebuild once so the next - # call can succeed. We do a single retry inside - # the lock to preserve message ordering for - # concurrent callers on this loop. - per_loop.pop(loop, None) + # PATCH: puller closed on us mid-send. Only + # drop the cache entry if it still points at + # the publisher we tried -- otherwise a peer + # task already replaced it with a healthy new + # one and we must not evict that. + current_entry = per_loop.get(loop) + if current_entry is not None and current_entry[0] is active_pub: + per_loop.pop(loop, None) tio_loop = salt.ext.tornado.ioloop.IOLoop.current() new_pub = salt.transport.ipc.IPCMessageClient( pull_uri, io_loop=tio_loop