| 1 | # SPDX-License-Identifier: Apache-2.0 |
| 2 | # SPDX-FileCopyrightText: Copyright contributors to the vLLM project |
| 3 | """P2PSession — bidirectional session combining client + server roles. |
| 4 | |
| 5 | A single P2PSession per remote peer handles BOTH directions of the P2P |
| 6 | protocol on one ControlConnection: it can request blocks from the peer |
| 7 | ("client" role, in :mod:`.client`) AND serve blocks to the peer |
| 8 | ("server" role, in :mod:`.server`). This module is the thin coordinator |
| 9 | that owns the connection, the handshake, send-gating, and the message |
| 10 | dispatch — each parsed message is forwarded to the corresponding role. |
| 11 | |
| 12 | Wire protocol is unchanged. Both sides advertise their NIXL metadata |
| 13 | via ConnectMsg when their session is connected; the peer's ConnectMsg |
| 14 | triggers transport.add_remote_peer; ConnectAckMsg confirms the peer |
| 15 | received our ConnectMsg, after which queued outgoing messages are flushed. |
| 16 | """ |
| 17 | |
| 18 | from __future__ import annotations |
| 19 | |
| 20 | import contextlib |
| 21 | from collections.abc import Sequence |
| 22 | from typing import TYPE_CHECKING, NamedTuple |
| 23 | |
| 24 | from vllm.logger import init_logger |
| 25 | from vllm.v1.kv_offload.base import OffloadKey |
| 26 | from vllm.v1.kv_offload.tiering.p2p.control.base import ControlConnection |
| 27 | from vllm.v1.kv_offload.tiering.p2p.session.client import ClientRole, LoadResult |
| 28 | from vllm.v1.kv_offload.tiering.p2p.session.protocol import ( |
| 29 | TYPE_KEY, |
| 30 | AbortAckMsg, |
| 31 | AbortFetchMsg, |
| 32 | ConnectAckMsg, |
| 33 | ConnectMsg, |
| 34 | DisconnectMsg, |
| 35 | FetchMsg, |
| 36 | LookupMsg, |
| 37 | LookupRespMsg, |
| 38 | TransferDoneMsg, |
| 39 | ) |
| 40 | from vllm.v1.kv_offload.tiering.p2p.session.server import ( |
| 41 | ServerRole, |
| 42 | StoreResult, |
| 43 | ) |
| 44 | |
| 45 | if TYPE_CHECKING: |
| 46 | from vllm.v1.kv_offload.base import ReqContext |
| 47 | from vllm.v1.kv_offload.tiering.base import JobId, ParentManager |
| 48 | from vllm.v1.kv_offload.tiering.p2p.data import DataTransport |
| 49 | |
| 50 | logger = init_logger(__name__) |
| 51 | |
| 52 | # Cap on consecutive non-protocol dispatch exceptions before we tear |
| 53 | # down the session. Protocol violations (ValueError) disconnect on the |
| 54 | # first occurrence; this threshold protects against repeated internal |
| 55 | # bugs that may indicate a peer-induced bad state. Reset on any |
| 56 | # successful dispatch. |
| 57 | _MAX_CONSECUTIVE_DISPATCH_ERRORS = 5 |
| 58 | |
| 59 | |
| 60 | class SessionPollResult(NamedTuple): |
| 61 | """Result of one P2PSession.poll() tick. |
| 62 | |
| 63 | `loads`/`stores` are the same per-role results the manager has always |
| 64 | consumed. `new_fetch_ids` reports kv_request_ids whose FetchMsg |
| 65 | arrived this tick — the manager uses them to bind kv_request_id → |
| 66 | session and replay any submit_store batches parked while no peer had |
| 67 | asked yet. Reporting (rather than calling back into the manager |
| 68 | mid-dispatch) keeps the dependency strictly top-down. |
| 69 | """ |
| 70 | |
| 71 | loads: list[LoadResult] |
| 72 | stores: list[StoreResult] |
| 73 | new_fetch_ids: list[str] |
| 74 | |
| 75 | |
| 76 | class SessionCloseResult(NamedTuple): |
| 77 | """Result of tearing down a P2PSession. |
| 78 | |
| 79 | `failed_jobs`/`failed_stores` are the in-flight jobs (client loads / |
| 80 | server stores) the manager must mark failed. `failed_req_ids` is every |
| 81 | client-side kv_request_id whose lookup() must fail (in-flight loads plus |
| 82 | unresolved probes); `failed_serves` is the server-side lookup state the |
| 83 | dead peer can no longer resolve. |
| 84 | """ |
| 85 | |
| 86 | failed_jobs: list[int] # client load job_ids |
| 87 | failed_req_ids: list[str] # client kv_request_ids (loads + probes) |
| 88 | failed_stores: list[int] # server store job_ids |
| 89 | failed_serves: list[ReqContext] # server-side lookup ctxs needing release |
| 90 | |
| 91 | |
| 92 | class P2PSession: |
| 93 | """Bidirectional session — coordinator over ClientRole + ServerRole. |
| 94 | |
| 95 | Lifecycle: |
| 96 | - Constructor with conn=None ⇒ pending. Accepts add_stored_blocks |
| 97 | but cannot send (used by the prefiller to buffer blocks before |
| 98 | the decoder connects). |
| 99 | - Constructor with conn != None ⇒ connected. Sends our own ConnectMsg |
| 100 | immediately; the peer's ConnectMsg arrives in poll() and is |
| 101 | dispatched to _on_connect (which calls transport.add_remote_peer |
| 102 | and replies with ConnectAckMsg). Outgoing sends are queued until |
| 103 | ConnectAckMsg confirms our metadata reached the peer. |
| 104 | - attach_connection(conn) on a pending session ⇒ same as above, |
| 105 | starting from pending. |
| 106 | """ |
| 107 | |
| 108 | def __init__( |
| 109 | self, |
| 110 | peer_id: str, |
| 111 | local_id: str, |
| 112 | transport: DataTransport, |
| 113 | local_block_len: int, |
| 114 | local_hash_seed: str, |
| 115 | conn: ControlConnection | None = None, |
| 116 | ) -> None: |
| 117 | self.peer_id = peer_id |
| 118 | self._local_id = local_id |
| 119 | self._transport = transport |
| 120 | self._local_block_len = local_block_len |
| 121 | self._local_hash_seed = local_hash_seed |
| 122 | self._conn: ControlConnection | None = None |
| 123 | |
| 124 | self._send_ready = False # True after the peer acked our ConnectMsg |
| 125 | # Msgs waiting to be sent on connection establishment |
| 126 | self._queued: list[dict] = [] |
| 127 | |
| 128 | # Consecutive non-protocol dispatch errors. Reset on success. |
| 129 | self._dispatch_error_count: int = 0 |
| 130 | |
| 131 | # kv_request_ids whose FetchMsg arrived during the current poll |
| 132 | # tick. Drained and returned in the next poll() result so the |
| 133 | # manager can bind kv_request_id → session and replay any |
| 134 | # submit_store batches parked before the binding existed. |
| 135 | self._new_fetch_ids: list[str] = [] |
| 136 | |
| 137 | self._client = ClientRole(peer_id=peer_id, send=self._send) |
| 138 | self._server = ServerRole( |
| 139 | peer_id=peer_id, |
| 140 | transport=transport, |
| 141 | send=self._send, |
| 142 | ) |
| 143 | |
| 144 | if conn is not None: |
| 145 | self.attach_connection(conn) |
| 146 | |
| 147 | # ------------------------------------------------------------------ |
| 148 | # Properties |
| 149 | # ------------------------------------------------------------------ |
| 150 | |
| 151 | @property |
| 152 | def alive(self) -> bool: |
| 153 | # Pending sessions (awaiting connection) are alive — only a |
| 154 | # closed real connection counts as dead. |
| 155 | return self._conn is None or self._conn.alive |
| 156 | |
| 157 | @property |
| 158 | def connected(self) -> bool: |
| 159 | return self._conn is not None |
| 160 | |
| 161 | @property |
| 162 | def ready(self) -> bool: |
| 163 | """True after the peer acked our ConnectMsg (we may send freely).""" |
| 164 | return self._send_ready |
| 165 | |
| 166 | @property |
| 167 | def has_pending_work(self) -> bool: |
| 168 | """True while inbound loads or outbound transfers are outstanding.""" |
| 169 | return self._client.has_active_loads or self._server.has_inflight_transfers |
| 170 | |
| 171 | # ------------------------------------------------------------------ |
| 172 | # Connection lifecycle |
| 173 | # ------------------------------------------------------------------ |
| 174 | |
| 175 | def attach_connection(self, conn: ControlConnection) -> None: |
| 176 | """Attach a connection to a pending session and announce ourselves. |
| 177 | |
| 178 | Symmetric: every side advertises its NIXL metadata on connect, so |
| 179 | whichever peer receives a session first can register the other. |
| 180 | """ |
| 181 | if self._conn is not None: |
| 182 | raise ValueError(f"P2PSession {self.peer_id}: already connected") |
| 183 | self._conn = conn |
| 184 | self._send_connect() |
| 185 | |
| 186 | # ------------------------------------------------------------------ |
| 187 | # Public API |
| 188 | # ------------------------------------------------------------------ |
| 189 | |
| 190 | def request_blocks( |
| 191 | self, |
| 192 | job_id: JobId, |
| 193 | kv_request_id: str, |
| 194 | keys: Sequence[OffloadKey], |
| 195 | block_ids: Sequence[int], |
| 196 | ) -> None: |
| 197 | """Send fetch to the peer.""" |
| 198 | self._client.request_blocks( |
| 199 | job_id, kv_request_id, keys, block_ids, send_ready=self._send_ready |
| 200 | ) |
| 201 | |
| 202 | def add_stored_blocks( |
| 203 | self, |
| 204 | kv_request_id: str, |
| 205 | keys: Sequence[OffloadKey], |
| 206 | block_ids: Sequence[int], |
| 207 | job_id: JobId, |
| 208 | ) -> None: |
| 209 | """New blocks stored locally — match against pending fetch demand.""" |
| 210 | self._server.add_stored_blocks(kv_request_id, keys, block_ids, job_id) |
| 211 | |
| 212 | def finish_request(self, kv_request_id: str) -> None: |
| 213 | """Called when the request is finishing locally. |
| 214 | |
| 215 | Finishes the client role (aborts any inbound load and drops any |
| 216 | pending symmetric-P2P lookup state) and finalizes any outbound |
| 217 | serving (server role) for this id. Roles that aren't active for |
| 218 | this id are silent no-ops. |
| 219 | """ |
| 220 | self._client.finish(kv_request_id) |
| 221 | self._server.finish(kv_request_id) |
| 222 | |
| 223 | def register_lookup(self, kv_request_id: str, key: bytes) -> bool | None: |
| 224 | """Register or resolve one (kv_request_id, key) probe. |
| 225 | |
| 226 | Called from the manager's lookup() for symmetric-P2P consumers |
| 227 | (``remote_kv_source`` sub-dict in kv_transfer_params). See |
| 228 | ``ClientRole.register_lookup`` for the state-machine contract. |
| 229 | """ |
| 230 | return self._client.register_lookup(kv_request_id, key) |
| 231 | |
| 232 | def flush_pending_lookups(self) -> None: |
| 233 | """Flush any aggregated symmetric-P2P lookups for this peer. |
| 234 | |
| 235 | Called once per scheduler step from the manager's |
| 236 | ``on_schedule_end()``. Send-gating is handled inside the |
| 237 | client's ``_send`` callback (queues until ConnectAckMsg). |
| 238 | """ |
| 239 | self._client.flush_pending_lookups() |
| 240 | |
| 241 | def serve_external_requests(self, parent: ParentManager) -> None: |
| 242 | """Resolve inbound peer lookups against the tiering manager. |
| 243 | |
| 244 | Delegates to the server role; the ``parent`` handle is valid |
| 245 | only for the duration of this call. |
| 246 | """ |
| 247 | self._server.serve_external_requests(parent) |
| 248 | |
| 249 | def poll(self) -> SessionPollResult: |
| 250 | """Process incoming messages, drive transfers, apply timeouts.""" |
| 251 | if self._conn is None: |
| 252 | # Pending session — store-job timeouts still apply so buffered |
| 253 | # jobs that never get picked up are surfaced as failures. |
| 254 | return SessionPollResult( |
| 255 | loads=[], |
| 256 | stores=self._server.collect_idle_timeouts(), |
| 257 | new_fetch_ids=[], |
| 258 | ) |
| 259 | |
| 260 | for msg in self._conn.recv(): |
| 261 | self._on_message(msg) |
| 262 | |
| 263 | loads = self._client.collect_results() |
| 264 | stores = self._server.collect_results() |
| 265 | self._server.drain_pending_aborts() |
| 266 | |
| 267 | new_fetch_ids = self._new_fetch_ids |
| 268 | self._new_fetch_ids = [] |
| 269 | return SessionPollResult( |
| 270 | loads=loads, stores=stores, new_fetch_ids=new_fetch_ids |
| 271 | ) |
| 272 | |
| 273 | def close(self) -> SessionCloseResult: |
| 274 | """Shut down. |
| 275 | |
| 276 | failed_jobs: client load job_ids to fail. |
| 277 | failed_req_ids: client kv_request_ids to fail — in-flight loads plus |
| 278 | requests with an unresolved symmetric-P2P probe toward the |
| 279 | now-dead peer. The manager fails these so the consumer's lookup() |
| 280 | falls back to local prefill instead of deferring forever on an |
| 281 | answer that can never arrive. |
| 282 | failed_stores: server store job_ids to fail. |
| 283 | failed_serves: synthetic lookup ctxs still owing |
| 284 | ``parent.on_request_finished`` (the manager flushes these on |
| 285 | its next ``serve_external_requests``). |
| 286 | """ |
| 287 | client_result = self._client.close() |
| 288 | failed_stores, failed_serves = self._server.close() |
| 289 | |
| 290 | if self._conn is not None: |
| 291 | with contextlib.suppress(Exception): |
| 292 | self._conn.send({TYPE_KEY: DisconnectMsg.TYPE}) |
| 293 | self._conn.close() |
| 294 | self._conn = None |
| 295 | |
| 296 | return SessionCloseResult( |
| 297 | failed_jobs=client_result.failed_jobs, |
| 298 | failed_req_ids=client_result.failed_req_ids, |
| 299 | failed_stores=failed_stores, |
| 300 | failed_serves=failed_serves, |
| 301 | ) |
| 302 | |
| 303 | # ------------------------------------------------------------------ |
| 304 | # Message dispatch |
| 305 | # ------------------------------------------------------------------ |
| 306 | |
| 307 | def _on_message(self, msg: dict) -> None: |
| 308 | msg_type = msg.get(TYPE_KEY) if isinstance(msg, dict) else msg |
| 309 | try: |
| 310 | self._dispatch_message(msg) |
| 311 | except ValueError as exc: |
| 312 | # Protocol contract violation from the peer — *Msg.validate() |
| 313 | # and handler-level checks raise ValueError. Retrying won't |
| 314 | # help and may corrupt session state, so disconnect now. |
| 315 | self._protocol_error(f"malformed {msg_type!r}: {exc}") |
| 316 | return |
| 317 | except Exception as exc: |
| 318 | # Anything else is most likely an internal bug rather than a |
| 319 | # peer fault. Log loudly with a traceback so it doesn't |
| 320 | # disappear, but don't kill the session on a single hiccup. |
| 321 | # Disconnect only if errors keep arriving — that pattern is |
| 322 | # consistent with a peer wedging us into a broken state. |
| 323 | self._dispatch_error_count += 1 |
| 324 | logger.exception( |
| 325 | "P2PSession %s: error handling message %r (count=%d): %s", |
| 326 | self.peer_id, |
| 327 | msg_type, |
| 328 | self._dispatch_error_count, |
| 329 | exc, |
| 330 | ) |
| 331 | if self._dispatch_error_count >= _MAX_CONSECUTIVE_DISPATCH_ERRORS: |
| 332 | self._protocol_error( |
| 333 | f"too many consecutive dispatch errors " |
| 334 | f"({self._dispatch_error_count})" |
| 335 | ) |
| 336 | return |
| 337 | self._dispatch_error_count = 0 |
| 338 | |
| 339 | def _protocol_error(self, reason: str) -> None: |
| 340 | """Log a protocol violation and disconnect. |
| 341 | |
| 342 | Best-effort sends ``DisconnectMsg`` so the peer learns why we're |
| 343 | going away, then marks the connection dead. The manager reaps |
| 344 | the session on the next poll via ``alive``. |
| 345 | """ |
| 346 | logger.error( |
| 347 | "P2PSession %s: protocol error: %s — disconnecting", |
| 348 | self.peer_id, |
| 349 | reason, |
| 350 | ) |
| 351 | if self._conn is not None: |
| 352 | with contextlib.suppress(Exception): |
| 353 | self._conn.send({TYPE_KEY: DisconnectMsg.TYPE}) |
| 354 | self._conn.mark_dead() |
| 355 | |
| 356 | def _dispatch_message(self, msg: dict) -> None: |
| 357 | # Drop messages buffered before disconnect: a poll batch can |
| 358 | # contain msg-after-DisconnectMsg, and dispatching them would |
| 359 | # mutate state on a dead session. |
| 360 | if self._conn is not None and not self._conn.alive: |
| 361 | return |
| 362 | msg_type = msg.get(TYPE_KEY) if isinstance(msg, dict) else None |
| 363 | if msg_type == ConnectMsg.TYPE: |
| 364 | self._on_connect(msg) |
| 365 | elif msg_type == ConnectAckMsg.TYPE: |
| 366 | ConnectAckMsg.validate(msg) |
| 367 | self._on_connect_ack() |
| 368 | elif msg_type == FetchMsg.TYPE: |
| 369 | FetchMsg.validate(msg) |
| 370 | kv_request_id = msg[FetchMsg.KV_REQUEST_ID] |
| 371 | keys = [ |
| 372 | OffloadKey(bh if isinstance(bh, bytes) else bytes(bh)) |
| 373 | for bh in msg[FetchMsg.KEYS] |
| 374 | ] |
| 375 | block_indexes = msg[FetchMsg.BLOCK_INDEXES] |
| 376 | round_seq = msg[FetchMsg.ROUND_SEQ] |
| 377 | # Run the server-role state machine inline as today — |
| 378 | # add_fetch_demand records demand against any blocks we've |
| 379 | # already seen in `available`. Report the kv_request_id so |
| 380 | # the manager (after poll() returns) can replay any parked |
| 381 | # submit_store batches; their add_stored_blocks calls hit |
| 382 | # the demand recorded here and submit transfers immediately. |
| 383 | self._server.on_fetch(kv_request_id, keys, block_indexes, round_seq) |
| 384 | self._new_fetch_ids.append(kv_request_id) |
| 385 | elif msg_type == AbortFetchMsg.TYPE: |
| 386 | AbortFetchMsg.validate(msg) |
| 387 | self._server.on_abort_fetch( |
| 388 | msg[AbortFetchMsg.KV_REQUEST_ID], |
| 389 | msg[AbortFetchMsg.ROUND_SEQ], |
| 390 | ) |
| 391 | elif msg_type == TransferDoneMsg.TYPE: |
| 392 | TransferDoneMsg.validate(msg) |
| 393 | self._client.on_transfer_done( |
| 394 | msg[TransferDoneMsg.KV_REQUEST_ID], |
| 395 | msg[TransferDoneMsg.SUCCESS], |
| 396 | msg[TransferDoneMsg.ROUND_SEQ], |
| 397 | ) |
| 398 | elif msg_type == AbortAckMsg.TYPE: |
| 399 | AbortAckMsg.validate(msg) |
| 400 | self._client.on_abort_ack( |
| 401 | msg[AbortAckMsg.KV_REQUEST_ID], |
| 402 | msg[AbortAckMsg.ROUND_SEQ], |
| 403 | ) |
| 404 | elif msg_type == LookupMsg.TYPE: |
| 405 | LookupMsg.validate(msg) |
| 406 | kv_request_id = msg[LookupMsg.KV_REQUEST_ID] |
| 407 | keys = [ |
| 408 | OffloadKey(bh if isinstance(bh, bytes) else bytes(bh)) |
| 409 | for bh in msg[LookupMsg.KEYS] |
| 410 | ] |
| 411 | self._server.on_lookup(kv_request_id, keys, msg[LookupMsg.ROUND_SEQ]) |
| 412 | elif msg_type == LookupRespMsg.TYPE: |
| 413 | LookupRespMsg.validate(msg) |
| 414 | kv_request_id = msg[LookupRespMsg.KV_REQUEST_ID] |
| 415 | keys = [ |
| 416 | OffloadKey(bh if isinstance(bh, bytes) else bytes(bh)) |
| 417 | for bh in msg[LookupRespMsg.KEYS] |
| 418 | ] |
| 419 | hits = msg[LookupRespMsg.HITS] |
| 420 | self._client.on_lookup_resp(kv_request_id, keys, hits) |
| 421 | elif msg_type == DisconnectMsg.TYPE: |
| 422 | if self._conn is not None: |
| 423 | self._conn.mark_dead() |
| 424 | else: |
| 425 | logger.warning( |
| 426 | "P2PSession %s: unknown message type %r", self.peer_id, msg_type |
| 427 | ) |
| 428 | |
| 429 | # ------------------------------------------------------------------ |
| 430 | # Handshake |
| 431 | # ------------------------------------------------------------------ |
| 432 | |
| 433 | def _on_connect(self, msg: dict) -> None: |
| 434 | # Validation failures here mean an incompatible or malicious peer. |
| 435 | # Mark the connection dead so the manager reaps the session; |
| 436 | # don't call add_remote_peer or send connect_ack. |
| 437 | if self._send_ready: |
| 438 | # We've already received connect_ack, so the handshake is |
| 439 | # complete. A second connect from the peer is a protocol |
| 440 | # violation — re-registering would corrupt transport state. |
| 441 | self._protocol_error("duplicate connect after handshake") |
| 442 | return |
| 443 | try: |
| 444 | ConnectMsg.validate(msg) |
| 445 | if msg[ConnectMsg.BLOCK_LEN] != self._local_block_len: |
| 446 | raise ValueError( |
| 447 | f"block_len mismatch from {self.peer_id}: " |
| 448 | f"remote={msg[ConnectMsg.BLOCK_LEN]}, " |
| 449 | f"local={self._local_block_len}" |
| 450 | ) |
| 451 | remote_fp = msg.get(ConnectMsg.CONFIG_FINGERPRINT, "") |
| 452 | local_fp = self._transport.config_fingerprint |
| 453 | if local_fp and remote_fp and remote_fp != local_fp: |
| 454 | raise ValueError( |
| 455 | f"config fingerprint mismatch from {self.peer_id}: " |
| 456 | f"remote={remote_fp!r}, local={local_fp!r}" |
| 457 | ) |
| 458 | if msg[ConnectMsg.HASH_SEED] != self._local_hash_seed: |
| 459 | raise ValueError( |
| 460 | f"hash seed mismatch from {self.peer_id}: " |
| 461 | f"remote={msg[ConnectMsg.HASH_SEED]!r}, " |
| 462 | f"local={self._local_hash_seed!r}. Ensure PYTHONHASHSEED " |
| 463 | "(if set) matches on all P2P peers." |
| 464 | ) |
| 465 | self._transport.add_remote_peer( |
| 466 | self.peer_id, |
| 467 | agent_metadata=msg[ConnectMsg.AGENT_METADATA], |
| 468 | base_addr=msg[ConnectMsg.BASE_ADDR], |
| 469 | num_blocks=msg[ConnectMsg.NUM_BLOCKS], |
| 470 | block_len=msg[ConnectMsg.BLOCK_LEN], |
| 471 | ) |
| 472 | except ValueError as exc: |
| 473 | logger.error("P2PSession %s: rejecting peer connect: %s", self.peer_id, exc) |
| 474 | if self._conn is not None: |
| 475 | self._conn.mark_dead() |
| 476 | return |
| 477 | |
| 478 | if self._conn is not None: |
| 479 | self._conn.send( |
| 480 | { |
| 481 | TYPE_KEY: ConnectAckMsg.TYPE, |
| 482 | ConnectAckMsg.PEER_ID: self._local_id, |
| 483 | } |
| 484 | ) |
| 485 | |
| 486 | def _on_connect_ack(self) -> None: |
| 487 | if self._queued: |
| 488 | logger.debug( |
| 489 | "P2PSession %s: connect_ack received, flushing %d queued msg(s)", |
| 490 | self.peer_id, |
| 491 | len(self._queued), |
| 492 | ) |
| 493 | self._send_ready = True |
| 494 | for queued in self._queued: |
| 495 | self._do_send(queued) |
| 496 | self._queued.clear() |
| 497 | |
| 498 | # ------------------------------------------------------------------ |
| 499 | # Send helpers |
| 500 | # ------------------------------------------------------------------ |
| 501 | |
| 502 | def _send_connect(self) -> None: |
| 503 | """Send our ConnectMsg announcing local NIXL metadata.""" |
| 504 | assert self._conn is not None |
| 505 | self._conn.send( |
| 506 | { |
| 507 | TYPE_KEY: ConnectMsg.TYPE, |
| 508 | ConnectMsg.PEER_ID: self._local_id, |
| 509 | ConnectMsg.AGENT_METADATA: self._transport.get_agent_metadata(), |
| 510 | ConnectMsg.BASE_ADDR: self._transport.base_addr, |
| 511 | ConnectMsg.NUM_BLOCKS: self._transport.num_blocks, |
| 512 | ConnectMsg.BLOCK_LEN: self._transport.block_len, |
| 513 | ConnectMsg.CONFIG_FINGERPRINT: self._transport.config_fingerprint, |
| 514 | ConnectMsg.HASH_SEED: self._local_hash_seed, |
| 515 | } |
| 516 | ) |
| 517 | |
| 518 | def _send(self, msg: dict) -> None: |
| 519 | if self._conn is None or not self._send_ready: |
| 520 | logger.debug( |
| 521 | "P2PSession %s: queueing %s (ready=%s queue_depth=%d)", |
| 522 | self.peer_id, |
| 523 | msg.get(TYPE_KEY), |
| 524 | self._send_ready, |
| 525 | len(self._queued) + 1, |
| 526 | ) |
| 527 | self._queued.append(msg) |
| 528 | return |
| 529 | self._do_send(msg) |
| 530 | |
| 531 | def _do_send(self, msg: dict) -> None: |
| 532 | if self._conn is None: |
| 533 | return |
| 534 | try: |
| 535 | self._conn.send(msg) |
| 536 | logger.debug("P2PSession %s: sent %s", self.peer_id, msg.get(TYPE_KEY)) |
| 537 | except Exception: |
| 538 | # A send failure means the connection is broken. Swallowing it |
| 539 | # silently strands every in-flight lookup/load toward this peer: |
| 540 | # the session stays alive, is never reaped, and the consumer's |
| 541 | # lookup() keeps returning RETRY until the HTTP client times out. |
| 542 | # Mark the connection dead so the manager reaps the session on |
| 543 | # its next poll and surfaces the stranded work as failures. |
| 544 | logger.warning( |
| 545 | "P2PSession %s: send of %s failed — marking connection dead", |
| 546 | self.peer_id, |
| 547 | msg.get(TYPE_KEY), |
| 548 | ) |
| 549 | self._conn.mark_dead() |