Skip to content
Merged
Show file tree
Hide file tree
Changes from 4 commits
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
32 changes: 31 additions & 1 deletion components/src/dynamo/sglang/_compat.py
Original file line number Diff line number Diff line change
Expand Up @@ -380,20 +380,48 @@ def filter_supported_async_generate_kwargs(
return {key: value for key, value in kwargs.items() if key in supported_kwarg_names}


def sglang_has_wide_mm_pad_values() -> bool:
Comment thread
maheshrbapatu marked this conversation as resolved.
Resolved
"""Detect the wide event-token format introduced by sglang#43043.

Both caller hashes and engine-owned identities produce wide values.
"""
try:
# Deferred: schedule_batch imports torch and other SGLang runtime modules.
from sglang.srt.managers.schedule_batch import MultimodalDataItem
except ImportError:
return False
return "pad_values" in MultimodalDataItem.__struct_fields__
Comment thread
maheshrbapatu marked this conversation as resolved.
Resolved


def sglang_uses_engine_owned_mm_identity(mm_processor: Any = None) -> bool:
"""Whether the active processor rejects caller-supplied image identities."""
if not sglang_has_wide_mm_pad_values():
return False
# Unknown processors on wide-pad builds must not receive identity overrides.
return mm_processor is None or bool(
getattr(mm_processor, "uses_wide_image_identity", False)
)


def supports_external_mm_hashes(engine: Any) -> bool:
"""Enable safe caller-provided MM hashes when this SGLang accepts them.

Supported SGLang releases apply caller hashes after some processors have
already built ``padded_input_ids``. Rebuild that derived field after
tokenization so the external hash, item pad value, and padded IDs remain
consistent. The repair is idempotent if upstream already rebuilt them.

Builds that own multimodal identity never get caller hashes.
"""
tokenizer_manager = getattr(engine, "tokenizer_manager", None)
mm_proc = getattr(tokenizer_manager, "mm_processor", None)
if sglang_uses_engine_owned_mm_identity(mm_proc):
return False
if "mm_hashes" not in filter_supported_async_generate_kwargs(
engine, {"mm_hashes": None}
):
return False

tokenizer_manager = getattr(engine, "tokenizer_manager", None)
tokenize_one = getattr(tokenizer_manager, "_tokenize_one_request", None)
if tokenize_one is None or getattr(
tokenize_one, "_dynamo_rebuilds_external_mm_padding", False
Expand Down Expand Up @@ -535,4 +563,6 @@ def supports_require_reasoning(engine: Any) -> bool:
"sglang_uses_mla_backend",
"supports_require_reasoning",
"supports_external_mm_hashes",
"sglang_uses_engine_owned_mm_identity",
"sglang_has_wide_mm_pad_values",
]
7 changes: 6 additions & 1 deletion components/src/dynamo/sglang/publisher.py
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,7 @@
)
from dynamo.llm import KvEventPublisher, WorkerMetricsPublisher
from dynamo.runtime import Endpoint
from dynamo.sglang._compat import override_server_args
from dynamo.sglang._compat import override_server_args, sglang_has_wide_mm_pad_values
from dynamo.sglang._disagg import SGLANG_WORKER_GROUP_ID_KEY, get_sglang_worker_group_id
from dynamo.sglang.args import Config
from dynamo.sglang.capacity import (
Expand Down Expand Up @@ -416,6 +416,10 @@ def init_kv_event_publish(self) -> List[KvEventPublisher]:
"sglang kv_events_config is set but missing 'endpoint'"
)
connect_ip = kv_event_connect_ip(base_ep)
# Event tokens carry SGLang's own multimodal pad values on builds
# with sgl-project/sglang#43043; the normalizer excludes them.
# Accepted caller hashes also produce wide values on non-Kimi models.
engine_owned_mm_identity = sglang_has_wide_mm_pad_values()

# Determine DP attention configuration
dp_ranks = get_local_dp_rank_range(self.server_args)
Expand Down Expand Up @@ -452,6 +456,7 @@ def init_kv_event_publish(self) -> List[KvEventPublisher]:
enable_local_indexer=self.dynamo_args.enable_local_indexer,
dp_rank=dp_rank,
kv_state_endpoint=self.dynamo_args.kv_state_endpoint,
engine_owned_mm_identity=engine_owned_mm_identity,
)
self.kv_publishers.append(publisher)

Expand Down
4 changes: 3 additions & 1 deletion components/src/dynamo/sglang/state_agent.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@
validate_advertise_host,
)
from dynamo.llm import KvStateAttachmentOwner
from dynamo.sglang._compat import resolved_server_args
from dynamo.sglang._compat import resolved_server_args, sglang_has_wide_mm_pad_values
from dynamo.sglang.capacity import model_card_dp_rank_bounds
from dynamo.sglang.kv_hints import kvcr_hint_endpoints

Expand Down Expand Up @@ -94,6 +94,7 @@ def state_agent_descriptors(engine, config) -> list[dict]:
raw_sources[rank] = source
if set(raw_sources) != set(sources):
raise ValueError("KVCR inventory requires a reported raw event source per rank")
engine_owned_mm_identity = sglang_has_wide_mm_pad_values()
descriptors = []
owners = set()
for rank, source in sorted(sources.items()):
Expand Down Expand Up @@ -148,6 +149,7 @@ def state_agent_descriptors(engine, config) -> list[dict]:
"raw_topic": topic,
"image_token_id": None,
"video_token_id": None,
"engine_owned_mm_identity": engine_owned_mm_identity,
"router_hint_source": {
"source_control_endpoint": endpoints[str(rank)],
"worker_type": role,
Expand Down
60 changes: 60 additions & 0 deletions components/src/dynamo/sglang/tests/test_sglang_publisher.py
Original file line number Diff line number Diff line change
Expand Up @@ -570,6 +570,66 @@ def __init__(self, **kwargs):
assert {call["kv_block_size"] for call in calls} == {32}


@pytest.mark.parametrize("engine_owned", [False, True])
def test_init_kv_event_publish_passes_engine_owned_mm_identity(
monkeypatch, engine_owned
):
calls = []

class FakeKvEventPublisher:
def __init__(self, **kwargs):
calls.append(kwargs)

monkeypatch.setattr(publisher_mod, "KvEventPublisher", FakeKvEventPublisher)
monkeypatch.setattr(
publisher_mod, "sglang_has_wide_mm_pad_values", lambda: engine_owned
)
monkeypatch.setattr(publisher_mod, "get_local_ip_auto", lambda: "127.0.0.1")
monkeypatch.setattr(
publisher_mod,
"ZmqEventPublisher",
SimpleNamespace(
offset_endpoint_port=staticmethod(
lambda base_ep, dp_rank: f"tcp://*:{5557 + dp_rank}"
)
),
)
monkeypatch.setattr(
publisher_mod,
"format_zmq_endpoint",
lambda endpoint, ip_address: endpoint.replace("*", ip_address),
)
# A non-leader node skips the scheduler metrics sockets.
server_args = SimpleNamespace(
kv_events_config='{"endpoint": "tcp://*:5557"}',
page_size=16,
dp_size=4,
enable_dp_attention=True,
nnodes=2,
node_rank=1,
)
config = SimpleNamespace(
server_args=server_args,
dynamo_args=SimpleNamespace(
enable_local_indexer=True,
kv_state_endpoint=None,
use_kv_events=True,
),
)
publisher = DynamoSglangPublisher(
engine=SimpleNamespace(),
config=config,
generate_endpoint=SimpleNamespace(),
component_gauges=SimpleNamespace(),
kv_worker_id=1234,
)

publisher.init_kv_event_publish()

assert len(calls) == 2
assert [call["engine_owned_mm_identity"] for call in calls] == [engine_owned] * 2


def test_init_kv_event_publish_subscribes_to_every_pure_dp_replica(monkeypatch):
calls = []

Expand Down
71 changes: 68 additions & 3 deletions components/src/dynamo/sglang/tests/test_sglang_unit.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@
import sys
import tomllib
from pathlib import Path
from types import SimpleNamespace
from types import ModuleType, SimpleNamespace
from unittest.mock import AsyncMock, Mock

import pytest
Expand All @@ -35,6 +35,8 @@
publish_server_args,
require_reasoning_kwargs,
resolved_server_args,
sglang_has_wide_mm_pad_values,
sglang_uses_engine_owned_mm_identity,
sglang_uses_mla_backend,
supports_external_mm_hashes,
)
Expand Down Expand Up @@ -1089,8 +1091,18 @@ async def async_generate(self, **kwargs):


@pytest.mark.asyncio
async def test_external_mm_hashes_rebuild_padding_after_hash_override():
item = SimpleNamespace(pad_value=91, offsets=[(1, 2)])
async def test_external_mm_hashes_rebuild_padding_after_hash_override(monkeypatch):
# Deferred so marker-only collection does not require SGLang runtime modules.
from sglang.srt.managers.schedule_batch import Modality, MultimodalDataItem

# The repair applies to SGLang builds that still derive pad values from
# caller hashes.
monkeypatch.setattr(
sglang_compat,
"sglang_uses_engine_owned_mm_identity",
lambda mm_processor=None: False,
)
item = MultimodalDataItem(modality=Modality.IMAGE, pad_value=91, offsets=[(1, 2)])
mm_inputs = SimpleNamespace(mm_items=[item], padded_input_ids=[1, 7, 7, 4])
tokenized = SimpleNamespace(input_ids=[1, 2, 3, 4], mm_inputs=mm_inputs)

Expand Down Expand Up @@ -1122,6 +1134,59 @@ async def async_generate(self):
assert not supports_external_mm_hashes(LegacyEngine())


def test_external_mm_hashes_are_disabled_for_engine_owned_mm_identity(monkeypatch):
# sgl-project/sglang#43043 builds derive wide pad values from their own
# identity; caller hashes no longer match Dynamo's pads and Kimi rejects them.
monkeypatch.setattr(
sglang_compat,
"sglang_uses_engine_owned_mm_identity",
lambda mm_processor=None: True,
)

class MmHashEngine:
async def async_generate(self, mm_hashes=None):
return None

assert not supports_external_mm_hashes(MmHashEngine())


@pytest.mark.parametrize("wide_schema", [False, True])
def test_mm_identity_and_pad_format_capabilities(monkeypatch, wide_schema):
fields = ("hash", "pad_value", "modality")
if wide_schema:
fields += ("pad_values",)
mod = ModuleType("sglang.srt.managers.schedule_batch")
mod.MultimodalDataItem = type(
"MultimodalDataItem", (), {"__struct_fields__": fields}
)
monkeypatch.setitem(sys.modules, "sglang.srt.managers.schedule_batch", mod)

class KimiProcessor:
uses_wide_image_identity = True

class CustomProcessor(KimiProcessor):
pass

class MmHashEngine:
def __init__(self, processor):
self.tokenizer_manager = SimpleNamespace(mm_processor=processor)

async def async_generate(self, mm_hashes=None):
return None

# The inherited capability survives external processor overrides.
custom = CustomProcessor()
assert sglang_uses_engine_owned_mm_identity(custom) is wide_schema
assert supports_external_mm_hashes(MmHashEngine(custom)) is not wide_schema
assert sglang_uses_engine_owned_mm_identity() is wide_schema

# Qwen accepts caller hashes, but its event tokens still use the wide format.
qwen = SimpleNamespace()
assert not sglang_uses_engine_owned_mm_identity(qwen)
assert supports_external_mm_hashes(MmHashEngine(qwen))
assert sglang_has_wide_mm_pad_values() is wide_schema


@pytest.mark.parametrize(
("request_data", "expected"),
[
Expand Down
7 changes: 6 additions & 1 deletion components/src/dynamo/sglang/tests/test_state_agent.py
Original file line number Diff line number Diff line change
Expand Up @@ -70,7 +70,11 @@ def fixture():
return engine, config, source, info


def test_descriptors_use_only_reported_sources():
@pytest.mark.parametrize("wide_pad_values", [False, True])
def test_descriptors_use_only_reported_sources(monkeypatch, wide_pad_values):
monkeypatch.setattr(
state_agent, "sglang_has_wide_mm_pad_values", lambda: wide_pad_values
)
engine, config, source, _ = fixture()
(result,) = state_agent_descriptors(engine, config)
assert result["cache_owner_id"] == source["cache_owner_id"]
Expand All @@ -80,6 +84,7 @@ def test_descriptors_use_only_reported_sources():
assert result["raw_zmq_endpoint"] == "tcp://worker:24020"
assert result["ingress_protocol"] == "sglang_hicache_v1"
assert result["kv_state_endpoint"] == "ns/backend/generate"
assert result["engine_owned_mm_identity"] is wide_pad_values
source["capabilities"] = ["router_hint"]
assert state_agent_descriptors(engine, config) == []

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -91,7 +91,7 @@ SGLang supports EPD, EP/D, E/PD, and E/P/D patterns. See [Multimodal Model Servi

Multimodal KV routing works with SGLang's aggregated worker topology. It is independent of the E/PD and E/P/D encoder-disaggregation patterns described later in this guide.

SGLang RadixAttention includes a per-media `pad_value` token in its prefix-cache key. Dynamo must use that same token in the routing view:
On SGLang 0.5.22, which Dynamo currently pins, RadixAttention includes a per-media `pad_value` token in its prefix-cache key. Dynamo uses the same token in the routing view:

1. The Rust frontend hashes each image or sampled video and calculates its expanded token count.
2. It derives `pad_value = MM_PAD_SHIFT_VALUE + (mm_hash % 2^30)`.
Expand All @@ -110,6 +110,18 @@ text-prefix routing without failing the request.

Step 4 needs an SGLang build that accepts `mm_hashes`. Dynamo's SGLang image carries the upstream patch that adds it; a custom build without it is still supported, but Dynamo detects the missing argument when the worker starts and routes on the text prefix alone, so media identity no longer contributes to cache overlap.

### Newer SGLang Image Identities

SGLang builds containing [sgl-project/sglang#43043](https://github.com/sgl-project/sglang/pull/43043) use four wide placeholder values per media item. These no longer match Dynamo's routing tokens, including for processors such as Qwen-VL that still accept `mm_hashes`. Kimi K2.5 and K3 processors also reject caller-supplied image hashes. Dynamo detects that processor capability and omits those overrides for these models.

On these builds, Dynamo accepts the wide values in KV events and indexes only complete text blocks before the first image or video block. The media block and all blocks chained after it are excluded from predicted cache hits, including in subsequent events. Other text stores, evictions, and cache clears continue to update the index. Requests still run, and SGLang continues to manage its own prefix cache, but image- and video-aware routing is unavailable. This applies both to URL passthrough and to `--frontend-decoding`.

Python SGLang workers detect the wide format automatically. A standalone indexer or selection service receiving engine events directly must register those sources with `engine_owned_mm_identity: true`; the field defaults to `false` for legacy sources. Without the flag, values exceeding 32 bits are still excluded, but smaller values in SGLang's new multimodal range cannot be distinguished from legacy routing tokens.

Exclusion tracking is bounded to 1,048,576 block IDs per event normalizer across cache tiers. If it fills up, Dynamo logs a warning and accepts only stores with no parent for the affected worker rank until an `AllBlocksCleared` event. Existing text evictions continue to apply. This fallback can miss additional text-prefix hits, but prevents an unknown-parent store from invalidating the relay's worker state.

### Launch an Aggregated Deployment

Launch an aggregated deployment with multimodal KV routing:

```bash
Expand All @@ -131,7 +143,7 @@ and sets `--router-mode kv` with a matching frontend and worker block size.

### Version Requirements

The Dynamo SGLang image includes both routing prerequisites:
The pinned Dynamo SGLang image includes these routing prerequisites:

- Dynamo is built with the `mm-routing` Rust feature.
- SGLang 0.5.13 or later includes `GenerateReqInput.mm_hashes` support. Dynamo currently pins 0.5.22.
Expand All @@ -142,7 +154,7 @@ from [sgl-project/sglang#25300](https://github.com/sgl-project/sglang/pull/25300
Without it, requests still complete but fall back to text-prefix-only routing.
Prefer upgrading to 0.5.13 or later instead of patching an installed package.

Enable `DYN_LOG=info,mm_routing=debug` to inspect media-token counts, multimodal hashes, the video processor contract, and the selected worker's overlap. A repeated request should select the same worker with high block overlap.
Enable `DYN_LOG=info,mm_routing=debug` to inspect media-token counts, multimodal hashes, the video processor contract, and the selected worker's overlap. On compatible builds, a repeated request should select the same worker with high block overlap. Builds using the newer identities only predict the text-prefix overlap described above.

For the user-facing workflow, see [Multimodal KV Routing](../../../../../use-cases/multimodal-serving/multimodal-kv-routing.md).

Expand Down
1 change: 1 addition & 0 deletions lib/backend-common/src/publisher.rs
Original file line number Diff line number Diff line change
Expand Up @@ -85,6 +85,7 @@ fn setup_kv_publishers(
video_token_id: None,
expected_rank: Some(dp_rank),
liveness: heartbeat_timeout.map(|timeout| (dp_rank, timeout)),
engine_owned_mm_identity: false,
recovery: None,
}),
None,
Expand Down
Loading
Loading