Skip to content

vllm.v1.kv_offload.tiering.p2p.session.protocol

P2P KV cache sharing protocol constants and documentation.

Protocol Overview

Two session types communicate over a bidirectional message channel: - P2PClientSession (requests blocks from a server) - P2PServerSession (serves blocks to a client)

Connection Lifecycle

  1. Client opens a connection and sends ConnectMsg with its identity, RDMA metadata, memory layout, and config fingerprint.
  2. Server validates block_len and config_fingerprint, registers the RDMA peer, and replies with ConnectAckMsg.
  3. Client receives ConnectAckMsg and transitions to ready state (flushes any queued messages).
  4. Either side may send DisconnectMsg to gracefully close.

Block Transfer Flow (happy path)

  1. Client sends FetchMsg with a kv_request_id and lists of block keys + remote indexes where it wants the data written.
  2. Server matches requested blocks against locally stored blocks:
  3. Blocks already available are transferred immediately via RDMA.
  4. Blocks not yet available are recorded as "demanded" and transferred when the server later stores them.
  5. When all blocks for a kv_request_id are transferred, the server sends TransferDoneMsg (success=True) to the client.
  6. Client reports the load job as complete.

Abort Flow (timeout path)

  1. If the client times out waiting for TransferDoneMsg, it sends AbortFetchMsg to cancel the request.
  2. Server cancels inflight transfers for that kv_request_id and replies with AbortAckMsg.
  3. Client receives AbortAckMsg and reports the load job as failed.
  4. If AbortAckMsg itself times out, the client fails the job anyway.

Message Format

All messages are dicts serialized with msgpack. Every message has a TYPE_KEY key identifying its type. Additional fields depend on the message type (see per-message class docstrings below).

Security

Sessions wrap all incoming message handling in try/except to guard against malformed messages from adversarial or buggy peers. Invalid messages are logged and dropped without crashing the session.

The config_fingerprint field in ConnectMsg ensures peers have compatible model configurations (model, dtype, block sizes). Mismatches are rejected during the handshake.

Classes:

  • AbortAckMsg –

    Server → Client: acknowledge cancellation.

  • AbortFetchMsg –

    Client → Server: cancel a pending request.

  • ConnectAckMsg –

    Server → Client: handshake acknowledgement.

  • ConnectMsg –

    Client → Server: initial handshake request.

  • DisconnectMsg –

    Either → Either: graceful connection close.

  • FetchMsg –

    Client → Server: request blocks by key.

  • TransferDoneMsg –

    Server → Client: all blocks transferred for a request.

AbortAckMsg

Server → Client: acknowledge cancellation.

Fields

KV_REQUEST_ID: The request that was cancelled.

Methods:

  • validate –

    Raise ValueError if any field has an invalid type or value.

Source code in vllm/v1/kv_offload/tiering/p2p/session/protocol.py 
234
235
236
237
238
239
240
241
242
243
244
245
246
247
class AbortAckMsg:
    """Server → Client: acknowledge cancellation.

    Fields:
        KV_REQUEST_ID: The request that was cancelled.
    """

    TYPE = "abort_ack"
    KV_REQUEST_ID = "kv_request_id"

    @staticmethod
    def validate(msg: dict) -> None:
        """Raise ValueError if any field has an invalid type or value."""
        _require(msg, AbortAckMsg.KV_REQUEST_ID, str)

validate(msg) staticmethod

Raise ValueError if any field has an invalid type or value.

Source code in vllm/v1/kv_offload/tiering/p2p/session/protocol.py 
244
245
246
247
@staticmethod
def validate(msg: dict) -> None:
    """Raise ValueError if any field has an invalid type or value."""
    _require(msg, AbortAckMsg.KV_REQUEST_ID, str)

AbortFetchMsg

Client → Server: cancel a pending request.

Fields

KV_REQUEST_ID: The request to cancel.

Methods:

  • validate –

    Raise ValueError if any field has an invalid type or value.

Source code in vllm/v1/kv_offload/tiering/p2p/session/protocol.py 
218
219
220
221
222
223
224
225
226
227
228
229
230
231
class AbortFetchMsg:
    """Client → Server: cancel a pending request.

    Fields:
        KV_REQUEST_ID: The request to cancel.
    """

    TYPE = "abort_fetch"
    KV_REQUEST_ID = "kv_request_id"

    @staticmethod
    def validate(msg: dict) -> None:
        """Raise ValueError if any field has an invalid type or value."""
        _require(msg, AbortFetchMsg.KV_REQUEST_ID, str)

validate(msg) staticmethod

Raise ValueError if any field has an invalid type or value.

Source code in vllm/v1/kv_offload/tiering/p2p/session/protocol.py 
228
229
230
231
@staticmethod
def validate(msg: dict) -> None:
    """Raise ValueError if any field has an invalid type or value."""
    _require(msg, AbortFetchMsg.KV_REQUEST_ID, str)

ConnectAckMsg

Server → Client: handshake acknowledgement.

Fields

PEER_ID: Server's peer identity string.

Methods:

  • validate –

    Raise ValueError if any field has an invalid type or value.

Source code in vllm/v1/kv_offload/tiering/p2p/session/protocol.py 
142
143
144
145
146
147
148
149
150
151
152
153
154
155
class ConnectAckMsg:
    """Server → Client: handshake acknowledgement.

    Fields:
        PEER_ID: Server's peer identity string.
    """

    TYPE = "connect_ack"
    PEER_ID = "peer_id"

    @staticmethod
    def validate(msg: dict) -> None:
        """Raise ValueError if any field has an invalid type or value."""
        _require(msg, ConnectAckMsg.PEER_ID, str)

validate(msg) staticmethod

Raise ValueError if any field has an invalid type or value.

Source code in vllm/v1/kv_offload/tiering/p2p/session/protocol.py 
152
153
154
155
@staticmethod
def validate(msg: dict) -> None:
    """Raise ValueError if any field has an invalid type or value."""
    _require(msg, ConnectAckMsg.PEER_ID, str)

ConnectMsg

Client → Server: initial handshake request.

Fields

PEER_ID: Local peer identity string. AGENT_METADATA: RDMA agent metadata (opaque bytes). BASE_ADDR: Base memory address of the KV block region. NUM_BLOCKS: Number of blocks in the KV block region. BLOCK_LEN: Size in bytes of each block (must match between peers). CONFIG_FINGERPRINT: SHA-256 prefix of the model configuration. Peers with different fingerprints are incompatible.

Methods:

  • validate –

    Raise ValueError if any field has an invalid type or value.

Source code in vllm/v1/kv_offload/tiering/p2p/session/protocol.py 
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
class ConnectMsg:
    """Client → Server: initial handshake request.

    Fields:
        PEER_ID: Local peer identity string.
        AGENT_METADATA: RDMA agent metadata (opaque bytes).
        BASE_ADDR: Base memory address of the KV block region.
        NUM_BLOCKS: Number of blocks in the KV block region.
        BLOCK_LEN: Size in bytes of each block (must match between peers).
        CONFIG_FINGERPRINT: SHA-256 prefix of the model configuration.
            Peers with different fingerprints are incompatible.
    """

    TYPE = "connect"
    PEER_ID = "peer_id"
    AGENT_METADATA = "agent_metadata"
    BASE_ADDR = "base_addr"
    NUM_BLOCKS = "num_blocks"
    BLOCK_LEN = "block_len"
    CONFIG_FINGERPRINT = "config_fingerprint"

    @staticmethod
    def validate(msg: dict) -> None:
        """Raise ValueError if any field has an invalid type or value."""
        _require(msg, ConnectMsg.PEER_ID, str)
        _require(msg, ConnectMsg.AGENT_METADATA, bytes)
        _require_non_neg_int(msg, ConnectMsg.BASE_ADDR)
        _require_pos_int(msg, ConnectMsg.NUM_BLOCKS)
        _require_pos_int(msg, ConnectMsg.BLOCK_LEN)

validate(msg) staticmethod

Raise ValueError if any field has an invalid type or value.

Source code in vllm/v1/kv_offload/tiering/p2p/session/protocol.py 
132
133
134
135
136
137
138
139
@staticmethod
def validate(msg: dict) -> None:
    """Raise ValueError if any field has an invalid type or value."""
    _require(msg, ConnectMsg.PEER_ID, str)
    _require(msg, ConnectMsg.AGENT_METADATA, bytes)
    _require_non_neg_int(msg, ConnectMsg.BASE_ADDR)
    _require_pos_int(msg, ConnectMsg.NUM_BLOCKS)
    _require_pos_int(msg, ConnectMsg.BLOCK_LEN)

DisconnectMsg

Either → Either: graceful connection close.

No additional fields beyond TYPE_KEY.

Source code in vllm/v1/kv_offload/tiering/p2p/session/protocol.py 
158
159
160
161
162
163
164
class DisconnectMsg:
    """Either → Either: graceful connection close.

    No additional fields beyond TYPE_KEY.
    """

    TYPE = "disconnect"

FetchMsg

Client → Server: request blocks by key.

Fields

KV_REQUEST_ID: Identifies this block transfer request. BLOCK_HASHES: List of block keys (OffloadKey bytes). BLOCK_INDEXES: List of remote block indexes (same length as BLOCK_HASHES).

Methods:

  • validate –

    Raise ValueError if any field has an invalid type or value.

Source code in vllm/v1/kv_offload/tiering/p2p/session/protocol.py 
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
class FetchMsg:
    """Client → Server: request blocks by key.

    Fields:
        KV_REQUEST_ID: Identifies this block transfer request.
        BLOCK_HASHES: List of block keys (OffloadKey bytes).
        BLOCK_INDEXES: List of remote block indexes (same length as BLOCK_HASHES).
    """

    TYPE = "fetch"
    KV_REQUEST_ID = "kv_request_id"
    BLOCK_HASHES = "block_hashes"
    BLOCK_INDEXES = "block_indexes"

    @staticmethod
    def validate(msg: dict) -> None:
        """Raise ValueError if any field has an invalid type or value."""
        _require(msg, FetchMsg.KV_REQUEST_ID, str)
        _require_list(msg, FetchMsg.BLOCK_HASHES)
        _require_list(msg, FetchMsg.BLOCK_INDEXES)
        hashes = msg[FetchMsg.BLOCK_HASHES]
        indexes = msg[FetchMsg.BLOCK_INDEXES]
        if len(hashes) != len(indexes):
            raise ValueError(
                f"block_hashes/block_indexes length mismatch: "
                f"{len(hashes)} vs {len(indexes)}"
            )
        for idx in indexes:
            if not isinstance(idx, int) or idx < 0:
                raise ValueError(f"block_indexes: invalid index {idx!r}")

validate(msg) staticmethod

Raise ValueError if any field has an invalid type or value.

Source code in vllm/v1/kv_offload/tiering/p2p/session/protocol.py 
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
@staticmethod
def validate(msg: dict) -> None:
    """Raise ValueError if any field has an invalid type or value."""
    _require(msg, FetchMsg.KV_REQUEST_ID, str)
    _require_list(msg, FetchMsg.BLOCK_HASHES)
    _require_list(msg, FetchMsg.BLOCK_INDEXES)
    hashes = msg[FetchMsg.BLOCK_HASHES]
    indexes = msg[FetchMsg.BLOCK_INDEXES]
    if len(hashes) != len(indexes):
        raise ValueError(
            f"block_hashes/block_indexes length mismatch: "
            f"{len(hashes)} vs {len(indexes)}"
        )
    for idx in indexes:
        if not isinstance(idx, int) or idx < 0:
            raise ValueError(f"block_indexes: invalid index {idx!r}")

TransferDoneMsg

Server → Client: all blocks transferred for a request.

Fields

KV_REQUEST_ID: The request that completed. SUCCESS: Whether the transfer completed successfully.

Methods:

  • validate –

    Raise ValueError if any field has an invalid type or value.

Source code in vllm/v1/kv_offload/tiering/p2p/session/protocol.py 
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
class TransferDoneMsg:
    """Server → Client: all blocks transferred for a request.

    Fields:
        KV_REQUEST_ID: The request that completed.
        SUCCESS: Whether the transfer completed successfully.
    """

    TYPE = "transfer_done"
    KV_REQUEST_ID = "kv_request_id"
    SUCCESS = "success"

    @staticmethod
    def validate(msg: dict) -> None:
        """Raise ValueError if any field has an invalid type or value."""
        _require(msg, TransferDoneMsg.KV_REQUEST_ID, str)
        _require(msg, TransferDoneMsg.SUCCESS, bool)

validate(msg) staticmethod

Raise ValueError if any field has an invalid type or value.

Source code in vllm/v1/kv_offload/tiering/p2p/session/protocol.py 
211
212
213
214
215
@staticmethod
def validate(msg: dict) -> None:
    """Raise ValueError if any field has an invalid type or value."""
    _require(msg, TransferDoneMsg.KV_REQUEST_ID, str)
    _require(msg, TransferDoneMsg.SUCCESS, bool)

_require(msg, key, typ, *, name='')

Raise ValueError if msg[key] is missing or not isinstance(typ).

Source code in vllm/v1/kv_offload/tiering/p2p/session/protocol.py 
74
75
76
77
78
79
def _require(msg: dict, key: str, typ: type, *, name: str = "") -> None:
    """Raise ValueError if msg[key] is missing or not isinstance(typ)."""
    val = msg.get(key)
    if not isinstance(val, typ):
        label = name or key
        raise ValueError(f"{label}: expected {typ.__name__}, got {type(val).__name__}")

_require_list(msg, key, *, name='')

Raise ValueError if msg[key] is not a list.

Source code in vllm/v1/kv_offload/tiering/p2p/session/protocol.py 
 98
 99
100
101
102
103
def _require_list(msg: dict, key: str, *, name: str = "") -> None:
    """Raise ValueError if msg[key] is not a list."""
    val = msg.get(key)
    if not isinstance(val, list):
        label = name or key
        raise ValueError(f"{label}: expected list, got {type(val).__name__}")

_require_non_neg_int(msg, key, *, name='')

Raise ValueError if msg[key] is not a non-negative int.

Source code in vllm/v1/kv_offload/tiering/p2p/session/protocol.py 
90
91
92
93
94
95
def _require_non_neg_int(msg: dict, key: str, *, name: str = "") -> None:
    """Raise ValueError if msg[key] is not a non-negative int."""
    val = msg.get(key)
    if not isinstance(val, int) or val < 0:
        label = name or key
        raise ValueError(f"{label}: expected non-negative int, got {val!r}")

_require_pos_int(msg, key, *, name='')

Raise ValueError if msg[key] is not a positive int.

Source code in vllm/v1/kv_offload/tiering/p2p/session/protocol.py 
82
83
84
85
86
87
def _require_pos_int(msg: dict, key: str, *, name: str = "") -> None:
    """Raise ValueError if msg[key] is not a positive int."""
    val = msg.get(key)
    if not isinstance(val, int) or val <= 0:
        label = name or key
        raise ValueError(f"{label}: expected positive int, got {val!r}")