[Performance][PD] Parallelize KV cache receive via ThreadPoolExecutor (vllm-project#10548)

### What this PR does / why we need it?

## Summary

This PR improves KV cache receive throughput in PD disaggregation by
running independent receive work concurrently while preserving FIFO
ordering for the same remote peer.

The original `KVCacheRecvingThread` processed every transfer task
sequentially:

```python
request_data = self.request_queue.get()
self._handle_request(request_data)
```

That makes unrelated transfers block behind each other. Under concurrent
PD traffic, tasks from different prefill ranks can queue behind a slow
transfer even though they target independent remote peers and can safely
progress in parallel.

The final implementation dispatches through a per-peer FIFO scheduler:

- requests for the same `(remote_host, remote_handshake_port)` are
serialized;
- requests for different peers can run concurrently in the existing
`ThreadPoolExecutor(max_workers=32)`;
- a busy peer yields after a small batch so it cannot monopolize
executor workers;
- request completion accounting waits for all submitted transfer tasks
before marking the request done.

## Main changes

1. **Parallel KV receive with per-peer FIFO**

`KVCacheRecvingThread.run()` now records a request as submitted and
queues it by remote peer. Each active peer has at most one handler
running, so REQ/REP ordering for that peer is preserved while
independent peers can transfer concurrently.

2. **Bounded peer handler batch**

`MAX_REQUESTS_PER_PEER_HANDLER = 5` prevents one busy peer from holding
an executor worker forever when other peers are already waiting.

3. **Thread-safe shared state**

- removes the shared `zmq.Poller` receive path and uses per-socket
`RCVTIMEO`;
   - protects `proc_not_transfer_request` with a lock;
- closes and discards failed REQ sockets instead of returning invalid
sockets to the pool;
- protects remote metadata cache population/readout with
`remote_metadata_lock`;
- tracks pending transfer tasks per request before calling
`task_tracker.update_done_task_count()`.

## Latest E2E validation

Test date: 2026-06-29.

Setup:

- 4-node DeepSeek V4 PD deployment, 32 NPUs total
- Prefill: DP4TP8
- Decode: TP4DP8
- Requests sent from `dev0` to `localhost:9000`; proxy forwards to
prefill/decode
- Workload: 500 requests, max concurrency 48, input length 256
- Baseline: PR changes fully reverted to merge-base for comparison
- Current: this PR head
- Timing data was collected with temporary measurement scaffolding
during validation; that scaffolding is not included in the final patch.

### Request-level result

| Metric | Baseline | Current | Change |
|---|---:|---:|---:|
| OK | 500/500 | 500/500 | no errors |
| Total time | 12.6s | 12.5s | -0.8% |
| Request p50 | 0.96s | 0.97s | roughly flat |
| Request p99 | 3.44s | 3.10s | -9.9% |

### KV receive timing

Aggregate across decode nodes, 2000 KV receive tasks per run:

| Metric | Baseline | Current | Change |
|---|---:|---:|---:|
| queue p50 | 18.14ms | 0.70ms | improved |
| queue p90 | 237.79ms | 1.87ms | improved |
| queue p99 | 603.63ms | 2.64ms | -99.56% |
| lifecycle p99 | 735.16ms | 512.80ms | -30.25% |
| active handler max | 1 | 7 | parallel receive active |

Per-node queue p99:

| Decode node | Baseline | Current | Change |
|---|---:|---:|---:|
| dev2 | 602.23ms | 4.11ms | -99.32% |
| dev3 | 595.88ms | 2.41ms | -99.60% |

The queue tail result is consistent with the earlier validation data:
queue p99 drops from hundreds of milliseconds to single-digit
milliseconds, showing that the tail was caused by serialized receive
handling.

## How was this patch tested?

- `pre-commit run ruff-check --all-files`
- `pre-commit run ruff-format --all-files`
- `git diff --check`
- `python3 -m py_compile
vllm_ascend/distributed/kv_transfer/kv_p2p/mooncake_connector.py
vllm_ascend/distributed/kv_transfer/kv_p2p/mooncake_hybrid_connector.py`
- `python3 -m pytest tests/ut/kv_offload/test_mooncake_connector.py
tests/ut/kv_offload/test_mooncake_hybrid_connector.py -q`
  - Result: 86 passed, 16 warnings
- Full 4-node E2E restart and proxy benchmark:
  - baseline: 500/500 OK
  - current: 500/500 OK
- Log scan after E2E found no `ZMQError`, `EFSM`, `Traceback`, duplicate
done signal, request loss, or address conflict errors.

### Does this PR introduce any user-facing change?

No. This changes internal KV receive scheduling and synchronization
only.

- vLLM version: v0.23.0
- vLLM main:
vllm-project/vllm@dc68bd8

---------

Signed-off-by: Zheng Shoujian <zheng.shoujian@outlook.com>

Author: jianzs

tests/ut/kv_offload/test_mooncake_connector.py

@@ -58,6 +58,7 @@
 patch("vllm.distributed.parallel_state._DCP", _mock_dcp_group).start()
 
 from vllm_ascend.distributed.kv_transfer.kv_p2p.mooncake_connector import (  # noqa: E402
+    MAX_REQUESTS_PER_PEER_HANDLER,
     GroupPull,
     KVCacheRecvingThread,
     KVCacheSendingThread,
@@ -510,6 +511,99 @@ def test_get_finished_requests(self, mock_tracker):
         result = self.thread.get_and_clear_finished_requests()
         self.assertEqual(result, {"req1", "req2"})
 
+    def test_submit_request_serializes_same_peer_fifo(self):
+        release_first_request = threading.Event()
+        first_request_started = threading.Event()
+        other_peer_started = threading.Event()
+        handled_requests: list[str] = []
+        active_by_peer: defaultdict[tuple[str, int], int] = defaultdict(int)
+        max_active_by_peer: defaultdict[tuple[str, int], int] = defaultdict(int)
+        state_lock = threading.Lock()
+
+        def handle_request(req_meta: dict[str, Any]):
+            peer_key = (req_meta["remote_host"], req_meta["remote_handshake_port"])
+            with state_lock:
+                active_by_peer[peer_key] += 1
+                max_active_by_peer[peer_key] = max(max_active_by_peer[peer_key], active_by_peer[peer_key])
+                handled_requests.append(req_meta["request_id"])
+
+            if req_meta["request_id"] == "same-peer-1":
+                first_request_started.set()
+                self.assertTrue(release_first_request.wait(timeout=2.0))
+            elif req_meta["request_id"] == "other-peer-1":
+                other_peer_started.set()
+
+            time.sleep(0.01)
+            with state_lock:
+                active_by_peer[peer_key] -= 1
+
+        self.thread._handle_request = handle_request  # type: ignore[method-assign]
+        same_peer_1 = {
+            "request_id": "same-peer-1",
+            "remote_host": "host-a",
+            "remote_handshake_port": 6000,
+            "all_task_done": False,
+        }
+        same_peer_2 = {
+            "request_id": "same-peer-2",
+            "remote_host": "host-a",
+            "remote_handshake_port": 6000,
+            "all_task_done": True,
+        }
+        other_peer = {
+            "request_id": "other-peer-1",
+            "remote_host": "host-b",
+            "remote_handshake_port": 6001,
+            "all_task_done": True,
+        }
+
+        try:
+            self.thread._submit_request(same_peer_1)
+            self.assertTrue(first_request_started.wait(timeout=1.0))
+            self.thread._submit_request(same_peer_2)
+            self.thread._submit_request(other_peer)
+
+            self.assertTrue(other_peer_started.wait(timeout=1.0))
+            time.sleep(0.05)
+            self.assertNotIn("same-peer-2", handled_requests)
+        finally:
+            release_first_request.set()
+            self.thread.executor.shutdown(wait=True, cancel_futures=True)
+
+        self.assertLess(handled_requests.index("same-peer-1"), handled_requests.index("same-peer-2"))
+        self.assertEqual(max_active_by_peer[("host-a", 6000)], 1)
+        self.assertEqual(max_active_by_peer[("host-b", 6001)], 1)
+
+    def test_peer_handler_yields_after_batch_limit(self):
+        peer_key = ("host-a", 6000)
+        requests = [
+            {
+                "request_id": f"req-{idx}",
+                "remote_host": peer_key[0],
+                "remote_handshake_port": peer_key[1],
+            }
+            for idx in range(MAX_REQUESTS_PER_PEER_HANDLER + 1)
+        ]
+        handled_requests: list[str] = []
+        self.thread.peer_request_queues[peer_key].extend(requests)
+        self.thread.active_peer_request_handlers.add(peer_key)
+        self.thread.executor = MagicMock()
+
+        def handle_request(req_meta: dict[str, Any]):
+            handled_requests.append(req_meta["request_id"])
+
+        self.thread._handle_request = handle_request  # type: ignore[method-assign]
+
+        self.thread._handle_peer_requests(peer_key)
+
+        self.assertEqual(handled_requests, [f"req-{idx}" for idx in range(MAX_REQUESTS_PER_PEER_HANDLER)])
+        self.assertEqual(
+            [req["request_id"] for req in self.thread.peer_request_queues[peer_key]],
+            [f"req-{MAX_REQUESTS_PER_PEER_HANDLER}"],
+        )
+        self.assertIn(peer_key, self.thread.active_peer_request_handlers)
+        self.thread.executor.submit.assert_called_once_with(self.thread._handle_peer_requests, peer_key)
+
 
 class TestSocketManagement(unittest.TestCase):
     def setUp(self):
@@ -534,7 +628,6 @@ def setUp(self):
             prefill_pp_layer_partition=None,
         )
         self.thread.remote_sockets = defaultdict(deque)
-        self.thread.remote_poller = MagicMock()
 
     @patch("vllm_ascend.distributed.kv_transfer.kv_p2p.mooncake_connector.zmq.Context")
     @patch("vllm_ascend.distributed.kv_transfer.kv_p2p.mooncake_connector.make_zmq_socket")
@@ -552,7 +645,8 @@ def test_get_remote_socket(self, mock_make_socket, mock_context):
         self.assertEqual(kwargs.get("path"), "tcp://test_host:12345")
         self.assertEqual(kwargs.get("socket_type"), zmq.REQ)  # type: ignore
         self.assertFalse(kwargs.get("bind", True))
-        self.thread.remote_poller.register.assert_called_with(mock_sock, zmq.POLLIN)  # type: ignore
+        mock_sock.setsockopt.assert_any_call(zmq.SNDTIMEO, int(self.thread.timeout * 1000))  # type: ignore
+        mock_sock.setsockopt.assert_any_call(zmq.RCVTIMEO, int(self.thread.timeout * 1000))  # type: ignore
 
     def test_return_socket_to_pool(self):
         mock_sock = MagicMock()
@@ -564,7 +658,6 @@ def test_return_socket_to_pool(self):
 
         self.assertEqual(len(self.thread.remote_sockets[test_path]), 1)
         self.assertEqual(self.thread.remote_sockets[test_path][0], mock_sock)
-        self.thread.remote_poller.register.assert_not_called()
 
 
 class TestCoreFunctionality(unittest.TestCase):
@@ -798,15 +891,15 @@ def test_get_remote_metadata_success(self, mock_recv, mock_send):
             patch.object(self.thread, "_get_remote_socket") as mock_get_socket,
             patch.object(self.thread, "_return_remote_socket") as mock_return_socket,
         ):
-            mock_socket = MagicMock()
+            mock_socket = MagicMock(spec=zmq.Socket)  # type: ignore[attr-defined]
             mock_get_socket.return_value = mock_socket
 
             self.thread._get_remote_metadata("host1", 5555)
 
             mock_get_socket.assert_called_once_with("host1", 5555)
             mock_return_socket.assert_called_once_with(mock_socket, "host1", 5555)
             mock_send.assert_called_once_with(mock_socket, self.thread.encoder.encode((GET_META_MSG, "")), "host1:5555")
-            mock_recv.assert_called_once_with(mock_socket, self.thread.remote_poller, "host1:5555")
+            mock_recv.assert_called_once_with(mock_socket, "host1:5555")
             self.assertEqual(self.thread.kv_caches_base_addr["remote_engine"][5555], [[0x3000], [0x4000]])
             self.assertEqual(self.thread.remote_block_stride_per_addr["remote_engine"][5555], [[1024]])
 
@@ -820,14 +913,15 @@ def test_get_remote_metadata_failure(self, mock_recv, mock_send):
             patch.object(self.thread, "_get_remote_socket") as mock_get_socket,
             patch.object(self.thread, "_return_remote_socket") as mock_return_socket,
         ):
-            mock_socket = MagicMock()
+            mock_socket = MagicMock(spec=zmq.Socket)  # type: ignore[attr-defined]
             mock_get_socket.return_value = mock_socket
 
             with self.assertRaises(Exception) as context:
                 self.thread._get_remote_metadata("host1", 5555)
 
             self.assertEqual(str(context.exception), "Network error")
-            mock_return_socket.assert_called_once()
+            mock_socket.close.assert_called_once()
+            mock_return_socket.assert_not_called()
 
 
 class TestMainThreadLoop(unittest.TestCase):
@@ -1335,7 +1429,7 @@ def test_request_finished_no_remote_decode(self):
 
     def test_get_transfer_block_ids_trims_attention_mtp_blocks(self):
         self.scheduler.group_transfer_info = [
-            types.SimpleNamespace(
+            types.SimpleNamespace(  # type: ignore[list-item]
                 tokens_per_block=16,
                 blocks_per_window=0,
                 is_state_group=False,
@@ -1348,7 +1442,7 @@ def test_get_transfer_block_ids_trims_attention_mtp_blocks(self):
 
     def test_get_transfer_block_ids_keeps_state_group(self):
         self.scheduler.group_transfer_info = [
-            types.SimpleNamespace(
+            types.SimpleNamespace(  # type: ignore[list-item]
                 tokens_per_block=16,
                 blocks_per_window=0,
                 is_state_group=True,
@@ -1361,7 +1455,7 @@ def test_get_transfer_block_ids_keeps_state_group(self):
 
     def test_get_transfer_block_ids_uses_compressed_prompt_len(self):
         self.scheduler.group_transfer_info = [
-            types.SimpleNamespace(
+            types.SimpleNamespace(  # type: ignore[list-item]
                 tokens_per_block=32,
                 blocks_per_window=0,
                 is_state_group=False,
@@ -1374,7 +1468,7 @@ def test_get_transfer_block_ids_uses_compressed_prompt_len(self):
 
     def test_get_transfer_block_ids_trims_sliding_window_mtp_blocks(self):
         self.scheduler.group_transfer_info = [
-            types.SimpleNamespace(
+            types.SimpleNamespace(  # type: ignore[list-item]
                 tokens_per_block=16,
                 blocks_per_window=3,
                 is_state_group=False,
@@ -1387,7 +1481,7 @@ def test_get_transfer_block_ids_trims_sliding_window_mtp_blocks(self):
 
     def test_get_swa_transfer_block_ids_clips_sliding_window_group(self):
         self.scheduler.group_transfer_info = [
-            types.SimpleNamespace(
+            types.SimpleNamespace(  # type: ignore[list-item]
                 tokens_per_block=16,
                 blocks_per_window=3,
                 is_state_group=False,
@@ -1400,7 +1494,7 @@ def test_get_swa_transfer_block_ids_clips_sliding_window_group(self):
 
     def test_get_swa_transfer_block_ids_drops_zero_from_sliding_window_tail(self):
         self.scheduler.group_transfer_info = [
-            types.SimpleNamespace(
+            types.SimpleNamespace(  # type: ignore[list-item]
                 tokens_per_block=16,
                 blocks_per_window=2,
                 is_state_group=False,
@@ -1413,7 +1507,7 @@ def test_get_swa_transfer_block_ids_drops_zero_from_sliding_window_tail(self):
 
     def test_transfer_block_ids_trims_mtp_before_swa_zero_filter(self):
         self.scheduler.group_transfer_info = [
-            types.SimpleNamespace(
+            types.SimpleNamespace(  # type: ignore[list-item]
                 tokens_per_block=16,
                 blocks_per_window=3,
                 is_state_group=False,
@@ -1578,20 +1672,15 @@ def test_ensure_zmq_send_retry_and_fail(self, mock_logger):
     def test_ensure_zmq_recv_success(self, mock_logger):
         mock_socket = MagicMock()
         mock_socket.recv.return_value = b"response"
-        mock_poller = MagicMock()
-        mock_poller.poll.return_value = [
-            (mock_socket, zmq.POLLIN)  # type: ignore
-        ]
-        data = ensure_zmq_recv(mock_socket, mock_poller, "tcp://localhost:1234")
+        data = ensure_zmq_recv(mock_socket, "tcp://localhost:1234")
         self.assertEqual(data, b"response")
 
     @patch("vllm_ascend.distributed.kv_transfer.kv_p2p.mooncake_connector.logger")
     def test_ensure_zmq_recv_timeout_and_fail(self, mock_logger):
         mock_socket = MagicMock()
-        mock_poller = MagicMock()
-        mock_poller.poll.return_value = []
+        mock_socket.recv.side_effect = zmq.ZMQError("Receive timeout")  # type: ignore
         with self.assertRaises(RuntimeError):
-            ensure_zmq_recv(mock_socket, mock_poller, "tcp://localhost:1234", timeout=0.01, max_retries=2)
+            ensure_zmq_recv(mock_socket, "tcp://localhost:1234", max_retries=2)
 
 
 class MockMooncakeAgentMetadata:

tests/ut/kv_offload/test_mooncake_hybrid_connector.py

@@ -1,6 +1,11 @@
 import sys
+import threading
+import time
 import types
 import unittest
+from collections import defaultdict, deque
+from concurrent.futures import ThreadPoolExecutor
+from typing import Any
 from unittest.mock import MagicMock
 
 fake_engine = types.ModuleType("mooncake.engine")
@@ -10,6 +15,8 @@
 from vllm.v1.request import RequestStatus  # noqa: E402
 
 from vllm_ascend.distributed.kv_transfer.kv_p2p.mooncake_hybrid_connector import (  # noqa: E402
+    MAX_REQUESTS_PER_PEER_HANDLER,
+    KVCacheRecvingThread,
     MooncakeConnectorScheduler,
 )
 
@@ -33,6 +40,114 @@ def __init__(
         self.output_token_ids = [101]
 
 
+class TestHybridKVCacheRecvingThreadDispatch(unittest.TestCase):
+    def _make_thread(self):
+        thread = object.__new__(KVCacheRecvingThread)
+        thread.executor = ThreadPoolExecutor(max_workers=2)
+        thread.peer_request_queues = defaultdict(deque)
+        thread.active_peer_request_handlers = set()
+        thread.peer_request_queues_lock = threading.Lock()
+        thread.request_task_counts = defaultdict(int)
+        thread.finished_request_markers = set()
+        thread.request_task_counts_lock = threading.Lock()
+        return thread
+
+    def test_submit_request_serializes_same_peer_fifo(self):
+        thread = self._make_thread()
+        release_first_request = threading.Event()
+        first_request_started = threading.Event()
+        other_peer_started = threading.Event()
+        handled_requests: list[str] = []
+        active_by_peer: defaultdict[tuple[str, int], int] = defaultdict(int)
+        max_active_by_peer: defaultdict[tuple[str, int], int] = defaultdict(int)
+        state_lock = threading.Lock()
+
+        def handle_request(req_meta: dict[str, Any]):
+            peer_key = (req_meta["remote_host"], req_meta["remote_handshake_port"])
+            with state_lock:
+                active_by_peer[peer_key] += 1
+                max_active_by_peer[peer_key] = max(max_active_by_peer[peer_key], active_by_peer[peer_key])
+                handled_requests.append(req_meta["request_id"])
+
+            if req_meta["request_id"] == "same-peer-1":
+                first_request_started.set()
+                self.assertTrue(release_first_request.wait(timeout=2.0))
+            elif req_meta["request_id"] == "other-peer-1":
+                other_peer_started.set()
+
+            time.sleep(0.01)
+            with state_lock:
+                active_by_peer[peer_key] -= 1
+
+        thread._handle_request = handle_request  # type: ignore[method-assign]
+        same_peer_1 = {
+            "request_id": "same-peer-1",
+            "remote_host": "host-a",
+            "remote_handshake_port": 6000,
+            "all_task_done": False,
+        }
+        same_peer_2 = {
+            "request_id": "same-peer-2",
+            "remote_host": "host-a",
+            "remote_handshake_port": 6000,
+            "all_task_done": True,
+        }
+        other_peer = {
+            "request_id": "other-peer-1",
+            "remote_host": "host-b",
+            "remote_handshake_port": 6001,
+            "all_task_done": True,
+        }
+
+        try:
+            thread._submit_request(same_peer_1)
+            self.assertTrue(first_request_started.wait(timeout=1.0))
+            thread._submit_request(same_peer_2)
+            thread._submit_request(other_peer)
+
+            self.assertTrue(other_peer_started.wait(timeout=1.0))
+            time.sleep(0.05)
+            self.assertNotIn("same-peer-2", handled_requests)
+        finally:
+            release_first_request.set()
+            thread.executor.shutdown(wait=True, cancel_futures=True)
+
+        self.assertLess(handled_requests.index("same-peer-1"), handled_requests.index("same-peer-2"))
+        self.assertEqual(max_active_by_peer[("host-a", 6000)], 1)
+        self.assertEqual(max_active_by_peer[("host-b", 6001)], 1)
+
+    def test_peer_handler_yields_after_batch_limit(self):
+        thread = self._make_thread()
+        peer_key = ("host-a", 6000)
+        requests = [
+            {
+                "request_id": f"req-{idx}",
+                "remote_host": peer_key[0],
+                "remote_handshake_port": peer_key[1],
+            }
+            for idx in range(MAX_REQUESTS_PER_PEER_HANDLER + 1)
+        ]
+        handled_requests: list[str] = []
+        thread.peer_request_queues[peer_key].extend(requests)
+        thread.active_peer_request_handlers.add(peer_key)
+        thread.executor = MagicMock()
+
+        def handle_request(req_meta: dict[str, Any]):
+            handled_requests.append(req_meta["request_id"])
+
+        thread._handle_request = handle_request  # type: ignore[method-assign]
+
+        thread._handle_peer_requests(peer_key)
+
+        self.assertEqual(handled_requests, [f"req-{idx}" for idx in range(MAX_REQUESTS_PER_PEER_HANDLER)])
+        self.assertEqual(
+            [req["request_id"] for req in thread.peer_request_queues[peer_key]],
+            [f"req-{MAX_REQUESTS_PER_PEER_HANDLER}"],
+        )
+        self.assertIn(peer_key, thread.active_peer_request_handlers)
+        thread.executor.submit.assert_called_once_with(thread._handle_peer_requests, peer_key)
+
+
 class TestMooncakeHybridConnectorScheduler(unittest.TestCase):
     def _make_scheduler(self):
         scheduler = object.__new__(MooncakeConnectorScheduler)