Skip to content

vllm.distributed.ec_transfer.ec_connector.cpu.protocol

Scheduler-side wire types for the ECCPUConnector.

Two peer-to-peer message types exchanged between consumer and producer over ZMQ (XferReq, XferAck) plus a peer-compatibility fingerprint (compute_ec_compatibility_hash) sent on every XferReq so a producer refuses to serve a mismatched peer.

Classes:

  • XferAck –

    Producer → Consumer: response to an XferReq.

  • XferReq –

    Consumer → Producer: I want to READ mm_hash; pin it and tell me where.

  • XferStatus –

    Outcome the producer reports for an XferReq.

Functions:

XferAck

Bases: Struct

Producer → Consumer: response to an XferReq.

On status == OK: producer has pinned source blocks and returns their indices plus fresh NIXL metadata for the consumer to register and READ. On any NACK_*: optional fields are empty; consumer falls back to local encode.

Source code in vllm/distributed/ec_transfer/ec_connector/cpu/protocol.py 
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
class XferAck(  # type: ignore[call-arg]
    msgspec.Struct,
    tag="ack",
    omit_defaults=True,
):
    """Producer → Consumer: response to an XferReq.

    On status == OK: producer has pinned source blocks and returns their
    indices plus fresh NIXL metadata for the consumer to register and READ.
    On any NACK_*: optional fields are empty; consumer falls back to local encode.
    """

    mm_hash: str
    status: XferStatus
    # Consumer's session_id echoed back from XferReq. Together with mm_hash it
    # identifies the exact transfer; the consumer uses both to build notif_msg.
    session_id: str = ""
    src_block_indices: list[int] = []
    agent_metadata: bytes = b""
    mem_descriptor: bytes = b""

XferReq

Bases: Struct

Consumer → Producer: I want to READ mm_hash; pin it and tell me where.

Source code in vllm/distributed/ec_transfer/ec_connector/cpu/protocol.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
class XferReq(  # type: ignore[call-arg]
    msgspec.Struct,
    tag="req",
    omit_defaults=True,
):
    """Consumer → Producer: I want to READ mm_hash; pin it and tell me where."""

    mm_hash: str
    compatibility_hash: str
    # Consumer's session identity. Together with mm_hash it uniquely identifies
    # this transfer on both sides: the producer keys its active xfers by
    # (session_id, mm_hash), and the NIXL notif_msg carries the same pair.
    session_id: str = ""
    connector_version: int = EC_CONNECTOR_VERSION

XferStatus

Bases: IntEnum

Outcome the producer reports for an XferReq.

Source code in vllm/distributed/ec_transfer/ec_connector/cpu/protocol.py 
22
23
24
25
26
27
28
29
30
class XferStatus(enum.IntEnum):
    """Outcome the producer reports for an XferReq."""

    OK = 0
    NACK_MISSING = 1  # producer no longer holds the encoding
    NACK_INCOMPAT = 2  # peer-compatibility hash mismatch
    NACK_VERSION = 3  # connector wire-version mismatch
    NACK_INTERNAL = 4  # producer hit an unexpected error
    NACK_NOT_READY = 5  # announced, but its GPU->mmap save has not landed yet

compute_ec_compatibility_hash(vllm_version, model, dtype, block_size_bytes)

Peer-compatibility fingerprint over the four factors that determine byte-layout compatibility. Producer NACKs any XferReq whose hash differs.

Source code in vllm/distributed/ec_transfer/ec_connector/cpu/protocol.py 
78
79
80
81
82
83
84
85
86
87
88
89
90
91
def compute_ec_compatibility_hash(
    vllm_version: str,
    model: str,
    dtype: str,
    block_size_bytes: int,
) -> str:
    """Peer-compatibility fingerprint over the four factors that determine
    byte-layout compatibility. Producer NACKs any XferReq whose hash differs.
    """
    factors = json.dumps(
        [vllm_version, model, dtype, block_size_bytes],
        separators=(",", ":"),
    )
    return hashlib.sha256(factors.encode("utf-8")).hexdigest()