| 1 | # SPDX-License-Identifier: Apache-2.0 |
| 2 | # SPDX-FileCopyrightText: Copyright contributors to the vLLM project |
| 3 | """Shared constants, lazy imports and helpers for the NIXL connector.""" |
| 4 | |
| 5 | import contextlib |
| 6 | from collections.abc import Iterator |
| 7 | from typing import Any |
| 8 | |
| 9 | import regex as re |
| 10 | import zmq |
| 11 | |
| 12 | from vllm.platforms import current_platform |
| 13 | from vllm.utils.network_utils import make_zmq_socket |
| 14 | from vllm.v1.kv_cache_interface import KVCacheSpec, UniformTypeKVCacheSpecs |
| 15 | |
| 16 | # Supported platforms and types of kv transfer buffer. |
| 17 | # {device: tuple of supported kv buffer types} |
| 18 | _NIXL_SUPPORTED_DEVICE = { |
| 19 | "cuda": ( |
| 20 | "cuda", |
| 21 | "cpu", |
| 22 | ), |
| 23 | "tpu": ("cpu",), |
| 24 | "xpu": ( |
| 25 | "cpu", |
| 26 | "xpu", |
| 27 | ), |
| 28 | "cpu": ("cpu",), |
| 29 | } |
| 30 | # support for oot platform by providing mapping in current_platform |
| 31 | _NIXL_SUPPORTED_DEVICE.update(current_platform.get_nixl_supported_devices()) |
| 32 | |
| 33 | |
| 34 | # TODO: merge with vllm.utils.network_utils.zmq_socket_ctx |
| 35 | @contextlib.contextmanager |
| 36 | def zmq_ctx(socket_type: Any, addr: str) -> Iterator[zmq.Socket]: |
| 37 | """Context manager for a ZMQ socket.""" |
| 38 | if socket_type not in (zmq.ROUTER, zmq.REQ): |
| 39 | raise ValueError(f"Unexpected socket type: {socket_type}") |
| 40 | |
| 41 | ctx: zmq.Context | None = None |
| 42 | try: |
| 43 | ctx = zmq.Context() # type: ignore[attr-defined] |
| 44 | yield make_zmq_socket( |
| 45 | ctx=ctx, path=addr, socket_type=socket_type, bind=socket_type == zmq.ROUTER |
| 46 | ) |
| 47 | finally: |
| 48 | if ctx is not None: |
| 49 | ctx.destroy(linger=0) |
| 50 | |
| 51 | |
| 52 | def get_representative_spec_type(spec: KVCacheSpec) -> type[KVCacheSpec]: |
| 53 | if isinstance(spec, UniformTypeKVCacheSpecs): |
| 54 | # All inner specs are the same type; pick any. |
| 55 | inner = next(iter(spec.kv_cache_specs.values())) |
| 56 | return type(inner) |
| 57 | return type(spec) |
| 58 | |
| 59 | |
| 60 | # Trailing 8-hex randomization suffix appended by |
| 61 | # ``input_processor.assign_request_id`` as ``-{random_uuid():.8}``. |
| 62 | _RANDOM_SUFFIX_RE = re.compile(r"-[0-9a-f]{8}$", re.IGNORECASE) |
| 63 | |
| 64 | |
| 65 | def get_base_request_id(request_id: str) -> str: |
| 66 | """Strip the per-request ``-<8 hex>`` randomization suffix, if present.""" |
| 67 | return _RANDOM_SUFFIX_RE.sub("", request_id) |