[Bugfix][KV Offload][P2P] Never block in ZmqConnection.send()

tests/v1/kv_offload/tiering/p2p/test_zmq_transport.py

@@ -10,6 +10,7 @@
 import pytest
 import zmq
 
+from vllm.v1.kv_offload.tiering.p2p.control import zmq as zmq_module
 from vllm.v1.kv_offload.tiering.p2p.control.zmq import (
     ZmqConnection,
     ZmqTransport,
@@ -230,6 +231,21 @@ def test_close_is_idempotent(self):
         transport.close()
         transport.close()  # should not raise
 
+    def test_send_raises_instead_of_blocking_when_queue_full(self, monkeypatch):
+        """A peer that never reads fills the queue; send() raises, not blocks."""
+        monkeypatch.setattr(zmq_module, "_SNDHWM", 2)
+        transport, _ = _make_transport()
+        try:
+            # Nothing listens on this port, so queued messages never drain.
+            conn = transport.connect(f"127.0.0.1:{_free_port()}")
+            start = time.monotonic()
+            with pytest.raises(ConnectionError, match="send queue"):
+                for _ in range(10):
+                    conn.send({"type": "test"})
+            assert time.monotonic() - start < 1.0
+        finally:
+            transport.close()
+
 
 class TestZmqReconnect:
     """Reconnecting to a peer whose connection died (real ZMQ sockets).

vllm/v1/kv_offload/tiering/p2p/control/zmq.py

@@ -27,6 +27,16 @@
 _HEARTBEAT_TIMEOUT_MS = 10000
 _HEARTBEAT_TTL_MS = 10000
 
+# Messages a DEALER queues for its peer before send() raises. Control messages
+# are small, so a full queue means the peer has stopped reading, not a burst.
+# Set explicitly so the fail-fast bound does not depend on libzmq's default.
+_SNDHWM = 1000
+# How long a closed DEALER keeps flushing queued messages, such as the
+# best-effort DisconnectMsg. Bounded: with the default (-1), messages queued
+# for an unreachable peer make Context.destroy() hang forever, since its
+# linger only applies to sockets that are still open.
+_DEALER_LINGER_MS = 500
+
 # Events that indicate a fatal connection failure — the peer will never
 # become reachable through this socket.
 _FATAL_MONITOR_EVENTS: frozenset[int] = frozenset()  # populated after import
@@ -86,13 +96,23 @@ def __init__(self, peer_id: str, sockets: _Sockets) -> None:
         self._inbox: list[dict] = []
 
     def send(self, msg: dict) -> None:
-        """Send a msgpack-encoded message to this peer."""
+        """Send a msgpack-encoded message to this peer.
+
+        Never blocks: callers hold the tiering manager lock. Raises
+        ConnectionError once _SNDHWM messages are queued for the peer, so the
+        owning session marks the connection dead instead of stalling.
+        """
         if not self.alive:
             raise RuntimeError(
                 f"ZmqConnection: send on closed connection to {self.peer_id}"
             )
         data = msgspec.msgpack.encode(msg)
-        self._sockets.dealer.send(data)
+        try:
+            self._sockets.dealer.send(data, zmq.NOBLOCK)
+        except zmq.Again as exc:
+            raise ConnectionError(
+                f"ZmqConnection: send queue to {self.peer_id} is full"
+            ) from exc
 
     def recv(self) -> Sequence[dict]:
         """Drain and return all buffered incoming messages."""
@@ -263,6 +283,10 @@ def _open_connection(
         dealer = self._zmq_ctx.socket(zmq.DEALER)
         _apply_heartbeat(dealer)
         dealer.identity = self._local_id.encode()
+        # ZMQ_IMMEDIATE stays 0: messages queue before the TCP connect
+        # completes, so the ConnectMsg sent right after connect() fits.
+        dealer.setsockopt(zmq.SNDHWM, _SNDHWM)
+        dealer.setsockopt(zmq.LINGER, _DEALER_LINGER_MS)
 
         # Unique per connection, not per peer: libzmq releases an inproc
         # endpoint on its reaper thread after the DEALER's close() has already