| 1 | # SPDX-License-Identifier: Apache-2.0 |
| 2 | # SPDX-FileCopyrightText: Copyright contributors to the vLLM project |
| 3 | import filecmp |
| 4 | import shutil |
| 5 | import tempfile |
| 6 | from pathlib import Path |
| 7 | from typing import Any |
| 8 | from unittest.mock import MagicMock |
| 9 | |
| 10 | import pytest |
| 11 | import torch |
| 12 | |
| 13 | from tests.v1.kv_connector.unit.utils import create_vllm_config |
| 14 | from vllm import LLM, SamplingParams |
| 15 | from vllm.config import KVTransferConfig |
| 16 | from vllm.distributed.kv_transfer.kv_connector.factory import KVConnectorFactory |
| 17 | from vllm.distributed.kv_transfer.kv_connector.v1 import KVConnectorRole |
| 18 | from vllm.distributed.kv_transfer.kv_connector.v1.base import ( |
| 19 | KVConnectorBase_V1, |
| 20 | SupportsHMA, |
| 21 | supports_hma, |
| 22 | ) |
| 23 | from vllm.distributed.kv_transfer.kv_connector.v1.metrics import ( |
| 24 | KVConnectorLogging, |
| 25 | KVConnectorStats, |
| 26 | ) |
| 27 | from vllm.distributed.kv_transfer.kv_connector.v1.multi_connector import ( |
| 28 | MultiConnector, |
| 29 | MultiKVConnectorPromMetrics, |
| 30 | MultiKVConnectorStats, |
| 31 | MultiKVConnectorWorkerMetadata, |
| 32 | ) |
| 33 | from vllm.distributed.kv_transfer.kv_connector.v1.nixl import ( |
| 34 | NixlKVConnectorStats, |
| 35 | ) |
| 36 | from vllm.v1.kv_cache_interface import KVCacheConfig |
| 37 | from vllm.v1.outputs import KVConnectorOutput, KVConnectorWorkerMetadata |
| 38 | |
| 39 | MODEL_NAME = "meta-llama/Llama-3.2-1B-Instruct" |
| 40 | |
| 41 | PROMPT_CONTEXT = "Hi " * 100 |
| 42 | PROMPTS = [ |
| 43 | PROMPT_CONTEXT + "Hello, my name is", |
| 44 | PROMPT_CONTEXT + "The capital of France is", |
| 45 | ] |
| 46 | |
| 47 | SAMPLING_PARAMS = SamplingParams(temperature=0, max_tokens=20) |
| 48 | |
| 49 | |
| 50 | # Test connector with custom stats for testing MultiConnector |
| 51 | class MockConnectorStats(KVConnectorStats): |
| 52 | """Mock stats class for testing.""" |
| 53 | |
| 54 | pass |
| 55 | |
| 56 | |
| 57 | class MockConnector(KVConnectorBase_V1): |
| 58 | """Mock connector for testing.""" |
| 59 | |
| 60 | _supports_divergent_local_hybrid_hits = False |
| 61 | |
| 62 | def __new__(cls, *args, **kwargs): |
| 63 | # mock all KVConnectorBase_V1 functions |
| 64 | mock = MagicMock(spec_set=KVConnectorBase_V1) |
| 65 | mock.supports_divergent_local_hybrid_hits = ( |
| 66 | cls._supports_divergent_local_hybrid_hits |
| 67 | ) |
| 68 | # Override just build_kv_connector_stats |
| 69 | mock.build_kv_connector_stats = cls.build_kv_connector_stats |
| 70 | mock.get_kv_connector_stats.return_value = None |
| 71 | mock.get_mem_pool_context.return_value = None |
| 72 | return mock |
| 73 | |
| 74 | @classmethod |
| 75 | def build_kv_connector_stats( |
| 76 | cls, data: dict[str, Any] | None = None |
| 77 | ) -> KVConnectorStats | None: |
| 78 | return MockConnectorStats(data=data) if data is not None else None |
| 79 | |
| 80 | def start_load_kv(self, forward_context, **kwargs): |
| 81 | pass |
| 82 | |
| 83 | def wait_for_layer_load(self, layer_name): |
| 84 | pass |
| 85 | |
| 86 | def save_kv_layer(self, layer_name, kv_layer, attn_metadata, **kwargs): |
| 87 | pass |
| 88 | |
| 89 | def wait_for_save(self): |
| 90 | pass |
| 91 | |
| 92 | def build_connector_meta(self, scheduler_output): |
| 93 | return None |
| 94 | |
| 95 | def get_num_new_matched_tokens(self, request, num_computed_tokens): |
| 96 | return (0, False) |
| 97 | |
| 98 | def update_state_after_alloc(self, request, blocks, num_tokens) -> None: |
| 99 | pass |
| 100 | |
| 101 | |
| 102 | class MockHMAConnector(KVConnectorBase_V1, SupportsHMA): |
| 103 | """Mock connector that supports HMA for testing.""" |
| 104 | |
| 105 | _supports_divergent_local_hybrid_hits = False |
| 106 | |
| 107 | def __new__(cls, *args, **kwargs): |
| 108 | mock = MagicMock(spec_set=cls) |
| 109 | mock.supports_divergent_local_hybrid_hits = ( |
| 110 | cls._supports_divergent_local_hybrid_hits |
| 111 | ) |
| 112 | mock.get_kv_connector_stats.return_value = None |
| 113 | return mock |
| 114 | |
| 115 | def start_load_kv(self, forward_context, **kwargs): |
| 116 | pass |
| 117 | |
| 118 | def wait_for_layer_load(self, layer_name): |
| 119 | pass |
| 120 | |
| 121 | def save_kv_layer(self, layer_name, kv_layer, attn_metadata, **kwargs): |
| 122 | pass |
| 123 | |
| 124 | def wait_for_save(self): |
| 125 | pass |
| 126 | |
| 127 | def build_connector_meta(self, scheduler_output): |
| 128 | return None |
| 129 | |
| 130 | def get_num_new_matched_tokens(self, request, num_computed_tokens): |
| 131 | return (0, False) |
| 132 | |
| 133 | def update_state_after_alloc(self, request, blocks, num_tokens) -> None: |
| 134 | pass |
| 135 | |
| 136 | def request_finished_all_groups(self, request, block_ids): |
| 137 | return (False, None) |
| 138 | |
| 139 | |
| 140 | class MockDivergentHMAConnector(MockHMAConnector): |
| 141 | _supports_divergent_local_hybrid_hits = True |
| 142 | |
| 143 | |
| 144 | # Register mock connectors |
| 145 | KVConnectorFactory.register_connector("MockConnector", __name__, MockConnector.__name__) |
| 146 | KVConnectorFactory.register_connector( |
| 147 | "MockHMAConnector", __name__, MockHMAConnector.__name__ |
| 148 | ) |
| 149 | KVConnectorFactory.register_connector( |
| 150 | "MockDivergentHMAConnector", |
| 151 | __name__, |
| 152 | MockDivergentHMAConnector.__name__, |
| 153 | ) |
| 154 | |
| 155 | |
| 156 | def test_register_finished_partial_tail_notifies_every_connector(): |
| 157 | connector = object.__new__(MultiConnector) |
| 158 | first = MagicMock(spec_set=KVConnectorBase_V1) |
| 159 | second = MagicMock(spec_set=KVConnectorBase_V1) |
| 160 | first.register_finished_partial_tail.return_value = True |
| 161 | second.register_finished_partial_tail.return_value = False |
| 162 | connector._connectors = [first, second] |
| 163 | request = MagicMock() |
| 164 | block_ids = ([1], [2]) |
| 165 | offloads = [(1, 2, 12)] |
| 166 | |
| 167 | assert connector.register_finished_partial_tail(request, block_ids, offloads) |
| 168 | first.register_finished_partial_tail.assert_called_once_with( |
| 169 | request, block_ids, offloads |
| 170 | ) |
| 171 | second.register_finished_partial_tail.assert_called_once_with( |
| 172 | request, block_ids, offloads |
| 173 | ) |
| 174 | |
| 175 | |
| 176 | @pytest.fixture |
| 177 | def mc() -> MultiConnector: |
| 178 | """MultiConnector using two mocked connectors.""" |
| 179 | mock_connector_config = { |
| 180 | "kv_connector": "MockConnector", |
| 181 | "kv_role": "kv_both", |
| 182 | "kv_connector_module_path": "tests.v1.kv_connector.unit.test_multi_connector", |
| 183 | } |
| 184 | |
| 185 | vllm_config = create_vllm_config( |
| 186 | kv_connector="MultiConnector", |
| 187 | kv_connector_extra_config={ |
| 188 | "connectors": [mock_connector_config, mock_connector_config], |
| 189 | }, |
| 190 | ) |
| 191 | |
| 192 | kv_cache_config = KVCacheConfig( |
| 193 | num_blocks=0, kv_cache_tensors=[], kv_cache_groups=[] |
| 194 | ) |
| 195 | |
| 196 | mc = MultiConnector( |
| 197 | vllm_config=vllm_config, |
| 198 | role=KVConnectorRole.WORKER, |
| 199 | kv_cache_config=kv_cache_config, |
| 200 | ) |
| 201 | |
| 202 | return mc |
| 203 | |
| 204 | |
| 205 | def test_multi_connector_mem_pool_context_none(mc: MultiConnector): |
| 206 | assert mc.get_mem_pool_context() is None |
| 207 | |
| 208 | |
| 209 | def test_multi_connector_forwards_mem_pool_context(mc: MultiConnector): |
| 210 | context = MagicMock() |
| 211 | provider = MagicMock(spec_set=KVConnectorBase_V1) |
| 212 | provider.get_mem_pool_context.return_value = context |
| 213 | mc._connectors = [mc._connectors[0], provider] |
| 214 | |
| 215 | assert mc.get_mem_pool_context() is context |
| 216 | provider.get_mem_pool_context.assert_called_once_with() |
| 217 | |
| 218 | |
| 219 | def test_multi_connector_rejects_multiple_mem_pool_contexts(mc: MultiConnector): |
| 220 | providers = [ |
| 221 | MagicMock(spec_set=KVConnectorBase_V1), |
| 222 | MagicMock(spec_set=KVConnectorBase_V1), |
| 223 | ] |
| 224 | for provider in providers: |
| 225 | provider.get_mem_pool_context.return_value = MagicMock() |
| 226 | mc._connectors = providers |
| 227 | |
| 228 | with pytest.raises( |
| 229 | ValueError, |
| 230 | match="Multiple connectors provide a KV cache memory pool", |
| 231 | ): |
| 232 | mc.get_mem_pool_context() |
| 233 | |
| 234 | |
| 235 | # Helper function to compare directories recursively |
| 236 | def _compare_directories(dir1: Path, dir2: Path) -> bool: |
| 237 | """Compares two directories recursively for identical content.""" |
| 238 | dcmp = filecmp.dircmp(dir1, dir2) |
| 239 | if dcmp.left_only or dcmp.right_only or dcmp.diff_files: |
| 240 | print(f"Differences found between {dir1} and {dir2}:") |
| 241 | print(f" Left only: {dcmp.left_only}") |
| 242 | print(f" Right only: {dcmp.right_only}") |
| 243 | print(f" Different files: {dcmp.diff_files}") |
| 244 | return False |
| 245 | for sub_dir in dcmp.common_dirs: |
| 246 | if not _compare_directories(dir1 / sub_dir, dir2 / sub_dir): |
| 247 | return False |
| 248 | return True |
| 249 | |
| 250 | |
| 251 | def test_multi_example_connector_consistency(): |
| 252 | """Tests that MultiConnector with two ExampleConnectors saves |
| 253 | identical KV cache data to separate storage locations. |
| 254 | """ |
| 255 | storage_1_path = Path("storage_1/") |
| 256 | storage_2_path = Path("storage_2/") |
| 257 | shutil.rmtree(storage_1_path, ignore_errors=True) |
| 258 | shutil.rmtree(storage_2_path, ignore_errors=True) |
| 259 | storage_1_path.mkdir() |
| 260 | storage_2_path.mkdir() |
| 261 | |
| 262 | # Configure MultiConnector with two ExampleConnectors |
| 263 | kv_transfer_config = KVTransferConfig( |
| 264 | kv_connector="MultiConnector", |
| 265 | kv_role="kv_both", |
| 266 | kv_connector_extra_config={ |
| 267 | "connectors": [ |
| 268 | { |
| 269 | "kv_connector": "TestExampleConnector", |
| 270 | "kv_role": "kv_both", |
| 271 | "kv_connector_extra_config": { |
| 272 | "shared_storage_path": str(storage_1_path), |
| 273 | "name": "storage1", |
| 274 | }, |
| 275 | "kv_connector_module_path": "tests.v1.kv_connector.unit.utils", |
| 276 | }, |
| 277 | { |
| 278 | "kv_connector": "TestExampleConnector", |
| 279 | "kv_role": "kv_both", |
| 280 | "kv_connector_extra_config": { |
| 281 | "shared_storage_path": str(storage_2_path), |
| 282 | "name": "storage2", |
| 283 | }, |
| 284 | "kv_connector_module_path": "tests.v1.kv_connector.unit.utils", |
| 285 | }, |
| 286 | ] |
| 287 | }, |
| 288 | ) |
| 289 | |
| 290 | llm = LLM( |
| 291 | model=MODEL_NAME, |
| 292 | enforce_eager=True, |
| 293 | block_size=16, |
| 294 | gpu_memory_utilization=0.5, |
| 295 | kv_transfer_config=kv_transfer_config, |
| 296 | async_scheduling=False, |
| 297 | ) |
| 298 | # Run generation - this should trigger saving KV cache |
| 299 | # Use a single prompt to avoid race conditions depending on the order of scheduling |
| 300 | _ = llm.generate(PROMPTS[0], SAMPLING_PARAMS) |
| 301 | |
| 302 | # --- Verification --- |
| 303 | |
| 304 | # Check that both storage directories were populated |
| 305 | local_subdirs = list(storage_1_path.iterdir()) |
| 306 | external_subdirs = list(storage_2_path.iterdir()) |
| 307 | |
| 308 | assert len(local_subdirs) > 0, ( |
| 309 | f"Local storage path {storage_1_path} is empty after generation." |
| 310 | ) |
| 311 | assert len(external_subdirs) > 0, ( |
| 312 | f"External storage path {storage_2_path} is empty after generation." |
| 313 | ) |
| 314 | assert len(local_subdirs) == len(external_subdirs), ( |
| 315 | f"Mismatch in number of cache entries: " |
| 316 | f"Local={len(local_subdirs)}, External={len(external_subdirs)}" |
| 317 | ) |
| 318 | |
| 319 | # The subdirectories should correspond to the prompt hashes |
| 320 | # Since prompts are the same, the hash directories should be the same name |
| 321 | local_subdir_names = sorted([d.name for d in local_subdirs]) |
| 322 | external_subdir_names = sorted([d.name for d in external_subdirs]) |
| 323 | assert local_subdir_names == external_subdir_names, ( |
| 324 | "Cache directory names do not match between local and external storage" |
| 325 | ) |
| 326 | |
| 327 | # Compare the contents of each corresponding cache directory |
| 328 | for subdir_name in local_subdir_names: |
| 329 | print(f"Comparing contents of cache directory: {subdir_name}") |
| 330 | assert _compare_directories( |
| 331 | storage_1_path / subdir_name, storage_2_path / subdir_name |
| 332 | ), ( |
| 333 | f"Contents differ for cache directory '{subdir_name}' between " |
| 334 | f"{storage_1_path} and {storage_2_path}" |
| 335 | ) |
| 336 | |
| 337 | events = get_connector_events() |
| 338 | storage1_scheduler_events = _ignore_event_collection(events["storage1-SCHEDULER"]) |
| 339 | storage2_scheduler_events = _ignore_event_collection(events["storage2-SCHEDULER"]) |
| 340 | # Initial events bind the cache manager, query completion counts, and exchange |
| 341 | # handshake metadata before the request is enqueued. |
| 342 | assert storage1_scheduler_events[:7] == [ |
| 343 | "bind_kv_cache_manager", |
| 344 | "get_finished_count", |
| 345 | "set_xfer_handshake_metadata_pp_aware", |
| 346 | "on_new_request", |
| 347 | "get_num_new_matched_tokens 0", |
| 348 | "update_state_after_alloc num_blocks=[7] 0", |
| 349 | "build_connector_meta", |
| 350 | ] |
| 351 | # First three events are from initialization. Layer hooks run before the |
| 352 | # deferred load starts after the forward pass. |
| 353 | expected_worker_prefix = [ |
| 354 | "get_mem_pool_context", |
| 355 | "register_kv_caches", |
| 356 | "set_host_xfer_buffer_ops", |
| 357 | "get_handshake_metadata", |
| 358 | "handle_preemptions", |
| 359 | "bind_connector_metadata", |
| 360 | "wait_for_layer_load", |
| 361 | "save_kv_layer", |
| 362 | ] |
| 363 | for connector_name in ("storage1-WORKER", "storage2-WORKER"): |
| 364 | worker_events = events[connector_name] |
| 365 | assert worker_events[: len(expected_worker_prefix)] == expected_worker_prefix |
| 366 | assert worker_events.index("start_load_kv") > worker_events.index( |
| 367 | "save_kv_layer" |
| 368 | ) |
| 369 | assert storage2_scheduler_events[:7] == [ |
| 370 | "bind_kv_cache_manager", |
| 371 | "get_finished_count", |
| 372 | "set_xfer_handshake_metadata_pp_aware", |
| 373 | "on_new_request", |
| 374 | "get_num_new_matched_tokens 0", |
| 375 | "update_state_after_alloc num_blocks=[7] 0", |
| 376 | "build_connector_meta", |
| 377 | ] |
| 378 | # Reset prefix cache or else we'll just get the tokens back from there. |
| 379 | llm.reset_prefix_cache() |
| 380 | |
| 381 | # Run generation again - this should trigger loading from the first |
| 382 | # connector. |
| 383 | _ = llm.generate(PROMPTS[1], SAMPLING_PARAMS) |
| 384 | |
| 385 | events = get_connector_events() |
| 386 | # get_num_new_matched_tokens will return new tokens from the first |
| 387 | # connector (first nonzero match is chosen), so update_state_after_alloc |
| 388 | # will report those external tokens on that one. Other connectors still |
| 389 | # receive the request's real blocks but with 0 external tokens. |
| 390 | storage1_scheduler_events = _events_from_request( |
| 391 | _ignore_event_collection(events["storage1-SCHEDULER"]) |
| 392 | ) |
| 393 | storage2_scheduler_events = _events_from_request( |
| 394 | _ignore_event_collection(events["storage2-SCHEDULER"]) |
| 395 | ) |
| 396 | assert storage1_scheduler_events[:4] == [ |
| 397 | "on_new_request", |
| 398 | "get_num_new_matched_tokens 0", |
| 399 | "update_state_after_alloc num_blocks=[7] 96", |
| 400 | "build_connector_meta", |
| 401 | ] |
| 402 | assert storage2_scheduler_events[:4] == [ |
| 403 | "on_new_request", |
| 404 | "get_num_new_matched_tokens 0", |
| 405 | "update_state_after_alloc num_blocks=[7] 0", |
| 406 | "build_connector_meta", |
| 407 | ] |
| 408 | |
| 409 | # Delete storage1 connector state |
| 410 | shutil.rmtree(storage_1_path) |
| 411 | |
| 412 | # Reset prefix cache or else we'll just get the tokens back from there. |
| 413 | llm.reset_prefix_cache() |
| 414 | |
| 415 | # Run generation again - this should trigger loading from the first |
| 416 | # connector. |
| 417 | _ = llm.generate(PROMPTS[0], SAMPLING_PARAMS) |
| 418 | |
| 419 | events = get_connector_events() |
| 420 | # get_num_new_matched_tokens will be called for both connectors but will |
| 421 | # return 0 from the first connector, while the second connector has a hit. |
| 422 | # Both connectors receive the request's real blocks, but only the chosen |
| 423 | # (second) connector reports external tokens. |
| 424 | storage1_scheduler_events = _events_from_request( |
| 425 | _ignore_event_collection(events["storage1-SCHEDULER"]) |
| 426 | ) |
| 427 | storage2_scheduler_events = _events_from_request( |
| 428 | _ignore_event_collection(events["storage2-SCHEDULER"]) |
| 429 | ) |
| 430 | assert storage1_scheduler_events[:4] == [ |
| 431 | "on_new_request", |
| 432 | "get_num_new_matched_tokens 0", |
| 433 | "update_state_after_alloc num_blocks=[7] 0", |
| 434 | "build_connector_meta", |
| 435 | ] |
| 436 | assert storage2_scheduler_events[:4] == [ |
| 437 | "on_new_request", |
| 438 | "get_num_new_matched_tokens 0", |
| 439 | "update_state_after_alloc num_blocks=[7] 96", |
| 440 | "build_connector_meta", |
| 441 | ] |
| 442 | |
| 443 | # Clean up |
| 444 | shutil.rmtree(storage_1_path) |
| 445 | shutil.rmtree(storage_2_path) |
| 446 | |
| 447 | |
| 448 | def _ignore_event_collection(events: list[str]) -> list[str]: |
| 449 | # Filter out per-step polling hooks that the scheduler calls repeatedly |
| 450 | # and which are not meaningful state transitions for these assertions. |
| 451 | ignored = {"get_kv_connector_stats", "has_pending_push_work", "take_events"} |
| 452 | return [event for event in events if event not in ignored] |
| 453 | |
| 454 | |
| 455 | def _events_from_request(events: list[str]) -> list[str]: |
| 456 | # The async engine-core can emit a trailing build_connector_meta from the |
| 457 | # previous generation's final (idle) scheduler step. Depending on timing it |
| 458 | # may be flushed into this window ahead of on_new_request, so anchor the |
| 459 | # comparison on the new request's first event to avoid a flaky ordering. |
| 460 | if "on_new_request" in events: |
| 461 | return events[events.index("on_new_request") :] |
| 462 | return events |
| 463 | |
| 464 | |
| 465 | def get_connector_events() -> dict[str, list[str]]: |
| 466 | # Read in connector events and reset the files. |
| 467 | import glob |
| 468 | |
| 469 | event_files = glob.glob(tempfile.gettempdir() + "/connector_*_events.log") |
| 470 | connector_events = {} |
| 471 | for fname in event_files: |
| 472 | name = fname.split("connector_")[1].split("_events.log")[0] |
| 473 | try: |
| 474 | with open(fname, "r+") as f: |
| 475 | connector_events[name] = [line.strip() for line in f if line.strip()] |
| 476 | f.truncate(0) |
| 477 | except Exception as e: |
| 478 | print(f"[ERROR] Could not read connector events for {name}: {e}") |
| 479 | |
| 480 | return connector_events |
| 481 | |
| 482 | |
| 483 | def test_engine_id_conflict(): |
| 484 | configs = [KVTransferConfig() for _ in range(2)] |
| 485 | ids = [config.engine_id for config in configs] |
| 486 | assert ids[0] != ids[1], ( |
| 487 | f"Engine IDs should be different for different configs. Got {ids}" |
| 488 | ) |
| 489 | |
| 490 | |
| 491 | def test_multi_connector_handle_preemptions_integration(): |
| 492 | """Integration test: verify MultiConnector delegates handle_preemptions |
| 493 | to all sub-connectors. |
| 494 | |
| 495 | Uses TestExampleConnector which logs all method calls to temp files. |
| 496 | This test directly calls handle_preemptions on a MultiConnector with |
| 497 | TestExampleConnector sub-connectors and verifies the calls are logged. |
| 498 | """ |
| 499 | from tests.v1.kv_connector.unit.utils import ( |
| 500 | create_scheduler, |
| 501 | create_vllm_config, |
| 502 | ) |
| 503 | |
| 504 | storage_path = Path(tempfile.mkdtemp()) |
| 505 | |
| 506 | try: |
| 507 | # Configure MultiConnector with two TestExampleConnectors |
| 508 | connectors_extra_config = { |
| 509 | "connectors": [ |
| 510 | { |
| 511 | "kv_connector": "TestExampleConnector", |
| 512 | "kv_role": "kv_both", |
| 513 | "kv_connector_extra_config": { |
| 514 | "shared_storage_path": str(storage_path / "s1"), |
| 515 | "name": "preempt1", |
| 516 | }, |
| 517 | "kv_connector_module_path": "tests.v1.kv_connector.unit.utils", |
| 518 | }, |
| 519 | { |
| 520 | "kv_connector": "TestExampleConnector", |
| 521 | "kv_role": "kv_both", |
| 522 | "kv_connector_extra_config": { |
| 523 | "shared_storage_path": str(storage_path / "s2"), |
| 524 | "name": "preempt2", |
| 525 | }, |
| 526 | "kv_connector_module_path": "tests.v1.kv_connector.unit.utils", |
| 527 | }, |
| 528 | ] |
| 529 | } |
| 530 | |
| 531 | vllm_config = create_vllm_config( |
| 532 | block_size=16, |
| 533 | max_num_batched_tokens=100, |
| 534 | kv_connector="MultiConnector", |
| 535 | kv_connector_extra_config=connectors_extra_config, |
| 536 | ) |
| 537 | |
| 538 | # Create scheduler - this initializes the MultiConnector with SCHEDULER role |
| 539 | scheduler = create_scheduler(vllm_config, num_blocks=10) |
| 540 | |
| 541 | # Clear any events from initialization |
| 542 | get_connector_events() |
| 543 | |
| 544 | # Directly call handle_preemptions on the scheduler's connector |
| 545 | # Note: handle_preemptions is normally a worker-side method, but we're |
| 546 | # testing the delegation behavior of MultiConnector here. |
| 547 | # The connector attribute contains the KV connector. |
| 548 | assert scheduler.connector is not None, "Scheduler should have a connector" |
| 549 | connector_md = scheduler.connector.build_connector_meta(scheduler.schedule()) |
| 550 | scheduler.connector.handle_preemptions(connector_md) |
| 551 | |
| 552 | # Verify both connectors received the handle_preemptions call |
| 553 | events = get_connector_events() |
| 554 | |
| 555 | # Both SCHEDULER-role connectors should have logged handle_preemptions |
| 556 | assert "handle_preemptions" in events.get("preempt1-SCHEDULER", []), ( |
| 557 | f"preempt1-SCHEDULER should have handle_preemptions call. " |
| 558 | f"Got events: {events}" |
| 559 | ) |
| 560 | assert "handle_preemptions" in events.get("preempt2-SCHEDULER", []), ( |
| 561 | f"preempt2-SCHEDULER should have handle_preemptions call. " |
| 562 | f"Got events: {events}" |
| 563 | ) |
| 564 | |
| 565 | finally: |
| 566 | # Cleanup |
| 567 | shutil.rmtree(storage_path, ignore_errors=True) |
| 568 | |
| 569 | |
| 570 | class TestMultiConnectorStats: |
| 571 | """Tests for MultiConnector stats reconstruction and operations.""" |
| 572 | |
| 573 | def test_build_kv_connector_stats_with_none(self): |
| 574 | """Test that build_kv_connector_stats returns empty stats when given None.""" |
| 575 | stats = MultiConnector.build_kv_connector_stats(data=None) |
| 576 | |
| 577 | assert stats is not None |
| 578 | assert isinstance(stats, MultiKVConnectorStats) |
| 579 | assert len(stats.data) == 0 |
| 580 | assert stats.is_empty() |
| 581 | |
| 582 | def test_build_kv_connector_stats_with_empty_dict(self): |
| 583 | """Test that build_kv_connector_stats returns empty stats with empty dict.""" |
| 584 | stats = MultiConnector.build_kv_connector_stats(data={}) |
| 585 | |
| 586 | assert stats is not None |
| 587 | assert isinstance(stats, MultiKVConnectorStats) |
| 588 | assert len(stats.data) == 0 |
| 589 | assert stats.is_empty() |
| 590 | |
| 591 | def test_build_kv_connector_stats_reconstructs_nixl_stats(self): |
| 592 | """Test that NixlConnector stats are properly reconstructed with |
| 593 | correct data.""" |
| 594 | serialized_data = { |
| 595 | "NixlConnector": { |
| 596 | "transfer_duration": [1.5, 2.3], |
| 597 | "post_duration": [0.1, 0.2], |
| 598 | "bytes_transferred": [1024, 2048], |
| 599 | "num_descriptors": [10, 20], |
| 600 | "num_failed_transfers": [], |
| 601 | "num_failed_notifications": [], |
| 602 | "num_failed_handshakes": [], |
| 603 | "num_kv_expired_reqs": [], |
| 604 | "num_notifications_after_expiry": [], |
| 605 | } |
| 606 | } |
| 607 | |
| 608 | stats = MultiConnector.build_kv_connector_stats(data=serialized_data) |
| 609 | |
| 610 | assert "NixlConnector" in stats.data |
| 611 | nixl_stats = stats.data["NixlConnector"] |
| 612 | assert isinstance(nixl_stats, NixlKVConnectorStats) |
| 613 | assert nixl_stats.data["transfer_duration"] == [1.5, 2.3] |
| 614 | assert nixl_stats.data["post_duration"] == [0.1, 0.2] |
| 615 | assert nixl_stats.data["bytes_transferred"] == [1024, 2048] |
| 616 | assert nixl_stats.data["num_descriptors"] == [10, 20] |
| 617 | |
| 618 | def test_build_kv_connector_stats_with_multiple_connectors(self): |
| 619 | """Test reconstruction with multiple connector types that have custom stats.""" |
| 620 | serialized_data = { |
| 621 | "NixlConnector": { |
| 622 | "transfer_duration": [1.5], |
| 623 | "post_duration": [0.1], |
| 624 | "bytes_transferred": [1024], |
| 625 | "num_descriptors": [10], |
| 626 | "num_failed_transfers": [], |
| 627 | "num_failed_notifications": [], |
| 628 | "num_failed_handshakes": [], |
| 629 | "num_kv_expired_reqs": [], |
| 630 | "num_notifications_after_expiry": [], |
| 631 | }, |
| 632 | "MockConnector": {"mock_field": [1, 2, 3]}, |
| 633 | } |
| 634 | |
| 635 | stats = MultiConnector.build_kv_connector_stats(data=serialized_data) |
| 636 | |
| 637 | assert stats is not None |
| 638 | assert isinstance(stats, MultiKVConnectorStats) |
| 639 | # Both connectors should be reconstructed |
| 640 | assert len(stats.data) == 2 |
| 641 | assert "NixlConnector" in stats.data |
| 642 | assert "MockConnector" in stats.data |
| 643 | assert isinstance(stats.data["NixlConnector"], NixlKVConnectorStats) |
| 644 | assert isinstance(stats.data["MockConnector"], MockConnectorStats) |
| 645 | # Verify data is preserved |
| 646 | assert stats.data["MockConnector"].data == {"mock_field": [1, 2, 3]} |
| 647 | assert stats.to_dict() == serialized_data |
| 648 | |
| 649 | def test_build_kv_connector_stats_raises_error_for_unknown_connector(self): |
| 650 | """Test that unknown connectors raise an error.""" |
| 651 | serialized_data = { |
| 652 | "UnknownConnector": {"some_field": [1, 2, 3]}, |
| 653 | "NixlConnector": { |
| 654 | "transfer_duration": [1.5], |
| 655 | "post_duration": [0.1], |
| 656 | "bytes_transferred": [1024], |
| 657 | "num_descriptors": [10], |
| 658 | "num_failed_transfers": [], |
| 659 | "num_failed_notifications": [], |
| 660 | "num_failed_handshakes": [], |
| 661 | "num_kv_expired_reqs": [], |
| 662 | "num_notifications_after_expiry": [], |
| 663 | }, |
| 664 | } |
| 665 | |
| 666 | with pytest.raises( |
| 667 | ValueError, match="Connector 'UnknownConnector' is not registered." |
| 668 | ): |
| 669 | MultiConnector.build_kv_connector_stats(data=serialized_data) |
| 670 | |
| 671 | def test_build_kv_connector_stats_with_already_instantiated_objects(self): |
| 672 | """Test that already-instantiated stats objects are preserved (same process).""" |
| 673 | # This simulates the in-process case where stats are not serialized |
| 674 | nixl_stats = NixlKVConnectorStats( |
| 675 | data={ |
| 676 | "transfer_duration": [1.5], |
| 677 | "post_duration": [0.1], |
| 678 | "bytes_transferred": [1024], |
| 679 | "num_descriptors": [10], |
| 680 | "num_failed_transfers": [], |
| 681 | "num_failed_notifications": [], |
| 682 | "num_failed_handshakes": [], |
| 683 | "num_kv_expired_reqs": [], |
| 684 | "num_notifications_after_expiry": [], |
| 685 | } |
| 686 | ) |
| 687 | mock_stats = MockConnectorStats(data={"mock_field": [1, 2, 3]}) |
| 688 | |
| 689 | data_with_objects = { |
| 690 | "NixlConnector": nixl_stats, |
| 691 | "MockConnector": mock_stats, |
| 692 | } |
| 693 | |
| 694 | stats = MultiConnector.build_kv_connector_stats(data=data_with_objects) |
| 695 | |
| 696 | assert stats is not None |
| 697 | assert isinstance(stats, MultiKVConnectorStats) |
| 698 | assert len(stats.data) == 2 |
| 699 | # Verify objects are preserved as-is |
| 700 | assert stats.data["NixlConnector"] is nixl_stats |
| 701 | assert stats.data["MockConnector"] is mock_stats |
| 702 | |
| 703 | def test_build_kv_connector_stats_with_mixed_objects_and_dicts(self): |
| 704 | """Test handling mixed already-instantiated and serialized stats.""" |
| 705 | # This can happen during transition or partial serialization |
| 706 | nixl_stats = NixlKVConnectorStats( |
| 707 | data={ |
| 708 | "transfer_duration": [1.5], |
| 709 | "post_duration": [0.1], |
| 710 | "bytes_transferred": [1024], |
| 711 | "num_descriptors": [10], |
| 712 | "num_failed_transfers": [], |
| 713 | "num_failed_notifications": [], |
| 714 | "num_failed_handshakes": [], |
| 715 | "num_kv_expired_reqs": [], |
| 716 | "num_notifications_after_expiry": [], |
| 717 | } |
| 718 | ) |
| 719 | |
| 720 | mixed_data = { |
| 721 | "NixlConnector": nixl_stats, # Already instantiated |
| 722 | "MockConnector": {"mock_field": [1, 2, 3]}, # Serialized |
| 723 | } |
| 724 | |
| 725 | stats = MultiConnector.build_kv_connector_stats(data=mixed_data) |
| 726 | |
| 727 | assert stats is not None |
| 728 | assert isinstance(stats, MultiKVConnectorStats) |
| 729 | assert len(stats.data) == 2 |
| 730 | # Instantiated object preserved |
| 731 | assert stats.data["NixlConnector"] is nixl_stats |
| 732 | # Serialized object reconstructed |
| 733 | assert isinstance(stats.data["MockConnector"], MockConnectorStats) |
| 734 | assert stats.data["MockConnector"].data == {"mock_field": [1, 2, 3]} |
| 735 | |
| 736 | def test_build_kv_connector_stats_skips_connectors_without_custom_stats(self): |
| 737 | """Test that connectors without custom stats (return None) are skipped.""" |
| 738 | # ExampleConnector doesn't override build_kv_connector_stats, |
| 739 | # so it returns None and should be skipped |
| 740 | serialized_data = { |
| 741 | "NixlConnector": { |
| 742 | "transfer_duration": [1.5], |
| 743 | "post_duration": [0.1], |
| 744 | "bytes_transferred": [1024], |
| 745 | "num_descriptors": [10], |
| 746 | "num_failed_transfers": [], |
| 747 | "num_failed_notifications": [], |
| 748 | "num_failed_handshakes": [], |
| 749 | "num_kv_expired_reqs": [], |
| 750 | "num_notifications_after_expiry": [], |
| 751 | }, |
| 752 | "ExampleConnector": {"some_field": [1, 2, 3]}, |
| 753 | } |
| 754 | |
| 755 | stats = MultiConnector.build_kv_connector_stats(data=serialized_data) |
| 756 | |
| 757 | assert stats is not None |
| 758 | assert isinstance(stats, MultiKVConnectorStats) |
| 759 | # Only NixlConnector should be reconstructed |
| 760 | assert len(stats.data) == 1 |
| 761 | assert "NixlConnector" in stats.data |
| 762 | assert isinstance(stats.data["NixlConnector"], NixlKVConnectorStats) |
| 763 | # ExampleConnector should be skipped (returns None) |
| 764 | assert "ExampleConnector" not in stats.data |
| 765 | |
| 766 | def test_prom_metrics_passes_flat_data_to_children(self): |
| 767 | child_metrics = MagicMock() |
| 768 | metrics = object.__new__(MultiKVConnectorPromMetrics) |
| 769 | metrics._prom_metrics = {"NixlConnector": child_metrics} |
| 770 | payload = {"NixlConnector": {"transfer_duration": [1.5]}} |
| 771 | |
| 772 | metrics.observe(payload, engine_idx=2) |
| 773 | |
| 774 | child_metrics.observe.assert_called_once_with({"transfer_duration": [1.5]}, 2) |
| 775 | |
| 776 | def test_aggregate_same_connector(self): |
| 777 | """Test aggregating stats from the same connector type.""" |
| 778 | stats1 = MultiKVConnectorStats( |
| 779 | data={ |
| 780 | "NixlConnector": NixlKVConnectorStats( |
| 781 | data={ |
| 782 | "transfer_duration": [1.0], |
| 783 | "post_duration": [0.1], |
| 784 | "bytes_transferred": [1024], |
| 785 | "num_descriptors": [10], |
| 786 | "num_failed_transfers": [], |
| 787 | "num_failed_notifications": [], |
| 788 | "num_failed_handshakes": [], |
| 789 | "num_kv_expired_reqs": [], |
| 790 | "num_notifications_after_expiry": [], |
| 791 | } |
| 792 | ) |
| 793 | } |
| 794 | ) |
| 795 | |
| 796 | stats2 = MultiKVConnectorStats( |
| 797 | data={ |
| 798 | "NixlConnector": NixlKVConnectorStats( |
| 799 | data={ |
| 800 | "transfer_duration": [2.0], |
| 801 | "post_duration": [0.2], |
| 802 | "bytes_transferred": [2048], |
| 803 | "num_descriptors": [20], |
| 804 | "num_failed_transfers": [], |
| 805 | "num_failed_notifications": [], |
| 806 | "num_failed_handshakes": [], |
| 807 | "num_kv_expired_reqs": [], |
| 808 | "num_notifications_after_expiry": [], |
| 809 | } |
| 810 | ) |
| 811 | } |
| 812 | ) |
| 813 | |
| 814 | result = stats1.aggregate(stats2) |
| 815 | |
| 816 | assert result is stats1 # Should return self |
| 817 | assert "NixlConnector" in result.data |
| 818 | nixl_stats = result.data["NixlConnector"] |
| 819 | assert nixl_stats.data["transfer_duration"] == [1.0, 2.0] |
| 820 | assert nixl_stats.data["post_duration"] == [0.1, 0.2] |
| 821 | assert nixl_stats.data["bytes_transferred"] == [1024, 2048] |
| 822 | assert nixl_stats.data["num_descriptors"] == [10, 20] |
| 823 | |
| 824 | def test_aggregate_new_connector(self): |
| 825 | """Test aggregating stats when a new connector type appears.""" |
| 826 | from vllm.distributed.kv_transfer.kv_connector.v1.metrics import ( |
| 827 | KVConnectorStats, |
| 828 | ) |
| 829 | |
| 830 | stats1 = MultiKVConnectorStats( |
| 831 | data={ |
| 832 | "NixlConnector": NixlKVConnectorStats( |
| 833 | data={ |
| 834 | "transfer_duration": [1.0], |
| 835 | "post_duration": [0.1], |
| 836 | "bytes_transferred": [1024], |
| 837 | "num_descriptors": [10], |
| 838 | "num_failed_transfers": [], |
| 839 | "num_failed_notifications": [], |
| 840 | "num_failed_handshakes": [], |
| 841 | "num_kv_expired_reqs": [], |
| 842 | "num_notifications_after_expiry": [], |
| 843 | } |
| 844 | ) |
| 845 | } |
| 846 | ) |
| 847 | |
| 848 | stats2 = MultiKVConnectorStats( |
| 849 | data={"ExampleConnector": KVConnectorStats(data={"field": [1, 2]})} |
| 850 | ) |
| 851 | |
| 852 | result = stats1.aggregate(stats2) |
| 853 | |
| 854 | assert "NixlConnector" in result.data |
| 855 | assert "ExampleConnector" in result.data |
| 856 | |
| 857 | def test_reduce(self): |
| 858 | """Test that reduce() correctly reduces all nested connector stats.""" |
| 859 | stats = MultiKVConnectorStats( |
| 860 | data={ |
| 861 | "NixlConnector": NixlKVConnectorStats( |
| 862 | data={ |
| 863 | "transfer_duration": [1.0, 2.0], |
| 864 | "post_duration": [0.1, 0.2], |
| 865 | "bytes_transferred": [1024, 2048], |
| 866 | "num_descriptors": [10, 20], |
| 867 | "num_failed_transfers": [], |
| 868 | "num_failed_notifications": [], |
| 869 | "num_failed_handshakes": [], |
| 870 | "num_kv_expired_reqs": [], |
| 871 | "num_notifications_after_expiry": [], |
| 872 | } |
| 873 | ) |
| 874 | } |
| 875 | ) |
| 876 | |
| 877 | reduced = stats.reduce() |
| 878 | |
| 879 | assert "NixlConnector" in reduced |
| 880 | assert isinstance(reduced["NixlConnector"], dict) |
| 881 | # Check that the stats were reduced (should have aggregated values) |
| 882 | assert "Num successful transfers" in reduced["NixlConnector"] |
| 883 | assert reduced["NixlConnector"]["Num successful transfers"] == 2 |
| 884 | |
| 885 | def test_log_renders_plain_scalars(self): |
| 886 | """Reduced stats must render as plain numbers in the CLI log, not |
| 887 | numpy reprs (eg np.float64(2.338)).""" |
| 888 | stats = MultiKVConnectorStats( |
| 889 | data={ |
| 890 | "NixlConnector": NixlKVConnectorStats( |
| 891 | data={ |
| 892 | "transfer_duration": [1.0, 2.0], |
| 893 | "post_duration": [0.1, 0.2], |
| 894 | "bytes_transferred": [1024, 2048], |
| 895 | "num_descriptors": [10, 20], |
| 896 | "num_failed_transfers": [], |
| 897 | "num_failed_notifications": [], |
| 898 | "num_failed_handshakes": [], |
| 899 | "num_kv_expired_reqs": [], |
| 900 | "num_notifications_after_expiry": [], |
| 901 | } |
| 902 | ) |
| 903 | } |
| 904 | ) |
| 905 | |
| 906 | kv_logging = KVConnectorLogging(kv_transfer_config=None) |
| 907 | kv_logging.transfer_stats_accumulator = stats |
| 908 | log_records: list[tuple] = [] |
| 909 | kv_logging.log(log_fn=lambda *args: log_records.append(args)) |
| 910 | |
| 911 | assert len(log_records) == 1 |
| 912 | msg = log_records[0][1] |
| 913 | assert "np.float" not in msg |
| 914 | assert "'Avg xfer time (ms)': 1500.0" in msg |
| 915 | # The accumulator is reset after logging. |
| 916 | assert kv_logging.transfer_stats_accumulator is None |
| 917 | |
| 918 | def test_reset(self): |
| 919 | """Test that reset() resets all nested connector stats.""" |
| 920 | stats = MultiKVConnectorStats( |
| 921 | data={ |
| 922 | "NixlConnector": NixlKVConnectorStats( |
| 923 | data={ |
| 924 | "transfer_duration": [1.0, 2.0], |
| 925 | "post_duration": [0.1, 0.2], |
| 926 | "bytes_transferred": [1024, 2048], |
| 927 | "num_descriptors": [10, 20], |
| 928 | "num_failed_transfers": [], |
| 929 | "num_failed_notifications": [], |
| 930 | "num_failed_handshakes": [], |
| 931 | "num_kv_expired_reqs": [], |
| 932 | "num_notifications_after_expiry": [], |
| 933 | } |
| 934 | ) |
| 935 | } |
| 936 | ) |
| 937 | |
| 938 | assert not stats.is_empty() |
| 939 | |
| 940 | stats.reset() |
| 941 | |
| 942 | # After reset, stats should be empty |
| 943 | assert stats.is_empty() |
| 944 | nixl_stats = stats.data["NixlConnector"] |
| 945 | assert len(nixl_stats.data["transfer_duration"]) == 0 |
| 946 | |
| 947 | def test_is_empty_with_multiple_connectors(self): |
| 948 | """Test is_empty() returns correct value with multiple connectors.""" |
| 949 | # All empty |
| 950 | stats = MultiKVConnectorStats( |
| 951 | data={ |
| 952 | "NixlConnector": NixlKVConnectorStats(data={}), |
| 953 | } |
| 954 | ) |
| 955 | # Initialize empty stats |
| 956 | stats.data["NixlConnector"].reset() |
| 957 | assert stats.is_empty() |
| 958 | |
| 959 | # One non-empty |
| 960 | stats.data["NixlConnector"].data["transfer_duration"].append(1.0) |
| 961 | assert not stats.is_empty() |
| 962 | |
| 963 | |
| 964 | def test_multi_connector_overrides_all_base_methods(): |
| 965 | """Ensure MultiConnector overrides all public methods from KVConnectorBase_V1.""" |
| 966 | # These are fine to inherit from KVConnectorBase_V1 |
| 967 | # TODO(https://github.com/vllm-project/vllm/pull/31811): Remove |
| 968 | # get_kv_connector_kv_cache_events from INHERITED_OK once implemented. |
| 969 | INHERITED_OK = { |
| 970 | "role", |
| 971 | "has_connector_metadata", |
| 972 | "get_kv_connector_kv_cache_events", |
| 973 | } |
| 974 | |
| 975 | base_members = { |
| 976 | name for name in dir(KVConnectorBase_V1) if not name.startswith("_") |
| 977 | } - KVConnectorBase_V1.__abstractmethods__ |
| 978 | |
| 979 | missing = [ |
| 980 | name |
| 981 | for name in sorted(base_members) |
| 982 | if name not in INHERITED_OK and name not in MultiConnector.__dict__ |
| 983 | ] |
| 984 | |
| 985 | if missing: |
| 986 | pytest.fail(f""" |
| 987 | MultiConnector does not override these KVConnectorBase_V1 methods: {missing} |
| 988 | |
| 989 | MultiConnector wraps other connectors and must delegate all methods. |
| 990 | Please add overrides that delegate to self._connectors. |
| 991 | |
| 992 | Options: |
| 993 | 1. Add delegation in MultiConnector (preferred) |
| 994 | 2. Add to INHERITED_OK if the base implementation works correctly |
| 995 | """) |
| 996 | |
| 997 | |
| 998 | def test_multi_connector_worker_metadata(mc): |
| 999 | class MockConnectorWorkerMetadata(KVConnectorWorkerMetadata): |
| 1000 | def __init__(self, data: set[str]): |
| 1001 | self.data = data |
| 1002 | |
| 1003 | class MockConnectorWorkerMetadata0(MockConnectorWorkerMetadata): |
| 1004 | def aggregate( |
| 1005 | self, other: KVConnectorWorkerMetadata |
| 1006 | ) -> KVConnectorWorkerMetadata: |
| 1007 | assert isinstance(other, MockConnectorWorkerMetadata) |
| 1008 | return MockConnectorWorkerMetadata0(data=self.data | other.data) |
| 1009 | |
| 1010 | class MockConnectorWorkerMetadata1(MockConnectorWorkerMetadata): |
| 1011 | def aggregate( |
| 1012 | self, other: KVConnectorWorkerMetadata |
| 1013 | ) -> KVConnectorWorkerMetadata: |
| 1014 | assert isinstance(other, MockConnectorWorkerMetadata) |
| 1015 | return MockConnectorWorkerMetadata1(data=self.data | other.data) |
| 1016 | |
| 1017 | # -------------------- test build_worker_connector_meta ------------------- |
| 1018 | |
| 1019 | # both connectors return None |
| 1020 | mc._connectors[0].build_connector_worker_meta.return_value = None |
| 1021 | mc._connectors[1].build_connector_worker_meta.return_value = None |
| 1022 | assert mc.build_connector_worker_meta() is None |
| 1023 | |
| 1024 | # only first connector returns None |
| 1025 | worker_meta1a = MockConnectorWorkerMetadata1({"1a"}) |
| 1026 | mc._connectors[0].build_connector_worker_meta.return_value = None |
| 1027 | mc._connectors[1].build_connector_worker_meta.return_value = worker_meta1a |
| 1028 | mc_worker_meta_none_1a = mc.build_connector_worker_meta() |
| 1029 | assert isinstance(mc_worker_meta_none_1a, MultiKVConnectorWorkerMetadata) |
| 1030 | assert mc_worker_meta_none_1a.metadata == (None, worker_meta1a) |
| 1031 | |
| 1032 | # only second connector returns None |
| 1033 | worker_meta0a = MockConnectorWorkerMetadata0({"0a"}) |
| 1034 | mc._connectors[0].build_connector_worker_meta.return_value = worker_meta0a |
| 1035 | mc._connectors[1].build_connector_worker_meta.return_value = None |
| 1036 | mc_worker_meta_0a_none = mc.build_connector_worker_meta() |
| 1037 | assert isinstance(mc_worker_meta_0a_none, MultiKVConnectorWorkerMetadata) |
| 1038 | assert mc_worker_meta_0a_none.metadata == (worker_meta0a, None) |
| 1039 | |
| 1040 | # both connectors do not return None |
| 1041 | worker_meta0b = MockConnectorWorkerMetadata0({"0b"}) |
| 1042 | worker_meta1b = MockConnectorWorkerMetadata1({"1b"}) |
| 1043 | mc._connectors[0].build_connector_worker_meta.return_value = worker_meta0b |
| 1044 | mc._connectors[1].build_connector_worker_meta.return_value = worker_meta1b |
| 1045 | mc_worker_meta_0b_1b = mc.build_connector_worker_meta() |
| 1046 | assert isinstance(mc_worker_meta_0b_1b, MultiKVConnectorWorkerMetadata) |
| 1047 | assert mc_worker_meta_0b_1b.metadata == (worker_meta0b, worker_meta1b) |
| 1048 | |
| 1049 | # ----------------------------- test aggregate ---------------------------- |
| 1050 | |
| 1051 | # aggregate ({"0a"}, None) and (None, {"1a"}) -> ({"0a"}, {"1a"}) |
| 1052 | mc_worker_meta_0a_1a = mc_worker_meta_0a_none.aggregate(mc_worker_meta_none_1a) |
| 1053 | assert isinstance(mc_worker_meta_0a_1a, MultiKVConnectorWorkerMetadata) |
| 1054 | assert mc_worker_meta_0a_1a.metadata == (worker_meta0a, worker_meta1a) |
| 1055 | |
| 1056 | # aggregate ({"0a"}, None) and ({"0b"}, None) -> ({"0a", "0b"}, None) |
| 1057 | mc._connectors[0].build_connector_worker_meta.return_value = worker_meta0b |
| 1058 | mc._connectors[1].build_connector_worker_meta.return_value = None |
| 1059 | mc_worker_meta_0b_none = mc.build_connector_worker_meta() |
| 1060 | mc_worker_meta_0a_0b = mc_worker_meta_0a_none.aggregate(mc_worker_meta_0b_none) |
| 1061 | assert isinstance(mc_worker_meta_0a_0b, MultiKVConnectorWorkerMetadata) |
| 1062 | assert mc_worker_meta_0a_0b.metadata[1] is None |
| 1063 | connector0_md = mc_worker_meta_0a_0b.metadata[0] |
| 1064 | assert isinstance(connector0_md, MockConnectorWorkerMetadata0) |
| 1065 | assert connector0_md.data == {"0a", "0b"} |
| 1066 | |
| 1067 | # aggregate ({"0a"}, {"1a"}) and ({"0b"}, {"1b"}) -> ({"0a", "0b"}, {"1a", "1b"}) |
| 1068 | mc_worker_meta_01a_01b = mc_worker_meta_0a_1a.aggregate(mc_worker_meta_0b_1b) |
| 1069 | assert isinstance(mc_worker_meta_01a_01b, MultiKVConnectorWorkerMetadata) |
| 1070 | metadata = mc_worker_meta_01a_01b.metadata |
| 1071 | assert len(metadata) == 2 |
| 1072 | connector0_md, connector1_md = metadata |
| 1073 | assert isinstance(connector0_md, MockConnectorWorkerMetadata0) |
| 1074 | assert isinstance(connector1_md, MockConnectorWorkerMetadata1) |
| 1075 | assert connector0_md.data == {"0a", "0b"} |
| 1076 | assert connector1_md.data == {"1a", "1b"} |
| 1077 | |
| 1078 | # ---------------------- test update_connector_output --------------------- |
| 1079 | |
| 1080 | def verify_worker_metadata(expected_metadata: MockConnectorWorkerMetadata | None): |
| 1081 | def _verify_worker_metadata(connector_output: KVConnectorOutput): |
| 1082 | worker_meta = connector_output.kv_connector_worker_meta |
| 1083 | if expected_metadata is None: |
| 1084 | assert worker_meta is None |
| 1085 | return |
| 1086 | |
| 1087 | assert isinstance(worker_meta, MockConnectorWorkerMetadata) |
| 1088 | assert type(worker_meta) is type(expected_metadata) |
| 1089 | assert expected_metadata.data == worker_meta.data |
| 1090 | |
| 1091 | return _verify_worker_metadata |
| 1092 | |
| 1093 | def assert_update_connector_output_called(mc: MultiConnector): |
| 1094 | for c in mc._connectors: |
| 1095 | c.update_connector_output.assert_called_once() |
| 1096 | c.update_connector_output.reset_mock() |
| 1097 | |
| 1098 | # no worker meta |
| 1099 | kv_connector_output = KVConnectorOutput() |
| 1100 | mc._connectors[0].update_connector_output.side_effect = verify_worker_metadata(None) |
| 1101 | mc._connectors[1].update_connector_output.side_effect = verify_worker_metadata(None) |
| 1102 | mc.update_connector_output(kv_connector_output) |
| 1103 | assert_update_connector_output_called(mc) |
| 1104 | |
| 1105 | # multi worker meta |
| 1106 | kv_connector_output.kv_connector_worker_meta = mc_worker_meta_01a_01b |
| 1107 | mc._connectors[0].update_connector_output.side_effect = verify_worker_metadata( |
| 1108 | connector0_md |
| 1109 | ) |
| 1110 | mc._connectors[1].update_connector_output.side_effect = verify_worker_metadata( |
| 1111 | connector1_md |
| 1112 | ) |
| 1113 | mc.update_connector_output(kv_connector_output) |
| 1114 | assert_update_connector_output_called(mc) |
| 1115 | assert kv_connector_output.kv_connector_worker_meta == mc_worker_meta_01a_01b |
| 1116 | |
| 1117 | |
| 1118 | def _make_multi_connector(connector_names: list[str]) -> MultiConnector: |
| 1119 | """Build a MultiConnector wrapping the given registered connectors.""" |
| 1120 | connectors = [ |
| 1121 | { |
| 1122 | "kv_connector": name, |
| 1123 | "kv_role": "kv_both", |
| 1124 | "kv_connector_module_path": "tests.v1.kv_connector.unit.test_multi_connector", # noqa: E501 |
| 1125 | } |
| 1126 | for name in connector_names |
| 1127 | ] |
| 1128 | vllm_config = create_vllm_config( |
| 1129 | kv_connector="MultiConnector", |
| 1130 | kv_connector_extra_config={"connectors": connectors}, |
| 1131 | ) |
| 1132 | kv_cache_config = KVCacheConfig( |
| 1133 | num_blocks=0, |
| 1134 | kv_cache_tensors=[], |
| 1135 | kv_cache_groups=[], |
| 1136 | ) |
| 1137 | return MultiConnector( |
| 1138 | vllm_config=vllm_config, |
| 1139 | role=KVConnectorRole.WORKER, |
| 1140 | kv_cache_config=kv_cache_config, |
| 1141 | ) |
| 1142 | |
| 1143 | |
| 1144 | def test_multi_connector_hma_support_detection(): |
| 1145 | """At runtime, _all_support_hma is True only when every sub-connector |
| 1146 | implements SupportsHMA. Test all combinations of HMA / non-HMA |
| 1147 | sub-connectors. |
| 1148 | """ |
| 1149 | assert supports_hma(MultiConnector) |
| 1150 | |
| 1151 | # -- All non-HMA connectors => _all_support_hma is False -- |
| 1152 | mc_none = _make_multi_connector(["MockConnector", "MockConnector"]) |
| 1153 | assert not supports_hma(mc_none._connectors[0]) |
| 1154 | assert not supports_hma(mc_none._connectors[1]) |
| 1155 | assert mc_none._all_support_hma is False |
| 1156 | |
| 1157 | # -- All HMA connectors => _all_support_hma is True -- |
| 1158 | mc_all = _make_multi_connector(["MockHMAConnector", "MockHMAConnector"]) |
| 1159 | assert supports_hma(mc_all._connectors[0]) |
| 1160 | assert supports_hma(mc_all._connectors[1]) |
| 1161 | assert mc_all._all_support_hma is True |
| 1162 | |
| 1163 | # -- Mixed: first HMA, second non-HMA => _all_support_hma is False -- |
| 1164 | mc_mixed1 = _make_multi_connector(["MockHMAConnector", "MockConnector"]) |
| 1165 | assert supports_hma(mc_mixed1._connectors[0]) |
| 1166 | assert not supports_hma(mc_mixed1._connectors[1]) |
| 1167 | assert mc_mixed1._all_support_hma is False |
| 1168 | |
| 1169 | # -- Mixed: first non-HMA, second HMA => _all_support_hma is False -- |
| 1170 | mc_mixed2 = _make_multi_connector(["MockConnector", "MockHMAConnector"]) |
| 1171 | assert not supports_hma(mc_mixed2._connectors[0]) |
| 1172 | assert supports_hma(mc_mixed2._connectors[1]) |
| 1173 | assert mc_mixed2._all_support_hma is False |
| 1174 | |
| 1175 | |
| 1176 | def test_divergent_local_hybrid_hit_capability_is_conservative(): |
| 1177 | all_supported = _make_multi_connector( |
| 1178 | ["MockDivergentHMAConnector", "MockDivergentHMAConnector"] |
| 1179 | ) |
| 1180 | assert all_supported.supports_divergent_local_hybrid_hits is True |
| 1181 | |
| 1182 | mixed = _make_multi_connector(["MockDivergentHMAConnector", "MockHMAConnector"]) |
| 1183 | assert mixed.supports_divergent_local_hybrid_hits is False |
| 1184 | |
| 1185 | |
| 1186 | @pytest.mark.skipif( |
| 1187 | not torch.cuda.is_available(), reason="Requires GPU to instantiate LLM" |
| 1188 | ) |
| 1189 | def test_multi_connector_mixed_hma_disables_hybrid_kv_cache(monkeypatch): |
| 1190 | """When MultiConnector wraps a mix of HMA (NixlConnector) and non-HMA |
| 1191 | (MockConnector) sub-connectors, verify that: |
| 1192 | 1. The scheduler's MultiConnector has _all_support_hma == False. |
| 1193 | 2. vLLM auto-disables the hybrid KV cache manager (no preference expressed by user) |
| 1194 | """ |
| 1195 | from unittest.mock import patch |
| 1196 | |
| 1197 | from tests.v1.kv_connector.unit.test_nixl_connector import FakeNixlWrapper |
| 1198 | |
| 1199 | monkeypatch.setenv("VLLM_ENABLE_V1_MULTIPROCESSING", "0") |
| 1200 | |
| 1201 | kv_transfer_config = KVTransferConfig( |
| 1202 | kv_connector="MultiConnector", |
| 1203 | kv_role="kv_both", |
| 1204 | kv_connector_extra_config={ |
| 1205 | "connectors": [ |
| 1206 | { |
| 1207 | "kv_connector": "NixlConnector", |
| 1208 | "kv_role": "kv_consumer", |
| 1209 | }, |
| 1210 | { |
| 1211 | "kv_connector": "MockConnector", |
| 1212 | "kv_role": "kv_both", |
| 1213 | "kv_connector_module_path": ( |
| 1214 | "tests.v1.kv_connector.unit.test_multi_connector" |
| 1215 | ), |
| 1216 | }, |
| 1217 | ], |
| 1218 | }, |
| 1219 | ) |
| 1220 | |
| 1221 | with patch( |
| 1222 | "vllm.distributed.kv_transfer.kv_connector.v1.nixl.base_worker.NixlWrapper", |
| 1223 | FakeNixlWrapper, |
| 1224 | ): |
| 1225 | llm = LLM( |
| 1226 | model="Qwen/Qwen3-0.6B", |
| 1227 | enforce_eager=True, |
| 1228 | gpu_memory_utilization=0.3, |
| 1229 | max_model_len=128, |
| 1230 | max_num_seqs=1, |
| 1231 | max_num_batched_tokens=128, |
| 1232 | kv_transfer_config=kv_transfer_config, |
| 1233 | ) |
| 1234 | try: |
| 1235 | # HMA should be auto-disabled when user has not expressed a preference. |
| 1236 | assert ( |
| 1237 | llm.llm_engine.vllm_config.scheduler_config.disable_hybrid_kv_cache_manager |
| 1238 | is True |
| 1239 | ) |
| 1240 | # The scheduler-side MultiConnector should detect the mixed |
| 1241 | # HMA support among its sub-connectors. |
| 1242 | scheduler = llm.llm_engine.engine_core.engine_core.scheduler |
| 1243 | mc = scheduler.connector |
| 1244 | assert isinstance(mc, MultiConnector) |
| 1245 | assert mc._all_support_hma is False |
| 1246 | finally: |
| 1247 | llm.llm_engine.engine_core.shutdown() |