Skip to content

vllm.model_executor.model_loader.weight_cache.seed

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.

Classes:

  • SeedConfig –

    One daemon's D2D seeding configuration: its listener, its source

  • SeedTransport –

    The daemon's D2D seeding endpoint: serves and fetches seedables.

  • Seedable –

    One kind of state a daemon can seed to or from a peer daemon.

SeedConfig

Bases: NamedTuple

One daemon's D2D seeding configuration: its listener, its source replica, and the shared token authenticating both.

Source code in vllm/model_executor/model_loader/weight_cache/seed.py 
77
78
79
80
81
82
83
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

SeedTransport

The daemon's D2D seeding endpoint: serves and fetches seedables.

Methods:

  • adopt –

    Best-effort: fetch and install seedable from the source daemon.

  • handle_request –

    Source side: answer one authenticated seed request.

  • request –

    Send one authenticated request to the source daemon.

  • start_listener –

    Start serving this rank's seedables to peer daemons.

Source code in vllm/model_executor/model_loader/weight_cache/seed.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
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)})

adopt(seedable, ctx)

Best-effort: fetch and install seedable from the source daemon.

Any failure degrades to the local path with a log line.

Source code in vllm/model_executor/model_loader/weight_cache/seed.py 
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
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)

handle_request(request)

Source side: answer one authenticated seed request.

Source code in vllm/model_executor/model_loader/weight_cache/seed.py 
132
133
134
135
136
137
138
139
140
141
142
143
144
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}

request(payload)

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 code in vllm/model_executor/model_loader/weight_cache/seed.py 
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
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

start_listener()

Start serving this rank's seedables to peer daemons.

Returns the bound address, or None when no listener is configured.

Source code in vllm/model_executor/model_loader/weight_cache/seed.py 
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
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

Seedable

Bases: Protocol

One kind of state a daemon can seed to or from a peer daemon.

Methods:

  • install –

    Target side: install a validated payload.

  • request_params –

    Target side: the key fields identifying what this rank needs.

  • resolve –

    Source side: produce the payload named by request.

  • validate_payload –

    Target side: validate the fetched payload, or raise ValueError.

Source code in vllm/model_executor/model_loader/weight_cache/seed.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
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."""
        ...

install(ctx, payload)

Target side: install a validated payload.

Source code in vllm/model_executor/model_loader/weight_cache/seed.py 
72
73
74
def install(self, ctx: Any, payload: Any) -> None:
    """Target side: install a validated payload."""
    ...

request_params(ctx)

Target side: the key fields identifying what this rank needs.

Source code in vllm/model_executor/model_loader/weight_cache/seed.py 
54
55
56
def request_params(self, ctx: Any) -> dict[str, Any]:
    """Target side: the key fields identifying what this rank needs."""
    ...

resolve(request)

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.

Source code in vllm/model_executor/model_loader/weight_cache/seed.py 
58
59
60
61
62
63
64
65
66
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.
    """
    ...

validate_payload(payload)

Target side: validate the fetched payload, or raise ValueError.

Source code in vllm/model_executor/model_loader/weight_cache/seed.py 
68
69
70
def validate_payload(self, payload: Any) -> Any:
    """Target side: validate the fetched payload, or raise ValueError."""
    ...