Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
77 changes: 72 additions & 5 deletions tests/test_config.py
Original file line number Diff line number Diff line change
Expand Up @@ -114,7 +114,32 @@ def test_per_request_spec_decode_metrics_requires_spec_decode():
)


def test_pd_dcp_interleave_size_is_adjusted_to_block_size(caplog):
@pytest.mark.parametrize(
"kv_transfer_config",
[
KVTransferConfig(
kv_connector="NixlConnector",
kv_role="kv_both",
),
KVTransferConfig(
kv_connector="MultiConnector",
kv_role="kv_both",
kv_connector_extra_config={
"connectors": [
{
"kv_connector": "NixlConnector",
"kv_role": "kv_both",
},
{
"kv_connector": "OffloadingConnector",
"kv_role": "kv_both",
},
]
},
),
],
)
def test_pd_dcp_interleave_size_is_adjusted_to_block_size(caplog, kv_transfer_config):
config = VllmConfig(
cache_config=CacheConfig(block_size=16),
device_config=DeviceConfig(device="cpu"),
Expand All @@ -124,10 +149,7 @@ def test_pd_dcp_interleave_size_is_adjusted_to_block_size(caplog):
cp_kv_cache_interleave_size=3,
distributed_executor_backend="mp",
),
kv_transfer_config=KVTransferConfig(
kv_connector="NixlConnector",
kv_role="kv_both",
),
kv_transfer_config=kv_transfer_config,
)

kv_cache_config = SimpleNamespace(
Expand All @@ -140,6 +162,51 @@ def test_pd_dcp_interleave_size_is_adjusted_to_block_size(caplog):
assert "automatically adjusted from 3 to block_size 16" in caplog.text


def test_kv_offloading_does_not_adjust_dcp_interleave_size():
config = VllmConfig(
cache_config=CacheConfig(block_size=16),
device_config=DeviceConfig(device="cpu"),
parallel_config=ParallelConfig(
tensor_parallel_size=2,
decode_context_parallel_size=2,
cp_kv_cache_interleave_size=1,
distributed_executor_backend="mp",
),
kv_transfer_config=KVTransferConfig(
kv_connector="OffloadingConnector",
kv_role="kv_both",
),
)

kv_cache_config = SimpleNamespace(
kv_cache_groups=[SimpleNamespace(kv_cache_spec=SimpleNamespace(block_size=16))]
)
config.adjust_dcp_kv_cache_interleave_size(kv_cache_config)

assert config.parallel_config.cp_kv_cache_interleave_size == 1


def test_kv_offloading_does_not_skip_dcp_interleave_validation():
config = SimpleNamespace(
cache_config=SimpleNamespace(
block_size=16,
mamba_cache_mode="none",
),
parallel_config=SimpleNamespace(
decode_context_parallel_size=2,
cp_kv_cache_interleave_size=3,
),
scheduler_config=SimpleNamespace(disable_chunked_mm_input=False),
kv_transfer_config=KVTransferConfig(
kv_connector="OffloadingConnector",
kv_role="kv_both",
),
)

with pytest.raises(AssertionError, match="divisible by"):
VllmConfig.validate_block_size(config)


def test_compile_config_repr_succeeds():
# setup: VllmBackend mutates the config object
config = VllmConfig()
Expand Down
32 changes: 16 additions & 16 deletions vllm/config/vllm.py
Original file line number Diff line number Diff line change
Expand Up @@ -2742,19 +2742,14 @@ def _validate_v1_model_runner(self) -> None:
def adjust_dcp_kv_cache_interleave_size(
self, kv_cache_config: "KVCacheConfig"
) -> None:
"""Normalize DCP interleave size against the resolved block_size for PD.
"""Normalize DCP interleave size against block_size for NIXL P/D.

Called by each worker (via ensure_kv_transfer_initialized), once it knows its
own final block_size via kv_cache_config.
"""
dcp_size = self.parallel_config.decode_context_parallel_size
if dcp_size <= 1:
return
# Get the kernel block_size, but don't use resolve_kv_cache_block_size to avoid
# scaling by dcp_size (we need the local block_size here).
local_block_size = min(
g.kv_cache_spec.block_size for g in kv_cache_config.kv_cache_groups
)
if self.parallel_config.dcp_kv_cache_interleave_size > 1 and (
self.parallel_config.cp_kv_cache_interleave_size
!= self.parallel_config.dcp_kv_cache_interleave_size
Expand All @@ -2768,11 +2763,17 @@ def adjust_dcp_kv_cache_interleave_size(
"deprecated when PCP is fully supported."
)

if (
self.kv_transfer_config is not None
and self.kv_transfer_config.kv_connector is not None
and self.parallel_config.cp_kv_cache_interleave_size != local_block_size
if self.kv_transfer_config is None or not self.kv_transfer_config.has_connector(
"NixlConnector"
):
return

# Get the kernel block_size, but don't use resolve_kv_cache_block_size to avoid
# scaling by dcp_size (we need the local block_size here).
local_block_size = min(
g.kv_cache_spec.block_size for g in kv_cache_config.kv_cache_groups
)
if self.parallel_config.cp_kv_cache_interleave_size != local_block_size:
interleave = self.parallel_config.cp_kv_cache_interleave_size
self.parallel_config.cp_kv_cache_interleave_size = local_block_size
logger.info_once(
Expand All @@ -2793,14 +2794,13 @@ def validate_block_size(self) -> None:
"""
block_size = self.cache_config.block_size

# Skip DCP interleave-size compatibility when a KV connector is configured:
# cp_kv_cache_interleave_size is pinned to block_size for PD by each worker
pd_active = (
# Skip DCP interleave-size compatibility for NIXL P/D: the interleave
# size is pinned to block_size by each worker.
nixl_pd_active = (
self.kv_transfer_config is not None
and self.kv_transfer_config.kv_connector is not None
and self.kv_transfer_config.is_kv_transfer_instance
and self.kv_transfer_config.has_connector("NixlConnector")
)
if self.parallel_config.decode_context_parallel_size > 1 and not pd_active:
if self.parallel_config.decode_context_parallel_size > 1 and not nixl_pd_active:
assert (
self.parallel_config.cp_kv_cache_interleave_size <= block_size
and block_size % self.parallel_config.cp_kv_cache_interleave_size == 0
Expand Down
2 changes: 1 addition & 1 deletion vllm/distributed/kv_transfer/kv_transfer_state.py
Original file line number Diff line number Diff line change
Expand Up @@ -85,7 +85,7 @@ def ensure_kv_transfer_initialized(
vllm_config.kv_transfer_config.is_kv_transfer_instance
and _KV_CONNECTOR_AGENT is None
):
# PD only supports an interleave_size equal to block_size.
# NIXL P/D requires an interleave_size equal to block_size.
vllm_config.adjust_dcp_kv_cache_interleave_size(kv_cache_config)
_sync_engine_id_across_tp(vllm_config)

Expand Down
Loading