vllm.distributed.ec_transfer.ec_connector.utils

EC connector helper utilities.

Classes:

• ECOutputAggregator –

  Merge every worker's EC connector output onto the single

• PlaceholderMetadataResolver –

  Resolves which processed keys a model needs published per modality.

Functions:

• collect_ec_item_metadata –

  Build one ec_transfer_params entry per feature for request_finished().

ECOutputAggregator

Merge every worker's EC connector output onto the single ModelRunnerOutput that reaches the scheduler.

Mirrors KVOutputAggregator: only output_rank's output is returned to the scheduler, but the EC connector may have run on any rank.

Source code in vllm/distributed/ec_transfer/ec_connector/utils.py

class ECOutputAggregator:
    """Merge every worker's EC connector output onto the single
    ModelRunnerOutput that reaches the scheduler.

    Mirrors KVOutputAggregator: only `output_rank`'s output is returned to the
    scheduler, but the EC connector may have run on any rank.
    """

    def aggregate(
        self, outputs: list[ModelRunnerOutput | None], output_rank: int = 0
    ) -> ModelRunnerOutput | None:
        output = outputs[output_rank]
        if not output:
            return None

        finished_sending = set[str]()
        finished_recving = set[str]()
        worker_meta = None
        connector_stats = None
        for model_runner_output in outputs:
            assert model_runner_output is not None
            ec_output = model_runner_output.ec_connector_output
            if not ec_output:
                continue

            finished_sending |= ec_output.finished_sending or set()
            finished_recving |= ec_output.finished_recving or set()

            if meta := ec_output.ec_connector_worker_meta:
                worker_meta = (
                    meta if worker_meta is None else worker_meta.aggregate(meta)
                )

            if stats := ec_output.ec_connector_stats:
                connector_stats = (
                    stats
                    if connector_stats is None
                    else connector_stats.aggregate(stats)
                )

        aggregated = ECConnectorOutput(
            finished_sending=finished_sending or None,
            finished_recving=finished_recving or None,
            ec_connector_stats=connector_stats,
            ec_connector_worker_meta=worker_meta,
        )
        if aggregated.is_empty():
            output.ec_connector_output = None
            return output

        # `output` is the shared empty output whenever `output_rank` had no work,
        # so attach through the copy-on-write helper.
        return ModelRunnerOutput.with_ec_conn_output(output, aggregated)

PlaceholderMetadataResolver

Resolves which processed keys a model needs published per modality.

Reads MultiModalDataParser.embedding_fields, the same declaration the consumer's parser requires, so the two cannot drift. An empty set means the modality cannot be delivered out of band, and the consumer will process the media itself.

Source code in vllm/distributed/ec_transfer/ec_connector/utils.py

class PlaceholderMetadataResolver:
    """Resolves which processed keys a model needs published per modality.

    Reads `MultiModalDataParser.embedding_fields`, the same declaration the
    consumer's parser requires, so the two cannot drift. An empty set means
    the modality cannot be delivered out of band, and the consumer will
    process the media itself.
    """

    def __init__(self, model_config: "ModelConfig") -> None:
        self._model_config = model_config
        self._cache: dict[str, set[str]] = {}

    def fields_for(self, modality: str) -> set[str]:
        if modality in self._cache:
            return self._cache[modality]

        fields: set[str] = set()
        try:
            from vllm.multimodal import MULTIMODAL_REGISTRY

            info = MULTIMODAL_REGISTRY.create_processor(self._model_config).info
            fields = info.data_parser.placeholder_metadata_fields(modality)
        except Exception:
            logger.warning(
                "Could not determine the placeholder metadata fields for "
                "modality %s; the consumer will preprocess the media itself.",
                modality,
                exc_info=True,
            )

        self._cache[modality] = fields
        return fields

collect_ec_item_metadata(mm_features, resolver)

Build one ec_transfer_params entry per feature for request_finished().

Keyed by mm_hash, each entry carries a metadata dict with whatever placeholder fields resolver says this model needs published for its modality, so a consumer can skip media preprocessing. item_indices identifies every occurrence in mm_features, including repeated hashes. Audio token counts come from the placeholder; other metadata requires data and falls back to raw media when it is unavailable on cache hits. A connector that also has transfer coordinates to report (e.g. NIXL peer_host/peer_port/size_bytes) merges those in alongside metadata, not into it.

Source code in vllm/distributed/ec_transfer/ec_connector/utils.py

def collect_ec_item_metadata(
    mm_features: "list[MultiModalFeatureSpec]",
    resolver: PlaceholderMetadataResolver,
) -> dict[str, dict[str, Any]]:
    """Build one `ec_transfer_params` entry per feature for `request_finished()`.

    Keyed by mm_hash, each entry carries a `metadata` dict with whatever
    placeholder fields `resolver` says this model needs published for its
    modality, so a consumer can skip media preprocessing. `item_indices`
    identifies every occurrence in `mm_features`, including repeated hashes.
    Audio token counts come from the placeholder; other metadata requires
    `data` and falls back to raw media when it is unavailable on cache hits.
    A connector that also has transfer coordinates to report
    (e.g. NIXL peer_host/peer_port/size_bytes) merges those in alongside
    `metadata`, not into it.
    """
    items: dict[str, dict[str, Any]] = {}
    for index, feature in enumerate(mm_features):
        metadata: dict[str, Any] = {}
        wanted = resolver.fields_for(feature.modality)
        if "audio_num_tokens" in wanted:
            metadata["audio_num_tokens"] = [feature.mm_position.get_num_embeds()]
        if feature.data is not None:
            for key, value in feature.data.get_data().items():
                if key not in wanted:
                    continue
                if isinstance(value, torch.Tensor):
                    metadata[key] = torch.atleast_1d(value).tolist()
                elif is_list_of(value, (int, float)):
                    # Some metadata (e.g. Qwen3-VL video timestamps) is
                    # produced as a plain list rather than a tensor.
                    metadata[key] = value
        entry = items.setdefault(feature.identifier, {"item_indices": []})
        entry["metadata"] = metadata
        entry["item_indices"].append(index)
    return items