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()