Skip to content

vllm.distributed.ec_transfer.ec_connector.mooncake.config

Classes:

  • MooncakeECConfig –

    Validated settings shared by the Scheduler and Worker roles.

MooncakeECConfig dataclass

Validated settings shared by the Scheduler and Worker roles.

control_port is the first TP-shard port after the DP offset; control_addr targets that shard, which advertises the full topology. control_host is the address this instance both advertises and binds, so reaching a Consumer means being routed to it rather than finding it on every interface.

Source code in vllm/distributed/ec_transfer/ec_connector/mooncake/config.py 
 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
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
@dataclass(frozen=True)
class MooncakeECConfig:
    """Validated settings shared by the Scheduler and Worker roles.

    ``control_port`` is the first TP-shard port after the DP offset;
    ``control_addr`` targets that shard, which advertises the full topology.
    ``control_host`` is the address this instance both advertises and binds,
    so reaching a Consumer means being routed to it rather than finding it on
    every interface.
    """

    is_producer: bool
    is_consumer: bool
    protocol: str
    buffer_device: str
    control_host: str
    control_port: int
    control_addr: str
    control_timeout_ms: int
    push_wait_timeout_s: float
    pool_size: int
    cross_encoder_cache: bool = False
    store_max_pending_items: int = 32
    store_max_pending_bytes: int = 2 * 1024**3
    store_read_buffer_bytes: int = 128 * 1024**2

    @classmethod
    def from_vllm_config(cls, vllm_config: VllmConfig) -> MooncakeECConfig:
        parallel_config = vllm_config.parallel_config
        ec_config = vllm_config.ec_transfer_config
        assert ec_config is not None

        if ec_config.is_ec_producer:
            if parallel_config.tensor_parallel_size > 1:
                raise ValueError(
                    "ECMooncakeConnector producers require tensor_parallel_size=1."
                )
            if parallel_config.pipeline_parallel_size > 1:
                raise ValueError(
                    "ECMooncakeConnector producers do not support pipeline parallelism."
                )
            if parallel_config.data_parallel_size > 1:
                raise ValueError(
                    "ECMooncakeConnector producers require data_parallel_size=1."
                )

        registered_buffer_size = _positive_int(
            "ec_buffer_size", ec_config.ec_buffer_size
        )
        get = ec_config.get_from_extra_config
        shared_reuse = get("cross_encoder_cache", False)
        if not isinstance(shared_reuse, bool):
            raise ValueError("cross_encoder_cache must be a boolean")
        if shared_reuse:
            if not ec_config.is_encode_only:
                raise ValueError(
                    "cross_encoder_cache requires an Encoder-only producer"
                )
            if not vllm_config.use_v2_model_runner:
                raise ValueError("cross_encoder_cache requires Model Runner V2")
            if vllm_config.lora_config is not None:
                raise ValueError("cross_encoder_cache does not support LoRA")
            mm_config = vllm_config.model_config.multimodal_config
            if mm_config is not None and mm_config.mm_processor_cache_gb == 0:
                raise ValueError(
                    "cross_encoder_cache requires mm_processor_cache_gb > 0 "
                    "to preserve content identifiers across Encoders and restarts"
                )
        control_port = int(ec_config.ec_port) + (
            parallel_config.data_parallel_index * parallel_config.tensor_parallel_size
        )
        highest_port = control_port + parallel_config.tensor_parallel_size - 1
        if not 1 <= control_port <= highest_port <= 65535:
            raise ValueError("ECMooncakeConnector ec_port must be in 1..65535.")

        return cls(
            is_producer=ec_config.is_ec_producer,
            is_consumer=ec_config.is_ec_consumer,
            protocol=str(get("mooncake_protocol", "rdma")),
            buffer_device=str(ec_config.ec_buffer_device or "cuda").lower(),
            control_host=str(ec_config.ec_ip),
            control_port=control_port,
            control_addr=make_zmq_path("tcp", ec_config.ec_ip, control_port),
            control_timeout_ms=max(
                1,
                math.ceil(
                    _positive_float("control_timeout_s", get("control_timeout_s", 30))
                    * 1000
                ),
            ),
            push_wait_timeout_s=_positive_float(
                "push_wait_timeout_s", get("push_wait_timeout_s", 60)
            ),
            pool_size=registered_buffer_size,
            cross_encoder_cache=shared_reuse,
            store_max_pending_items=_positive_int(
                "store_max_pending_items", get("store_max_pending_items", 32)
            ),
            store_max_pending_bytes=_positive_int(
                "store_max_pending_bytes", get("store_max_pending_bytes", 2 * 1024**3)
            ),
            store_read_buffer_bytes=_positive_int(
                "store_read_buffer_bytes", get("store_read_buffer_bytes", 128 * 1024**2)
            ),
        )