| 1 | # SPDX-License-Identifier: Apache-2.0 |
| 2 | # SPDX-FileCopyrightText: Copyright contributors to the vLLM project |
| 3 | """Abstract base classes for the P2P control-plane transport. |
| 4 | |
| 5 | The control plane handles peer discovery, connection lifecycle, and |
| 6 | message routing. It is message-content agnostic — it moves opaque |
| 7 | dicts between peers without interpreting them. |
| 8 | |
| 9 | Architecture |
| 10 | ------------ |
| 11 | |
| 12 | ControlTransport (one per node) |
| 13 | ├── listen for inbound connections |
| 14 | ├── connect() to outbound peers |
| 15 | └── poll() → new ControlConnections |
| 16 | |
| 17 | ControlConnection (one per peer) |
| 18 | ├── send(msg) — enqueue a message to the peer |
| 19 | ├── recv() — drain buffered inbound messages |
| 20 | ├── mark_dead() — signal that the peer is gone |
| 21 | └── close() — tear down the connection |
| 22 | |
| 23 | Threading model: all I/O is driven by the caller invoking poll(). |
| 24 | No background threads. poll() must be called periodically to: |
| 25 | - receive messages (buffered per-connection) |
| 26 | - accept new inbound peers |
| 27 | - detect disconnections |
| 28 | |
| 29 | Implementor contracts |
| 30 | --------------------- |
| 31 | |
| 32 | - ControlConnection.send() must not block. Messages are serialized |
| 33 | and queued for the next I/O pass. |
| 34 | - ControlConnection.recv() returns all messages received since the |
| 35 | last call (may be empty). Messages are dicts (already deserialized). |
| 36 | - ControlConnection.alive returns False after mark_dead() or close(). |
| 37 | - ControlTransport.poll() returns newly accepted connections only — |
| 38 | not previously returned ones. Each connection appears exactly once. |
| 39 | - ControlTransport.connect() creates an outbound connection to a peer |
| 40 | identified by peer_id (format: "host:port"). Raises on failure. |
| 41 | - Messages may arrive from unknown peers (new inbound connections). |
| 42 | The transport creates a ControlConnection and returns it from poll() |
| 43 | with the first message(s) already in its recv() buffer. |
| 44 | """ |
| 45 | |
| 46 | from __future__ import annotations |
| 47 | |
| 48 | from abc import ABC, abstractmethod |
| 49 | from collections.abc import Sequence |
| 50 | |
| 51 | |
| 52 | class ControlConnection(ABC): |
| 53 | """Bidirectional message channel to a single remote peer. |
| 54 | |
| 55 | Lifecycle: |
| 56 | 1. Created by ControlTransport (connect() or poll()) |
| 57 | 2. Used for send/recv by sessions |
| 58 | 3. Marked dead on disconnect (mark_dead()) |
| 59 | 4. Cleaned up with close() |
| 60 | |
| 61 | Once mark_dead() is called, alive becomes False and the owning |
| 62 | session should stop using this connection. close() releases |
| 63 | underlying resources (sockets, monitors). |
| 64 | """ |
| 65 | |
| 66 | def __init__(self, peer_id: str) -> None: |
| 67 | self.peer_id = peer_id |
| 68 | |
| 69 | @property |
| 70 | @abstractmethod |
| 71 | def alive(self) -> bool: |
| 72 | """True if the connection is usable. False after mark_dead/close.""" |
| 73 | ... |
| 74 | |
| 75 | @abstractmethod |
| 76 | def send(self, msg: dict) -> None: |
| 77 | """Enqueue a message for delivery to the peer. |
| 78 | |
| 79 | Must not block. Serialization happens internally. |
| 80 | Raises on closed connection. |
| 81 | """ |
| 82 | ... |
| 83 | |
| 84 | @abstractmethod |
| 85 | def recv(self) -> Sequence[dict]: |
| 86 | """Drain and return all messages received since the last call. |
| 87 | |
| 88 | Returns an empty sequence if no messages are pending. |
| 89 | The returned sequence is read-only — callers must not mutate it. |
| 90 | Messages are dicts deserialized from the wire format. |
| 91 | """ |
| 92 | ... |
| 93 | |
| 94 | @abstractmethod |
| 95 | def mark_dead(self) -> None: |
| 96 | """Mark this connection as dead (peer disconnected). |
| 97 | |
| 98 | After this call, alive returns False. The session should |
| 99 | stop using this connection and the transport will clean it up. |
| 100 | Resources are not released here — close() must still run. |
| 101 | """ |
| 102 | ... |
| 103 | |
| 104 | @abstractmethod |
| 105 | def close(self) -> None: |
| 106 | """Release all resources (sockets, monitors). |
| 107 | |
| 108 | Idempotent. After close(), alive returns False. |
| 109 | """ |
| 110 | ... |
| 111 | |
| 112 | |
| 113 | class ControlTransport(ABC): |
| 114 | """Manages peer connections and drives all control-plane I/O. |
| 115 | |
| 116 | Owns the listening socket and all active connections. |
| 117 | The caller must invoke poll() periodically to process I/O. |
| 118 | |
| 119 | Lifecycle: |
| 120 | 1. Constructed with a local identity and listen address |
| 121 | 2. connect() to reach remote peers |
| 122 | 3. poll() to accept inbound peers and process messages |
| 123 | 4. close() to shut down |
| 124 | """ |
| 125 | |
| 126 | @abstractmethod |
| 127 | def connect(self, peer_id: str) -> ControlConnection: |
| 128 | """Create an outbound connection to a remote peer. |
| 129 | |
| 130 | Args: |
| 131 | peer_id: Remote peer identity (format: "host:port"). |
| 132 | |
| 133 | Returns: |
| 134 | A new ControlConnection ready for send/recv. |
| 135 | |
| 136 | The connection's send queue is live immediately — messages |
| 137 | sent before the remote peer's poll() will be buffered. |
| 138 | |
| 139 | A connection to peer_id that is registered but no longer alive is |
| 140 | retired and replaced; a live one is a duplicate and must not be |
| 141 | replaced. |
| 142 | |
| 143 | """ |
| 144 | ... |
| 145 | |
| 146 | @abstractmethod |
| 147 | def poll(self) -> Sequence[ControlConnection]: |
| 148 | """Process all pending I/O and return newly accepted connections. |
| 149 | |
| 150 | This is the main I/O driver. Each call: |
| 151 | - Receives messages from all connected peers (buffered in |
| 152 | each connection's recv() queue) |
| 153 | - Accepts new inbound peers and creates connections for them |
| 154 | (first message already in recv() buffer) |
| 155 | - Detects disconnections and marks connections dead |
| 156 | |
| 157 | Returns: |
| 158 | Newly accepted inbound connections (not previously returned). |
| 159 | The returned sequence is read-only — callers must not mutate |
| 160 | it. The caller is responsible for creating sessions for |
| 161 | these connections. |
| 162 | |
| 163 | """ |
| 164 | ... |
| 165 | |
| 166 | @abstractmethod |
| 167 | def close(self) -> None: |
| 168 | """Shut down the transport and all connections. |
| 169 | |
| 170 | Closes the listening socket and all active connections. |
| 171 | Idempotent. |
| 172 | """ |
| 173 | ... |