| 1 | # SPDX-License-Identifier: Apache-2.0 |
| 2 | # SPDX-FileCopyrightText: Copyright contributors to the vLLM project |
| 3 | """Client-role state machine for a single peer session. |
| 4 | |
| 5 | Handles outgoing fetch requests, abort-on-timeout, abort-ack timeout, |
| 6 | and produces ``LoadResult`` for completed loads. The session coordinator |
| 7 | parses wire messages and dispatches typed arguments here; this module |
| 8 | never touches ``ControlConnection`` directly — it emits via the ``send`` |
| 9 | callback injected by the coordinator (which gates on ConnectAck). |
| 10 | """ |
| 11 | |
| 12 | from __future__ import annotations |
| 13 | |
| 14 | import time |
| 15 | from collections.abc import Callable, Sequence |
| 16 | from dataclasses import dataclass, field |
| 17 | from typing import TYPE_CHECKING, NamedTuple |
| 18 | |
| 19 | from vllm.logger import init_logger |
| 20 | from vllm.v1.kv_offload.base import OffloadKey |
| 21 | from vllm.v1.kv_offload.tiering.p2p.session.protocol import ( |
| 22 | TYPE_KEY, |
| 23 | AbortFetchMsg, |
| 24 | FetchMsg, |
| 25 | LookupMsg, |
| 26 | ) |
| 27 | |
| 28 | if TYPE_CHECKING: |
| 29 | from vllm.v1.kv_offload.tiering.base import JobId |
| 30 | |
| 31 | logger = init_logger(__name__) |
| 32 | |
| 33 | _LOAD_TIMEOUT_S = 30.0 |
| 34 | _ABORT_ACK_TIMEOUT_S = 10.0 |
| 35 | |
| 36 | |
| 37 | @dataclass |
| 38 | class _InboundLoadState: |
| 39 | """Client-role state for a single in-flight load request. |
| 40 | |
| 41 | Lives in ``_ClientRequestState.loads`` keyed by round_seq for the |
| 42 | duration of a fetch; the owning kv_request_id is the outer dict key. |
| 43 | """ |
| 44 | |
| 45 | job_id: int # opaque ID assigned by the manager to this load request |
| 46 | submitted_at: float |
| 47 | aborted_at: float | None = None |
| 48 | |
| 49 | |
| 50 | @dataclass |
| 51 | class _ClientRequestState: |
| 52 | """Per-kv_request_id client-side state. |
| 53 | |
| 54 | One entry per kv_request_id we're driving. Lookup-phase fields are |
| 55 | used only by symmetric P2P (``do_p2p_fetch``); PD-only loads leave |
| 56 | ``probes``/``unsent`` empty and drive just ``phase`` and ``loads``. An |
| 57 | entry is dropped once every field is idle — see ``ClientRole._maybe_prune``. |
| 58 | """ |
| 59 | |
| 60 | # -- Lookup phase (symmetric P2P only; untouched for PD) -- |
| 61 | # Probe outcome per OffloadKey: None while in-flight (registered/sent |
| 62 | # but unresolved), True/False once a LookupRespMsg lands. There is no |
| 63 | # timeout — finish (via finish_request) is guaranteed after |
| 64 | # the request's lookup() calls and clears every probe, so an |
| 65 | # unanswered probe simply stays None until then. |
| 66 | probes: dict[OffloadKey, bool | None] = field(default_factory=dict) |
| 67 | # OffloadKeys registered but not yet flushed onto the wire. Drained and |
| 68 | # cleared by the next flush_pending_lookups. |
| 69 | unsent: list[OffloadKey] = field(default_factory=list) |
| 70 | # Current lookup round. LookupMsgs carry it, each fetch closes it and |
| 71 | # advances it, so every round's supply/demand/completion is isolated |
| 72 | # on the wire. PD clients never probe and stay on round 0. |
| 73 | round_seq: int = 0 |
| 74 | # This id ran the symmetric lookup phase (register_lookup); a fetch |
| 75 | # with keys then requires every key to be a confirmed probe. PD |
| 76 | # loads never probe. |
| 77 | probed: bool = False |
| 78 | |
| 79 | # The peer holds lookup state no FetchMsg has closed: a LookupMsg |
| 80 | # was flushed since the last fetch. finish owes a terminal empty |
| 81 | # FetchMsg while set, so the peer releases parked supply. |
| 82 | peer_lookup_open: bool = False |
| 83 | # In-flight loads keyed by the round their fetch carried. The |
| 84 | # scheduler submits loads incrementally as chunks resolve, so several |
| 85 | # can be in flight at once; TransferDone/AbortAck match by round. |
| 86 | loads: dict[int, _InboundLoadState] = field(default_factory=dict) |
| 87 | |
| 88 | |
| 89 | class LoadResult(NamedTuple): |
| 90 | """Result from a session poll, client side.""" |
| 91 | |
| 92 | job_id: int |
| 93 | kv_request_id: str |
| 94 | success: bool |
| 95 | |
| 96 | |
| 97 | class ClientCloseResult(NamedTuple): |
| 98 | """What the manager must fail when a peer session is torn down. |
| 99 | |
| 100 | ``failed_jobs`` is the ``job_id`` of every load still in flight; each |
| 101 | becomes a ``JobResult(success=False)``. ``failed_req_ids`` is every |
| 102 | kv_request_id whose lookup() would otherwise defer forever on the dead |
| 103 | peer — the in-flight loads plus any request holding an unresolved |
| 104 | symmetric-P2P probe. ``failed_jobs`` is the subset of ``failed_req_ids`` |
| 105 | that had a load job. |
| 106 | """ |
| 107 | |
| 108 | failed_jobs: list[int] |
| 109 | failed_req_ids: list[str] |
| 110 | |
| 111 | |
| 112 | class ClientRole: |
| 113 | """Client-side load state machine for one peer session. |
| 114 | |
| 115 | The coordinator owns the connection and the send-gating; this role |
| 116 | is given a ``send`` callback and a ``peer_id`` for log messages and |
| 117 | is otherwise self-contained. |
| 118 | """ |
| 119 | |
| 120 | def __init__(self, peer_id: str, send: Callable[[dict], None]) -> None: |
| 121 | self._peer_id = peer_id |
| 122 | self._send = send |
| 123 | # All per-kv_request_id state lives here. Entries are created |
| 124 | # lazily by request_blocks / register_lookup and dropped by |
| 125 | # _maybe_prune once every field is idle. |
| 126 | self._requests: dict[str, _ClientRequestState] = {} |
| 127 | # kv_request_ids with unsent lookup keys for the next flush to |
| 128 | # visit — the work-list that keeps flush_pending_lookups from |
| 129 | # scanning every request each scheduler step. Mirrors the server's |
| 130 | # _serve_pending. Populated by register_lookup, drained by |
| 131 | # flush_pending_lookups, and discarded on finish/close. |
| 132 | self._flush_pending: set[str] = set() |
| 133 | # kv_request_ids with at least one fetch in flight or aborting — the |
| 134 | # work-list collect_results walks for timeouts, and the |
| 135 | # has_active_loads predicate, instead of scanning every request. |
| 136 | # Kept in exact sync with ``st.loads``. |
| 137 | self._active_loads: set[str] = set() |
| 138 | self._completed_loads: list[LoadResult] = [] |
| 139 | |
| 140 | # ------------------------------------------------------------------ |
| 141 | # State helpers |
| 142 | # ------------------------------------------------------------------ |
| 143 | |
| 144 | def _get_or_create_request(self, kv_request_id: str) -> _ClientRequestState: |
| 145 | """Get or create the state entry for a kv_request_id.""" |
| 146 | st = self._requests.get(kv_request_id) |
| 147 | if st is None: |
| 148 | st = _ClientRequestState() |
| 149 | self._requests[kv_request_id] = st |
| 150 | return st |
| 151 | |
| 152 | def _maybe_prune(self, kv_request_id: str) -> None: |
| 153 | """Drop the entry once it holds no live load or lookup state. |
| 154 | |
| 155 | ``peer_lookup_open`` is only read by ``finish``, and every path |
| 156 | that clears the last probe (fetch / finish / close) also settles |
| 157 | it, so dropping on emptiness never loses a flag still in use. |
| 158 | """ |
| 159 | st = self._requests.get(kv_request_id) |
| 160 | if st is not None and not st.loads and not st.probes and not st.unsent: |
| 161 | del self._requests[kv_request_id] |
| 162 | |
| 163 | def _on_load_terminal(self, kv_request_id: str, st: _ClientRequestState) -> None: |
| 164 | """Wind down id-level state once no load remains in flight.""" |
| 165 | if st.loads: |
| 166 | return |
| 167 | self._active_loads.discard(kv_request_id) |
| 168 | self._maybe_prune(kv_request_id) |
| 169 | |
| 170 | @property |
| 171 | def has_active_loads(self) -> bool: |
| 172 | """True if any kv_request_id has a fetch in flight.""" |
| 173 | return bool(self._active_loads) |
| 174 | |
| 175 | # ------------------------------------------------------------------ |
| 176 | # Public API |
| 177 | # ------------------------------------------------------------------ |
| 178 | |
| 179 | def request_blocks( |
| 180 | self, |
| 181 | job_id: JobId, |
| 182 | kv_request_id: str, |
| 183 | keys: Sequence[OffloadKey], |
| 184 | block_ids: Sequence[int], |
| 185 | send_ready: bool, |
| 186 | ) -> None: |
| 187 | """Send the FetchMsg closing the current lookup round. |
| 188 | |
| 189 | The scheduler may submit several loads per kv_request_id as its |
| 190 | matched prefix resolves incrementally; each fetch carries the |
| 191 | round it closes so the loads stay independent on the wire. |
| 192 | """ |
| 193 | logger.debug( |
| 194 | "P2PSession %s: request_blocks job_id=%d kv_request_id=%s " |
| 195 | "blocks=%d ready=%s", |
| 196 | self._peer_id, |
| 197 | job_id, |
| 198 | kv_request_id, |
| 199 | len(block_ids), |
| 200 | send_ready, |
| 201 | ) |
| 202 | st = self._get_or_create_request(kv_request_id) |
| 203 | round_seq = st.round_seq |
| 204 | st.round_seq += 1 |
| 205 | st.loads[round_seq] = _InboundLoadState( |
| 206 | job_id=job_id, |
| 207 | submitted_at=time.monotonic(), |
| 208 | ) |
| 209 | self._active_loads.add(kv_request_id) |
| 210 | st.peer_lookup_open = False |
| 211 | self._send( |
| 212 | { |
| 213 | TYPE_KEY: FetchMsg.TYPE, |
| 214 | FetchMsg.KV_REQUEST_ID: kv_request_id, |
| 215 | FetchMsg.KEYS: list(keys), |
| 216 | FetchMsg.BLOCK_INDEXES: [int(idx) for idx in block_ids], |
| 217 | FetchMsg.ROUND_SEQ: round_seq, |
| 218 | } |
| 219 | ) |
| 220 | # Issuing the fetch closes this lookup round, so drop all probe |
| 221 | # state. Once the peer serves the fetch both sides unpin, so a |
| 222 | # stale cached True would let a later lookup() return HIT for a |
| 223 | # block the producer may have evicted; clearing forces a fresh |
| 224 | # probe under the next round. |
| 225 | if st.probed and keys: |
| 226 | assert st.probes, ( |
| 227 | f"symmetric fetch for {kv_request_id} has keys but no probes" |
| 228 | ) |
| 229 | assert all(st.probes.get(key) is True for key in keys) |
| 230 | st.probes.clear() |
| 231 | |
| 232 | def finish(self, kv_request_id: str) -> None: |
| 233 | """Finish a request: abort in-flight loads and release lookup state. |
| 234 | |
| 235 | Called from the session's ``finish_request``. Sends an |
| 236 | AbortFetchMsg per load not already aborting, and — independently — |
| 237 | the terminal empty FetchMsg when the peer still holds lookup |
| 238 | state no fetch has closed (its "request finished" signal: it |
| 239 | releases lookup state, drains parked supply, and fires |
| 240 | ``cb.finish_request``). A later round's supply can be parked |
| 241 | while an earlier round's load is still in flight, so both can be |
| 242 | owed at once. |
| 243 | |
| 244 | Then drop all probe/lookup state. Keep aborted loads until their abort |
| 245 | acknowledgement or timeout produces a terminal result. |
| 246 | """ |
| 247 | st = self._requests.get(kv_request_id) |
| 248 | if st is None: |
| 249 | return |
| 250 | if st.loads: |
| 251 | for round_seq, load in st.loads.items(): |
| 252 | if load.aborted_at is not None: |
| 253 | continue |
| 254 | self._send( |
| 255 | { |
| 256 | TYPE_KEY: AbortFetchMsg.TYPE, |
| 257 | AbortFetchMsg.KV_REQUEST_ID: kv_request_id, |
| 258 | AbortFetchMsg.ROUND_SEQ: round_seq, |
| 259 | } |
| 260 | ) |
| 261 | load.aborted_at = time.monotonic() |
| 262 | if st.peer_lookup_open: |
| 263 | st.peer_lookup_open = False |
| 264 | self._send( |
| 265 | { |
| 266 | TYPE_KEY: FetchMsg.TYPE, |
| 267 | FetchMsg.KV_REQUEST_ID: kv_request_id, |
| 268 | FetchMsg.KEYS: [], |
| 269 | FetchMsg.BLOCK_INDEXES: [], |
| 270 | FetchMsg.ROUND_SEQ: st.round_seq, |
| 271 | } |
| 272 | ) |
| 273 | st.probes.clear() |
| 274 | st.unsent.clear() |
| 275 | self._flush_pending.discard(kv_request_id) |
| 276 | self._maybe_prune(kv_request_id) |
| 277 | |
| 278 | def on_transfer_done( |
| 279 | self, kv_request_id: str, success: bool, round_seq: int |
| 280 | ) -> None: |
| 281 | """Handle a TransferDoneMsg from the peer.""" |
| 282 | st = self._requests.get(kv_request_id) |
| 283 | load = st.loads.pop(round_seq, None) if st is not None else None |
| 284 | if st is not None and load is not None: |
| 285 | self._completed_loads.append( |
| 286 | LoadResult( |
| 287 | job_id=load.job_id, |
| 288 | kv_request_id=kv_request_id, |
| 289 | success=success, |
| 290 | ) |
| 291 | ) |
| 292 | self._on_load_terminal(kv_request_id, st) |
| 293 | else: |
| 294 | # No matching in-flight load: either a duplicate |
| 295 | # transfer_done from the peer (protocol violation) or a |
| 296 | # benign race with a local cancel/abort/timeout that |
| 297 | # already popped the entry. We don't track terminated ids, |
| 298 | # so we can't tell — log so it's findable. |
| 299 | logger.warning( |
| 300 | "P2PSession %s: transfer_done for unknown kv_request_id=%s " |
| 301 | "round=%s (duplicate from peer, or raced with local " |
| 302 | "cancel/timeout)", |
| 303 | self._peer_id, |
| 304 | kv_request_id, |
| 305 | round_seq, |
| 306 | ) |
| 307 | |
| 308 | def on_abort_ack(self, kv_request_id: str, round_seq: int) -> None: |
| 309 | """Handle an AbortAckMsg from the peer.""" |
| 310 | st = self._requests.get(kv_request_id) |
| 311 | load = st.loads.pop(round_seq, None) if st is not None else None |
| 312 | if st is not None and load is not None: |
| 313 | logger.warning( |
| 314 | "P2PSession %s: load request %s (job_id=%d) " |
| 315 | "completed with failure after abort", |
| 316 | self._peer_id, |
| 317 | kv_request_id, |
| 318 | load.job_id, |
| 319 | ) |
| 320 | self._completed_loads.append( |
| 321 | LoadResult( |
| 322 | job_id=load.job_id, |
| 323 | kv_request_id=kv_request_id, |
| 324 | success=False, |
| 325 | ) |
| 326 | ) |
| 327 | self._on_load_terminal(kv_request_id, st) |
| 328 | else: |
| 329 | # See on_transfer_done: same ambiguity (duplicate ack |
| 330 | # vs. raced with local cancel/timeout that already popped). |
| 331 | logger.warning( |
| 332 | "P2PSession %s: abort_ack for unknown kv_request_id=%s " |
| 333 | "round=%s (duplicate from peer, or raced with local " |
| 334 | "cancel/timeout)", |
| 335 | self._peer_id, |
| 336 | kv_request_id, |
| 337 | round_seq, |
| 338 | ) |
| 339 | |
| 340 | # ------------------------------------------------------------------ |
| 341 | # Symmetric-P2P lookup (do_p2p_fetch=true) |
| 342 | # ------------------------------------------------------------------ |
| 343 | |
| 344 | def register_lookup(self, kv_request_id: str, key: bytes) -> bool | None: |
| 345 | """Register or resolve one (kv_request_id, key) probe. |
| 346 | |
| 347 | Idempotent across scheduler steps: |
| 348 | - First call: creates a pending entry, returns None. |
| 349 | - Subsequent calls while in-flight: returns None. |
| 350 | - Once a LookupRespMsg has resolved the entry: returns the cached |
| 351 | bool result on every call without popping it. |
| 352 | |
| 353 | A resolved entry is retained until a fetch closes the round |
| 354 | (``request_blocks`` clears all probes) or the request finishes |
| 355 | (``finish`` clears all entries for the id). A request's |
| 356 | block set can be re-probed across steps, so popping on read would |
| 357 | make a repeat probe of an already-resolved key look brand-new and |
| 358 | re-queue it, emitting a redundant LookupMsg for an answer we |
| 359 | already hold. Keeping the entry until fetch makes repeat probes |
| 360 | free; clearing at fetch forces a fresh probe under the next |
| 361 | round, since the block is unpinned once served. |
| 362 | """ |
| 363 | st = self._get_or_create_request(kv_request_id) |
| 364 | st.probed = True |
| 365 | okey = OffloadKey(key) |
| 366 | if okey in st.probes: |
| 367 | return st.probes[okey] |
| 368 | st.probes[okey] = None |
| 369 | st.unsent.append(okey) |
| 370 | self._flush_pending.add(kv_request_id) |
| 371 | logger.debug( |
| 372 | "P2P LOOKUP client %s: REGISTER kv_request_id=%s key=%s (unsent=%d)", |
| 373 | self._peer_id, |
| 374 | kv_request_id, |
| 375 | key.hex()[:16], |
| 376 | len(st.unsent), |
| 377 | ) |
| 378 | return None |
| 379 | |
| 380 | def flush_pending_lookups(self) -> None: |
| 381 | """Send a LookupMsg for each kv_request_id with unsent entries. |
| 382 | |
| 383 | Called once per scheduler step from the manager's |
| 384 | ``on_schedule_end()``. A request's block set may be discovered |
| 385 | across several scheduler steps, so more than one LookupMsg can |
| 386 | go out per kv_request_id — one per step that registered new |
| 387 | keys, all tagged with the current round. register_lookup() |
| 388 | de-dups in-flight and already-resolved (req_id, key) pairs, so |
| 389 | each LookupMsg carries only the keys first probed in that step. |
| 390 | Send-gating is handled by the injected ``_send`` callback |
| 391 | (queues until ConnectAckMsg if needed). |
| 392 | |
| 393 | Only requests that registered new keys since the last flush are |
| 394 | visited — the ``_flush_pending`` work-list avoids scanning every |
| 395 | live request each scheduler step. |
| 396 | """ |
| 397 | for req_id in self._flush_pending: |
| 398 | st = self._requests.get(req_id) |
| 399 | if st is None or not st.unsent: |
| 400 | continue |
| 401 | # The peer now holds lookup state for this id; finish owes a |
| 402 | # terminal empty FetchMsg until a fetch closes it. |
| 403 | st.peer_lookup_open = True |
| 404 | logger.debug( |
| 405 | "P2P LOOKUP client %s: SEND LookupMsg kv_request_id=%s keys=%d", |
| 406 | self._peer_id, |
| 407 | req_id, |
| 408 | len(st.unsent), |
| 409 | ) |
| 410 | self._send( |
| 411 | { |
| 412 | TYPE_KEY: LookupMsg.TYPE, |
| 413 | LookupMsg.KV_REQUEST_ID: req_id, |
| 414 | LookupMsg.KEYS: list(st.unsent), |
| 415 | LookupMsg.ROUND_SEQ: st.round_seq, |
| 416 | } |
| 417 | ) |
| 418 | st.unsent = [] |
| 419 | self._flush_pending.clear() |
| 420 | |
| 421 | def on_lookup_resp( |
| 422 | self, |
| 423 | kv_request_id: str, |
| 424 | keys: Sequence[bytes], |
| 425 | hits: Sequence[bool], |
| 426 | ) -> None: |
| 427 | """Apply per-pair hit/miss results from a peer. |
| 428 | |
| 429 | Pairs that don't match a known entry (already cancelled or |
| 430 | never asked) are silently dropped — the producer is free to |
| 431 | split or coalesce responses. |
| 432 | """ |
| 433 | n_hit = sum(1 for hit in hits if hit) |
| 434 | logger.debug( |
| 435 | "P2P LOOKUP client %s: RECV LookupRespMsg kv_request_id=%s " |
| 436 | "keys=%d hits=%d misses=%d", |
| 437 | self._peer_id, |
| 438 | kv_request_id, |
| 439 | len(keys), |
| 440 | n_hit, |
| 441 | len(hits) - n_hit, |
| 442 | ) |
| 443 | st = self._requests.get(kv_request_id) |
| 444 | if st is None: |
| 445 | return |
| 446 | for h, hit in zip(keys, hits): |
| 447 | key = OffloadKey(h) |
| 448 | if key in st.probes: |
| 449 | st.probes[key] = hit |
| 450 | |
| 451 | def collect_results(self) -> list[LoadResult]: |
| 452 | """Walk load timeouts and drain completed loads. |
| 453 | |
| 454 | Loads past ``_LOAD_TIMEOUT_S`` get an AbortFetchMsg sent and |
| 455 | enter the aborting phase. Aborting loads past |
| 456 | ``_ABORT_ACK_TIMEOUT_S`` are surfaced as failed. |
| 457 | |
| 458 | Lookups have no timeout: an unanswered probe stays None (RETRY) |
| 459 | until finish_request clears it — see ``_ClientRequestState.probes``. |
| 460 | """ |
| 461 | now = time.monotonic() |
| 462 | to_remove: list[tuple[str, int]] = [] |
| 463 | for req_id in self._active_loads: |
| 464 | st = self._requests[req_id] |
| 465 | assert st.loads |
| 466 | for round_seq, load in st.loads.items(): |
| 467 | if load.aborted_at is None: |
| 468 | if now - load.submitted_at >= _LOAD_TIMEOUT_S: |
| 469 | load.aborted_at = now |
| 470 | logger.warning( |
| 471 | "P2PSession %s: %s round=%s timed out, sending abort", |
| 472 | self._peer_id, |
| 473 | req_id, |
| 474 | round_seq, |
| 475 | ) |
| 476 | self._send( |
| 477 | { |
| 478 | TYPE_KEY: AbortFetchMsg.TYPE, |
| 479 | AbortFetchMsg.KV_REQUEST_ID: req_id, |
| 480 | AbortFetchMsg.ROUND_SEQ: round_seq, |
| 481 | } |
| 482 | ) |
| 483 | else: |
| 484 | if now - load.aborted_at >= _ABORT_ACK_TIMEOUT_S: |
| 485 | to_remove.append((req_id, round_seq)) |
| 486 | self._completed_loads.append( |
| 487 | LoadResult( |
| 488 | job_id=load.job_id, |
| 489 | kv_request_id=req_id, |
| 490 | success=False, |
| 491 | ) |
| 492 | ) |
| 493 | logger.warning( |
| 494 | "P2PSession %s: abort_ack timed out for " |
| 495 | "kv_request_id=%s round=%s", |
| 496 | self._peer_id, |
| 497 | req_id, |
| 498 | round_seq, |
| 499 | ) |
| 500 | for req_id, round_seq in to_remove: |
| 501 | st = self._requests[req_id] |
| 502 | st.loads.pop(round_seq) |
| 503 | self._on_load_terminal(req_id, st) |
| 504 | |
| 505 | results = self._completed_loads |
| 506 | self._completed_loads = [] |
| 507 | return results |
| 508 | |
| 509 | def close(self) -> ClientCloseResult: |
| 510 | """Tear down, reporting work the dead peer can no longer complete. |
| 511 | |
| 512 | A request is failed if it has a load in flight (its job fails) or |
| 513 | holds an unresolved symmetric-P2P probe (its lookup() would defer |
| 514 | forever on an answer that can never arrive). See ``ClientCloseResult``. |
| 515 | """ |
| 516 | failed_jobs = [ |
| 517 | load.job_id for st in self._requests.values() for load in st.loads.values() |
| 518 | ] |
| 519 | failed_req_ids = [ |
| 520 | req_id |
| 521 | for req_id, st in self._requests.items() |
| 522 | if st.loads or any(hit is None for hit in st.probes.values()) |
| 523 | ] |
| 524 | self._requests.clear() |
| 525 | self._flush_pending.clear() |
| 526 | self._active_loads.clear() |
| 527 | self._completed_loads.clear() |
| 528 | return ClientCloseResult(failed_jobs=failed_jobs, failed_req_ids=failed_req_ids) |