[Bugfix][KV Offload][P2P] Stop DEALER monitor before closing its reader

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

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