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..d037b8d2c181 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,127 @@ 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: + # 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 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. 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 + ) + 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,