[Fast Start] Add daemon-to-daemon warmup cache seeding for Preload (#63088)

Signed-off-by: Isotr0py <Isotr0py@outlook.com>
Co-authored-by: Xun Sun <UNIDY2002@outlook.com>
Co-authored-by: liusy58 <mg21330037@smail.nju.edu.cn>
Co-authored-by: Kimi Code <noreply@moonshot.cn>

docs/features/preload.md

@@ -70,6 +70,50 @@ loading and before binding its socket, then writes the cache:
 vllm preload --model /path/to/model --tensor-parallel-size 4
 ```
 
+### Seeding warmup caches between daemons (D2D)
+
+A fresh daemon replica — a new node joining a fleet, a restarted preload
+set — would normally pay the autotune pass itself. With a seed source
+configured, each daemon rank fetches its rank-matched warmup caches from the
+source replica over an authenticated TCP control plane before tuning,
+validates them, and installs them into its own on-disk cache, so its warmup
+skips the profiling sweep. Any failure (unreachable, missing, incompatible)
+falls back to local tuning.
+
+Source replica:
+
+```bash
+vllm preload --model /path/to/model --tensor-parallel-size 4 \
+    --seed-listen-host 0.0.0.0 --seed-listen-port 29600 \
+    --seed-token "$TOKEN"
+```
+
+Mirror replica (possibly on another node):
+
+```bash
+vllm preload --model /path/to/model --tensor-parallel-size 4 \
+    --seed-source-host <source-host> --seed-source-port 29600 \
+    --seed-token "$TOKEN"
+```
+
+Each rank binds and dials `base_port + group offset + its global rank`, so one
+`host:base_port` pair covers a whole replica, including the draft daemon group
+when speculative decoding is configured.
+
+The exchange is keyed by the on-disk cache layout: a request names the cache
+by kind, configuration hash, DP rank and world rank, and the source resolves
+it to files under its cache root. Before installing, the target rejects a
+payload whose FlashInfer environment metadata (GPU, CUDA/cuDNN stack) differs
+from its own — such a table would never load anyway, and a hard conflict is
+never overwritten, which would poison the cache file into re-tuning on every
+start. Tables are shareable across vLLM versions as long as the configuration
+hash matches.
+
+The channel is a generic seeding point: warmup caches are the first kind, and
+further daemon state (e.g. Triton warmup caches or weight manifests) plugs in
+as additional kinds — small JSON payloads travel inline, while large binary
+state rides the data plane.
+
 ## How it works
 
 1. `vllm preload` spawns one daemon process per GPU. Each daemon loads its
@@ -193,6 +237,10 @@ Socket paths are derived from the GPU UUID, so they are stable regardless of
   passes, and a table tuned on the draft model would key on the draft's
   config, which no engine looks up), so those deployments' first engine
   tunes and fills the cache instead.
+- **Warmup cache seeding**: only the v1 per-rank FlashInfer autotune cache is
+  seeded; single-rank daemons use FlashInfer's managed v2 store and always
+  tune in place. Seeding requires a trusted network — the TCP channel is
+  authenticated by a shared token but unencrypted.
 - **Quantization**: every quantization method in the model must declare
   support for pre-processed weights (the daemon transfers weights *after*
   quantization post-processing). Unsupported methods raise
@@ -214,6 +262,13 @@ processes owned by the same user:
   directory is additionally rejected if it is group/world accessible.
 - On Linux the daemon verifies the connecting peer's UID via `SO_PEERCRED`.
 
+The optional daemon-to-daemon listener (`--seed-listen-*`) is a
+separate TCP channel: requests are JSON-only and size-capped, authenticate
+with the shared `--seed-token` (constant-time comparison), and
+an unauthenticated peer that stops sending is cut off by a whole-message
+deadline instead of blocking the accept loop. The channel is unencrypted —
+bind it to a trusted network.
+
 See the [security documentation](../usage/security.md) for vLLM's general
 threat model.
 

tests/model_executor/model_loader/test_weight_cache.py

@@ -7,6 +7,7 @@ warm restarts (weights mapped from the daemon via CUDA IPC) must both serve
 identical outputs.
 """
 
+import os
 import shutil
 import subprocess
 import sys
@@ -15,7 +16,7 @@ import threading
 import time
 import urllib.error
 import urllib.request
-from dataclasses import dataclass, field
+from dataclasses import dataclass, field, replace
 from typing import Any
 
 import pytest
@@ -36,6 +37,8 @@ class WeightCacheDaemon:
         model: str,
         tp_size: int,
         extra_args: list[str] | None = None,
+        env_extra: dict[str, str] | None = None,
+        log_file: str | None = None,
     ):
         # Short base path: Unix socket paths are limited to ~107 characters.
         self.socket_dir = tempfile.mkdtemp(prefix="vllm_ipc_")
@@ -58,21 +61,41 @@ class WeightCacheDaemon:
             str(self.health_port),
             *(extra_args or []),
         ]
+        self._env_extra = env_extra
+        self._log_file = log_file
+        self._log_chunks: list[str] = []
         self._proc: subprocess.Popen | None = None
 
     def __enter__(self) -> "WeightCacheDaemon":
+        env = {**os.environ, **self._env_extra} if self._env_extra is not None else None
         self._proc = subprocess.Popen(
-            self._cmd, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, text=True
+            self._cmd,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.STDOUT,
+            text=True,
+            env=env,
         )
-        threading.Thread(target=self._drain_stderr, daemon=True).start()
+        threading.Thread(target=self._drain_output, daemon=True).start()
         return self
 
     def __exit__(self, *exc_info) -> None:
         self._stop()
 
-    def _drain_stderr(self) -> None:
-        assert self._proc is not None and self._proc.stderr is not None
-        self._proc.stderr.read()
+    @property
+    def logs(self) -> str:
+        return "".join(self._log_chunks)
+
+    def _drain_output(self) -> None:
+        assert self._proc is not None and self._proc.stdout is not None
+        if self._log_file is None:
+            for line in self._proc.stdout:
+                self._log_chunks.append(line)
+            return
+        with open(self._log_file, "w") as out:
+            for line in self._proc.stdout:
+                self._log_chunks.append(line)
+                out.write(line)
+                out.flush()
 
     def check_health(self) -> None:
         url = f"http://127.0.0.1:{self.health_port}/health"
@@ -270,6 +293,85 @@ def test_ipc_cache_cold_start_and_warm_restart_tp2_ep(vllm_runner, tmp_path):
     run_cold_start_and_warm_restart(vllm_runner, DSV2_EP_CASE, tmp_path)
 
 
+@create_new_process_for_each_test()
+@pytest.mark.distributed(num_gpus=2)
+def test_d2d_warmup_cache_seeding(vllm_runner, tmp_path):
+    """A mirror daemon adopts the source's autotune cache over D2D, skips
+    re-tuning, and serves outputs identical to the default-loader baseline."""
+    if not current_platform.is_cuda_alike():
+        pytest.skip("Weight cache IPC sharing requires CUDA or ROCm")
+    if not current_platform.is_device_capability_family(100):
+        pytest.skip("Kimi K3 IPC weight cache requires an SM100 MXFP4 backend")
+    from vllm.utils.flashinfer import has_flashinfer
+
+    if not (has_flashinfer() and current_platform.has_device_capability(90)):
+        pytest.skip("FlashInfer autotune requires FlashInfer and SM90+")
+
+    case = replace(
+        K3_CASE,
+        llm_kwargs={**K3_CASE.llm_kwargs, "tensor_parallel_size": 2},
+    )
+    baseline_outputs = generate(vllm_runner, case, None, fallback=True)
+    assert all(text for _, texts in baseline_outputs for text in texts)
+
+    token = "d2d-test-token"
+    base_port = get_open_port()
+    log_dir = os.environ.get("D2D_E2E_LOG_DIR")
+    log_file = (
+        (lambda name: os.path.join(log_dir, name)) if log_dir else lambda name: None
+    )
+    # The v1 per-rank cache path needs a multi-rank tuning group (v2 is gated
+    # to single-rank), so both daemons run at tp=2.
+    with WeightCacheDaemon(
+        case.model,
+        tp_size=2,
+        extra_args=[
+            *case.daemon_args,
+            "--seed-listen-host",
+            "127.0.0.1",
+            "--seed-listen-port",
+            str(base_port),
+            "--seed-token",
+            token,
+        ],
+        env_extra={"VLLM_FLASHINFER_AUTOTUNE_CACHE_DIR": str(tmp_path / "source")},
+        log_file=log_file("source.log"),
+    ) as source:
+        source.wait_until_ready()
+        # The source had an empty cache dir, so it must have tuned in place.
+        assert "Running FlashInfer autotune" in source.logs
+
+        with WeightCacheDaemon(
+            case.model,
+            tp_size=2,
+            extra_args=[
+                *case.daemon_args,
+                "--seed-source-host",
+                "127.0.0.1",
+                "--seed-source-port",
+                str(base_port),
+                "--seed-token",
+                token,
+            ],
+            env_extra={"VLLM_FLASHINFER_AUTOTUNE_CACHE_DIR": str(tmp_path / "mirror")},
+            log_file=log_file("mirror.log"),
+        ) as mirror:
+            mirror.wait_until_ready()
+            # The mirror adopted both ranks' tables and skipped the sweep.
+            assert mirror.logs.count("Adopted remote flashinfer_autotune seed") == 2
+            assert "Running FlashInfer autotune" not in mirror.logs
+            mirror_tables = sorted(
+                (tmp_path / "mirror").rglob("autotune_configs_dp0_rank*.json")
+            )
+            assert len(mirror_tables) == 2
+
+            mirror_outputs = generate(
+                vllm_runner, case, mirror.socket_dir, fallback=False
+            )
+
+    assert mirror_outputs == baseline_outputs
+
+
 def run_cold_start_and_warm_restart(vllm_runner, case: ModelCase, tmp_path, capfd=None):
     # Baseline: plain disk loading with the default loader.
     baseline_outputs = generate(
@@ -802,3 +904,168 @@ def test_ipc_cache_key_tracks_kv_quantization(
     assert len(requested_keys) == 2
     for key in requested_keys:
         assert server.cache_key.mismatched_fields(key) == mismatched_fields
+
+
+def _warmup_cache_request(config_hash: str = "a" * 64, **overrides) -> dict:
+    request = {
+        "kind": "flashinfer_autotune",
+        "config_hash": config_hash,
+        "dp_rank": 0,
+        "rank": 0,
+    }
+    request.update(overrides)
+    return request
+
+
+def test_warmup_cache_request_validation_and_miss(tmp_path, monkeypatch):
+    """The source validates the wire key before touching the filesystem."""
+    monkeypatch.setenv("VLLM_FLASHINFER_AUTOTUNE_CACHE_DIR", str(tmp_path))
+    from vllm.model_executor.model_loader.weight_cache.seed import SeedTransport
+    from vllm.model_executor.model_loader.weight_cache.warmup_cache import (
+        FlashInferWarmupCache,
+    )
+
+    def handle(request):
+        return SeedTransport([FlashInferWarmupCache()]).handle_request(request)
+
+    assert handle("not a dict")["status"] == "error"
+    assert handle(_warmup_cache_request(kind="bogus"))["status"] == "error"
+    # A request missing the kind entirely is rejected, not resolved.
+    assert handle({"token": "t"})["status"] == "error"
+    # A path traversal or malformed hash must never reach the filesystem.
+    assert handle(_warmup_cache_request(config_hash="../x"))["status"] == "error"
+    assert handle(_warmup_cache_request(config_hash="A" * 64))["status"] == "error"
+    # A garbage rank just names a file that does not exist: a clean miss.
+    assert handle(_warmup_cache_request(rank=-1))["status"] == "miss"
+    # A well-formed request for a cache that does not exist is a clean miss.
+    assert handle(_warmup_cache_request())["status"] == "miss"
+
+
+def test_warmup_cache_metadata_gate():
+    """Tables flashinfer would refuse to load are rejected before install."""
+    pytest.importorskip("flashinfer")
+    from flashinfer.autotuner.autotuner import _collect_metadata
+
+    from vllm.model_executor.model_loader.weight_cache.warmup_cache import (
+        FlashInferWarmupCache,
+    )
+
+    kind = FlashInferWarmupCache()
+    filename = "autotune_configs_dp0_rank0.json"
+
+    matching = {filename: {"_metadata": _collect_metadata(), "records": {}}}
+    assert kind.validate_payload(matching) == matching
+
+    # Pre-metadata flashinfer files are accepted, like load_configs does.
+    legacy: dict[str, dict] = {filename: {"records": {}}}
+    assert kind.validate_payload(legacy) == legacy
+
+    foreign = {"_metadata": {**_collect_metadata(), "gpu": "Not A Real GPU"}}
+    with pytest.raises(ValueError, match="environment mismatch"):
+        kind.validate_payload({filename: foreign})
+    with pytest.raises(ValueError, match="JSON object"):
+        kind.validate_payload({filename: [1, 2]})
+    with pytest.raises(ValueError, match="exactly one"):
+        kind.validate_payload({filename: {}, "extra.json": {}})
+
+
+def test_remote_warmup_cache_listener(tmp_path, monkeypatch):
+    """Peers need the token, and only get warmup caches for their rank."""
+    import json
+    import socket
+
+    monkeypatch.setenv("VLLM_FLASHINFER_AUTOTUNE_CACHE_DIR", str(tmp_path))
+    from vllm.model_executor.model_loader.weight_cache.protocol import (
+        recv_json,
+        send_json,
+    )
+    from vllm.model_executor.model_loader.weight_cache.seed import (
+        SeedConfig,
+        SeedTransport,
+    )
+    from vllm.model_executor.model_loader.weight_cache.warmup_cache import (
+        FlashInferWarmupCache,
+    )
+
+    config_hash = "b" * 64
+    table = {"records": {"op": {"shape": 1}}}
+    cache_dir = tmp_path / config_hash
+    cache_dir.mkdir()
+    (cache_dir / "autotune_configs_dp0_rank0.json").write_text(json.dumps(table))
+
+    transport = SeedTransport(
+        [FlashInferWarmupCache()],
+        SeedConfig(listen=("127.0.0.1", 0), token="test-secret"),
+    )
+    address = transport.start_listener()
+    assert address is not None
+    try:
+        cases = [
+            (_warmup_cache_request(config_hash, token="wrong"), "error"),
+            (
+                _warmup_cache_request(config_hash, kind="weights", token="test-secret"),
+                "error",
+            ),
+            (_warmup_cache_request(config_hash, token="test-secret"), "ok"),
+            (_warmup_cache_request("f" * 64, token="test-secret"), "miss"),
+        ]
+        for request, expected in cases:
+            with socket.create_connection(address, timeout=2.0) as peer:
+                send_json(peer, request)
+                response = recv_json(peer)
+            assert response["status"] == expected
+            if expected == "ok":
+                assert response["payload"] == {"autotune_configs_dp0_rank0.json": table}
+    finally:
+        transport.close()
+
+
+def test_transport_cuts_off_trickling_peer(monkeypatch):
+    """A peer that never finishes its message cannot block the accept loop."""
+    import socket
+    import time
+
+    import vllm.model_executor.model_loader.weight_cache.seed as seed_mod
+    from vllm.model_executor.model_loader.weight_cache.protocol import (
+        recv_json,
+        send_json,
+    )
+    from vllm.model_executor.model_loader.weight_cache.seed import (
+        SeedConfig,
+        SeedTransport,
+    )
+
+    monkeypatch.setattr(seed_mod, "_REMOTE_REQUEST_TIMEOUT_S", 0.5)
+    transport = SeedTransport([], SeedConfig(listen=("127.0.0.1", 0), token="t"))
+    address = transport.start_listener()
+    assert address is not None
+    try:
+        with socket.create_connection(address) as idle:
+            idle.sendall(b"\x00")  # a partial frame, then silence
+            time.sleep(1.5)  # past the whole-message deadline
+            with socket.create_connection(address, timeout=2.0) as peer:
+                send_json(peer, {"cmd": "anything", "token": "t"})
+                response = recv_json(peer)
+            # The trickler was cut off and this peer got served (unknown
+            # command, but the accept loop answered).
+            assert response["status"] == "error"
+    finally:
+        transport.close()
+
+
+def test_warmup_cache_install_writes_table(tmp_path, monkeypatch):
+    """Install writes the fetched table to the rank's on-disk cache path."""
+    import json
+
+    from vllm.model_executor.model_loader.weight_cache.warmup_cache import (
+        FlashInferWarmupCache,
+    )
+    from vllm.model_executor.warmup import flashinfer_autotune_cache as fi_cache
+
+    target = tmp_path / "autotune_configs_dp0_rank0.json"
+    monkeypatch.setattr(
+        fi_cache, "resolve_rank_flashinfer_autotune_file", lambda _: target
+    )
+    table = {"records": {"op": {"shape": 1}}}
+    FlashInferWarmupCache().install(None, {"autotune_configs_dp0_rank0.json": table})
+    assert json.loads(target.read_text()) == table

tests/model_executor/test_flashinfer_autotune_warmup.py

@@ -176,7 +176,11 @@ def test_flashinfer_autotune_lifecycle(
     model = runner.get_model()
     cache_path = tmp_path / "autotune.json"
     rank_cache_path = cache_path.with_name(f"autotune_dp0_rank{rank}.json")
-    rank_cache_path.write_text(json.dumps({"_metadata": {}, "op": 1}))
+    # A matching cache lets every rank skip the tuning sweep, so the dummy
+    # failure case must run on a cache miss to reach the sweep at all.
+    cache_hit = failure != "dummy"
+    if cache_hit:
+        rank_cache_path.write_text(json.dumps({"_metadata": {}, "op": 1}))
     events = Mock()
     events.tuner.load_configs.return_value = True
     events.tuner._dirty = True
@@ -226,9 +230,9 @@ def test_flashinfer_autotune_lifecycle(
     monkeypatch.setattr(fi_utils, "has_flashinfer_autotune_v2", lambda: has_v2)
     monkeypatch.setattr(fi_utils, "autotune", events.autotune)
     monkeypatch.setattr(warmup.envs, "VLLM_FLASHINFER_AUTOTUNE_SKIP_OPS", skip_ops)
-    resolve_file = Mock(return_value=cache_path)
+    resolve_file = Mock(return_value=rank_cache_path)
     resolve_root = Mock(return_value=str(tmp_path))
-    monkeypatch.setattr(warmup, "resolve_flashinfer_autotune_file", resolve_file)
+    monkeypatch.setattr(warmup, "resolve_rank_flashinfer_autotune_file", resolve_file)
     monkeypatch.setattr(warmup, "resolve_flashinfer_autotune_v2_root", resolve_root)
 
     def gather(out, value, *, group):
@@ -257,22 +261,21 @@ def test_flashinfer_autotune_lifecycle(
     else:
         resolve_root.assert_not_called()
         resolve_file.assert_called_once_with(runner)
-        expected.extend(
-            [
-                call.tuner.load_configs(str(rank_cache_path)),
-                call.autotune(tune_mode=True, **kwargs),
-            ]
-        )
+        if cache_hit:
+            expected.append(call.tuner.load_configs(str(rank_cache_path)))
+        expected.append(call.autotune(tune_mode=True, **kwargs))
     expected.extend(
         [call.set_group(world.cpu_group if world_size > 1 else None), call.enter()]
     )
     if hisparse:
         expected.append(call.hisparse(runner))
-    for name, arg, kwargs in [
-        ("dummy", runner, {"skip_attn": hisparse}),
+    warmup_calls: list[tuple[str, object, dict]] = [
         ("replayssm", runner, {}),
         ("kimi", model, {}),
-    ]:
+    ]
+    if use_v2 or not cache_hit:
+        warmup_calls.insert(0, ("dummy", runner, {"skip_attn": hisparse}))
+    for name, arg, kwargs in warmup_calls:
         expected.append(getattr(call, name)(arg, **kwargs))
         if failure == name:
             break
@@ -567,8 +570,8 @@ def autotune_run(monkeypatch, tmp_path):
         )
         monkeypatch.setattr(
             warmup,
-            "resolve_flashinfer_autotune_file",
-            lambda runner: tmp_path / "autotune_configs.json",
+            "resolve_rank_flashinfer_autotune_file",
+            lambda runner: run.cache_path(run.rank),
         )
         monkeypatch.setattr(
             warmup, "_flashinfer_autotune_skip_ops", lambda runner: None

vllm/entrypoints/cli/preload.py

@@ -100,6 +100,37 @@ def _start_health_server(
     return server, thread
 
 
+def _validate_seed_args(
+    args: argparse.Namespace, *, world_size: int, num_groups: int
+) -> None:
+    """Validate the seed CLI options before any daemon spawns."""
+    for flag, host, port in (
+        ("--seed-listen", args.seed_listen_host, args.seed_listen_port),
+        ("--seed-source", args.seed_source_host, args.seed_source_port),
+    ):
+        if (host is None) != (port is None):
+            raise ValueError(f"{flag}-host and {flag}-port must be provided together")
+    has_endpoint = any(
+        value is not None
+        for value in (
+            args.seed_listen_host,
+            args.seed_listen_port,
+            args.seed_source_host,
+            args.seed_source_port,
+        )
+    )
+    if has_endpoint != bool(args.seed_token):
+        raise ValueError("Preload daemons must set --seed-token to enable D2D transfer")
+    for base_port in (args.seed_listen_port, args.seed_source_port):
+        if base_port is None:
+            continue
+        max_port = base_port + num_groups * world_size - 1
+        if not 1 <= base_port <= 65535 or max_port > 65535:
+            raise ValueError(
+                f"Daemon-to-daemon port range {base_port}-{max_port} exceeds 65535"
+            )
+
+
 class PreloadSubcommand(CLISubcommand):
     """The `preload` subcommand for the vLLM CLI."""
 
@@ -143,6 +174,39 @@ class PreloadSubcommand(CLISubcommand):
             default="0.0.0.0",
             help="Host for the parent /health endpoint (default: 0.0.0.0).",
         )
+        parser.add_argument(
+            "--seed-listen-host",
+            type=str,
+            default=None,
+            help="Host to bind for authenticated daemon-to-daemon requests.",
+        )
+        parser.add_argument(
+            "--seed-listen-port",
+            type=int,
+            default=None,
+            help="Base port for daemon-to-daemon requests; each daemon binds "
+            "base_port + group offset + its global rank.",
+        )
+        parser.add_argument(
+            "--seed-source-host",
+            type=str,
+            default=None,
+            help="Host of a source daemon replica to seed warmup caches from.",
+        )
+        parser.add_argument(
+            "--seed-source-port",
+            type=int,
+            default=None,
+            help="Base port of the source daemon replica; each daemon dials "
+            "base_port + group offset + its global rank.",
+        )
+        parser.add_argument(
+            "--seed-token",
+            type=str,
+            default=None,
+            help="Shared secret for daemon-to-daemon requests. The JSON "
+            "connection is unencrypted; use a trusted network.",
+        )
 
     @staticmethod
     def cmd(args: argparse.Namespace) -> None:
@@ -153,6 +217,7 @@ class PreloadSubcommand(CLISubcommand):
             get_draft_daemon_config,
             plan_local_ranks,
         )
+        from vllm.model_executor.model_loader.weight_cache.seed import SeedConfig
         from vllm.model_executor.model_loader.weight_cache.utils import (
             format_daemon_role,
         )
@@ -179,6 +244,7 @@ class PreloadSubcommand(CLISubcommand):
         tp_size = parallel_config.tensor_parallel_size
         pp_size = parallel_config.pipeline_parallel_size
         dp_size = parallel_config.data_parallel_size
+        world_size = dp_size * pp_size * tp_size
 
         placements = plan_local_ranks(parallel_config)
         local_world_size = len(placements)
@@ -248,32 +314,54 @@ class PreloadSubcommand(CLISubcommand):
                 groups, placements
             )
         }
-        procs = [
-            ctx.Process(
-                target=_run_daemon,
-                args=(
-                    dp_rank * pp_size * tp_size + pp_rank * tp_size + tp_rank,
-                    local_rank,
-                    config,
-                    init_method,
-                    args.weight_cache_socket_dir,
-                    ready_queue,
-                    is_draft,
-                    dp_rank,
-                    pp_rank,
-                ),
-                name=(
-                    f"vllm-weight-cache-{format_daemon_role(is_draft)}-"
-                    f"{(dp_rank * pp_size * tp_size + pp_rank * tp_size + tp_rank)}"
-                ),
-            )
-            for (is_draft, config, init_method), (
-                local_rank,
-                dp_rank,
-                pp_rank,
-                tp_rank,
-            ) in product(groups, placements)
-        ]
+        _validate_seed_args(args, world_size=world_size, num_groups=len(groups))
+
+        def _seed_endpoint(
+            host: str | None, base_port: int | None, offset: int
+        ) -> tuple[str, int] | None:
+            # Validated above: host and port are provided together.
+            if host is None or base_port is None:
+                return None
+            return host, base_port + offset
+
+        procs = []
+        for group_index, (is_draft, config, init_method) in enumerate(groups):
+            # Draft daemons live past the target group's port block so both
+            # roles can be reached over a single seed host:base_port.
+            port_offset = group_index * world_size
+            for local_rank, dp_rank, pp_rank, tp_rank in placements:
+                global_rank = dp_rank * pp_size * tp_size + pp_rank * tp_size + tp_rank
+                offset = port_offset + global_rank
+                seed_config = SeedConfig(
+                    listen=_seed_endpoint(
+                        args.seed_listen_host, args.seed_listen_port, offset
+                    ),
+                    source=_seed_endpoint(
+                        args.seed_source_host, args.seed_source_port, offset
+                    ),
+                    token=args.seed_token,
+                )
+                procs.append(
+                    ctx.Process(
+                        target=_run_daemon,
+                        args=(
+                            global_rank,
+                            local_rank,
+                            config,
+                            init_method,
+                            args.weight_cache_socket_dir,
+                            ready_queue,
+                            is_draft,
+                            dp_rank,
+                            pp_rank,
+                            seed_config,
+                        ),
+                        name=(
+                            f"vllm-weight-cache-{format_daemon_role(is_draft)}-"
+                            f"{global_rank}"
+                        ),
+                    )
+                )
         health_state = _HealthState(procs)
         health_server = None
         health_thread = None

vllm/model_executor/model_loader/weight_cache/daemon.py

@@ -119,15 +119,19 @@ from vllm.model_executor.model_loader.weight_cache.protocol import (
     send_msg,
     verify_peer_is_owner,
 )
+from vllm.model_executor.model_loader.weight_cache.seed import (
+    SeedConfig,
+    SeedTransport,
+)
 from vllm.model_executor.model_loader.weight_cache.utils import (
-    build_warmup_runner,
     export_model_attrs,
     format_daemon_role,
     is_draft_model_cacheable,
 )
-from vllm.model_executor.warmup.kernel_warmup import flashinfer_autotune
+from vllm.model_executor.model_loader.weight_cache.warmup_cache import (
+    FlashInferWarmupCache,
+)
 from vllm.platforms import current_platform
-from vllm.utils.flashinfer import has_flashinfer
 from vllm.utils.mem_utils import format_gib
 from vllm.utils.torch_utils import set_default_torch_dtype
 from vllm.v1.worker.workspace import init_workspace_manager
@@ -212,10 +216,13 @@ class WeightCacheDaemon:
         is_draft: bool = False,
         dp_rank: int = 0,
         pp_rank: int = 0,
+        seed_config: SeedConfig | None = None,
     ):
         parallel_config = vllm_config.parallel_config
         if parallel_config.enable_eplb:
             raise ValueError("The weight cache daemon does not support EPLB.")
+        self.warmup_caches = [FlashInferWarmupCache()]
+        self.d2d = SeedTransport(self.warmup_caches, seed_config)
         self.tp_size = parallel_config.tensor_parallel_size
         self.pp_size = parallel_config.pipeline_parallel_size
         self.dp_size = parallel_config.data_parallel_size
@@ -277,39 +284,9 @@ class WeightCacheDaemon:
         )
 
     def warmup(self) -> None:
-        """Run the FlashInfer autotune pass against the cached model."""
-        vllm_config = self.vllm_config
-        # Mirrors the gate in the engine's kernel warmup.
-        if (
-            vllm_config.kernel_config.enable_flashinfer_autotune is False
-            or not has_flashinfer()
-            or not current_platform.has_device_capability(90)
-            or is_draft_model_cacheable(vllm_config.speculative_config)
-        ):
-            return
-        try:
-            current_platform.update_block_size_for_backend(vllm_config)
-            runner = build_warmup_runner(
-                vllm_config,
-                self.local_rank,
-                is_draft=self.is_draft,
-                model_config=self.model_config,
-            )
-            assert self.model is not None, "warmup ran before load_model"
-            runner.load_model(model=self.model)
-            # The V2 runner requires skip_attn when no KV cache exists
-            flashinfer_autotune(runner, skip_attn=vllm_config.use_v2_model_runner)
-            logger.info(
-                "Weight cache %s daemon rank %d tuned FlashInfer; the tuned "
-                "configs are in the on-disk autotune cache",
-                self.role,
-                self.global_rank,
-            )
-        finally:
-            # Free the runner and the dummy-run activations; the daemon's
-            # weights are untouched.
-            gc.collect()
-            torch.accelerator.empty_cache()
+        """Run each warmup cache's pass (adopting remote caches via D2D)."""
+        for cache in self.warmup_caches:
+            cache.warmup(self, self.d2d)
 
     def load_model(self) -> None:
         torch.accelerator.set_device_index(self.local_rank)
@@ -418,6 +395,7 @@ class WeightCacheDaemon:
         server.bind(socket_path)
         os.chmod(socket_path, 0o600)
         server.listen()
+        self.d2d.start_listener()
         logger.info(
             "Weight cache %s daemon rank %d serving on %s",
             self.role,
@@ -445,6 +423,7 @@ class WeightCacheDaemon:
                             send_msg(conn, {"status": "error", "message": str(e)})
         finally:
             server.close()
+            self.d2d.close()
             if os.path.exists(socket_path):
                 os.unlink(socket_path)
             os.close(lock_fd)
@@ -553,6 +532,7 @@ def _run_daemon(
     is_draft: bool = False,
     dp_rank: int = 0,
     pp_rank: int = 0,
+    seed_config: SeedConfig | None = None,
 ) -> None:
     # Spawned daemons start with logging unconfigured, like engine workers.
     if logging_config := getattr(vllm_config, "logging_config", None):
@@ -566,6 +546,7 @@ def _run_daemon(
         is_draft,
         dp_rank,
         pp_rank,
+        seed_config=seed_config,
     )
     daemon.load_model()
     daemon.warmup()

vllm/model_executor/model_loader/weight_cache/protocol.py

@@ -19,6 +19,7 @@ import socket
 import stat
 import struct
 import tempfile
+import time
 from dataclasses import dataclass, fields
 from typing import Any, NamedTuple
 
@@ -46,6 +47,9 @@ _LEN_STRUCT = struct.Struct("!Q")
 # Sanity bound for a single message. IPC handles are tiny; only small
 # non-CUDA tensors are ever shipped by value.
 MAX_MSG_SIZE = 1 << 34
+# Bound for one JSON message on the remote warmup-cache control plane.
+# Warmup cache tables are small JSON documents.
+MAX_JSON_MSG_SIZE = 1 << 26
 
 
 def _current_uid() -> int:
@@ -452,9 +456,45 @@ def recv_msg(sock: socket.socket) -> Any:
     return pickle.loads(_recv_exact(sock, length))
 
 
-def _recv_exact(sock: socket.socket, num_bytes: int) -> bytes:
+def send_json(sock: socket.socket, obj: Any) -> None:
+    payload = json.dumps(obj).encode()
+    if len(payload) > MAX_JSON_MSG_SIZE:
+        raise ValueError("JSON message exceeds configured limit")
+    sock.sendall(_LEN_STRUCT.pack(len(payload)))
+    sock.sendall(payload)
+
+
+def recv_json(
+    sock: socket.socket,
+    max_size: int = MAX_JSON_MSG_SIZE,
+    deadline: float | None = None,
+) -> Any:
+    """Receive one JSON message.
+
+    Args:
+        sock: Connected socket.
+        max_size: Largest payload accepted.
+        deadline: ``time.monotonic()`` by which the whole message must have
+            arrived; a per-read timeout alone lets a peer trickle bytes
+            forever.
+
+    """
+    (length,) = _LEN_STRUCT.unpack(_recv_exact(sock, _LEN_STRUCT.size, deadline))
+    if length > max_size:
+        raise ValueError(f"Message size {length} exceeds limit {max_size}")
+    return json.loads(_recv_exact(sock, length, deadline))
+
+
+def _recv_exact(
+    sock: socket.socket, num_bytes: int, deadline: float | None = None
+) -> bytes:
     buf = bytearray()
     while len(buf) < num_bytes:
+        if deadline is not None:
+            remaining = deadline - time.monotonic()
+            if remaining <= 0:
+                raise TimeoutError("Peer did not send its message in time")
+            sock.settimeout(remaining)
         chunk = sock.recv(num_bytes - len(buf))
         if not chunk:
             raise ConnectionError("Socket closed while receiving message")

vllm/model_executor/model_loader/weight_cache/seed.py

@@ -0,0 +1,232 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""Daemon-to-daemon seeding over an authenticated JSON channel.
+
+A :class:`Seedable` is one kind of state a daemon can seed to or from a peer
+daemon: warmup caches today, weight shards later. The target names what it
+wants by kind and key fields; the source resolves the key to a payload; the
+target validates and installs it:
+
+    request:  {"kind": "flashinfer_autotune", <key fields>..., "token": "..."}
+    response: {"status": "ok", "payload": <kind-specific>}
+              {"status": "miss"} | {"status": "error", "message": "..."}
+
+Payloads are JSON values. Large binary state (weights) must not be inlined:
+the payload then describes the transfer (e.g. a storage manifest) and the
+kind's ``install`` drives the data plane itself.
+
+The connection is unencrypted beyond the shared token; keep the listener on a
+trusted network.
+"""
+
+import contextlib
+import hmac
+import socket
+import threading
+import time
+from collections.abc import Iterable
+from typing import Any, NamedTuple, Protocol
+
+from vllm.logger import init_logger
+from vllm.model_executor.model_loader.weight_cache.protocol import (
+    recv_json,
+    send_json,
+)
+
+logger = init_logger(__name__)
+
+# Requests carry a kind and small key fields; payloads flow in responses.
+_MAX_REMOTE_REQUEST_SIZE = 64 << 10
+_REQUEST_TIMEOUT_S = 30.0
+# An unauthenticated peer gets a whole-message deadline, not a per-read
+# timeout: per-read timeouts let a peer trickle bytes forever and stall the
+# serial accept loop. Authenticated handlers (e.g. weight manifests) may be
+# slow, so the budget relaxes once the token checks out.
+_REMOTE_REQUEST_TIMEOUT_S = 10.0
+_AUTHENTICATED_TIMEOUT_S = 300.0
+
+
+class Seedable(Protocol):
+    """One kind of state a daemon can seed to or from a peer daemon."""
+
+    kind: str
+
+    def request_params(self, ctx: Any) -> dict[str, Any]:
+        """Target side: the key fields identifying what this rank needs."""
+        ...
+
+    def resolve(self, request: dict[str, Any]) -> Any | None:
+        """Source side: produce the payload named by ``request``.
+
+        Validate the wire key first — raise an error on any malformed field
+        (``ValueError`` for bad values, ``KeyError`` for missing ones), before
+        touching local state. The transport turns it into an error response.
+        Return None on a miss.
+        """
+        ...
+
+    def validate_payload(self, payload: Any) -> Any:
+        """Target side: validate the fetched payload, or raise ValueError."""
+        ...
+
+    def install(self, ctx: Any, payload: Any) -> None:
+        """Target side: install a validated payload."""
+        ...
+
+
+class SeedConfig(NamedTuple):
+    """One daemon's D2D seeding configuration: its listener, its source
+    replica, and the shared token authenticating both."""
+
+    listen: tuple[str, int] | None = None
+    source: tuple[str, int] | None = None
+    token: str | None = None
+
+
+class SeedTransport:
+    """The daemon's D2D seeding endpoint: serves and fetches seedables."""
+
+    def __init__(
+        self,
+        seedables: Iterable[Seedable],
+        seed_config: SeedConfig | None = None,
+    ):
+        seed_config = seed_config or SeedConfig()
+        listen, source, token = seed_config
+        if (listen is not None or source is not None) and not token:
+            raise ValueError("A daemon-to-daemon listener/source requires a token")
+        self._seedables = list(seedables)
+        self._seed_config = seed_config
+        self._server: socket.socket | None = None
+
+    def adopt(self, seedable: Seedable, ctx: Any) -> None:
+        """Best-effort: fetch and install ``seedable`` from the source daemon.
+
+        Any failure degrades to the local path with a log line.
+        """
+        response = self.request(
+            {"kind": seedable.kind, **seedable.request_params(ctx)},
+        )
+        if response is None:
+            return
+        status = response.get("status")
+        if status != "ok":
+            log = logger.info if status == "miss" else logger.warning
+            log(
+                "No remote %s seed adopted (status %s); using the local path",
+                seedable.kind,
+                status,
+            )
+            return
+        try:
+            seedable.install(ctx, seedable.validate_payload(response.get("payload")))
+        except Exception as e:
+            logger.warning(
+                "Rejected the remote %s seed; using the local path: %s",
+                seedable.kind,
+                e,
+            )
+            return
+        logger.info("Adopted remote %s seed", seedable.kind)
+
+    def handle_request(self, request: Any) -> dict[str, Any]:
+        """Source side: answer one authenticated seed request."""
+        kind_name = request.get("kind") if isinstance(request, dict) else None
+        seedable = next((s for s in self._seedables if s.kind == kind_name), None)
+        if seedable is None:
+            return {"status": "error", "message": "unsupported seed kind"}
+        try:
+            payload = seedable.resolve(request)
+        except (OSError, KeyError, TypeError, ValueError) as e:
+            return {"status": "error", "message": str(e)}
+        if payload is None:
+            return {"status": "miss"}
+        return {"status": "ok", "payload": payload}
+
+    def request(self, payload: dict[str, Any]) -> dict[str, Any] | None:
+        """Send one authenticated request to the source daemon.
+
+        Returns the response, or None with a log line on any transport
+        failure; the caller falls back to its local path then.
+        """
+        source = self._seed_config.source
+        if source is None:
+            return None
+        label = payload.get("kind", "payload")
+        try:
+            with socket.create_connection(source, timeout=_REQUEST_TIMEOUT_S) as sock:
+                send_json(sock, {**payload, "token": self._seed_config.token})
+                response = recv_json(sock)
+        except (OSError, ValueError) as e:
+            logger.warning(
+                "Cannot fetch %s from %s over D2D; falling back locally: %s",
+                label,
+                source,
+                e,
+            )
+            return None
+        if not isinstance(response, dict):
+            logger.warning(
+                "Source %s sent a malformed response for %s; falling back locally",
+                source,
+                label,
+            )
+            return None
+        return response
+
+    def start_listener(self) -> tuple[str, int] | None:
+        """Start serving this rank's seedables to peer daemons.
+
+        Returns the bound address, or None when no listener is configured.
+        """
+        listen = self._seed_config.listen
+        if listen is None:
+            return None
+        family = socket.AF_INET6 if ":" in listen[0] else socket.AF_INET
+        self._server = socket.create_server(listen, family=family)
+        threading.Thread(
+            target=self._serve,
+            name="vllm-weight-cache-d2d",
+            daemon=True,
+        ).start()
+        host, port = self._server.getsockname()[:2]
+        return host, port
+
+    def close(self) -> None:
+        if self._server is not None:
+            self._server.close()
+            self._server = None
+
+    def _serve(self) -> None:
+        assert self._server is not None
+        while True:
+            try:
+                conn, _ = self._server.accept()
+            except OSError:
+                return
+            with conn:
+                try:
+                    request = recv_json(
+                        conn,
+                        max_size=_MAX_REMOTE_REQUEST_SIZE,
+                        deadline=time.monotonic() + _REMOTE_REQUEST_TIMEOUT_S,
+                    )
+                    presented = (
+                        request.get("token") if isinstance(request, dict) else None
+                    )
+                    if (
+                        not self._seed_config.token
+                        or not isinstance(presented, str)
+                        or not hmac.compare_digest(presented, self._seed_config.token)
+                    ):
+                        send_json(
+                            conn,
+                            {"status": "error", "message": "Unauthorized request"},
+                        )
+                        continue
+                    # Authenticated handlers (e.g. weight manifests) may be slow.
+                    conn.settimeout(_AUTHENTICATED_TIMEOUT_S)
+                    send_json(conn, self.handle_request(request))
+                except (OSError, ValueError) as e:
+                    with contextlib.suppress(OSError):
+                        send_json(conn, {"status": "error", "message": str(e)})

vllm/model_executor/model_loader/weight_cache/warmup_cache.py

@@ -0,0 +1,183 @@
+# SPDX-License-Identifier: Apache-2.0
+# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+"""Daemon warmup caches, seeded daemon-to-daemon over the shared transport.
+
+Each warmup-cache object (currently only :class:`FlashInferWarmupCache`)
+drives its own warmup pass on the daemon; the seeding flow itself
+(:class:`seed.SeedTransport`) is shared with future kinds such as weight
+shards. A target daemon installs its rank-matched cache before tuning, so
+both the daemon's own warmup and the later engine start skip re-tuning. The
+disk layout is the index: a request names the cache by the same key the
+layout uses (kind, config hash, dp rank, world rank) and the source resolves
+it to files under the cache root. There is no separate artifact store or
+registry.
+
+The FlashInfer payload is a ``{filename: parsed JSON}`` map rather than a
+single document because upcoming kinds (Triton tuned configs) consist of
+several JSON files per cache.
+
+TODO: install valid files individually instead of falling back to local
+warmup when any file of a multi-file kind fails validation.
+"""
+
+import gc
+import json
+from typing import TYPE_CHECKING, Any, Protocol
+
+import regex as re
+import torch
+
+from vllm.logger import init_logger
+from vllm.model_executor.model_loader.weight_cache.seed import Seedable
+from vllm.model_executor.model_loader.weight_cache.utils import (
+    build_warmup_runner,
+    is_draft_model_cacheable,
+)
+from vllm.platforms import current_platform
+from vllm.utils.flashinfer import has_flashinfer
+
+if TYPE_CHECKING:
+    from vllm.model_executor.model_loader.weight_cache.daemon import (
+        WeightCacheDaemon,
+    )
+    from vllm.model_executor.model_loader.weight_cache.seed import (
+        SeedTransport,
+    )
+    from vllm.v1.worker.gpu_model_runner import GPUModelRunner
+
+logger = init_logger(__name__)
+
+_CONFIG_HASH_RE = re.compile(r"[0-9a-f]{64}")
+
+
+class WarmupCache(Seedable, Protocol):
+    """One daemon warmup cache: its warmup pass and its cache's semantics.
+
+    The warmup layer; the D2D exchange lives in seed.py (SeedTransport).
+    """
+
+    def warmup(self, daemon: "WeightCacheDaemon", transport: "SeedTransport") -> None:
+        """Run this rank's warmup pass against the cached model."""
+        ...
+
+
+class FlashInferWarmupCache(WarmupCache):
+    """FlashInfer's per-rank autotune table (a single JSON file)."""
+
+    kind = "flashinfer_autotune"
+
+    def warmup(self, daemon: "WeightCacheDaemon", transport: "SeedTransport") -> None:
+        vllm_config = daemon.vllm_config
+        # Mirrors the gate in the engine's kernel warmup.
+        if (
+            vllm_config.kernel_config.enable_flashinfer_autotune is False
+            or not has_flashinfer()
+            or not current_platform.has_device_capability(90)
+            or is_draft_model_cacheable(vllm_config.speculative_config)
+        ):
+            return
+        from vllm.model_executor.warmup.kernel_warmup import flashinfer_autotune
+
+        try:
+            current_platform.update_block_size_for_backend(vllm_config)
+            runner = build_warmup_runner(
+                vllm_config,
+                daemon.local_rank,
+                is_draft=daemon.is_draft,
+                model_config=daemon.model_config,
+            )
+            assert daemon.model is not None, "warmup ran before load_model"
+            runner.load_model(model=daemon.model)
+            transport.adopt(self, runner)
+            # The V2 runner requires skip_attn when no KV cache exists
+            flashinfer_autotune(runner, skip_attn=vllm_config.use_v2_model_runner)
+            logger.info(
+                "Weight cache %s daemon rank %d tuned FlashInfer; the tuned "
+                "configs are in the on-disk autotune cache",
+                daemon.role,
+                daemon.global_rank,
+            )
+        finally:
+            # Free the runner and the dummy-run activations; the daemon's
+            # weights are untouched.
+            gc.collect()
+            torch.accelerator.empty_cache()
+
+    def request_params(self, runner: "GPUModelRunner") -> dict[str, Any]:
+        from vllm.distributed.parallel_state import get_world_group
+        from vllm.model_executor.warmup.flashinfer_autotune_cache import (
+            flashinfer_autotune_cache_hash,
+        )
+
+        return {
+            "config_hash": flashinfer_autotune_cache_hash(runner),
+            "dp_rank": runner.vllm_config.parallel_config.data_parallel_rank,
+            "rank": get_world_group().rank_in_group,
+        }
+
+    def resolve(self, request: dict[str, Any]) -> dict[str, Any] | None:
+        from vllm.model_executor.warmup.flashinfer_autotune_cache import (
+            flashinfer_autotune_filename,
+            flashinfer_autotune_root,
+        )
+
+        config_hash = request.get("config_hash")
+        if not isinstance(config_hash, str) or not _CONFIG_HASH_RE.fullmatch(
+            config_hash
+        ):
+            raise ValueError("invalid config_hash")
+
+        # Garbage ranks simply name a file that does not exist (a miss); the
+        # hex check above is what keeps the request inside the cache root.
+        filename = flashinfer_autotune_filename(request["dp_rank"], request["rank"])
+        path = flashinfer_autotune_root() / config_hash / filename
+        if not path.is_file():
+            return None
+        return {filename: json.loads(path.read_bytes())}
+
+    def validate_payload(self, payload: Any) -> dict[str, Any]:
+        if not isinstance(payload, dict) or len(payload) != 1:
+            raise ValueError("expected exactly one autotune table")
+        ((filename, table),) = payload.items()
+        if not isinstance(table, dict):
+            raise ValueError("autotune table must be a JSON object")
+        if not filename.startswith("autotune_configs_"):
+            raise ValueError(f"unexpected file {filename!r}")
+        # Reject tables flashinfer would refuse to load: a hard _metadata
+        # conflict would also never be overwritten, poisoning this cache
+        # path into re-tuning on every start.
+        saved_meta = table.get("_metadata")
+        if saved_meta is None:
+            # Pre-metadata flashinfer file; load_configs accepts these.
+            return payload
+        try:
+            from flashinfer.autotuner.autotuner import (
+                _classify_metadata_mismatches,
+                _collect_metadata,
+            )
+        except ImportError:
+            logger.warning(
+                "flashinfer metadata classifier unavailable; "
+                "accepting the remote autotune table unchecked"
+            )
+            return payload
+        hard, soft = _classify_metadata_mismatches(saved_meta, _collect_metadata())
+        mismatches = {**hard, **soft}
+        if mismatches:
+            details = ", ".join(
+                f"{key}: remote={old} vs local={new}"
+                for key, (old, new) in mismatches.items()
+            )
+            raise ValueError(f"autotune table environment mismatch: {details}")
+        return payload
+
+    def install(self, runner: "GPUModelRunner", payload: dict[str, Any]) -> None:
+        from vllm.model_executor.warmup.flashinfer_autotune_cache import (
+            resolve_rank_flashinfer_autotune_file,
+            write_flashinfer_autotune_cache,
+        )
+
+        (data,) = payload.values()
+        cache_path = resolve_rank_flashinfer_autotune_file(runner)
+        write_flashinfer_autotune_cache(cache_path, json.dumps(data).encode())
+        logger.info("Installed remote FlashInfer autotune table: %s", cache_path)

vllm/model_executor/warmup/flashinfer_autotune_cache.py

@@ -38,26 +38,44 @@ def flashinfer_autotune_cache_hash(runner: "GPUModelRunner") -> str:
     return hashlib.sha256(config_hash.encode()).hexdigest()
 
 
-def resolve_flashinfer_autotune_file(runner: "GPUModelRunner") -> Path:
+def flashinfer_autotune_root() -> Path:
+    """Root directory of the FlashInfer autotune cache (no runner needed)."""
     override_dir = envs.VLLM_FLASHINFER_AUTOTUNE_CACHE_DIR
     if override_dir:
-        root = Path(override_dir).expanduser()
-    else:
-        from flashinfer.jit import env as flashinfer_jit_env
+        return Path(override_dir).expanduser()
+    from flashinfer.jit import env as flashinfer_jit_env
 
-        flashinfer_workspace = flashinfer_jit_env.FLASHINFER_WORKSPACE_DIR
-        root = (
-            Path(envs.VLLM_CACHE_ROOT)
-            / "flashinfer_autotune_cache"
-            / flashinfer_workspace.parent.name
-            / flashinfer_workspace.name
-        )
+    flashinfer_workspace = flashinfer_jit_env.FLASHINFER_WORKSPACE_DIR
+    return (
+        Path(envs.VLLM_CACHE_ROOT)
+        / "flashinfer_autotune_cache"
+        / flashinfer_workspace.parent.name
+        / flashinfer_workspace.name
+    )
 
-    output_dir = root / flashinfer_autotune_cache_hash(runner)
+
+def flashinfer_autotune_filename(dp_rank: int, rank: int) -> str:
+    """Per-rank cache filename, shared by the engine reader and D2D exchange."""
+    return f"autotune_configs_dp{dp_rank}_rank{rank}.json"
+
+
+def resolve_flashinfer_autotune_file(runner: "GPUModelRunner") -> Path:
+    output_dir = flashinfer_autotune_root() / flashinfer_autotune_cache_hash(runner)
     output_dir.mkdir(parents=True, exist_ok=True)
     return output_dir / "autotune_configs.json"
 
 
+def resolve_rank_flashinfer_autotune_file(runner: "GPUModelRunner") -> Path:
+    """The per-rank variant of :func:`resolve_flashinfer_autotune_file`."""
+    from vllm.distributed.parallel_state import get_world_group
+
+    path = resolve_flashinfer_autotune_file(runner)
+    dp_rank = runner.vllm_config.parallel_config.data_parallel_rank
+    return path.with_name(
+        flashinfer_autotune_filename(dp_rank, get_world_group().rank_in_group)
+    )
+
+
 def use_flashinfer_autotune_v2(runner: "GPUModelRunner") -> bool:
     from vllm.distributed.parallel_state import get_node_count, get_world_group
     from vllm.utils.flashinfer import has_flashinfer_autotune_v2

vllm/model_executor/warmup/kernel_warmup.py

@@ -22,8 +22,8 @@ from vllm.model_executor.warmup.b12x_warmup import b12x_warmup
 from vllm.model_executor.warmup.cutedsl_warmup import cutedsl_warmup
 from vllm.model_executor.warmup.deep_gemm_warmup import deep_gemm_warmup
 from vllm.model_executor.warmup.flashinfer_autotune_cache import (
-    resolve_flashinfer_autotune_file,
     resolve_flashinfer_autotune_v2_root,
+    resolve_rank_flashinfer_autotune_file,
     use_flashinfer_autotune_v2,
 )
 from vllm.model_executor.warmup.flashinfer_sparse_mla_warmup import (
@@ -405,6 +405,27 @@ def _all_ranks_have_matching_cache(path: Path, group) -> bool:
     return fingerprint is not None and all(f == fingerprint for f in gathered)
 
 
+def _load_shared_autotune_cache(cache_path: Path, tuner, tune_group) -> bool:
+    """Load the per-rank autotune cache into the tuner, group-wide.
+
+    Returns True only when every rank of the tune group had a matching file
+    and loaded it: the dummy-run sweep contains collectives, so skipping it
+    is safe only if the decision is unanimous.
+    """
+    if not _all_ranks_have_matching_cache(cache_path, tune_group):
+        return False
+    loaded = tuner.load_configs(str(cache_path))
+    if tune_group.world_size > 1:
+        loaded_by_rank: list[bool | None] = [None] * tune_group.world_size
+        torch.distributed.all_gather_object(
+            loaded_by_rank, loaded, group=tune_group.cpu_group
+        )
+        loaded = all(loaded_by_rank)
+    if not loaded:
+        tuner.clear_cache()
+    return loaded
+
+
 def flashinfer_autotune(runner: "GPUModelRunner", *, skip_attn: bool = False) -> None:
     """Autotune FlashInfer operations.
     FlashInfer have many implementations for the same operation,
@@ -455,6 +476,7 @@ def flashinfer_autotune(runner: "GPUModelRunner", *, skip_attn: bool = False) ->
         autotune_kwargs["skip_ops"] = skip_ops
 
     use_autotune_v2 = use_flashinfer_autotune_v2(runner)
+    cache_hit = False
     if use_autotune_v2:
         from flashinfer import autotune_v2, autotune_v2_reload
 
@@ -469,12 +491,7 @@ def flashinfer_autotune(runner: "GPUModelRunner", *, skip_attn: bool = False) ->
             mode="tune", cache_root=cache_root, **autotune_kwargs
         )
     else:
-        cache_path = resolve_flashinfer_autotune_file(runner)
-        # The world group only spans DP ranks when vLLM folds DP into it.
-        dp_rank = runner.vllm_config.parallel_config.data_parallel_rank
-        cache_path = cache_path.with_name(
-            f"{cache_path.stem}_dp{dp_rank}_rank{world.rank_in_group}{cache_path.suffix}"
-        )
+        cache_path = resolve_rank_flashinfer_autotune_file(runner)
         if is_leader:
             logger.info_once(
                 "Using FlashInfer autotune cache file: %s. Pass "
@@ -482,16 +499,7 @@ def flashinfer_autotune(runner: "GPUModelRunner", *, skip_attn: bool = False) ->
                 cache_path,
             )
 
-        if _all_ranks_have_matching_cache(cache_path, tune_group):
-            loaded = tuner.load_configs(str(cache_path))
-            if tune_group.world_size > 1:
-                loaded_by_rank: list[bool | None] = [None] * tune_group.world_size
-                torch.distributed.all_gather_object(
-                    loaded_by_rank, loaded, group=tune_group.cpu_group
-                )
-                loaded = all(loaded_by_rank)
-            if not loaded:
-                tuner.clear_cache()
+        cache_hit = _load_shared_autotune_cache(cache_path, tuner, tune_group)
         autotune_context = fi_utils.autotune(tune_mode=True, **autotune_kwargs)
 
     group = tune_group.cpu_group if tune_group.world_size > 1 else None
@@ -506,7 +514,8 @@ def flashinfer_autotune(runner: "GPUModelRunner", *, skip_attn: bool = False) ->
                 # HiSparse hot-buffer attention is bounded by decode batch
                 # size, not the prefill-sized batch used for the full model.
                 autotune_hisparse_flashinfer_attention(runner)
-            _run_flashinfer_autotune_dummy_runs(runner, skip_attn=skip_attn)
+            if not cache_hit:
+                _run_flashinfer_autotune_dummy_runs(runner, skip_attn=skip_attn)
             replayssm_autotune_warmup(runner)
             _autotune_kimi_k3_kda_qkvg(runner.get_model())
     finally: