Skip to content

vllm.distributed.ec_transfer.ec_connector.cpu.common

Shared types for the ECCPUConnector scheduler and worker delegates.

Classes:

Functions:

ECCPUConnectorMetadata dataclass

Bases: ECConnectorMetadata

Per-step scheduler → worker payload for the ECCPUConnector.

Populated by ECCPUScheduler.build_connector_meta; consumed by ECCPUWorker via the mixin's bind_connector_metadata.

Source code in vllm/distributed/ec_transfer/ec_connector/cpu/common.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
@dataclass
class ECCPUConnectorMetadata(ECConnectorMetadata):
    """Per-step scheduler → worker payload for the ECCPUConnector.

    Populated by `ECCPUScheduler.build_connector_meta`; consumed by
    `ECCPUWorker` via the mixin's `bind_connector_metadata`.
    """

    # Producer role: mm_hashes the scheduler has just allocated CPU
    # blocks for this step; the worker's save_caches copies
    # encoder_cache[mm_hash] → mmap at these block IDs.
    saves: dict[str, list[int]] = field(default_factory=dict)

    # Consumer role: mm_hashes whose bytes are available in the local mmap,
    # mapped to (transfer_id, block_ids, shape); the worker's start_load_caches
    # copies mmap[block_ids] → GPU encoder_cache and reports the transfer_id
    # once its copy lands. The id is minted per dispatch, so two loads of the
    # same mm_hash in different steps are distinct and a late report from the
    # earlier one cannot release the later one's pin.
    loads: dict[str, tuple[int, list[int], tuple[int, int]]] = field(
        default_factory=dict
    )

ECCPUWorkerMetadata dataclass

Bases: ECConnectorWorkerMetadata

Per-step worker → scheduler payload for the ECCPUConnector.

Reports the GPU copies that completed this step: saved mm_hashes become safe to mark ready, and loaded transfers become safe to unpin once every participating rank has reported them. Built by ECCPUWorker.build_connector_worker_meta; consumed by ECCPUScheduler.update_connector_output.

Loads are reported by transfer id rather than mm_hash because every rank copies the same blocks: aggregate concatenates, so an id appears once per reporting rank and the scheduler can count participants off the list.

Source code in vllm/distributed/ec_transfer/ec_connector/cpu/common.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
@dataclass
class ECCPUWorkerMetadata(ECConnectorWorkerMetadata):
    """Per-step worker → scheduler payload for the ECCPUConnector.

    Reports the GPU copies that completed this step: saved mm_hashes become
    safe to mark ready, and loaded transfers become safe to unpin once every
    participating rank has reported them. Built by
    `ECCPUWorker.build_connector_worker_meta`; consumed by
    `ECCPUScheduler.update_connector_output`.

    Loads are reported by transfer id rather than mm_hash because every rank
    copies the same blocks: `aggregate` concatenates, so an id appears once
    per reporting rank and the scheduler can count participants off the list.
    """

    completed_saves: list[str] = field(default_factory=list)
    completed_loads: list[int] = field(default_factory=list)

    def aggregate(self, other: ECConnectorWorkerMetadata) -> ECConnectorWorkerMetadata:
        assert isinstance(other, ECCPUWorkerMetadata)
        self.completed_saves.extend(other.completed_saves)
        self.completed_loads.extend(other.completed_loads)
        return self

create_ec_shared_region(vllm_config)

Build the EC mmap region from vllm_config.

Both ECCPUScheduler and ECCPUWorker call this to get the same shared region (same engine_id, same block_size_bytes).

Source code in vllm/distributed/ec_transfer/ec_connector/cpu/common.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
def create_ec_shared_region(vllm_config: "VllmConfig") -> ECSharedRegion:
    """Build the EC mmap region from `vllm_config`.

    Both `ECCPUScheduler` and `ECCPUWorker` call this to get the same
    shared region (same engine_id, same block_size_bytes).
    """
    ec_config = vllm_config.ec_transfer_config
    assert ec_config is not None, "ec_transfer_config required to build region"

    dp_index = vllm_config.parallel_config.data_parallel_index
    engine_id = f"{vllm_config.instance_id}_dp{dp_index}"

    dtype = vllm_config.model_config.dtype
    hidden_dim = vllm_config.model_config.get_inputs_embeds_size()
    element_size = torch.empty(0, dtype=dtype).element_size()
    block_size_bytes = hidden_dim * element_size

    cpu_bytes = ec_config.ec_connector_extra_config.get("ec_cpu_bytes")
    if not cpu_bytes:
        raise ValueError("ec_cpu_bytes must be specified in ec_connector_extra_config")
    cpu_bytes = int(cpu_bytes)
    num_blocks = cpu_bytes // block_size_bytes

    return ECSharedRegion(
        engine_id=engine_id,
        num_blocks=num_blocks,
        block_size_bytes=block_size_bytes,
    )