@tduncan / Nvidia-Optimization-Stack-VLLM / commits / 0d72627565

[Perf][EPD] Batch image requests per encoder (#57095)

Signed-off-by: Tianyu Guo <guoty@inferact.ai> Signed-off-by: Zhou ziheng <jiaranran2@gmail.com> Co-authored-by: Zhou ziheng <jiaranran2@gmail.com> Co-authored-by: OpenAI Codex <codex@openai.com>

Tianyu Guo committed Sep 17, 2026 at 18:45 UTC 0d726275653800e82c324e1d32b074f782e89315

6 files changed +329 -59

examples/disaggregated/disaggregated_encoder/README.md

+16

@@ -106,6 +106,22 @@ If you enable prefill instance (`--prefill-servers-urls` not disabled), you will

106 | `--decode-servers-urls` | Comma-separated list of decode endpoints. Non-stream and stream paths both round-robin over this list. |

107 | `--host`, `--port` | Bind address for the proxy itself (defaults: `0.0.0.0:8000`). |

108

109 +The proxy batches images from the same user request assigned to the same encoder.

110 +Set the proxy environment variable `ENCODER_MAX_BATCH_SIZE` to limit the number

111 +of images per encoder subrequest. It defaults to `0` (unlimited); `1` sends each

112 +image separately. Audio and video remain separate subrequests.

113 +

114 +For encoders with a smaller `--limit-mm-per-prompt` image limit than P/PD, set

115 +`ENCODER_MAX_BATCH_SIZE` no higher than the smallest encoder image limit.

116 +For example, for encoders configured with `--limit-mm-per-prompt '{"image": 2}'`:

117 +

118 +```bash

119 +ENCODER_MAX_BATCH_SIZE=2 python disagg_epd_proxy.py \

120 + --encode-servers-urls "http://e1:8001,http://e2:8002" \

121 + --prefill-servers-urls disable \

122 + --decode-servers-urls "http://pd1:8003"

123 +```

124 +

125 Example usage:

126 For E + PD setup:

127

examples/disaggregated/disaggregated_encoder/disagg_epd_proxy.py

+88 -38

@@ -10,8 +10,8 @@ clusters:

11 For MM input we:

12 1. Extract *every* image/audio/video item.

13 - 2. Fire N concurrent requests to the encoder cluster

14 - (one request per item, with **all text removed**).

13 + 2. Send concurrent encoder requests with all text removed,

14 + grouping images assigned to the same encoder.

15 3. Wait for all of them to succeed.

16 4. Forward the *original* request to a decode server.

17 """

@@ -99,6 +99,11 @@ def validate_ec_consumer_routing(

99 # Diagnostic switch: forward the original request to the decoder so the

100 # only difference from the rewrite path is the rewrite itself.

101 NO_REWRITE = False

102 +# Maximum images per encoder subrequest; 0 leaves batches unlimited.

103 +ENCODER_MAX_BATCH_SIZE = int(os.getenv("ENCODER_MAX_BATCH_SIZE", "0"))

104 +if ENCODER_MAX_BATCH_SIZE < 0:

105 + raise ValueError("ENCODER_MAX_BATCH_SIZE must be non-negative")

106 +

107 # Decode-side retries for a retryable internal error (`finish_reason="error"`,

108 # e.g. an encoder embedding the connector could not deliver). Re-issuing runs

109 # the encode again, which produces a fresh transfer.

@@ -222,7 +227,7 @@ async def fanout_encoder_primer(

227 req_id: str,

228 consumer_zmq: str | None = None,

229 ) -> tuple[dict[int, dict], dict[str, Any]]:

225 - """1. Build one request *per MM item* with all text removed.

230 + """1. Group images by encoder, retaining per-item round-robin assignment.

231 2. Send them concurrently to the encode cluster.

232 3. Raise if any of them fails.

233

@@ -248,6 +253,7 @@ async def fanout_encoder_primer(

253 item_uuids: dict[int, str] = {}

254 item_transfer_ids: dict[int, str] = {}

255 item_meta: dict[int, dict] = {}

256 + transfer_items: dict[int, dict[str, str]] = {}

257 ec_params: dict[str, Any] = {}

258

259 # Round-robin over encode servers to distribute load a bit. The cursor

@@ -259,29 +265,45 @@ async def fanout_encoder_primer(

265 e_urls, encoder_rr_idx, len(mm_items)

266 )

267

268 + groups: list[tuple[str, list[int]]] = []

269 + image_groups: dict[str, list[int]] = {}

270 for idx, (item, target_url) in enumerate(zip(mm_items, url_cycle)):

271 + if item["type"] == "image_url":

272 + indices = image_groups.get(target_url)

273 + if indices is None or (

274 + ENCODER_MAX_BATCH_SIZE and len(indices) >= ENCODER_MAX_BATCH_SIZE

275 + ):

276 + indices = []

277 + image_groups[target_url] = indices

278 + groups.append((target_url, indices))

279 + indices.append(idx)

280 + else:

281 + groups.append((target_url, [idx]))

282 +

283 + for target_url, indices in groups:

284 # Derive a *child* request id: <parent>:<index>:<random-short>

264 - child_req_id = f"{req_id}:{idx}:{uuid.uuid4().hex[:6]}"

285 + child_req_id = f"{req_id}:{indices[0]}:{uuid.uuid4().hex[:6]}"

286 headers = {"x-request-id": child_req_id, "Content-Type": "application/json"}

287

288 # With --no-rewrite the decoder still receives the raw image and derives

289 # the cache key by hashing it, so the encoder must do the same -- passing

290 # a uuid here would make the two disagree and silently defeat the EC

291 # transfer, leaving the decoder to encode the image itself.

271 - item_uuid = None if NO_REWRITE else content_uuid(item)

272 - if item_uuid is not None:

273 - item_uuids[idx] = item_uuid

274 - transfer_id = uuid.uuid4().hex

275 - item_transfer_ids[idx] = transfer_id

292 + content = []

293 + for idx in indices:

294 + item = mm_items[idx]

295 + item_uuid = None if NO_REWRITE else content_uuid(item)

296 + if item_uuid is not None:

297 + item_uuids[idx] = item_uuid

298 + item_transfer_ids[idx] = uuid.uuid4().hex

299 + content.append(item if item_uuid is None else {**item, "uuid": item_uuid})

300

301 encoder_req = {

302 "model": orig_request.get("model"),

303 "messages": [

304 {

305 "role": "user",

282 - "content": [

283 - item if item_uuid is None else {**item, "uuid": item_uuid}

284 - ],

306 + "content": content,

307 },

308 ],

309 # No max_tokens cap: the encoder instance never samples, it finishes

@@ -297,17 +319,13 @@ async def fanout_encoder_primer(

319 if key in orig_request:

320 encoder_req[key] = orig_request[key]

321 if consumer_zmq is not None:

300 - # No mm_hash here on purpose. The encoder's own

301 - # `mm_features[i].identifier` is derived from the uuid *and* the

302 - # engine's media_io_kwargs / mm_processor_kwargs, so this proxy

303 - # cannot know it before the encoder runs. Sending the bare uuid

304 - # would fail the connector's hash match and make the producer

305 - # invent its own transfer id, which the consumer could then never

306 - # cancel. Omitting it lets the connector match by position, which

307 - # is exact: one encoder request carries exactly one item.

322 + # The engine may rehash UUIDs with processing options. Match by

323 + # position instead; batches contain only images in input order.

324 encoder_req["ec_transfer_params"] = {

325 "consumer_zmq": consumer_zmq,

310 - "ec_items": [{"transfer_id": transfer_id}],

326 + "ec_items": [

327 + {"transfer_id": item_transfer_ids[idx]} for idx in indices

328 + ],

329 }

330 tasks.append(

331 encode_session.post(

@@ -320,7 +338,8 @@ async def fanout_encoder_primer(

338 results = await asyncio.gather(*tasks, return_exceptions=True)

339

340 # Fail fast if any sub-request failed

323 - for idx, r in enumerate(results):

341 + for (_, indices), r in zip(groups, results):

342 + idx = indices[0]

343 if isinstance(r, Exception):

344 logger.error(

345 "[%s] Encoder request #%d raised exception: %s",

@@ -357,18 +376,45 @@ async def fanout_encoder_primer(

376 logger.warning("[%s] Could not read encoder metadata #%d", req_id, idx)

377 params = {}

378 if params:

360 - # One encoder request carries exactly one item, so there is a

361 - # single reported entry. Do not key it by this proxy's uuid: when

362 - # media_io_kwargs or mm_processor_kwargs are set the engine

363 - # re-hashes the uuid together with them, so the encoder's own

364 - # `mm_features[i].identifier` is a derived value this proxy cannot

365 - # predict. Fall back to the sole entry, and carry the key the

366 - # encoder actually used through as `ec_mm_hash`.

367 - ec_mm_hash = item_uuids.get(idx)

368 - reported = params.get(ec_mm_hash)

369 - if reported is None and len(params) == 1:

370 - ((ec_mm_hash, reported),) = params.items()

371 - if reported:

379 + by_index = {}

380 + for mm_hash, reported in params.items():

381 + if len(indices) == 1:

382 + break

383 + for local_index in reported.get("item_indices", []):

384 + if type(local_index) is not int or not 0 <= local_index < len(

385 + indices

386 + ):

387 + raise HTTPException(502, "Invalid encoder item index")

388 + if local_index in by_index:

389 + raise HTTPException(502, "Duplicate encoder item index")

390 + by_index[local_index] = (mm_hash, reported)

391 + for local_index, idx in enumerate(indices):

392 + matched = by_index.get(local_index)

393 + if matched is None:

394 + # Compatibility with encoders without position metadata.

395 + mm_hash = item_uuids.get(idx)

396 + if mm_hash in params:

397 + matched = (mm_hash, params[mm_hash])

398 + elif len(indices) == 1 and len(params) == 1:

399 + matched = next(iter(params.items()))

400 + elif (

401 + mm_items[idx]["type"] == "video_url"

402 + and (orig_request.get("mm_processor_kwargs") or {}).get(

403 + "use_audio_in_video"

404 + )

405 + and len(params) == 2

406 + and consumer_zmq is None

407 + and all(

408 + entry.keys() <= {"metadata", "item_indices"}

409 + for entry in params.values()

410 + )

411 + ):

412 + # Keep raw video for its audio/video features when no transfer

413 + # handles need to be forwarded.

414 + continue

415 + else:

416 + raise HTTPException(502, "Encoder metadata cannot be matched")

417 + ec_mm_hash, reported = matched

418 metadata = reported.get("metadata") or {}

419 if metadata:

420 item_meta[idx] = {

@@ -384,12 +430,16 @@ async def fanout_encoder_primer(

430 # looks it up by mm_hash on the request, so carry it through.

431 ec_params[ec_mm_hash] = reported

432 if NO_REWRITE and consumer_zmq is not None:

387 - ec_params.setdefault("ec_items", []).append(

388 - {"mm_hash": ec_mm_hash, "transfer_id": item_transfer_ids[idx]}

389 - )

433 + transfer_items[idx] = {

434 + "mm_hash": ec_mm_hash,

435 + "transfer_id": item_transfer_ids[idx],

436 + }

437 +

438 + if transfer_items:

439 + ec_params["ec_items"] = [transfer_items[idx] for idx in sorted(transfer_items)]

440

441 logger.info(

392 - "[%s] All %d encoder requests completed successfully", req_id, len(mm_items)

442 + "[%s] All %d encoder requests completed successfully", req_id, len(groups)

443 )

444 return item_meta, ec_params

445

tests/v1/ec_connector/unit/test_ec_example_connector.py

+15

@@ -16,6 +16,7 @@ from vllm.distributed.ec_transfer.ec_connector.example_connector import (

16 ECExampleConnectorMetadata,

17 MMMeta,

18 )

19 +from vllm.distributed.ec_transfer.ec_connector.utils import collect_ec_item_metadata

20 from vllm.multimodal.inputs import MultiModalFeatureSpec, PlaceholderRange

21 from vllm.v1.core.sched.output import SchedulerOutput

@@ -41,6 +42,20 @@ class MockRequest:

42 return self._token_counts[input_id]

45 +def test_metadata_preserves_repeated_item_positions():

46 + request = MockRequest("request", ["A", "B", "A"], [4, 4, 4])

47 + resolver = Mock()

48 + resolver.fields_for.return_value = {"image_grid_thw"}

49 + for feature in request.mm_features:

50 + feature.data = Mock()

51 + feature.data.get_data.return_value = {"image_grid_thw": torch.tensor([1, 2, 2])}

52 + items = collect_ec_item_metadata(request.mm_features, resolver)

53 + assert items == {

54 + "A": {"metadata": {"image_grid_thw": [1, 2, 2]}, "item_indices": [0, 2]},

55 + "B": {"metadata": {"image_grid_thw": [1, 2, 2]}, "item_indices": [1]},

56 + }

57 +

58 +

59 @pytest.fixture

60 def temp_storage(tmp_path):

61 """Fixture providing temporary storage path."""

tests/v1/ec_connector/unit/test_epd_proxy_retry.py

+200 -16

@@ -9,6 +9,8 @@ change to them is what these tests catch.

9 import asyncio

10 import importlib.util

11 from pathlib import Path

12 +from types import SimpleNamespace

13 +from typing import Any

15 import httpx

16 import msgspec

@@ -83,6 +85,128 @@ class _EncoderSession:

85 return _EncoderResponse(self._replies.pop(0))

88 +@pytest.mark.parametrize("no_rewrite", [False, True])

89 +@pytest.mark.parametrize(

90 + "batch_size, expected_sizes", [(0, [3, 3]), (1, [1] * 6), (2, [2, 2, 1, 1])]

91 +)

92 +def test_image_batches_preserve_item_identity(

93 + proxy, monkeypatch, no_rewrite, batch_size, expected_sizes

94 +):

95 + """Rehashed and repeated images retain metadata and per-occurrence transfers."""

96 + from vllm.distributed.ec_transfer.ec_connector.mooncake.scheduler import (

97 + ECMooncakeScheduler,

98 + )

99 +

100 + seen = []

101 +

102 + class Session:

103 + async def post(self, url, data=None, headers=None):

104 + body = msgspec.json.decode(data)

105 + seen.append((url, body))

106 + params: dict[str, dict[str, Any]] = {}

107 + for index, item in enumerate(body["messages"][0]["content"]):

108 + image = item["image_url"]["url"]

109 + entry = params.setdefault(

110 + image + "-processed",

111 + {

112 + "metadata": {"image_grid_thw": [1, 2, ord(image)]},

113 + "item_indices": [],

114 + "peer_port": 1234,

115 + },

116 + )

117 + entry["item_indices"].append(index)

118 + # Response ordering must not affect the image-to-metadata mapping.

119 + return _EncoderResponse(dict(reversed(list(params.items()))))

120 +

121 + monkeypatch.setattr(proxy, "encode_session", Session())

122 + monkeypatch.setattr(proxy, "NO_REWRITE", no_rewrite)

123 + monkeypatch.setattr(proxy, "ENCODER_MAX_BATCH_SIZE", batch_size)

124 + monkeypatch.setattr(proxy, "encoder_rr_idx", 0)

125 + images = ["A", "A", "C", "D", "A", "F"]

126 + body = {

127 + "messages": [

128 + {

129 + "role": "user",

130 + "content": [

131 + {"type": "image_url", "image_url": {"url": image}}

132 + for image in images

133 + ],

134 + }

135 + ],

136 + "mm_processor_kwargs": {"max_pixels": 262144},

137 + }

138 + meta, handles = asyncio.run(

139 + proxy.fanout_encoder_primer(

140 + body, ["http://e0", "http://e1"], "r", "tcp://consumer:1"

141 + )

142 + )

143 + assert [len(b["messages"][0]["content"]) for _, b in seen] == expected_sizes

144 + assert (

145 + len(

146 + {

147 + x["transfer_id"]

148 + for _, b in seen

149 + for x in b["ec_transfer_params"]["ec_items"]

150 + }

151 + )

152 + == 6

153 + )

154 + for url, batch in seen:

155 + assert batch["mm_processor_kwargs"] == body["mm_processor_kwargs"]

156 + for item in batch["messages"][0]["content"]:

157 + assert item["image_url"]["url"] in (

158 + {"A", "C"} if url.startswith("http://e0/") else {"A", "D", "F"}

159 + )

160 + assert [meta[i]["image_grid_thw"][-1] for i in range(6)] == list(map(ord, images))

161 + assert [meta[i]["ec_mm_hash"] for i in range(6)] == [

162 + x + "-processed" for x in images

163 + ]

164 + assert handles["A-processed"]["peer_port"] == 1234

165 + assert "ec_transfer_params" not in body

166 +

167 + transfers_by_encoder: list[list[dict[str, str]]] = [[], []]

168 + for url, batch in seen:

169 + rank = 0 if url.startswith("http://e0/") else 1

170 + transfers_by_encoder[rank].extend(batch["ec_transfer_params"]["ec_items"])

171 + expected_transfers = [

172 + transfers_by_encoder[i % 2][i // 2]["transfer_id"] for i in range(len(images))

173 + ]

174 + params = (

175 + handles

176 + if no_rewrite

177 + else proxy.rewrite_for_decode(body, meta)["ec_transfer_params"]

178 + )

179 + request = SimpleNamespace(

180 + ec_transfer_params=params,

181 + mm_features=[

182 + SimpleNamespace(identifier=image + "-processed") for image in images

183 + ],

184 + )

185 + assert [

186 + ECMooncakeScheduler._request_transfer_id(request, i) for i in range(len(images))

187 + ] == expected_transfers

188 +

189 +

190 +def test_batch_rejects_ambiguous_metadata(proxy, monkeypatch):

191 + monkeypatch.setattr(

192 + proxy,

193 + "encode_session",

194 + _EncoderSession([{"unknown": {"metadata": {"image_grid_thw": [1, 2, 2]}}}]),

195 + )

196 + body = {

197 + "messages": [

198 + {

199 + "content": [

200 + {"type": "image_url", "image_url": {"url": image}}

201 + for image in ["A", "B"]

202 + ]

203 + }

204 + ]

205 + }

206 + with pytest.raises(proxy.HTTPException, match="cannot be matched"):

207 + asyncio.run(proxy.fanout_encoder_primer(body, ["http://e0"], "r"))

208 +

209 +

210 def test_a_decode_retry_does_not_inherit_the_previous_handles(proxy, monkeypatch):

211 """The retry loop re-enters `prepare_for_decode` with the same body.

212

@@ -187,20 +311,21 @@ async def test_http_roundtrip_preserves_payload_and_response_bytes(

311 body = await request.json()

312 seen[stage].append(body)

313 if stage == "encode":

190 - item = body["messages"][0]["content"][0]

191 - mm_hash = item["uuid"]

192 - if body.get("mm_processor_kwargs"):

193 - mm_hash += "-processed"

194 - return web.json_response(

195 - {

196 - "ec_transfer_params": {

197 - mm_hash: {

198 - "metadata": {"image_grid_thw": [[1, 2, 2]]},

199 - "peer_port": len(seen[stage]),

200 - }

201 - }

202 - }

203 - )

314 + params: dict[str, dict[str, Any]] = {}

315 + for index, item in enumerate(body["messages"][0]["content"]):

316 + mm_hash = item["uuid"]

317 + if body.get("mm_processor_kwargs"):

318 + mm_hash += "-processed"

319 + entry = params.setdefault(

320 + mm_hash,

321 + {

322 + "metadata": {"image_grid_thw": [[1, 2, 2]]},

323 + "peer_port": len(seen[stage]),

324 + "item_indices": [],

325 + },

326 + )

327 + entry["item_indices"].append(index)

328 + return web.json_response({"ec_transfer_params": params})

329 if stage == "prefill":

330 assert body["max_tokens"] == 1 and body["stream"] is False

331 return web.json_response(

@@ -263,7 +388,7 @@ async def test_http_roundtrip_preserves_payload_and_response_bytes(

388 finally:

389 await proxy.on_shutdown()

390

266 - assert len(seen["encode"]) == 4

391 + assert len(seen["encode"]) == 2

392 assert len(seen["decode"]) == 2

393 for requests in seen.values():

394 for forwarded in requests:

@@ -296,7 +421,7 @@ async def test_http_roundtrip_preserves_payload_and_response_bytes(

421 if request_options:

422 mm_hash += "-processed"

423 assert set(final["ec_transfer_params"]) - {"ec_items"} == {mm_hash}

299 - assert final["ec_transfer_params"][mm_hash]["peer_port"] in (3, 4)

424 + assert final["ec_transfer_params"][mm_hash]["peer_port"] == 2

425 if prefill:

426 assert final["kv_transfer_params"] == {"remote_block_ids": [2]}

427

@@ -324,3 +449,62 @@ def test_pooled_connections_are_retired_before_the_server_closes_them(

449 assert proxy.decode_session.connector._keepalive_timeout == pooled_for

450 finally:

451 asyncio.run(proxy.on_shutdown())

452 +

453 +

454 +@pytest.mark.parametrize(

455 + "no_rewrite, transfer",

456 + [(False, None), (True, None), (False, "push"), (False, "handle")],

457 +)

458 +def test_video_audio_fallback(proxy, monkeypatch, no_rewrite, transfer):

459 + """Keep raw video and neighboring images, but never discard transfer handles."""

460 + import copy

461 +

462 + from vllm.distributed.ec_transfer.ec_connector.utils import collect_ec_item_metadata

463 +

464 + video = {"type": "video_url", "video_url": {"url": "video-with-audio"}}

465 + images = [{"type": "image_url", "image_url": {"url": key}} for key in ("A", "B")]

466 + body: dict[str, Any] = {

467 + "messages": [{"content": [images[0], video, images[1]]}],

468 + "mm_processor_kwargs": {"use_audio_in_video": True},

469 + }

470 + original = copy.deepcopy(body)

471 + # Collector indices describe two processed features from one video item.

472 + video_metadata = collect_ec_item_metadata(

473 + [SimpleNamespace(identifier=key, data=None) for key in ("video", "audio")], None

474 + )

475 + if transfer == "handle":

476 + video_metadata["audio"]["transfer_id"] = "reservation"

477 + image_metadata = {

478 + key: {"metadata": {"image_grid_thw": [1, 2, size]}, "item_indices": [i]}

479 + for i, (key, size) in enumerate((("A", 2), ("B", 4)))

480 + }

481 + # Images form the first group, the singleton video the second.

482 + monkeypatch.setattr(proxy, "ENCODER_MAX_BATCH_SIZE", 0)

483 + monkeypatch.setattr(proxy, "NO_REWRITE", no_rewrite)

484 + replies = _EncoderSession([image_metadata, video_metadata])

485 + monkeypatch.setattr(proxy, "encode_session", replies)

486 + consumer = "tcp://consumer:1" if transfer == "push" else None

487 + preparation = proxy.prepare_for_decode(body, "r", ["http://e0"], "", consumer)

488 + if transfer:

489 + with pytest.raises(proxy.HTTPException, match="cannot be matched"):

490 + asyncio.run(preparation)

491 + return

492 +

493 + prepared, _, _ = asyncio.run(preparation)

494 + assert body == original

495 + assert prepared["mm_processor_kwargs"] == body["mm_processor_kwargs"]

496 + final = prepared["messages"][0]["content"]

497 + assert final[1] == video

498 + handles = prepared["ec_transfer_params"]

499 + assert not video_metadata.keys() & handles.keys()

500 + if no_rewrite:

501 + assert final == original["messages"][0]["content"]

502 + assert "ec_items" not in handles

503 + else:

504 + for position, image, key in ((0, images[0], "A"), (2, images[1], "B")):

505 + assert final[position] == {

506 + "type": "image_embeds",

507 + "image_embeds": image_metadata[key]["metadata"],

508 + "uuid": proxy.content_uuid(image),

509 + }

510 + assert [item["mm_hash"] for item in handles["ec_items"]] == ["A", "B"]

tests/v1/ec_connector/unit/test_scheduler_nixl_producer.py

+3 -1

@@ -67,6 +67,7 @@ def test_request_finished_producer_emits_params(monkeypatch):

67 assert params == {

68 "h1": {

69 "metadata": {},

70 + "item_indices": [0],

71 "peer_host": "1.2.3.4",

72 "peer_port": 5601,

73 "size_bytes": 2 * 32 * 2,

@@ -91,6 +92,7 @@ def test_request_finished_announces_not_ready_entry(monkeypatch):

92 assert params == {

93 "h1": {

94 "metadata": {},

95 + "item_indices": [0],

96 "peer_host": "1.2.3.4",

97 "peer_port": 5601,

98 "size_bytes": 2 * 32 * 2,

@@ -110,5 +112,5 @@ def test_request_finished_skips_unallocated_entry(monkeypatch):

112 assert delay is False

113 # The item's placeholder metadata is still reported even though there's

114 # no cache entry to transfer (empty "metadata": no fields, no transfer).

113 - assert params == {"h1": {"metadata": {}}}

115 + assert params == {"h1": {"metadata": {}, "item_indices": [0]}}

116 s.shutdown()

vllm/distributed/ec_transfer/ec_connector/utils.py

+7 -4

@@ -60,15 +60,16 @@ def collect_ec_item_metadata(

61 Keyed by mm_hash, each entry carries a `metadata` dict with whatever

62 placeholder fields `resolver` says this model needs published for its

63 - modality, so a consumer can skip the image transform. `data` is None for

64 - items served from the processor cache, in which case the metadata is

63 + modality, so a consumer can skip the image transform. `item_indices`

64 + identifies every occurrence in `mm_features`, including repeated hashes.

65 + `data` is None for items served from the processor cache, so the metadata is

66 unavailable here and the consumer has to fall back to processing the

67 media itself. A connector that also has transfer coordinates to report

68 (e.g. NIXL peer_host/peer_port/size_bytes) merges those in alongside

69 `metadata`, not into it.

70 """

71 items: dict[str, dict[str, Any]] = {}

71 - for feature in mm_features:

72 + for index, feature in enumerate(mm_features):

73 metadata: dict[str, Any] = {}

74 if feature.data is not None:

75 wanted = resolver.fields_for(feature.modality)

@@ -81,7 +82,9 @@ def collect_ec_item_metadata(

82 # Some metadata (e.g. Qwen3-VL video timestamps) is

83 # produced as a plain list rather than a tensor.

84 metadata[key] = value

84 - items[feature.identifier] = {"metadata": metadata}

85 + entry = items.setdefault(feature.identifier, {"item_indices": []})

86 + entry["metadata"] = metadata

87 + entry["item_indices"].append(index)

88 return items