Merge branch 'main' into fix/per-prompt-mm-features

Author: albertoperdomo2

docs/disaggregation.md

@@ -539,9 +539,30 @@ Specifies which KV transfer protocol the sidecar uses to coordinate prefill/deco
 | `mooncake` | `MooncakeConnector` | [Mooncake](https://github.com/kvcache-ai/Mooncake) KV transfer using RDMA |
 | `offloading` | `OffloadingConnector` | KV transfer over the vLLM CPU offloading tier. The decoder pulls KV from the prefiller via the `p2p` secondary tier. |
 
-With `offloading`, the sidecar dispatches prefill and decode concurrently. It injects role-keyed `kv_transfer_params`: the prefiller receives `{"decode": {"kv_request_id": <id>}}` (no peer address), and the decoder receives `{"prefill": {"kv_request_id": <id>, "remote_host": <prefiller host>, "remote_port": <p2p-connector-port>}}` so it can pull KV from the prefiller. The prefiller host comes from the `x-prefiller-host-port` header; the port is `--p2p-connector-port`.
-
-When the request also carries the `x-kv-cache-source-host-port` header (set by the EPP `p2p-source-producer` to a peer holding more cached prefix than the pod computing the prefix), the sidecar injects an additional `p2p` key so vLLM pulls that cached prefix over the P2P tier instead of recomputing it. Under disaggregation the prefiller leg carries `{"decode": {...}, "p2p": {"kv_request_id": <own id>, "remote_host": <source host>, "remote_port": <p2p-connector-port>}}` (the only supported multi-key combination); without a prefiller the decoder-only request carries `{"p2p": {...}}` alone. A malformed or disallowed source header is ignored and the request proceeds unchanged, as is any source header on a connector that cannot pull over the P2P tier: only `offloading`, or NIXLv2 with `--enable-p2p-pull`, honors it. For the pulled blocks to be servable, the source pod must offload its generated (decode-phase) KV: set `offload_prompt_only: false` in its `kv_connector_extra_config` (the default `true` offloads only prefill blocks).
+With `offloading`, the sidecar dispatches prefill and decode concurrently. It
+injects role-keyed `kv_transfer_params`, each key named for the remote party it
+describes: the prefiller receives `{"remote_decoder": {"kv_request_id": <id>}}`
+(no peer address), and the decoder receives `{"remote_prefiller":
+{"kv_request_id": <id>, "remote_host": <prefiller host>, "remote_port":
+<p2p-connector-port>}}` so it can pull KV from the prefiller. The prefiller host
+comes from the `x-prefiller-host-port` header; the port is
+`--p2p-connector-port`.
+
+When the request also carries the `x-kv-cache-source-host-port` header (set by
+the EPP `p2p-source-producer` to a peer holding more cached prefix than the pod
+computing the prefix), the sidecar injects an additional `remote_kv_source` key
+so vLLM pulls that cached prefix over the P2P tier instead of recomputing it.
+Under disaggregation the prefiller leg carries `{"remote_decoder": {...},
+"remote_kv_source": {"kv_request_id": <own id>, "remote_host": <source host>,
+"remote_port": <p2p-connector-port>}}` (the only supported multi-key
+combination); without a prefiller the decoder-only request carries
+`{"remote_kv_source": {...}}` alone. A malformed or disallowed source header is
+ignored and the request proceeds unchanged, as is any source header on a
+connector that cannot pull over the P2P tier: only `offloading`, or NIXLv2 with
+`--enable-p2p-pull`, honors it. For the pulled blocks to be servable, the source
+pod must offload its generated (decode-phase) KV: set `offload_prompt_only:
+false` in its `kv_connector_extra_config` (the default `true` offloads only
+prefill blocks).
 
 Both prefill and decode pods require the following `--kv-transfer-config`:
 
@@ -559,7 +580,28 @@ Both prefill and decode pods require the following `--kv-transfer-config`:
 
 `host` must be the pod's own IP at runtime (use the Kubernetes downward API env var `status.podIP`). `port` must match `--p2p-connector-port` (default `7777`). `cpu_bytes_to_use` controls the CPU KV offload buffer size; size it to hold the KV for the expected concurrent in-flight transfers. `OffloadingConnector` is available in vLLM nightly builds from 2026-06-30 onward (commit `bec232a`, [PR #42285](https://github.com/vllm-project/vllm/pull/42285)).
 
-**Restriction:** `--kv-connector=offloading` requires `--data-parallel-size=1`. Wide-EP pods (DP > 1) are rejected at startup: every DP rank would bind the same `POD_IP:<p2p-connector-port>`. DP-aware support is not yet implemented.
+**Data parallelism:** the P2P tier supports `--data-parallel-size` N > 1 when
+each pod is a complete DP group (the per-pod DP deployment).
+
+- vLLM gives each DP replica its own P2P listener and offload region:
+  replica `i` serves on `<p2p-connector-port>+i`, where `i` is the
+  **global** `data_parallel_index`
+  ([PR #47636](https://github.com/vllm-project/vllm/pull/47636),
+  [PR #47987](https://github.com/vllm-project/vllm/pull/47987)). Engines
+  without those changes bind every replica to the same
+  `POD_IP:<p2p-connector-port>`, and DP > 1 fails at engine startup.
+- The sidecar serves rank `r` on its own port + `r`, so the routed
+  endpoint's port names the target rank. The sidecar injects
+  `remote_port` = `--p2p-connector-port` + `r`; a port outside the rank
+  range falls back to rank 0.
+- The endpoint port encodes the pod-local rank, which matches the global
+  index only when the pod is a whole DP group. In multi-pod DP groups (for
+  example LWS wide-EP, where pod `k`'s replicas hold global indices
+  `k*N..k*N+N-1` behind the same serving ports), the global index cannot
+  be read from the port and would have to be supplied with the request, so
+  those deployments are not covered.
+- Every replica maps its own offload region, so the pod's `/dev/shm` must
+  exceed N x `cpu_bytes_to_use`.
 
 ### General Sidecar Flags
 
@@ -576,7 +618,7 @@ Both prefill and decode pods require the following `--kv-transfer-config`:
 |---|---|---|---|---|
 | `mooncake` | `--mooncake-bootstrap-port` | `MOONCAKE_BOOTSTRAP_PORT` | `8998` | Port used to query the Mooncake bootstrap endpoint on prefill pods. Corresponds to vLLM's `VLLM_MOONCAKE_BOOTSTRAP_PORT`. |
 | `sglang` | — | `SGLANG_BOOTSTRAP_PORT` | `8998` | Port used for the SGLang bootstrap endpoint on prefill pods. |
-| `offloading` | `--p2p-connector-port` | `P2P_CONNECTOR_PORT` | `7777` | Prefiller's OffloadingConnector P2P tier listening port, injected as `remote_port` on the decode leg so the decoder can pull KV. |
+| `offloading` | `--p2p-connector-port` | `P2P_CONNECTOR_PORT` | `7777` | Prefiller's OffloadingConnector P2P tier listening port (rank-0 port under data parallelism), injected as `remote_port` on the decode leg so the decoder can pull KV. |
 | `nixlv2` | `--enable-p2p-pull` | — | `false` | Declare the OffloadingConnector P2P tier available for cached-prefix pulls when the PD connector is NIXLv2, i.e. the engines run `MultiConnector(NixlConnector + OffloadingConnector)`. NIXL moves KV prefill to decode while the OffloadingConnector pulls the cached prefix named by `x-kv-cache-source-host-port`. Rejected at startup with any other connector; `offloading` provides the tier natively and needs no flag. |
 
 ---

pkg/sidecar/proxy/connector_nixlv2_p2p_test.go

@@ -103,8 +103,8 @@ var _ = Describe("NIXL Connector with P2P pull", func() {
 		// NIXL fields still drive the NixlConnector under MultiConnector.
 		Expect(kv).To(HaveKeyWithValue(requestFieldDoRemoteDecode, true))
 		Expect(kv).To(HaveKeyWithValue(requestFieldDoRemotePrefill, false))
-		// The p2p block drives the OffloadingConnector's cached-prefix pull.
-		p2p, ok := kv[requestFieldP2PParams].(map[string]any)
+		// The remote_kv_source block drives the OffloadingConnector's cached-prefix pull.
+		p2p, ok := kv[requestFieldRemoteKVSource].(map[string]any)
 		Expect(ok).To(BeTrue())
 		Expect(p2p[requestFieldKVRequestID]).ToNot(BeEmpty())
 		Expect(p2p[requestFieldRemoteHost]).To(Equal("10.9.9.9"))
@@ -120,7 +120,7 @@ var _ = Describe("NIXL Connector with P2P pull", func() {
 			routing.KVCacheSourceHeader:   kvCacheSource,
 		})
 
-		Expect(prefillKV()).ToNot(HaveKey(requestFieldP2PParams))
+		Expect(prefillKV()).ToNot(HaveKey(requestFieldRemoteKVSource))
 	})
 
 	It("does not compose a p2p pull when the source is the prefiller itself", func() {
@@ -133,7 +133,7 @@ var _ = Describe("NIXL Connector with P2P pull", func() {
 			routing.KVCacheSourceHeader:   prefillHostPort,
 		})
 
-		Expect(prefillKV()).ToNot(HaveKey(requestFieldP2PParams))
+		Expect(prefillKV()).ToNot(HaveKey(requestFieldRemoteKVSource))
 	})
 
 	// The parallel-dispatch (MoRI-IO WRITE) path builds the prefill leg in a
@@ -157,10 +157,10 @@ var _ = Describe("NIXL Connector with P2P pull", func() {
 		body, _ := io.ReadAll(resp.Body) //nolint:errcheck
 		Expect(resp.StatusCode).To(Equal(http.StatusOK), string(body))
 
-		// Prefill leg keeps the NIXL WRITE fields and gains the composed p2p block.
+		// Prefill leg keeps the NIXL WRITE fields and gains the composed remote_kv_source block.
 		pkv := kvParams(env.prefillHandler, 0)
 		Expect(pkv).To(HaveKeyWithValue(requestFieldDoRemoteDecode, true))
-		p2p, ok := pkv[requestFieldP2PParams].(map[string]any)
+		p2p, ok := pkv[requestFieldRemoteKVSource].(map[string]any)
 		Expect(ok).To(BeTrue())
 		Expect(p2p[requestFieldRemoteHost]).To(Equal("10.9.9.9"))
 		Expect(p2p[requestFieldRemotePort]).To(BeNumerically("==", p2pConnectorPort))

pkg/sidecar/proxy/connector_p2p.go

@@ -21,8 +21,11 @@ import (
 	"encoding/json"
 	"fmt"
 	"io"
+	"net"
 	"net/http"
 	"os"
+	"strconv"
+	"strings"
 	"time"
 
 	"go.opentelemetry.io/otel/attribute"
@@ -57,10 +60,11 @@ func (s *Server) handleP2P(w http.ResponseWriter, r *http.Request, prefillPodHos
 	}
 
 	kvRequestID := newUUID()
+	prefillP2PPort := s.p2pPortFor(prefillPodHostPort)
 	s.logger.Info("running P2P protocol",
 		"prefill_host", extractHost(prefillPodHostPort),
 		"kv_request_id", kvRequestID,
-		"p2p_connector_port", s.config.P2PConnectorPort)
+		"p2p_connector_port", prefillP2PPort)
 
 	// Prefill leg: store KV under kv_request_id, no peer address. Capped to a
 	// single output token so the prefiller returns as soon as KV is stored.
@@ -69,7 +73,7 @@ func (s *Server) handleP2P(w http.ResponseWriter, r *http.Request, prefillPodHos
 		prefillData[k] = v
 	}
 	prefillKVParams := map[string]any{
-		requestFieldP2PDecodeParams: map[string]any{
+		requestFieldRemoteDecoder: map[string]any{
 			requestFieldKVRequestID: kvRequestID,
 		},
 	}
@@ -95,10 +99,10 @@ func (s *Server) handleP2P(w http.ResponseWriter, r *http.Request, prefillPodHos
 		decodeData[k] = v
 	}
 	decodeData[requestFieldKVTransferParams] = map[string]any{
-		requestFieldP2PPrefillParams: map[string]any{
+		requestFieldRemotePrefiller: map[string]any{
 			requestFieldKVRequestID: kvRequestID,
 			requestFieldRemoteHost:  extractHost(prefillPodHostPort),
-			requestFieldRemotePort:  s.config.P2PConnectorPort,
+			requestFieldRemotePort:  prefillP2PPort,
 		},
 	}
 
@@ -213,38 +217,77 @@ func (s *Server) handleP2PConcurrentRequests(w http.ResponseWriter, r *http.Requ
 // KVConnector is offloading, or is composed alongside NIXL via MultiConnector
 // (declared with --enable-p2p-pull) when the PD connector is NIXLv2. On any
 // other connector --enable-p2p-pull has no effect, since no MultiConnector
-// routes the p2p params to an OffloadingConnector.
+// routes the remote_kv_source params to an OffloadingConnector.
 func (s *Server) p2pPullAvailable() bool {
 	return s.config.KVConnector == KVConnectorOffloading ||
 		(s.config.EnableP2PPull && s.config.KVConnector == KVConnectorNIXLV2)
 }
 
-// addP2PPullToPrefill adds the OffloadingConnector p2p pull block to a prefill
+// addP2PPullToPrefill adds the OffloadingConnector P2P pull block to a prefill
 // leg's kv_transfer_params so the prefiller pulls cached prefix from
 // kvCacheSource while keeping its own computed blocks available for the
 // decoder. It is a no-op when no source is set or the source resolves to the
-// prefiller itself, since there is nothing to pull from oneself. The p2p key
-// composes with NIXL params: vLLM's MultiConnector routes it to the
-// OffloadingConnector and the NIXL fields to the NixlConnector.
+// prefiller itself, since there is nothing to pull from oneself. The
+// remote_kv_source key composes with NIXL params: vLLM's MultiConnector
+// routes it to the OffloadingConnector and the NIXL fields to the
+// NixlConnector.
 func (s *Server) addP2PPullToPrefill(prefillKVParams map[string]any, kvCacheSource, prefillPodHostPort string) {
 	if kvCacheSource != "" && extractHost(kvCacheSource) != extractHost(prefillPodHostPort) {
-		prefillKVParams[requestFieldP2PParams] = s.p2pSourceParams(kvCacheSource)
+		prefillKVParams[requestFieldRemoteKVSource] = s.p2pSourceParams(kvCacheSource)
 	}
 }
 
-// p2pSourceParams builds the kv_transfer_params.p2p block for a pull from
-// sourceHostPort's OffloadingConnector P2P tier. The kv_request_id is its
+// p2pSourceParams builds the kv_transfer_params.remote_kv_source block for a
+// pull from sourceHostPort's OffloadingConnector P2P tier. The kv_request_id is its
 // own fresh UUID: in P2P mode it is consumer-side only.
 func (s *Server) p2pSourceParams(sourceHostPort string) map[string]any {
 	return map[string]any{
 		requestFieldKVRequestID: newUUID(),
 		requestFieldRemoteHost:  extractHost(sourceHostPort),
-		requestFieldRemotePort:  s.config.P2PConnectorPort,
+		requestFieldRemotePort:  s.p2pPortFor(sourceHostPort),
 	}
 }
 
+// p2pPortFor resolves the P2P tier control port on the target endpoint. The
+// sidecar serves rank r on <base port>+r (data_parallel.go), so the routed
+// endpoint's port encodes the pod-local rank. vLLM binds the tier at
+// <p2p-connector-port> + the global data_parallel_index, so the mapping is
+// correct when each pod is its own DP group (local rank == global index).
+// Multi-pod DP groups (e.g. LWS wide-EP) need the source's global rank
+// supplied per request instead; this derivation is the per-pod fallback. A
+// port outside the rank range (or unparsable) falls back to the base P2P
+// port, which is rank 0's tier.
+func (s *Server) p2pPortFor(targetHostPort string) int {
+	base := s.config.P2PConnectorPort
+	if s.config.DataParallelSize <= 1 || s.dpBasePort == 0 {
+		return base
+	}
+	// Backward compatible behavior: trim `http:` prefix (see createProxyHandler).
+	targetHostPort, _ = strings.CutPrefix(targetHostPort, "http://")
+	_, portStr, err := net.SplitHostPort(targetHostPort)
+	if err != nil {
+		s.logger.V(logging.DEBUG).Info("P2P target has no parsable port, using base P2P port",
+			"target", targetHostPort, "error", err)
+		return base
+	}
+	port, err := strconv.Atoi(portStr)
+	if err != nil {
+		s.logger.V(logging.DEBUG).Info("P2P target port is not numeric, using base P2P port",
+			"target", targetHostPort, "error", err)
+		return base
+	}
+	rank := port - s.dpBasePort
+	if rank < 0 || rank >= s.config.DataParallelSize {
+		s.logger.V(logging.DEBUG).Info("P2P target port outside the DP rank range, using base P2P port",
+			"target", targetHostPort, "dpBasePort", s.dpBasePort,
+			"dataParallelSize", s.config.DataParallelSize)
+		return base
+	}
+	return base + rank
+}
+
 // decodeWithP2PSource serves a decoder-only request through the local vLLM
-// with kv_transfer_params.p2p injected, so the engine looks up and pulls
+// with kv_transfer_params.remote_kv_source injected, so the engine looks up and pulls
 // cached prefix blocks from the peer at sourceHostPort instead of recomputing
 // them. It replaces any client-supplied kv_transfer_params (the sidecar owns
 // that field) and routes through dispatchDecode so chunked decode still
@@ -266,12 +309,12 @@ func (s *Server) decodeWithP2PSource(w http.ResponseWriter, r *http.Request, sou
 	p2pParams := s.p2pSourceParams(sourceHostPort)
 	// Rebuild kv_transfer_params from scratch: the sidecar owns this field, so
 	// client-supplied keys are dropped rather than forwarded to vLLM.
-	requestData[requestFieldKVTransferParams] = map[string]any{requestFieldP2PParams: p2pParams}
+	requestData[requestFieldKVTransferParams] = map[string]any{requestFieldRemoteKVSource: p2pParams}
 
 	s.logger.Info("running P2P source protocol",
 		"source_host", extractHost(sourceHostPort),
 		"kv_request_id", p2pParams[requestFieldKVRequestID],
-		"p2p_connector_port", s.config.P2PConnectorPort)
+		"p2p_connector_port", p2pParams[requestFieldRemotePort])
 
 	newBody, err := json.Marshal(requestData)
 	if err != nil {