diff --git a/docs/features/kv_offloading_usage.md b/docs/features/kv_offloading_usage.md index cff65753d9..8a48d11be7 100644 --- a/docs/features/kv_offloading_usage.md +++ b/docs/features/kv_offloading_usage.md @@ -81,6 +81,8 @@ vllm serve \ Each entry in `secondary_tiers` is a dict with a required `type` field plus tier-specific fields. +The filesystem and object-store tiers can publish hash-only `BlockStored` KV events for blocks they successfully store, tagged with a stable per-tier `medium` (`FS` for the filesystem tier, `OBJ` for the object-store tier). Set `enable_kv_events: true` in the tier's entry to opt in; events are published only when KV cache events are also enabled globally via `--kv-events-config`. + ### Filesystem (FS) The filesystem tier (`type: "fs"`) writes blocks to a directory on local storage. @@ -91,6 +93,7 @@ The filesystem tier (`type: "fs"`) writes blocks to a directory on local storage | `root_dir` | yes | — | Base directory; vLLM creates subdirectories beneath it (see [On-Disk Layout](#on-disk-layout)). | | `n_read_threads` | no | `16` | Read-priority I/O threads (load path). | | `n_write_threads` | no | `16` | Write-priority I/O threads (store path). | +| `enable_kv_events` | no | `false` | Publish `BlockStored` KV events (medium `FS`) for successfully stored blocks. Requires KV cache events to be enabled globally. | Each thread group prefers its own queue but pulls from the other when its primary queue is empty, so a write-heavy or read-heavy burst won't leave the off-priority queue waiting. Size the totals to your storage's effective concurrency. @@ -120,6 +123,31 @@ To enable KV cache sharing between multiple vLLM instances using the same `root_ PYTHONHASHSEED=0 vllm serve ... ``` +### Object Store (OBJ) + +The object-store tier (`type: "obj"`) offloads blocks to an S3-compatible object store through the NIXL OBJ backend. + +| Key | Required | Default | Notes | +| --- | --- | --- | --- | +| `type` | yes | — | Must be `obj`. | +| `store_config` | yes | — | Object store connection parameters (see below). | +| `prefix` | no | `""` | Key prefix prepended to all object keys. | +| `io_threads` | no | `4` | Number of NIXL OBJ backend I/O threads. | +| `enable_kv_events` | no | `false` | Publish `BlockStored` KV events (medium `OBJ`) for successfully stored blocks. Requires KV cache events to be enabled globally. | + +`store_config` fields: + +| Key | Required | Default | Notes | +| --- | --- | --- | --- | +| `bucket` | yes | — | Bucket name. | +| `endpoint_override` | yes | — | Object store endpoint host; the URL scheme is set separately via `scheme`. | +| `scheme` | no | `http` | `http` or `https`. | +| `access_key`, `secret_key`, `session_token` | no | `""` | Explicit credentials. When left empty, the NIXL OBJ plugin falls back to the AWS SDK default credential provider chain (IAM roles, environment variables, credential files), which enables workload-identity auth on Kubernetes. | +| `region` | no | `""` | Bucket region, if the endpoint requires one. | +| `ca_bundle` | no | `""` | CA bundle path for TLS verification. | + +Object keys follow the same run-configuration digest scheme as the filesystem tier (see [On-Disk Layout](#on-disk-layout)) and are stored under the optional `prefix`. The [Cross-Process Sharing](#cross-process-sharing) requirement (`PYTHONHASHSEED`) applies to shared buckets as well, so instances sharing a bucket produce identical keys for identical content. At startup the tier probes object store connectivity and fails fast with a configuration error if the bucket is unreachable. + ### P2P (Including P/D) The P2P tier (`type: "p2p"`) shares completed KV blocks between vLLM instances over RDMA via NIXL. Each instance binds a control socket on `host:port` and exchanges blocks directly with peers — no shared filesystem required. diff --git a/tests/v1/kv_offload/tiering/test_fs_tier.py b/tests/v1/kv_offload/tiering/test_fs_tier.py index 680a958478..c90a4dad5e 100644 --- a/tests/v1/kv_offload/tiering/test_fs_tier.py +++ b/tests/v1/kv_offload/tiering/test_fs_tier.py @@ -18,8 +18,10 @@ import numpy as np import pytest import torch +from vllm.distributed.kv_events import MEDIUM_FS from vllm.v1.kv_offload.base import ( LookupResult, + OffloadingEvent, OffloadKey, ReqContext, ScheduleEndContext, @@ -58,6 +60,16 @@ _MOCK_OFFLOADING_SPEC.kv_cache_config = _MOCK_KV_CACHE_CONFIG _MOCK_OFFLOADING_SPEC.block_size_factor = 1 +def _make_offloading_spec(enable_kv_cache_events: bool) -> MagicMock: + """Mock spec with an explicit global KV events flag.""" + spec = MagicMock() + spec.vllm_config = _MOCK_VLLM_CONFIG + spec.kv_cache_config = _MOCK_KV_CACHE_CONFIG + spec.block_size_factor = 1 + spec.kv_events_config.enable_kv_cache_events = enable_kv_cache_events + return spec + + def key(n: int) -> OffloadKey: return make_offload_key(n.to_bytes(8, "big"), 0) @@ -164,6 +176,23 @@ def fs_tier(tmp_path): tier.shutdown() +@pytest.fixture +def fs_tier_with_events(tmp_path): + tensor = _page_aligned_zero_tensor(4, _BLOCK_ELEMENTS) + mock_view = memoryview(tensor.numpy()) + tier = FileSystemTierManager( + offloading_spec=_make_offloading_spec(enable_kv_cache_events=True), + primary_kv_view=mock_view, + tier_type="fs", + root_dir=str(tmp_path), + n_read_threads=4, + n_write_threads=4, + enable_kv_events=True, + ) + yield tier + tier.shutdown() + + # --------------------------------------------------------------------------- # Tests # --------------------------------------------------------------------------- @@ -396,3 +425,170 @@ def test_batch_lookup_dispatch(fs_tier, monkeypatch, use_c_ext): results = lookup_and_wait(tier, [key(1), key(2)]) assert results == [LookupResult.HIT, LookupResult.MISS] + + +# --------------------------------------------------------------------------- +# KV events +# --------------------------------------------------------------------------- + + +def test_successful_store_emits_stored_event(fs_tier_with_events): + """A completed store job emits one stored event with the job's keys.""" + tier = fs_tier_with_events + keys = [key(1), key(2)] + tier.submit_store(make_job(1, keys, [0, 1])) + assert all(r.success for r in drain(tier)) + + events = list(tier.take_events()) + assert len(events) == 1 + assert events[0].keys == keys + # Literal medium pins the wire contract, not just the constant choice. + assert events[0].medium == "FS" + assert not events[0].removed + # take_events drains the buffer. + assert list(tier.take_events()) == [] + + +def test_load_job_emits_no_event(fs_tier_with_events): + tier = fs_tier_with_events + tier.submit_store(make_job(1, [key(1)], [0])) + results = drain(tier) + assert len(results) == 1 + assert results[0].success + list(tier.take_events()) + + tier.submit_load(make_job(2, [key(1)], [1], is_promotion=True)) + results = drain(tier) + assert len(results) == 1 + assert results[0].success + assert list(tier.take_events()) == [] + + +def test_mixed_job_results_emit_event_only_for_successful_job( + fs_tier_with_events, monkeypatch +): + """With a failed and a successful store job in flight, exactly one event + is emitted and its keys belong to the successful job.""" + import vllm.v1.kv_offload.tiering.fs.manager as mgr_mod + + tier = fs_tier_with_events + failing_path = tier.file_mapper.get_file_name(key(1)) + original_store_block = mgr_mod.store_block + + def flaky_store_block(dest_path, *args, **kwargs): + if dest_path == failing_path: + raise OSError("injected store failure") + return original_store_block(dest_path, *args, **kwargs) + + monkeypatch.setattr(mgr_mod, "store_block", flaky_store_block) + + tier.submit_store(make_job(1, [key(1)], [0])) + tier.submit_store(make_job(2, [key(2)], [1])) + results = drain(tier) + assert len(results) == 2 + by_id = {r.job_id: r for r in results} + assert not by_id[1].success + assert by_id[2].success + + events = list(tier.take_events()) + assert len(events) == 1 + assert events[0].keys == [key(2)] + + +def test_partially_failed_store_emits_no_event(fs_tier_with_events, monkeypatch): + """A store job with any failed block emits no event for the whole job.""" + import vllm.v1.kv_offload.tiering.fs.manager as mgr_mod + + tier = fs_tier_with_events + failing_path = tier.file_mapper.get_file_name(key(2)) + original_store_block = mgr_mod.store_block + + def flaky_store_block(dest_path, *args, **kwargs): + if dest_path == failing_path: + raise OSError("injected store failure") + return original_store_block(dest_path, *args, **kwargs) + + monkeypatch.setattr(mgr_mod, "store_block", flaky_store_block) + + tier.submit_store(make_job(1, [key(1), key(2)], [0, 1])) + results = drain(tier) + assert len(results) == 1 + assert not results[0].success + assert list(tier.take_events()) == [] + assert tier._store_job_keys == {} + + +def test_events_disabled_by_default(fs_tier): + tier, _ = fs_tier + tier.submit_store(make_job(1, [key(1)], [0])) + results = drain(tier) + assert len(results) == 1 + assert results[0].success + assert tier.events is None + assert tier._store_job_keys == {} + assert list(tier.take_events()) == [] + + +def test_events_require_global_kv_events_flag(tmp_path): + """Tier-level opt-in alone is not enough; the global flag gates events.""" + tensor = _page_aligned_zero_tensor(4, _BLOCK_ELEMENTS) + tier = FileSystemTierManager( + offloading_spec=_make_offloading_spec(enable_kv_cache_events=False), + primary_kv_view=memoryview(tensor.numpy()), + tier_type="fs", + root_dir=str(tmp_path), + enable_kv_events=True, + ) + try: + assert tier.events is None + tier.submit_store(make_job(1, [key(1)], [0])) + results = drain(tier) + assert len(results) == 1 + assert results[0].success + assert list(tier.take_events()) == [] + assert tier._store_job_keys == {} + finally: + tier.shutdown() + + +def test_cascade_store_emits_fs_event_through_tiering_manager(tmp_path): + """A GPU->CPU->fs cascade surfaces the tier-owned FS stored event via the + TieringOffloadingManager's aggregated take_events().""" + from vllm.v1.kv_offload.tiering.manager import ( + CPUPrimaryTierOffloadingManager, + TieringOffloadingManager, + ) + + tensor = _page_aligned_zero_tensor(4, _BLOCK_ELEMENTS) + view = memoryview(tensor.numpy()) + mock_region = MagicMock() + mock_region.create_kv_memoryview.return_value = view + primary = CPUPrimaryTierOffloadingManager(num_blocks=4, mmap_region=mock_region) + tier = FileSystemTierManager( + offloading_spec=_make_offloading_spec(enable_kv_cache_events=True), + primary_kv_view=primary.get_kv_memoryview(), + tier_type="fs", + root_dir=str(tmp_path), + enable_kv_events=True, + ) + manager = TieringOffloadingManager(primary_tier=primary, secondary_tiers=[tier]) + try: + keys = [key(1), key(2)] + manager.on_new_request(_CTX) + assert manager.prepare_store(keys, _CTX) is not None + manager.complete_store(keys, _CTX) # cascades to the fs tier + + events: list[OffloadingEvent] = [] + ctx = ScheduleEndContext(new_req_ids=[], preempted_req_ids=()) + deadline = time.monotonic() + 5.0 + while time.monotonic() < deadline and not events: + manager.on_schedule_end(ctx) + events.extend(manager.take_events()) + time.sleep(0.01) + + fs_events = [e for e in events if e.medium == MEDIUM_FS] + assert len(fs_events) == 1 + assert set(fs_events[0].keys) == set(keys) + assert not fs_events[0].removed + finally: + tier.shutdown() diff --git a/tests/v1/kv_offload/tiering/test_obj_tier.py b/tests/v1/kv_offload/tiering/test_obj_tier.py index f429d6cf62..37687adce0 100644 --- a/tests/v1/kv_offload/tiering/test_obj_tier.py +++ b/tests/v1/kv_offload/tiering/test_obj_tier.py @@ -179,8 +179,19 @@ class MockNixlAgent: # --------------------------------------------------------------------------- +def _make_events_spec(enable_kv_cache_events: bool) -> SimpleNamespace: + """Offloading spec stub with an explicit global KV events flag.""" + return SimpleNamespace( + vllm_config=_make_vllm_config(), + kv_cache_config=SimpleNamespace(kv_cache_groups=[]), + kv_events_config=SimpleNamespace(enable_kv_cache_events=enable_kv_cache_events), + ) + + def _make_tier( num_blocks: int = 4, + offloading_spec: SimpleNamespace = _OFFLOADING_SPEC, + **tier_kwargs, ) -> tuple[ObjectStoreSecondaryTierManager, MockNixlAgent]: """Create a tier backed by a fresh MockNixlAgent.""" mock_agent = MockNixlAgent() @@ -194,11 +205,12 @@ def _make_tier( ), ): tier = ObjectStoreSecondaryTierManager( - offloading_spec=_OFFLOADING_SPEC, + offloading_spec=offloading_spec, primary_kv_view=view, tier_type="obj", store_config=_STORE_CONFIG, prefix=_RUN_PREFIX, + **tier_kwargs, ) return tier, mock_agent @@ -421,6 +433,100 @@ class TestMockObjTierShutdown: tier.shutdown() # must not raise +class TestObjTierKVEvents: + def setup_method(self): + self.tier, self.agent = _make_tier( + offloading_spec=_make_events_spec(enable_kv_cache_events=True), + enable_kv_events=True, + ) + + def test_successful_store_emits_stored_event(self): + """A completed store transfer emits one stored event with the job's keys.""" + keys = [key(1), key(2)] + self.tier.submit_store(make_job(1, keys, [0, 1])) + assert all(r.success for r in drain(self.tier)) + + events = list(self.tier.take_events()) + assert len(events) == 1 + assert events[0].keys == keys + # Literal medium pins the wire contract, not just the constant choice. + assert events[0].medium == "OBJ" + assert not events[0].removed + # take_events drains the buffer. + assert list(self.tier.take_events()) == [] + + def test_mixed_job_results_emit_event_only_for_successful_job(self): + """With a failed and a successful store job resolving in the same + poll, exactly one event is emitted and its keys belong to the + successful job.""" + original = self.agent.check_xfer_state + self.agent.check_xfer_state = lambda h: "ERR" if h._id == 0 else original(h) + self.tier.submit_store(make_job(1, [key(1)], [0])) # handle 0: fails + self.tier.submit_store(make_job(2, [key(2)], [1])) # handle 1: succeeds + results = drain(self.tier) + by_id = {r.job_id: r for r in results} + assert not by_id[1].success + assert by_id[2].success + + events = list(self.tier.take_events()) + assert len(events) == 1 + assert events[0].keys == [key(2)] + assert self.tier._store_job_keys == {} + + def test_load_job_emits_no_event(self): + self.tier.submit_store(make_job(1, [key(1)], [0])) + results = drain(self.tier) + assert len(results) == 1 + assert results[0].success + list(self.tier.take_events()) + + self.tier.submit_load(make_job(2, [key(1)], [0])) + results = drain(self.tier) + assert len(results) == 1 + assert results[0].success + assert list(self.tier.take_events()) == [] + + def test_failed_transfer_emits_no_event(self): + self.agent.check_xfer_state = lambda h: "ERR" + self.tier.submit_store(make_job(1, [key(1)], [0])) + results = drain(self.tier) + assert not results[0].success + assert list(self.tier.take_events()) == [] + assert self.tier._store_job_keys == {} + + def test_submission_failure_emits_no_event(self): + self.agent.make_prepped_xfer = lambda *a, **k: None + self.tier.submit_store(make_job(1, [key(1)], [0])) + results = list(self.tier.get_finished_jobs()) + assert not results[0].success + assert list(self.tier.take_events()) == [] + assert self.tier._store_job_keys == {} + + def test_events_disabled_by_default(self): + tier, _ = _make_tier() + tier.submit_store(make_job(1, [key(1)], [0])) + results = drain(tier) + assert len(results) == 1 + assert results[0].success + assert tier.events is None + assert tier._store_job_keys == {} + assert list(tier.take_events()) == [] + + def test_events_require_global_kv_events_flag(self): + """Tier-level opt-in alone is not enough; the global flag gates events.""" + tier, _ = _make_tier( + offloading_spec=_make_events_spec(enable_kv_cache_events=False), + enable_kv_events=True, + ) + tier.submit_store(make_job(1, [key(1)], [0])) + results = drain(tier) + assert len(results) == 1 + assert results[0].success + assert tier.events is None + assert tier._store_job_keys == {} + assert list(tier.take_events()) == [] + + class TestObjStoreConfig: def test_explicit_credentials_included(self): cfg = ObjStoreConfig( diff --git a/vllm/distributed/kv_events.py b/vllm/distributed/kv_events.py index a7d83bb378..c2faf34095 100644 --- a/vllm/distributed/kv_events.py +++ b/vllm/distributed/kv_events.py @@ -44,6 +44,8 @@ class KVCacheEvent( MEDIUM_GPU = "GPU" MEDIUM_CPU = "CPU" +MEDIUM_FS = "FS" +MEDIUM_OBJ = "OBJ" class BlockStored(KVCacheEvent): diff --git a/vllm/v1/kv_offload/tiering/fs/manager.py b/vllm/v1/kv_offload/tiering/fs/manager.py index a38ac8eb40..d8d1700285 100644 --- a/vllm/v1/kv_offload/tiering/fs/manager.py +++ b/vllm/v1/kv_offload/tiering/fs/manager.py @@ -19,7 +19,7 @@ import functools import json import os from collections.abc import Iterable -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, ClassVar try: from vllm.fs_io_C import batch_lookup as batch_lookup_C @@ -30,11 +30,18 @@ except ImportError: from typing_extensions import override +from vllm.distributed.kv_events import MEDIUM_FS from vllm.logger import init_logger -from vllm.v1.kv_offload.base import LookupResult, OffloadKey, ReqContext +from vllm.v1.kv_offload.base import ( + LookupResult, + OffloadingEvent, + OffloadKey, + ReqContext, +) from vllm.v1.kv_offload.file_mapper import FileMapper from vllm.v1.kv_offload.tiering.async_lookup import AsyncLookupManager from vllm.v1.kv_offload.tiering.base import ( + JobId, JobMetadata, JobResult, RequestOffloadingContext, @@ -92,6 +99,8 @@ class FileSystemTierManager(SecondaryTierManager): content. """ + medium: ClassVar[str] = MEDIUM_FS + def __init__( self, offloading_spec: "OffloadingSpec", @@ -100,6 +109,7 @@ class FileSystemTierManager(SecondaryTierManager): root_dir: str, n_read_threads: int = 16, n_write_threads: int = 16, + enable_kv_events: bool = False, ): """ Args: @@ -110,9 +120,26 @@ class FileSystemTierManager(SecondaryTierManager): root_dir: Root directory for block files. n_read_threads: Number of read-priority I/O threads. n_write_threads: Number of write-priority I/O threads. + enable_kv_events: Emit BlockStored KV events for blocks + successfully stored to this tier. Effective only when KV + cache events are enabled globally (kv_events_config). """ super().__init__(offloading_spec, primary_kv_view, tier_type) + self.events: list[OffloadingEvent] | None = None + if enable_kv_events: + if offloading_spec.kv_events_config.enable_kv_cache_events: + self.events = [] + else: + logger.warning( + "enable_kv_events is set on secondary tier '%s' but KV " + "cache events are disabled globally; the tier will not " + "emit events.", + tier_type, + ) + # Keys of in-flight store jobs, tracked only when events are enabled. + self._store_job_keys: dict[JobId, list[OffloadKey]] = {} + # Extract block size from primary view assert primary_kv_view.strides is not None, ( "primary_kv_view.strides cannot be None" @@ -157,6 +184,8 @@ class FileSystemTierManager(SecondaryTierManager): @override def submit_store(self, job_metadata: JobMetadata) -> None: + if self.events is not None: + self._store_job_keys[job_metadata.job_id] = list(job_metadata.keys) tasks = ( functools.partial( store_block, @@ -188,10 +217,22 @@ class FileSystemTierManager(SecondaryTierManager): """ Collect completed jobs from the finished-jobs queue. """ - return ( - JobResult(job_id=job_id, success=success) - for job_id, success in self._pool.get_finished() - ) + results = [] + for job_id, success in self._pool.get_finished(): + if self.events is not None: + keys = self._store_job_keys.pop(job_id, None) + if success and keys: + self.events.append( + OffloadingEvent(keys=keys, medium=self.medium, removed=False) + ) + results.append(JobResult(job_id=job_id, success=success)) + return results + + @override + def take_events(self) -> Iterable[OffloadingEvent]: + if self.events is not None: + yield from self.events + self.events.clear() @override def drain_jobs(self) -> None: diff --git a/vllm/v1/kv_offload/tiering/obj/manager.py b/vllm/v1/kv_offload/tiering/obj/manager.py index 6060370ea0..954f86ed16 100644 --- a/vllm/v1/kv_offload/tiering/obj/manager.py +++ b/vllm/v1/kv_offload/tiering/obj/manager.py @@ -5,15 +5,22 @@ import ctypes import time from collections.abc import Iterable -from typing import TYPE_CHECKING, NamedTuple +from typing import TYPE_CHECKING, ClassVar, NamedTuple +from vllm.distributed.kv_events import MEDIUM_OBJ from vllm.distributed.nixl_utils import NixlWrapper as nixl_agent from vllm.distributed.nixl_utils import nixl_agent_config from vllm.logger import init_logger -from vllm.v1.kv_offload.base import LookupResult, OffloadKey, ReqContext +from vllm.v1.kv_offload.base import ( + LookupResult, + OffloadingEvent, + OffloadKey, + ReqContext, +) from vllm.v1.kv_offload.file_mapper import FileMapper from vllm.v1.kv_offload.tiering.async_lookup import AsyncLookupManager from vllm.v1.kv_offload.tiering.base import ( + JobId, JobMetadata, JobResult, RequestOffloadingContext, @@ -90,6 +97,8 @@ class ObjectStoreSecondaryTierManager(SecondaryTierManager): primary tier. Object keys are formed as ``{prefix}/{hash_shard}/{hash}.bin``. """ + medium: ClassVar[str] = MEDIUM_OBJ + def __init__( self, offloading_spec: "OffloadingSpec", @@ -98,8 +107,36 @@ class ObjectStoreSecondaryTierManager(SecondaryTierManager): store_config: dict, prefix: str = "", io_threads: int = 4, + enable_kv_events: bool = False, ): + """ + Args: + offloading_spec: Offloading configuration. + primary_kv_view: Memoryview of the primary tier's CPU KV cache. + tier_type: Tier type identifier, set by SecondaryTierFactory. + store_config: Object store connection parameters (see ObjStoreConfig). + prefix: Key prefix prepended to all object keys. + io_threads: Number of NIXL I/O threads. + enable_kv_events: Emit BlockStored KV events for blocks + successfully stored to this tier. Effective only when KV + cache events are enabled globally (kv_events_config). + """ super().__init__(offloading_spec, primary_kv_view, tier_type) + + self.events: list[OffloadingEvent] | None = None + if enable_kv_events: + if offloading_spec.kv_events_config.enable_kv_cache_events: + self.events = [] + else: + logger.warning( + "enable_kv_events is set on secondary tier '%s' but KV " + "cache events are disabled globally; the tier will not " + "emit events.", + tier_type, + ) + # Keys of in-flight store jobs, tracked only when events are enabled. + self._store_job_keys: dict[JobId, list[OffloadKey]] = {} + agent_config = nixl_agent_config(backends=[]) self._agent = nixl_agent("ObjAgent", agent_config) obj_config = ObjStoreConfig(**store_config) @@ -231,6 +268,8 @@ class ObjectStoreSecondaryTierManager(SecondaryTierManager): return LookupResult.HIT if result else LookupResult.MISS def submit_store(self, job_metadata: JobMetadata) -> None: + if self.events is not None: + self._store_job_keys[job_metadata.job_id] = list(job_metadata.keys) obj_keys = (self._file_mapper.get_file_name(k) for k in job_metadata.keys) self._submit_transfer( job_metadata.job_id, job_metadata.block_ids, obj_keys, NIXL_WRITE @@ -279,8 +318,20 @@ class ObjectStoreSecondaryTierManager(SecondaryTierManager): self._poll_active_transfers() results = self._pending_results self._pending_results = [] + if self.events is not None: + for result in results: + keys = self._store_job_keys.pop(result.job_id, None) + if result.success and keys: + self.events.append( + OffloadingEvent(keys=keys, medium=self.medium, removed=False) + ) return results + def take_events(self) -> Iterable[OffloadingEvent]: + if self.events is not None: + yield from self.events + self.events.clear() + def drain_jobs(self) -> None: """Block until every submitted transfer has completed or failed.