Skip to content

vllm.distributed.ec_transfer.ec_connector.cpu.data.base

Abstract base class for the EC data-plane transport (NIXL/UCX).

Classes:

  • DataTransport –

    Interface wrapping the NIXL data-plane for EC block transfers.

DataTransport

Bases: ABC

Interface wrapping the NIXL data-plane for EC block transfers.

Peer lifecycle

add_remote_peer() — register a peer and store its dlist handle internally remove_remote_peer() — deregister and release the stored handle post_read() — issue a READ using the stored handle for agent_name

The dlist handle (prepped xfer descriptor list) is an implementation detail of the transport and is never exposed to callers.

Methods:

  • add_remote_peer –

    Register a remote peer, prep its READ-source dlist, and return agent_name.

  • check_xfer_state –

    Return the NIXL state string: 'DONE', 'PROC', or an error string.

  • deregister –

    Deregister all memory and tear down the NIXL agent.

  • get_agent_metadata –

    Return this agent's serialized NIXL metadata blob.

  • get_mem_descriptor –

    Return the msgpack-encoded block descriptor list for this agent.

  • get_new_notifs –

    Drain NIXL completion notifications addressed to this agent.

  • post_read –

    Issue a consumer-initiated READ using the stored dlist for agent_name.

  • release_xfer_handle –

    Release a transfer handle returned by post_read.

  • remove_remote_peer –

    Deregister a previously registered remote peer and release its handle.

Source code in vllm/distributed/ec_transfer/ec_connector/cpu/data/base.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
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
class DataTransport(ABC):
    """Interface wrapping the NIXL data-plane for EC block transfers.

    Peer lifecycle:
      add_remote_peer()  — register a peer and store its dlist handle internally
      remove_remote_peer() — deregister and release the stored handle
      post_read()        — issue a READ using the stored handle for agent_name

    The dlist handle (prepped xfer descriptor list) is an implementation detail
    of the transport and is never exposed to callers.
    """

    @abstractmethod
    def get_agent_metadata(self) -> bytes:
        """Return this agent's serialized NIXL metadata blob."""

    @abstractmethod
    def get_mem_descriptor(self) -> bytes:
        """Return the msgpack-encoded block descriptor list for this agent."""

    @abstractmethod
    def add_remote_peer(self, metadata: bytes, mem_descriptor: bytes) -> str:
        """Register a remote peer, prep its READ-source dlist, and return agent_name.

        The dlist handle is stored internally and used by post_read().
        """

    @abstractmethod
    def remove_remote_peer(self, agent_name: str) -> None:
        """Deregister a previously registered remote peer and release its handle."""

    @abstractmethod
    def post_read(
        self,
        local_indices: list[int],
        agent_name: str,
        remote_indices: list[int],
        notif_msg: bytes,
    ) -> Any:
        """Issue a consumer-initiated READ using the stored dlist for agent_name.

        Returns a transfer handle for use with check_xfer_state / release_xfer_handle.
        """

    @abstractmethod
    def get_new_notifs(self) -> dict[str, list[bytes]]:
        """Drain NIXL completion notifications addressed to this agent."""

    @abstractmethod
    def check_xfer_state(self, handle: Any) -> str:
        """Return the NIXL state string: 'DONE', 'PROC', or an error string."""

    @abstractmethod
    def release_xfer_handle(self, handle: Any) -> None:
        """Release a transfer handle returned by post_read."""

    @abstractmethod
    def deregister(self) -> None:
        """Deregister all memory and tear down the NIXL agent."""

add_remote_peer(metadata, mem_descriptor) abstractmethod

Register a remote peer, prep its READ-source dlist, and return agent_name.

The dlist handle is stored internally and used by post_read().

Source code in vllm/distributed/ec_transfer/ec_connector/cpu/data/base.py 
29
30
31
32
33
34
@abstractmethod
def add_remote_peer(self, metadata: bytes, mem_descriptor: bytes) -> str:
    """Register a remote peer, prep its READ-source dlist, and return agent_name.

    The dlist handle is stored internally and used by post_read().
    """

check_xfer_state(handle) abstractmethod

Return the NIXL state string: 'DONE', 'PROC', or an error string.

Source code in vllm/distributed/ec_transfer/ec_connector/cpu/data/base.py 
57
58
59
@abstractmethod
def check_xfer_state(self, handle: Any) -> str:
    """Return the NIXL state string: 'DONE', 'PROC', or an error string."""

deregister() abstractmethod

Deregister all memory and tear down the NIXL agent.

Source code in vllm/distributed/ec_transfer/ec_connector/cpu/data/base.py 
65
66
67
@abstractmethod
def deregister(self) -> None:
    """Deregister all memory and tear down the NIXL agent."""

get_agent_metadata() abstractmethod

Return this agent's serialized NIXL metadata blob.

Source code in vllm/distributed/ec_transfer/ec_connector/cpu/data/base.py 
21
22
23
@abstractmethod
def get_agent_metadata(self) -> bytes:
    """Return this agent's serialized NIXL metadata blob."""

get_mem_descriptor() abstractmethod

Return the msgpack-encoded block descriptor list for this agent.

Source code in vllm/distributed/ec_transfer/ec_connector/cpu/data/base.py 
25
26
27
@abstractmethod
def get_mem_descriptor(self) -> bytes:
    """Return the msgpack-encoded block descriptor list for this agent."""

get_new_notifs() abstractmethod

Drain NIXL completion notifications addressed to this agent.

Source code in vllm/distributed/ec_transfer/ec_connector/cpu/data/base.py 
53
54
55
@abstractmethod
def get_new_notifs(self) -> dict[str, list[bytes]]:
    """Drain NIXL completion notifications addressed to this agent."""

post_read(local_indices, agent_name, remote_indices, notif_msg) abstractmethod

Issue a consumer-initiated READ using the stored dlist for agent_name.

Returns a transfer handle for use with check_xfer_state / release_xfer_handle.

Source code in vllm/distributed/ec_transfer/ec_connector/cpu/data/base.py 
40
41
42
43
44
45
46
47
48
49
50
51
@abstractmethod
def post_read(
    self,
    local_indices: list[int],
    agent_name: str,
    remote_indices: list[int],
    notif_msg: bytes,
) -> Any:
    """Issue a consumer-initiated READ using the stored dlist for agent_name.

    Returns a transfer handle for use with check_xfer_state / release_xfer_handle.
    """

release_xfer_handle(handle) abstractmethod

Release a transfer handle returned by post_read.

Source code in vllm/distributed/ec_transfer/ec_connector/cpu/data/base.py 
61
62
63
@abstractmethod
def release_xfer_handle(self, handle: Any) -> None:
    """Release a transfer handle returned by post_read."""

remove_remote_peer(agent_name) abstractmethod

Deregister a previously registered remote peer and release its handle.

Source code in vllm/distributed/ec_transfer/ec_connector/cpu/data/base.py 
36
37
38
@abstractmethod
def remove_remote_peer(self, agent_name: str) -> None:
    """Deregister a previously registered remote peer and release its handle."""