From b6c8ba4bd6870b4f321f394e3fe228a847c637d5 Mon Sep 17 00:00:00 2001 From: Maroon Ayoub Date: Mon, 26 Oct 2026 08:12:54 -0700 Subject: [PATCH] [Bugfix][KV Offload][P2P] Stop DEALER monitor before closing its reader libzmq sends monitor events with a blocking send from its only I/O thread. ZmqConnection.close() closed the monitor's reader while the DEALER could still raise events: EVENT_CLOSED when a connect still in progress is torn down (a departed pod or lost node), and handshake or reconnect events while the DEALER lingers to flush queued messages. The send then never returns and every P2P control socket in the engine stops until restart. Stop the DEALER's monitor before closing the reader. The DEALER keeps its bounded LINGER, so a final DisconnectMsg still reaches the peer. The close tests keep a wedged transport referenced: garbage collection would terminate its context, which blocks forever, so a regression hung the test run instead of failing. Signed-off-by: Maroon Ayoub --- .../tiering/p2p/test_zmq_transport.py | 99 +++++++++++++++++++ vllm/v1/kv_offload/tiering/p2p/control/zmq.py | 4 + 2 files changed, 103 insertions(+) diff --git a/tests/v1/kv_offload/tiering/p2p/test_zmq_transport.py b/tests/v1/kv_offload/tiering/p2p/test_zmq_transport.py index eebb9c01ba1e..d7bf17fb925d 100644 --- a/tests/v1/kv_offload/tiering/p2p/test_zmq_transport.py +++ b/tests/v1/kv_offload/tiering/p2p/test_zmq_transport.py @@ -4,6 +4,7 @@ from __future__ import annotations +import select import socket import time @@ -17,6 +18,11 @@ _Sockets, ) +# Transports whose libzmq I/O thread is blocked. Terminating their context +# blocks forever, and garbage collection would terminate it, so they stay +# referenced until the interpreter exits (pyzmq skips the term at exit). +_WEDGED_TRANSPORTS: list[ZmqTransport] = [] + def _free_port() -> int: """Find a free TCP port.""" @@ -353,6 +359,99 @@ def test_inbound_message_survives_dead_registration(self): transport_a.close() transport_b.close() + def test_close_with_unsent_message_keeps_transport_usable(self, monkeypatch): + """A connection closed with an unsent message does not wedge the transport. + + A closed DEALER that still holds a message keeps reconnecting to an + unresponsive peer, and its monitor events then block libzmq's I/O + thread, so no socket in the context can send or receive again. + """ + apply_heartbeat = zmq_module._apply_heartbeat + + def apply_heartbeat_fast_handshake(sock: zmq.Socket) -> None: + apply_heartbeat(sock) + sock.setsockopt(zmq.HANDSHAKE_IVL, 200) + + monkeypatch.setattr( + zmq_module, "_apply_heartbeat", apply_heartbeat_fast_handshake + ) + # Accepts TCP connections but never answers the ZMTP greeting. + silent = socket.socket() + silent.bind(("127.0.0.1", 0)) + silent.listen(8) + transport_a, port_a = _make_transport() + transport_b, _ = _make_transport() + wedged = True + try: + stale = transport_b.connect(f"127.0.0.1:{silent.getsockname()[1]}") + stale.send({"type": "stale"}) + stale.close() + # Several ZMTP handshake timeouts on the closed DEALER. + time.sleep(1.0) + + conn = transport_b.connect(f"127.0.0.1:{port_a}") + conn.send({"seq": 1}) + inbound = _wait_for_inbound(transport_a)[0] + assert _wait_for_messages(transport_a, inbound, 1) == [{"seq": 1}] + wedged = False + finally: + transport_a.close() + silent.close() + if wedged: + _WEDGED_TRANSPORTS.append(transport_b) + else: + transport_b.close() + + def test_close_while_connecting_keeps_transport_usable(self): + """Closing a connection whose TCP connect never completed keeps the + transport usable. + + A peer address with no host behind it (a departed pod, a lost node) + never answers the SYN. Tearing down that connect raises EVENT_CLOSED + on libzmq's I/O thread, and a monitor event sent after its reader is + closed blocks that thread, so no socket in the context can send or + receive again. + """ + # Linux drops SYNs to a listener whose accept queue is full, so a + # connect to it stays in progress like one to an address with no host. + full = socket.socket() + full.bind(("127.0.0.1", 0)) + full.listen(0) + addr = full.getsockname() + filler = socket.create_connection(addr, timeout=1.0) + probe = socket.socket() + probe.setblocking(False) + probe.connect_ex(addr) + _, connected, _ = select.select([], [probe], [], 0.5) + probe.close() + if connected: + filler.close() + full.close() + pytest.skip("kernel completes connects to a full accept queue") + + transport_a, port_a = _make_transport() + transport_b, _ = _make_transport() + wedged = True + try: + pending = transport_b.connect(f"127.0.0.1:{addr[1]}") + # Let the I/O thread start the TCP connect. + time.sleep(0.2) + pending.close() + + conn = transport_b.connect(f"127.0.0.1:{port_a}") + conn.send({"seq": 1}) + inbound = _wait_for_inbound(transport_a)[0] + assert _wait_for_messages(transport_a, inbound, 1) == [{"seq": 1}] + wedged = False + finally: + transport_a.close() + filler.close() + full.close() + if wedged: + _WEDGED_TRANSPORTS.append(transport_b) + else: + transport_b.close() + class TestExpandedMonitorEvents: """The monitor mask includes EVENT_DISCONNECTED and EVENT_CLOSED so diff --git a/vllm/v1/kv_offload/tiering/p2p/control/zmq.py b/vllm/v1/kv_offload/tiering/p2p/control/zmq.py index ee0fa819f3db..a64f0463844f 100644 --- a/vllm/v1/kv_offload/tiering/p2p/control/zmq.py +++ b/vllm/v1/kv_offload/tiering/p2p/control/zmq.py @@ -132,6 +132,10 @@ def close(self) -> None: self._closed = True self._dead = True logger.info("ZmqConnection: closing connection to %s", self.peer_id) + # libzmq sends monitor events with a blocking send from its I/O + # thread, so an event raised after the reader is closed (EVENT_CLOSED + # for a connect still in progress) blocks every socket in the context. + self._sockets.dealer.disable_monitor() self._sockets.monitor.close() self._sockets.dealer.close()