Skip to content

vllm.v1.kv_offload.tiering.p2p.control.base

Abstract base classes for the P2P control-plane transport.

The control plane handles peer discovery, connection lifecycle, and message routing. It is message-content agnostic — it moves opaque dicts between peers without interpreting them.

Architecture

ControlTransport (one per node)
    ├── listen for inbound connections
    ├── connect() to outbound peers
    └── poll() → new ControlConnections

ControlConnection (one per peer)
    ├── send(msg)   — enqueue a message to the peer
    ├── recv()      — drain buffered inbound messages
    ├── mark_dead() — signal that the peer is gone
    └── close()     — tear down the connection

Threading model: all I/O is driven by the caller invoking poll(). No background threads. poll() must be called periodically to: - receive messages (buffered per-connection) - accept new inbound peers - detect disconnections

Implementor contracts

  • ControlConnection.send() must not block. Messages are serialized and queued for the next I/O pass.
  • ControlConnection.recv() returns all messages received since the last call (may be empty). Messages are dicts (already deserialized).
  • ControlConnection.alive returns False after mark_dead() or close().
  • ControlTransport.poll() returns newly accepted connections only — not previously returned ones. Each connection appears exactly once.
  • ControlTransport.connect() creates an outbound connection to a peer identified by peer_id (format: "host:port"). Raises on failure.
  • Messages may arrive from unknown peers (new inbound connections). The transport creates a ControlConnection and returns it from poll() with the first message(s) already in its recv() buffer.

Classes:

ControlConnection

Bases: ABC

Bidirectional message channel to a single remote peer.

Lifecycle
  1. Created by ControlTransport (connect() or poll())
  2. Used for send/recv by sessions
  3. Marked dead on disconnect (mark_dead())
  4. Cleaned up with close()

Once mark_dead() is called, alive becomes False and the owning session should stop using this connection. close() releases underlying resources (sockets, monitors).

Methods:

  • close –

    Release all resources (sockets, monitors).

  • mark_dead –

    Mark this connection as dead (peer disconnected).

  • recv –

    Drain and return all messages received since the last call.

  • send –

    Enqueue a message for delivery to the peer.

Attributes:

  • alive (bool) –

    True if the connection is usable. False after mark_dead/close.

Source code in vllm/v1/kv_offload/tiering/p2p/control/base.py 
 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
class ControlConnection(ABC):
    """Bidirectional message channel to a single remote peer.

    Lifecycle:
        1. Created by ControlTransport (connect() or poll())
        2. Used for send/recv by sessions
        3. Marked dead on disconnect (mark_dead())
        4. Cleaned up with close()

    Once mark_dead() is called, alive becomes False and the owning
    session should stop using this connection. close() releases
    underlying resources (sockets, monitors).
    """

    def __init__(self, peer_id: str) -> None:
        self.peer_id = peer_id

    @property
    @abstractmethod
    def alive(self) -> bool:
        """True if the connection is usable. False after mark_dead/close."""
        ...

    @abstractmethod
    def send(self, msg: dict) -> None:
        """Enqueue a message for delivery to the peer.

        Must not block. Serialization happens internally.
        Raises on closed connection.
        """
        ...

    @abstractmethod
    def recv(self) -> Sequence[dict]:
        """Drain and return all messages received since the last call.

        Returns an empty sequence if no messages are pending.
        The returned sequence is read-only — callers must not mutate it.
        Messages are dicts deserialized from the wire format.
        """
        ...

    @abstractmethod
    def mark_dead(self) -> None:
        """Mark this connection as dead (peer disconnected).

        After this call, alive returns False. The session should
        stop using this connection and the transport will clean it up.
        """
        ...

    @abstractmethod
    def close(self) -> None:
        """Release all resources (sockets, monitors).

        Idempotent. After close(), alive returns False.
        """
        ...

alive abstractmethod property

True if the connection is usable. False after mark_dead/close.

close() abstractmethod

Release all resources (sockets, monitors).

Idempotent. After close(), alive returns False.

Source code in vllm/v1/kv_offload/tiering/p2p/control/base.py 
103
104
105
106
107
108
109
@abstractmethod
def close(self) -> None:
    """Release all resources (sockets, monitors).

    Idempotent. After close(), alive returns False.
    """
    ...

mark_dead() abstractmethod

Mark this connection as dead (peer disconnected).

After this call, alive returns False. The session should stop using this connection and the transport will clean it up.

Source code in vllm/v1/kv_offload/tiering/p2p/control/base.py 
 94
 95
 96
 97
 98
 99
100
101
@abstractmethod
def mark_dead(self) -> None:
    """Mark this connection as dead (peer disconnected).

    After this call, alive returns False. The session should
    stop using this connection and the transport will clean it up.
    """
    ...

recv() abstractmethod

Drain and return all messages received since the last call.

Returns an empty sequence if no messages are pending. The returned sequence is read-only — callers must not mutate it. Messages are dicts deserialized from the wire format.

Source code in vllm/v1/kv_offload/tiering/p2p/control/base.py 
84
85
86
87
88
89
90
91
92
@abstractmethod
def recv(self) -> Sequence[dict]:
    """Drain and return all messages received since the last call.

    Returns an empty sequence if no messages are pending.
    The returned sequence is read-only — callers must not mutate it.
    Messages are dicts deserialized from the wire format.
    """
    ...

send(msg) abstractmethod

Enqueue a message for delivery to the peer.

Must not block. Serialization happens internally. Raises on closed connection.

Source code in vllm/v1/kv_offload/tiering/p2p/control/base.py 
75
76
77
78
79
80
81
82
@abstractmethod
def send(self, msg: dict) -> None:
    """Enqueue a message for delivery to the peer.

    Must not block. Serialization happens internally.
    Raises on closed connection.
    """
    ...

ControlTransport

Bases: ABC

Manages peer connections and drives all control-plane I/O.

Owns the listening socket and all active connections. The caller must invoke poll() periodically to process I/O.

Lifecycle
  1. Constructed with a local identity and listen address
  2. connect() to reach remote peers
  3. poll() to accept inbound peers and process messages
  4. close() to shut down

Methods:

  • close –

    Shut down the transport and all connections.

  • connect –

    Create an outbound connection to a remote peer.

  • poll –

    Process all pending I/O and return newly accepted connections.

Source code in vllm/v1/kv_offload/tiering/p2p/control/base.py 
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
156
157
158
159
160
161
162
163
164
165
166
class ControlTransport(ABC):
    """Manages peer connections and drives all control-plane I/O.

    Owns the listening socket and all active connections.
    The caller must invoke poll() periodically to process I/O.

    Lifecycle:
        1. Constructed with a local identity and listen address
        2. connect() to reach remote peers
        3. poll() to accept inbound peers and process messages
        4. close() to shut down
    """

    @abstractmethod
    def connect(self, peer_id: str) -> ControlConnection:
        """Create an outbound connection to a remote peer.

        Args:
            peer_id: Remote peer identity (format: "host:port").

        Returns:
            A new ControlConnection ready for send/recv.

        The connection's send queue is live immediately — messages
        sent before the remote peer's poll() will be buffered.
        """
        ...

    @abstractmethod
    def poll(self) -> Sequence[ControlConnection]:
        """Process all pending I/O and return newly accepted connections.

        This is the main I/O driver. Each call:
          - Receives messages from all connected peers (buffered in
            each connection's recv() queue)
          - Accepts new inbound peers and creates connections for them
            (first message already in recv() buffer)
          - Detects disconnections and marks connections dead

        Returns:
            Newly accepted inbound connections (not previously returned).
            The returned sequence is read-only — callers must not mutate
            it. The caller is responsible for creating sessions for
            these connections.
        """
        ...

    @abstractmethod
    def close(self) -> None:
        """Shut down the transport and all connections.

        Closes the listening socket and all active connections.
        Idempotent.
        """
        ...

close() abstractmethod

Shut down the transport and all connections.

Closes the listening socket and all active connections. Idempotent.

Source code in vllm/v1/kv_offload/tiering/p2p/control/base.py 
159
160
161
162
163
164
165
166
@abstractmethod
def close(self) -> None:
    """Shut down the transport and all connections.

    Closes the listening socket and all active connections.
    Idempotent.
    """
    ...

connect(peer_id) abstractmethod

Create an outbound connection to a remote peer.

Parameters:

  • peer_id

    (str) –

    Remote peer identity (format: "host:port").

Returns:

The connection's send queue is live immediately — messages sent before the remote peer's poll() will be buffered.

Source code in vllm/v1/kv_offload/tiering/p2p/control/base.py 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
@abstractmethod
def connect(self, peer_id: str) -> ControlConnection:
    """Create an outbound connection to a remote peer.

    Args:
        peer_id: Remote peer identity (format: "host:port").

    Returns:
        A new ControlConnection ready for send/recv.

    The connection's send queue is live immediately — messages
    sent before the remote peer's poll() will be buffered.
    """
    ...

poll() abstractmethod

Process all pending I/O and return newly accepted connections.

This is the main I/O driver. Each call: - Receives messages from all connected peers (buffered in each connection's recv() queue) - Accepts new inbound peers and creates connections for them (first message already in recv() buffer) - Detects disconnections and marks connections dead

Returns:

Source code in vllm/v1/kv_offload/tiering/p2p/control/base.py 
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
@abstractmethod
def poll(self) -> Sequence[ControlConnection]:
    """Process all pending I/O and return newly accepted connections.

    This is the main I/O driver. Each call:
      - Receives messages from all connected peers (buffered in
        each connection's recv() queue)
      - Accepts new inbound peers and creates connections for them
        (first message already in recv() buffer)
      - Detects disconnections and marks connections dead

    Returns:
        Newly accepted inbound connections (not previously returned).
        The returned sequence is read-only — callers must not mutate
        it. The caller is responsible for creating sessions for
        these connections.
    """
    ...