diff --git a/docs/disaggregation.md b/docs/disaggregation.md index 0a38201c72..4405bd26e2 100644 --- a/docs/disaggregation.md +++ b/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": }}` (no peer address), and the decoder receives `{"prefill": {"kv_request_id": , "remote_host": , "remote_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": , "remote_host": , "remote_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": }}` +(no peer address), and the decoder receives `{"remote_prefiller": +{"kv_request_id": , "remote_host": , "remote_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": , "remote_host": , +"remote_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`: diff --git a/pkg/sidecar/proxy/connector_nixlv2_p2p_test.go b/pkg/sidecar/proxy/connector_nixlv2_p2p_test.go index c6a2f5b43c..8381517121 100644 --- a/pkg/sidecar/proxy/connector_nixlv2_p2p_test.go +++ b/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)) diff --git a/pkg/sidecar/proxy/connector_p2p.go b/pkg/sidecar/proxy/connector_p2p.go index 9893524004..f66368bf48 100644 --- a/pkg/sidecar/proxy/connector_p2p.go +++ b/pkg/sidecar/proxy/connector_p2p.go @@ -69,7 +69,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,7 +95,7 @@ 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, @@ -213,27 +213,28 @@ 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{ @@ -244,7 +245,7 @@ func (s *Server) p2pSourceParams(sourceHostPort string) map[string]any { } // 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,7 +267,7 @@ 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), diff --git a/pkg/sidecar/proxy/connector_p2p_source_test.go b/pkg/sidecar/proxy/connector_p2p_source_test.go index 83d6e0bbed..50f42df029 100644 --- a/pkg/sidecar/proxy/connector_p2p_source_test.go +++ b/pkg/sidecar/proxy/connector_p2p_source_test.go @@ -73,7 +73,7 @@ var _ = Describe("P2P KV cache source header", func() { return sendBody(proxyBaseAddr, chatCompletionsRequestBodyWithMaxCompletionTokens, headers) } - It("should inject p2p params on the local request without disaggregation", func() { + It("should inject remote_kv_source params on the local request without disaggregation", func() { proxyBaseAddr := testInfo.startProxy() sendRequest(proxyBaseAddr, map[string]string{routing.KVCacheSourceHeader: kvCacheSource}) @@ -84,9 +84,9 @@ var _ = Describe("P2P KV cache source header", func() { kvParams, ok := dreq[requestFieldKVTransferParams].(map[string]any) Expect(ok).To(BeTrue()) - Expect(kvParams).ToNot(HaveKey(requestFieldP2PDecodeParams)) - Expect(kvParams).ToNot(HaveKey(requestFieldP2PPrefillParams)) - p2p, ok := kvParams[requestFieldP2PParams].(map[string]any) + Expect(kvParams).ToNot(HaveKey(requestFieldRemoteDecoder)) + Expect(kvParams).ToNot(HaveKey(requestFieldRemotePrefiller)) + p2p, ok := kvParams[requestFieldRemoteKVSource].(map[string]any) Expect(ok).To(BeTrue()) Expect(p2p[requestFieldKVRequestID]).ToNot(BeEmpty()) Expect(p2p[requestFieldRemoteHost]).To(Equal("10.9.9.9")) @@ -99,7 +99,7 @@ var _ = Describe("P2P KV cache source header", func() { <-testInfo.stoppedCh }) - It("should add p2p params to the prefill leg under disaggregation", func() { + It("should add remote_kv_source params to the prefill leg under disaggregation", func() { proxyBaseAddr := testInfo.startProxy() prefillHostPort := testInfo.prefillBackend.URL[len("http://"):] @@ -112,37 +112,38 @@ var _ = Describe("P2P KV cache source header", func() { return len(testInfo.prefillHandler.GetCompletionRequests()) }).Should(Equal(1)) - // Prefill leg: decode + p2p, each with its own kv_request_id. + // Prefill leg: remote_decoder + remote_kv_source, each with its own + // kv_request_id. preq := testInfo.prefillHandler.GetCompletionRequests()[0] prefillKVParams, ok := preq[requestFieldKVTransferParams].(map[string]any) Expect(ok).To(BeTrue()) - decodeParams, ok := prefillKVParams[requestFieldP2PDecodeParams].(map[string]any) + decodeParams, ok := prefillKVParams[requestFieldRemoteDecoder].(map[string]any) Expect(ok).To(BeTrue()) - p2p, ok := prefillKVParams[requestFieldP2PParams].(map[string]any) + p2p, ok := prefillKVParams[requestFieldRemoteKVSource].(map[string]any) Expect(ok).To(BeTrue()) Expect(p2p[requestFieldKVRequestID]).ToNot(BeEmpty()) Expect(p2p[requestFieldKVRequestID]).ToNot(Equal(decodeParams[requestFieldKVRequestID])) Expect(p2p[requestFieldRemoteHost]).To(Equal("10.9.9.9")) Expect(p2p[requestFieldRemotePort]).To(BeNumerically("==", p2pConnectorPort)) - // Decode leg: prefill only, never prefill + p2p. + // Decode leg: remote_prefiller only, never remote_kv_source. decodeReqs := testInfo.decodeHandler.GetCompletionRequests() Expect(decodeReqs).To(HaveLen(1)) decodeKVParams, ok := decodeReqs[0][requestFieldKVTransferParams].(map[string]any) Expect(ok).To(BeTrue()) - Expect(decodeKVParams).To(HaveKey(requestFieldP2PPrefillParams)) - Expect(decodeKVParams).ToNot(HaveKey(requestFieldP2PParams)) + Expect(decodeKVParams).To(HaveKey(requestFieldRemotePrefiller)) + Expect(decodeKVParams).ToNot(HaveKey(requestFieldRemoteKVSource)) testInfo.cancelFn() <-testInfo.stoppedCh }) - It("should not add p2p to the prefill leg when the source is the prefiller itself", func() { + It("should not add remote_kv_source params to the prefill leg when the source is the prefiller itself", func() { proxyBaseAddr := testInfo.startProxy() prefillHostPort := testInfo.prefillBackend.URL[len("http://"):] // The source resolves to the selected prefiller - there is nothing to - // pull from itself, so the prefill leg carries decode only. + // pull from itself, so the prefill leg carries remote_decoder only. sendRequest(proxyBaseAddr, map[string]string{ routing.PrefillEndpointHeader: prefillHostPort, routing.KVCacheSourceHeader: prefillHostPort, @@ -155,14 +156,14 @@ var _ = Describe("P2P KV cache source header", func() { preq := testInfo.prefillHandler.GetCompletionRequests()[0] prefillKVParams, ok := preq[requestFieldKVTransferParams].(map[string]any) Expect(ok).To(BeTrue()) - Expect(prefillKVParams).To(HaveKey(requestFieldP2PDecodeParams)) - Expect(prefillKVParams).ToNot(HaveKey(requestFieldP2PParams)) + Expect(prefillKVParams).To(HaveKey(requestFieldRemoteDecoder)) + Expect(prefillKVParams).ToNot(HaveKey(requestFieldRemoteKVSource)) testInfo.cancelFn() <-testInfo.stoppedCh }) - It("should not inject p2p params without the header", func() { + It("should not inject remote_kv_source params without the header", func() { proxyBaseAddr := testInfo.startProxy() sendRequest(proxyBaseAddr, nil) @@ -198,9 +199,9 @@ var _ = Describe("P2P KV cache source header", func() { Expect(decodeReqs).To(HaveLen(1)) kvParams, ok := decodeReqs[0][requestFieldKVTransferParams].(map[string]any) Expect(ok).To(BeTrue()) - // Only the sidecar-owned p2p key survives; the client's keys are gone. + // Only the sidecar-owned remote_kv_source key survives; the client's keys are gone. Expect(kvParams).To(HaveLen(1)) - p2p, ok := kvParams[requestFieldP2PParams].(map[string]any) + p2p, ok := kvParams[requestFieldRemoteKVSource].(map[string]any) Expect(ok).To(BeTrue()) Expect(p2p[requestFieldRemoteHost]).To(Equal("10.9.9.9")) @@ -221,7 +222,7 @@ var _ = Describe("P2P KV cache source header", func() { <-testInfo.stoppedCh }) - It("should not inject p2p params when the source is the local pod", func() { + It("should not inject remote_kv_source params when the source is the local pod", func() { GinkgoT().Setenv("POD_IP", "10.9.9.9") proxyBaseAddr := testInfo.startProxy() @@ -247,7 +248,7 @@ var _ = Describe("P2P KV cache source header", func() { for _, dreq := range decodeReqs { kvParams, ok := dreq[requestFieldKVTransferParams].(map[string]any) Expect(ok).To(BeTrue()) - p2p, ok := kvParams[requestFieldP2PParams].(map[string]any) + p2p, ok := kvParams[requestFieldRemoteKVSource].(map[string]any) Expect(ok).To(BeTrue()) ids = append(ids, p2p[requestFieldKVRequestID]) } diff --git a/pkg/sidecar/proxy/connector_p2p_test.go b/pkg/sidecar/proxy/connector_p2p_test.go index 52a444a986..842422897f 100644 --- a/pkg/sidecar/proxy/connector_p2p_test.go +++ b/pkg/sidecar/proxy/connector_p2p_test.go @@ -60,7 +60,7 @@ var _ = Describe("P2P Connector", func() { return len(testInfo.prefillHandler.GetCompletionRequests()) }).Should(Equal(1)) - // Prefill leg: kv_transfer_params.decode carries only kv_request_id, + // Prefill leg: kv_transfer_params.remote_decoder carries only kv_request_id, // with no peer address. prefillReqs := testInfo.prefillHandler.GetCompletionRequests() Expect(prefillReqs).To(HaveLen(1)) @@ -69,8 +69,8 @@ var _ = Describe("P2P Connector", func() { Expect(preq).To(HaveKey(requestFieldKVTransferParams)) prefillKVParams, ok := preq[requestFieldKVTransferParams].(map[string]any) Expect(ok).To(BeTrue()) - Expect(prefillKVParams).ToNot(HaveKey(requestFieldP2PPrefillParams)) - prefillDecode, ok := prefillKVParams[requestFieldP2PDecodeParams].(map[string]any) + Expect(prefillKVParams).ToNot(HaveKey(requestFieldRemotePrefiller)) + prefillDecode, ok := prefillKVParams[requestFieldRemoteDecoder].(map[string]any) Expect(ok).To(BeTrue()) prefillKVRequestID := prefillDecode[requestFieldKVRequestID] Expect(prefillKVRequestID).ToNot(BeEmpty()) @@ -82,7 +82,7 @@ var _ = Describe("P2P Connector", func() { Expect(preq).To(HaveKeyWithValue(requestFieldMaxCompletionTokens, BeNumerically("==", 1))) Expect(preq[requestFieldStream]).To(BeFalse()) - // Decode leg: kv_transfer_params.prefill carries the prefiller's + // Decode leg: kv_transfer_params.remote_prefiller carries the prefiller's // OffloadingConnector P2P tier address plus the matching kv_request_id. Expect(testInfo.decodeHandler.RequestCount.Load()).To(BeNumerically("==", 1)) decodeReqs := testInfo.decodeHandler.GetCompletionRequests() @@ -92,8 +92,8 @@ var _ = Describe("P2P Connector", func() { Expect(dreq).To(HaveKey(requestFieldKVTransferParams)) decodeKVParams, ok := dreq[requestFieldKVTransferParams].(map[string]any) Expect(ok).To(BeTrue()) - Expect(decodeKVParams).ToNot(HaveKey(requestFieldP2PDecodeParams)) - decodePrefill, ok := decodeKVParams[requestFieldP2PPrefillParams].(map[string]any) + Expect(decodeKVParams).ToNot(HaveKey(requestFieldRemoteDecoder)) + decodePrefill, ok := decodeKVParams[requestFieldRemotePrefiller].(map[string]any) Expect(ok).To(BeTrue()) Expect(decodePrefill[requestFieldKVRequestID]).To(Equal(prefillKVRequestID)) Expect(decodePrefill[requestFieldRemoteHost]).To(Equal(extractHost(prefillHostPort))) diff --git a/pkg/sidecar/proxy/proxy.go b/pkg/sidecar/proxy/proxy.go index ff508f00a0..19e2c5792c 100644 --- a/pkg/sidecar/proxy/proxy.go +++ b/pkg/sidecar/proxy/proxy.go @@ -92,11 +92,13 @@ const ( requestFieldRemoteBootstrapAddr = "remote_bootstrap_addr" // OffloadingConnector kv_transfer_params fields. The role is encoded by the - // nesting key: "decode" on the prefiller leg, "prefill" on the decoder leg. - requestFieldP2PDecodeParams = "decode" - requestFieldP2PPrefillParams = "prefill" - requestFieldP2PParams = "p2p" - requestFieldKVRequestID = "kv_request_id" + // nesting key, named for the remote party it describes: "remote_decoder" on + // the prefiller leg, "remote_prefiller" on the decoder leg, "remote_kv_source" + // for a symmetric cached-prefix pull. + requestFieldRemoteDecoder = "remote_decoder" + requestFieldRemotePrefiller = "remote_prefiller" + requestFieldRemoteKVSource = "remote_kv_source" + requestFieldKVRequestID = "kv_request_id" KVConnectorNIXLV2 = constants.KVConnectorNIXLV2 KVConnectorSharedStorage = constants.KVConnectorSharedStorage diff --git a/release-notes.d/unreleased/1937.md b/release-notes.d/unreleased/1937.md index 02fcf8f9c9..4fe908e4bc 100644 --- a/release-notes.d/unreleased/1937.md +++ b/release-notes.d/unreleased/1937.md @@ -4,4 +4,4 @@ url: https://github.com/llm-d/llm-d-router/pull/1937 author: nilig date: 2026-07-14 --- -The routing sidecar honors the `x-kv-cache-source-host-port` header, injecting `kv_transfer_params.p2p` so vLLM pulls cached prefix KV from the named peer over the OffloadingConnector P2P tier instead of recomputing it. +The routing sidecar honors the `x-kv-cache-source-host-port` header, injecting `kv_transfer_params.remote_kv_source` so vLLM pulls cached prefix KV from the named peer over the OffloadingConnector P2P tier instead of recomputing it.