diff --git a/salt/channel/client.py b/salt/channel/client.py index 804fa4cd54d0..0091fe0a0251 100644 --- a/salt/channel/client.py +++ b/salt/channel/client.py @@ -408,6 +408,36 @@ def close(self): self._closing = True self.transport.close() + async def close_async(self): + """Async-aware close. + + When the caller is running on the ioloop that owns the underlying + transport, this awaits the transport's ``close_async`` (which in + turn awaits its ``_send_recv`` task's exit future) before + returning. That ordering matters: sync ``close()``'s + same-thread + loop-running fallback runs teardown immediately, + which can race the still-running ``_send_recv`` task and leave + the ``zmq.Context`` alive on refs the task holds -- the Context + is then finalized from a later ioloop callback and can wedge in + ``zmq_ctx_term()``. Awaiting ``close_async`` lets the send/recv + task drain the shutdown sentinel and release its socket + reference first, so the subsequent teardown is a clean + release-of-last-references rather than a race. + + Falls back to sync ``close()`` for transports (e.g. TCP) that + do not implement ``close_async``. Ioloop-owning callers such + as ``salt.minion``'s reconnect path should prefer this method. + """ + if self._closing: + return + log.debug("Async-closing %s instance", self.__class__.__name__) + self._closing = True + close_async = getattr(self.transport, "close_async", None) + if close_async is None: + self.transport.close() + else: + await close_async() + def __enter__(self): return self @@ -419,7 +449,7 @@ async def __aenter__(self): return self async def __aexit__(self, *_): - self.close() + await self.close_async() class AsyncPubChannel: diff --git a/salt/minion.py b/salt/minion.py index ab833c93db57..965b4b8ce71b 100644 --- a/salt/minion.py +++ b/salt/minion.py @@ -2018,7 +2018,20 @@ async def connect_master(self, failed=False): if hasattr(self.pub_channel, "close"): self.pub_channel.close() if hasattr(self, "req_channel") and self.req_channel: - self.req_channel.close() + # Wait for the underlying transport's ``_send_recv`` task to + # drain its shutdown sentinel and release the socket + # reference before we drop our reference to the channel. + # Otherwise the Context stays alive on the task's coroutine + # locals and is finalized later from a plain ioloop + # callback, where pyzmq's ``Context.__del__`` can wedge in + # ``zmq_ctx_term()``. ``close_async`` is available on + # ``AsyncReqChannel``; guard so this still works if a + # third-party channel subclass only exposes sync ``close``. + close_async = getattr(self.req_channel, "close_async", None) + if close_async is not None: + await close_async() + else: + self.req_channel.close() self.req_channel = None # Consider refactoring so that eval_master does not have a subtle side-effect on the contents of the opts array @@ -4655,7 +4668,18 @@ async def handle_event(self, package): if hasattr(self.pub_channel, "close"): self.pub_channel.close() if hasattr(self, "req_channel") and self.req_channel: - self.req_channel.close() + # See ``connect_master`` for why ``close_async`` is + # preferred here over the sync ``close``: the + # transport's send/recv task must drain before we + # drop our reference to the channel, or the + # underlying ``zmq.Context`` gets finalized from a + # later ioloop callback and can wedge in + # ``zmq_ctx_term()``. + close_async = getattr(self.req_channel, "close_async", None) + if close_async is not None: + await close_async() + else: + self.req_channel.close() self.req_channel = None # if eval_master finds a new master for us, self.connected diff --git a/salt/transport/zeromq.py b/salt/transport/zeromq.py index c0fc71881db5..c07243bae7cc 100644 --- a/salt/transport/zeromq.py +++ b/salt/transport/zeromq.py @@ -2322,6 +2322,66 @@ async def _reconnect(self): self.socket = None await self.connect() + async def close_async(self): + """Preferred close for ioloop-owning callers. + + The sync ``close()``'s same-thread + loop-running branch cannot + await the running ``_send_recv`` task (blocking would deadlock + the loop it is on), so it runs ``_sync_teardown`` immediately. + ``_send_recv`` may still be holding a reference to ``socket`` + in its coroutine locals when that teardown runs, which leaves + the ``Context`` refcount above zero. When ``_send_recv`` + eventually exits and drops its socket reference, the + ``Context`` is finalized -- from an ioloop callback, on the + loop's own thread -- and pyzmq's ``Context.__del__`` can wedge + the loop in ``zmq_ctx_term()``. + + ``close_async`` fixes the race by actually awaiting + ``_send_recv_exit_future`` before the teardown runs. Once + ``_send_recv`` has drained the shutdown sentinel and returned, + the socket / context are the only remaining references; the + explicit ``socket.close()`` + ``context.destroy(linger=1000)`` + below then release them deterministically, with the bounded + linger acting as a safety net so the destroy never blocks the + caller indefinitely. + + Callers on the ioloop thread that are about to drop the + underlying client reference (e.g. ``salt.minion`` on its + reconnect path) should use this instead of the sync + ``close()``. + """ + if self._closing: + return + self._closing = True + if hasattr(self, "_queue") and self._queue is not None: + try: + self._queue.put_nowait((None, None)) + except Exception: # pylint: disable=broad-except + pass + socket = self.socket + context = self.context + exit_future = self._send_recv_exit_future + self.socket = None + self.context = None + self._send_recv_exit_future = None + if exit_future is not None: + try: + await asyncio.wait_for(asyncio.shield(exit_future), timeout=5) + except (asyncio.TimeoutError, asyncio.CancelledError): + pass + except Exception: # pylint: disable=broad-except + log.debug("RequestClient graceful drain failed", exc_info=True) + if socket is not None: + try: + socket.close() + except Exception: # pylint: disable=broad-except + pass + if context is not None and not context.closed: + try: + context.destroy(linger=1000) + except Exception: # pylint: disable=broad-except + pass + async def send(self, load, timeout=60): """ Return a future which will be completed when the message has a response diff --git a/tests/pytests/scenarios/reauth/test_reauth.py b/tests/pytests/scenarios/reauth/test_reauth.py index 7808302bf498..02ad59576b84 100644 --- a/tests/pytests/scenarios/reauth/test_reauth.py +++ b/tests/pytests/scenarios/reauth/test_reauth.py @@ -106,6 +106,13 @@ def test_reauth(salt_cli, salt_minion, salt_master, timeout, event_listener): def test_presence_events(salt_cli, salt_minion, salt_master, event_listener): + # ``test_reauth`` above tears the minion down at end (``stop_event.set()`` + # -> ``minion_func`` finally -> ``salt_minion.terminate()``). The + # package-scoped ``salt_minion`` fixture does not auto-restart, so + # this second test in the file runs against a dead minion. Bring it + # back up before pinging. + if not salt_minion.is_running(): + salt_minion.start() # On slow runners (FIPS/Arm64 in particular) the first ping after # master+minion startup can exceed the factory-level 30 s CLI timeout # while the minion finishes auth/reauth and the presence machinery