Merge branch 'main' into fix/2065-experimental-flow-control-env-var

Author: elinacse

docs/disaggregation.md

@@ -580,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
 
@@ -597,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_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.
@@ -98,7 +102,7 @@ func (s *Server) handleP2P(w http.ResponseWriter, r *http.Request, prefillPodHos
 		requestFieldRemotePrefiller: map[string]any{
 			requestFieldKVRequestID: kvRequestID,
 			requestFieldRemoteHost:  extractHost(prefillPodHostPort),
-			requestFieldRemotePort:  s.config.P2PConnectorPort,
+			requestFieldRemotePort:  prefillP2PPort,
 		},
 	}
 
@@ -240,10 +244,48 @@ 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.remote_kv_source injected, so the engine looks up and pulls
 // cached prefix blocks from the peer at sourceHostPort instead of recomputing
@@ -272,7 +314,7 @@ func (s *Server) decodeWithP2PSource(w http.ResponseWriter, r *http.Request, sou
 	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 {

pkg/sidecar/proxy/connector_p2p_test.go

@@ -147,3 +147,55 @@ var _ = DescribeTable("p2pPullAvailable",
 	Entry("the flag has no effect on sglang", KVConnectorSGLang, true, false),
 	Entry("the flag has no effect on shared-storage", KVConnectorSharedStorage, true, false),
 )
+
+var _ = DescribeTable("p2pPortFor",
+	func(dpSize, dpBasePort int, target string, want int) {
+		s := &Server{
+			dpBasePort: dpBasePort,
+			config:     Config{P2PConnectorPort: 7777, DataParallelSize: dpSize},
+		}
+		Expect(s.p2pPortFor(target)).To(Equal(want))
+	},
+	Entry("single DP uses the base port regardless of target", 1, 8000, "10.0.0.5:8003", 7777),
+	Entry("rank 0 target uses the base port", 4, 8000, "10.0.0.5:8000", 7777),
+	Entry("rank 2 target offsets by 2", 4, 8000, "10.0.0.5:8002", 7779),
+	Entry("last rank target offsets by dpSize-1", 4, 8000, "10.0.0.5:8003", 7780),
+	Entry("port below the base falls back to the base port", 4, 8000, "10.0.0.5:7999", 7777),
+	Entry("port beyond the rank range falls back to the base port", 4, 8000, "10.0.0.5:8004", 7777),
+	Entry("target without a port falls back to the base port", 4, 8000, "10.0.0.5", 7777),
+	Entry("unparsable port falls back to the base port", 4, 8000, "10.0.0.5:http", 7777),
+	Entry("zero base port disables derivation", 4, 0, "10.0.0.5:8002", 7777),
+	Entry("scheme-prefixed target derives the rank", 4, 8000, "http://10.0.0.5:8002", 7779),
+)
+
+var _ = Describe("p2pSourceParams", func() {
+	It("offsets remote_port by the source endpoint's DP rank", func() {
+		s := &Server{
+			dpBasePort: 8000,
+			config:     Config{P2PConnectorPort: 7777, DataParallelSize: 4},
+		}
+		params := s.p2pSourceParams("10.0.0.9:8002")
+		Expect(params[requestFieldRemoteHost]).To(Equal("10.0.0.9"))
+		Expect(params[requestFieldRemotePort]).To(Equal(7779))
+		Expect(params[requestFieldKVRequestID]).ToNot(BeEmpty())
+	})
+
+	It("keeps rank derivation on Clone, which rank servers rely on", func() {
+		s := &Server{
+			dpBasePort: 8000,
+			config:     Config{P2PConnectorPort: 7777, DataParallelSize: 4},
+		}
+		params := s.Clone().p2pSourceParams("10.0.0.9:8002")
+		Expect(params[requestFieldRemotePort]).To(Equal(7779))
+	})
+
+	It("derives both host and port from a scheme-prefixed source", func() {
+		s := &Server{
+			dpBasePort: 8000,
+			config:     Config{P2PConnectorPort: 7777, DataParallelSize: 4},
+		}
+		params := s.p2pSourceParams("http://10.0.0.9:8002")
+		Expect(params[requestFieldRemoteHost]).To(Equal("10.0.0.9"))
+		Expect(params[requestFieldRemotePort]).To(Equal(7779))
+	})
+})

pkg/sidecar/proxy/options.go

@@ -252,7 +252,7 @@ func (opts *Options) AddFlags(fs *pflag.FlagSet) {
 	fs.IntVar(&opts.MooncakeBootstrapPort, mooncakeBootstrapPortFlag, opts.MooncakeBootstrapPort,
 		"the port used to query the Mooncake bootstrap endpoint on prefill pods (only used with --kv-connector=mooncake)")
 	fs.IntVar(&opts.P2PConnectorPort, p2pConnectorPortFlag, opts.P2PConnectorPort,
-		"the prefiller's OffloadingConnector P2P tier listening port, injected as remote_port on the decode leg (used with --kv-connector=offloading or --enable-p2p-pull)")
+		"the prefiller's OffloadingConnector P2P tier listening port, injected as remote_port on the decode leg; with --data-parallel-size > 1 this is the rank-0 port and rank r uses port+r (used with --kv-connector=offloading or --enable-p2p-pull)")
 	fs.BoolVar(&opts.EnableP2PPull, enableP2PPull, opts.EnableP2PPull,
 		"declare the OffloadingConnector P2P tier available for cached-prefix pulls when the PD connector is NIXL, i.e. engines run MultiConnector(NixlConnector + OffloadingConnector). Rejected with any other --kv-connector; offloading provides the tier natively without this flag.")
 	fs.BoolVar(&opts.SecureServing, secureServing, opts.SecureServing, "Enables secure proxy. Defaults to true.")
@@ -532,11 +532,11 @@ func (opts *Options) Validate() error {
 	if opts.P2PConnectorPort < 1 || opts.P2PConnectorPort > 65535 {
 		return fmt.Errorf("--p2p-connector-port must be between 1 and 65535, got %d", opts.P2PConnectorPort)
 	}
-
-	// offloading does not support wide-EP: every DP rank would bind the same
-	// POD_IP:<p2p-connector-port>. DP-aware support is not yet implemented.
-	if opts.KVConnector == KVConnectorOffloading && opts.DataParallelSize > 1 {
-		return fmt.Errorf("--kv-connector=offloading does not support --data-parallel-size > 1 (got %d)", opts.DataParallelSize)
+	// The injected port is offset by the target's DP rank, so the highest
+	// rank's port must stay in range too.
+	if opts.DataParallelSize > 1 && opts.P2PConnectorPort+opts.DataParallelSize-1 > 65535 {
+		return fmt.Errorf("--p2p-connector-port %d plus data-parallel rank %d exceeds 65535",
+			opts.P2PConnectorPort, opts.DataParallelSize-1)
 	}
 
 	// --enable-p2p-pull composes the OffloadingConnector P2P tier alongside NIXL

pkg/sidecar/proxy/options_test.go

@@ -626,12 +626,12 @@ func TestP2PConnectorPort(t *testing.T) {
 }
 
 func TestValidateOffloadingDP(t *testing.T) {
-	t.Run("rejects offloading with data-parallel-size > 1", func(t *testing.T) {
+	t.Run("allows offloading with data-parallel-size > 1", func(t *testing.T) {
 		opts := NewOptions()
 		opts.KVConnector = KVConnectorOffloading
 		opts.DataParallelSize = 2
 		require.NoError(t, opts.Complete())
-		require.ErrorContains(t, opts.Validate(), "--kv-connector=offloading does not support --data-parallel-size > 1")
+		require.NoError(t, opts.Validate())
 	})
 
 	t.Run("allows offloading with data-parallel-size 1", func(t *testing.T) {
@@ -641,6 +641,24 @@ func TestValidateOffloadingDP(t *testing.T) {
 		require.NoError(t, opts.Complete())
 		require.NoError(t, opts.Validate())
 	})
+
+	t.Run("rejects a rank port beyond 65535", func(t *testing.T) {
+		opts := NewOptions()
+		opts.KVConnector = KVConnectorOffloading
+		opts.DataParallelSize = 4
+		opts.P2PConnectorPort = 65533
+		require.NoError(t, opts.Complete())
+		require.ErrorContains(t, opts.Validate(), "exceeds 65535")
+	})
+
+	t.Run("allows the highest rank port at 65535", func(t *testing.T) {
+		opts := NewOptions()
+		opts.KVConnector = KVConnectorOffloading
+		opts.DataParallelSize = 4
+		opts.P2PConnectorPort = 65532
+		require.NoError(t, opts.Complete())
+		require.NoError(t, opts.Validate())
+	})
 }
 
 func TestValidateEnableP2PPull(t *testing.T) {

pkg/sidecar/proxy/proxy.go

@@ -26,6 +26,7 @@ import (
 	"net/http"
 	"net/http/httputil"
 	"net/url"
+	"strconv"
 	"strings"
 	"time"
 
@@ -210,6 +211,8 @@ type Config struct {
 
 	// P2PConnectorPort is the prefiller's OffloadingConnector P2P tier listening port,
 	// injected as remote_port on the decode leg so the decoder can pull KV from it.
+	// With data parallelism it is the rank-0 port: rank r's tier listens on
+	// P2PConnectorPort+r and the injected port is offset by the target's rank.
 	// Meaningful with --kv-connector=offloading or --enable-p2p-pull.
 	P2PConnectorPort int
 
@@ -332,6 +335,11 @@ type Server struct {
 
 	prefillSamplerFn func(n int) int // allow test override
 
+	// dpBasePort is the rank-0 proxy port. Rank clones override config.Port
+	// (data_parallel.go), so rank derivation from a routed endpoint's port
+	// needs the pre-clone base. 0 disables derivation.
+	dpBasePort int
+
 	config Config
 }
 
@@ -353,6 +361,9 @@ func NewProxy(config Config) *Server {
 		forwardDataParallel: true,
 		prefillSamplerFn:    rand.IntN,
 	}
+	if basePort, err := strconv.Atoi(config.Port); err == nil {
+		server.dpBasePort = basePort
+	}
 
 	server.setKVConnector()
 	if config.UseTLSForPrefiller {
@@ -421,6 +432,7 @@ func (s *Server) Clone() *Server {
 		dataParallelProxies: s.dataParallelProxies,
 		forwardDataParallel: s.forwardDataParallel,
 		prefillSamplerFn:    s.prefillSamplerFn,
+		dpBasePort:          s.dpBasePort,
 		config:              s.config,
 	}
 }

pkg/sidecar/proxy/proxy_helpers.go

@@ -12,6 +12,7 @@ import (
 	"net/http"
 	"net/http/httputil"
 	"net/url"
+	"strings"
 	"syscall"
 	"time"
 
@@ -196,9 +197,12 @@ func cloneRequestWithBody(ctx context.Context, r *http.Request, body []byte) *ht
 	return cloned
 }
 
-// extractHost returns the host part of a host:port string. If parsing
-// fails (e.g. no port), the input is returned as-is.
+// extractHost returns the host part of a host:port string. A `http://`
+// prefix is trimmed first, matching createProxyHandler's tolerance for
+// routed endpoint values. If parsing fails (e.g. no port), the input is
+// returned as-is.
 func extractHost(hostWithPort string) string {
+	hostWithPort, _ = strings.CutPrefix(hostWithPort, "http://")
 	host, _, err := net.SplitHostPort(hostWithPort)
 	if err != nil {
 		return hostWithPort