Skip to content

vllm.distributed.ec_transfer.ec_connector.cpu.connector

ECCPUConnector — CPU encoder-cache offloading.

A role-routed shell: one instance per process. The scheduler delegate owns the mmap region and offload bookkeeping; the worker delegate owns the mmap view and GPU copy plumbing. An ec_both instance reuses encoder outputs it has already offloaded to CPU instead of recomputing them.

Classes:

  • ECCPUConnector –

    EC connector that offloads encoder cache to a shared CPU mmap region.

ECCPUConnector

Bases: ECConnectorBase

EC connector that offloads encoder cache to a shared CPU mmap region.

Source code in vllm/distributed/ec_transfer/ec_connector/cpu/connector.py 
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 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
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 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
class ECCPUConnector(ECConnectorBase):
    """EC connector that offloads encoder cache to a shared CPU mmap region."""

    def __init__(self, vllm_config: "VllmConfig", role: ECConnectorRole) -> None:
        super().__init__(vllm_config=vllm_config, role=role)

        self.connector_worker = None
        self.connector_scheduler = None

        if role == ECConnectorRole.WORKER:
            self.connector_worker = self._make_worker(vllm_config)
        elif role == ECConnectorRole.SCHEDULER:
            self.connector_scheduler = self._make_scheduler(vllm_config)
        else:
            raise ValueError(f"Unknown ECConnectorRole: {role}")

    # Construction seams.
    def _make_worker(self, vllm_config: "VllmConfig"):
        # Deferred import: the worker module touches torch/CUDA at import time
        # via the region, so keep that cost off the scheduler path.
        from vllm.distributed.ec_transfer.ec_connector.cpu.worker import (
            ECCPUWorker,
        )

        return ECCPUWorker(vllm_config)

    def _make_scheduler(self, vllm_config: "VllmConfig"):
        from vllm.distributed.ec_transfer.ec_connector.cpu.scheduler import (
            ECCPUScheduler,
        )

        return ECCPUScheduler(vllm_config)

    # Worker-side forwarders.
    def start_load_caches(
        self, encoder_cache: dict[str, torch.Tensor], **kwargs
    ) -> None:
        assert self.connector_worker is not None
        metadata = self._get_connector_metadata()
        assert isinstance(metadata, ECCPUConnectorMetadata)
        self.connector_worker.start_load_caches(
            encoder_cache, connector_metadata=metadata
        )

    def save_caches(
        self, encoder_cache: dict[str, torch.Tensor], mm_hash: str, **kwargs
    ) -> None:
        assert self.connector_worker is not None
        metadata = self._get_connector_metadata()
        assert isinstance(metadata, ECCPUConnectorMetadata)
        self.connector_worker.save_caches(
            encoder_cache, mm_hash, connector_metadata=metadata
        )

    # Scheduler-side forwarders.
    def has_cache_item(self, identifier: str) -> bool:
        assert self.connector_scheduler is not None
        return self.connector_scheduler.has_cache_item(identifier)

    def ensure_cache_available(
        self, request: "Request", num_computed_tokens: int
    ) -> bool:
        assert self.connector_scheduler is not None
        return self.connector_scheduler.ensure_cache_available(
            request, num_computed_tokens
        )

    def update_state_after_alloc(self, request: "Request", index: int) -> None:
        assert self.connector_scheduler is not None
        self.connector_scheduler.update_state_after_alloc(request, index)

    def build_connector_meta(
        self, scheduler_output: "SchedulerOutput"
    ) -> ECCPUConnectorMetadata:
        assert self.connector_scheduler is not None
        return self.connector_scheduler.build_connector_meta(scheduler_output)

    def get_finished(
        self, finished_req_ids: set[str]
    ) -> tuple[set[str] | None, set[str] | None]:
        if self.connector_worker is not None:
            self.connector_worker.flush_saves()
        return None, None

    # Shared.
    def shutdown(self) -> None:
        if self.connector_scheduler is not None:
            self.connector_scheduler.shutdown()
        if self.connector_worker is not None:
            self.connector_worker.shutdown()