diff --git a/.github/workflows/e2e-gpu-job.yml b/.github/workflows/e2e-gpu-job.yml index 0e003efdd8..2f3fdae268 100644 --- a/.github/workflows/e2e-gpu-job.yml +++ b/.github/workflows/e2e-gpu-job.yml @@ -89,6 +89,11 @@ on: type: string default: "" description: "Prebuilt TokenSpeed carrier image to extract onto the runner (resolved by detect-changes); empty = build from source" + servicer_unit_tests: + required: false + type: boolean + default: false + description: "Also run grpc_servicer/tests here, where the importorskip('vllm') cases meet a real vLLM; set on one vLLM lane" jobs: run: @@ -174,6 +179,14 @@ jobs: pip install wheel/*.whl bash scripts/ci_install_e2e_deps.sh ${{ inputs.extra_deps }} + # The servicer's vLLM-gated unit tests (importorskip("vllm")) only meet a + # real vLLM here; the CPU unit-test job skips them. One lane opts in. + - name: Run vLLM-gated servicer unit tests + if: inputs.servicer_unit_tests + run: | + python3 -c "import vllm, msgspec" # the gates these tests importorskip on + pytest -q grpc_servicer/tests -p no:cacheprovider -rs + - name: Setup Oracle if: inputs.setup_agentic_deps run: | diff --git a/.github/workflows/pr-test-rust.yml b/.github/workflows/pr-test-rust.yml index 1999f27beb..cfcad09ed1 100644 --- a/.github/workflows/pr-test-rust.yml +++ b/.github/workflows/pr-test-rust.yml @@ -257,8 +257,12 @@ jobs: rm -f crates/grpc_client/python/smg_grpc_proto/proto mkdir -p crates/grpc_client/python/smg_grpc_proto/proto cp crates/grpc_client/proto/*.proto crates/grpc_client/python/smg_grpc_proto/proto/ - pip install ./crates/grpc_client/python - pytest -q grpc_servicer/tests + # msgspec and pydantic gate the sidecar-protocol, fake-redis and + # limit-options tests. Modules gated on zmq or vllm still skip here + # (-rs lists them): the zmq ones import engine servicers that need + # torch, and the vllm cases run on the vLLM e2e lane. + pip install ./crates/grpc_client/python msgspec pydantic + pytest -q grpc_servicer/tests -rs # CPU-only unit tests for the e2e harness itself (connection-mode parsing, # ZMQ/EPD command builders, ZMQ-lane collection filter). --noconftest keeps @@ -673,6 +677,7 @@ jobs: test_dirs: ${{ matrix.test_dirs || 'e2e_test/chat_completions' }} min_selected: ${{ matrix.min_selected }} tokenspeed_prebuilt_image: ${{ matrix.engine == 'tokenspeed' && needs.detect-changes.outputs.tokenspeed-image || '' }} + servicer_unit_tests: ${{ matrix.engine == 'vllm' }} secrets: inherit # Worker-side media processing: the vLLM gRPC worker fetches and diff --git a/grpc_servicer/README.md b/grpc_servicer/README.md index 9661f72cf3..e5879b5579 100644 --- a/grpc_servicer/README.md +++ b/grpc_servicer/README.md @@ -70,6 +70,29 @@ e2e lanes run the multimodal suites in this mode, and `crates/multimodal/scripts/check_worker_anchor_parity.py` checks that a spec's anchor is the token vLLM expands. +To move fetching and processing out of the vLLM process, run the GPU-free +sidecar next to a private Redis and point the worker at it +(`pip install smg-grpc-servicer[vllm,vllm-redis]`): + +```bash +python -m smg_grpc_servicer.vllm.mm_sidecar --model Qwen/Qwen3-VL-8B-Instruct \ + --redis-url redis://127.0.0.1:6379/0 --allowed-media-domains example.com +SMG_VLLM_MM_PROCESSOR=redis SMG_VLLM_MM_REDIS_URL=redis://127.0.0.1:6379/0 \ + vllm serve Qwen/Qwen3-VL-8B-Instruct --grpc +``` + +The sidecar and the worker must agree on model, vLLM version, dtype, video +backend, media/processor kwargs and `--limit-mm-per-prompt` (pass the flag to +both processes; the sidecar's limit is the one that applies, the limit is +resolved per modality before hashing so equivalent spellings match, and the key +namespace is derived from all of these): the worker advertises +`mm_processor=redis` only while a sidecar with a matching fingerprint keeps its +`hello` key alive, and rejects results that disagree. Jobs and results travel over Redis lists under +`smg:mm:v1:{namespace}`; results carry full tensors keyed by a per-attempt job +id and expire after 120 s. Knobs: `SMG_VLLM_MM_SIDECAR_TIMEOUT_MS` (30000), +`SMG_VLLM_MM_SIDECAR_MAX_QUEUE` (256, fail fast when the queue is deeper), +`SMG_VLLM_MM_SIDECAR_NAMESPACE` (override the derived namespace). + ### MLX ```bash diff --git a/grpc_servicer/pyproject.toml b/grpc_servicer/pyproject.toml index 97177ce255..98f48d9817 100644 --- a/grpc_servicer/pyproject.toml +++ b/grpc_servicer/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "smg-grpc-servicer" -version = "0.10.0" +version = "0.11.0" description = "SMG gRPC servicer implementations for LLM inference engines (vLLM, MLX, TokenSpeed, SGLang)" requires-python = ">=3.10" dependencies = [ @@ -31,6 +31,8 @@ classifiers = [ [project.optional-dependencies] vllm = ["vllm>=0.19.0", "pyzmq>=25.0.0", "msgspec>=0.18.0"] +# Redis media-processing sidecar backend for vLLM (SMG_VLLM_MM_PROCESSOR=redis). +vllm-redis = ["redis>=5.0", "msgspec>=0.18.0"] # 0.5.18 dropped the server_args parameter from # create_load_snapshot_reader(); the floor tracks that signature, which the # request manager calls unconditionally. (Earlier floors recorded the shmem diff --git a/grpc_servicer/smg_grpc_servicer/mm_sidecar_protocol.py b/grpc_servicer/smg_grpc_servicer/mm_sidecar_protocol.py new file mode 100644 index 0000000000..f5288206d8 --- /dev/null +++ b/grpc_servicer/smg_grpc_servicer/mm_sidecar_protocol.py @@ -0,0 +1,192 @@ +"""Servicer <-> sidecar protocol for worker-side media processing over Redis. + +Engine-free: no vLLM import. Redis is a job transport plus the sidecar's own +cache; the servicer always receives full tensors, so nothing here has to stay +consistent with vLLM's processor caches. + +Keys under ``smg:mm:v1:{namespace}``: +- ``:hello`` HASH, refreshed by the sidecar (fingerprint + schemes) +- ``:jobs`` LIST, servicer LPUSH / sidecar BRPOP +- ``:res:{job_id}`` LIST, sidecar LPUSH + EXPIRE / servicer BRPOP +""" + +from __future__ import annotations + +import hashlib +from collections.abc import Mapping + +import msgspec + +SCHEMA_VERSION = 1 +KEY_ROOT = "smg:mm:v1" + +ENV_REDIS_URL = "SMG_VLLM_MM_REDIS_URL" +ENV_TIMEOUT_MS = "SMG_VLLM_MM_SIDECAR_TIMEOUT_MS" +ENV_MAX_QUEUE = "SMG_VLLM_MM_SIDECAR_MAX_QUEUE" +ENV_NAMESPACE = "SMG_VLLM_MM_SIDECAR_NAMESPACE" + +DEFAULT_REDIS_URL = "redis://127.0.0.1:6379/0" +DEFAULT_TIMEOUT_MS = 30_000 +DEFAULT_MAX_QUEUE = 256 + +HELLO_TTL_S = 15 +HELLO_REFRESH_S = 5 +RESULT_TTL_S = 120 + +# Client-caused failures become INVALID_ARGUMENT (terminal 400). +CLIENT_ERROR_CODES = frozenset( + { + "fetch_failed", + "domain_not_allowed", + "scheme_not_accepted", + "limit_exceeded", + "decode_failed", + "placeholder_mismatch", + } +) +# Everything else is UNAVAILABLE (503) so the router retries elsewhere. +RETRYABLE_ERROR_CODES = frozenset({"expired", "processor_error", "fingerprint_mismatch"}) + + +class Fingerprint(msgspec.Struct, frozen=True): + """What must match between a worker and the sidecar processing for it.""" + + model: str + vllm_version: str + dtype: str + video_backend: str + media_io_kwargs: str + mm_processor_kwargs: str + # The sidecar's limit is the only one applied on the redis path. + limit_per_prompt: str + + def namespace(self) -> str: + """Key namespace: fleets that would fail `mismatches` never share keys.""" + joined = "|".join(getattr(self, name) for name in self.__struct_fields__) + return hashlib.sha256(joined.encode()).hexdigest()[:16] + + def mismatches(self, other: Fingerprint) -> list[str]: + return [ + name for name in self.__struct_fields__ if getattr(self, name) != getattr(other, name) + ] + + def to_hello(self) -> dict[str, str]: + return {name: getattr(self, name) for name in self.__struct_fields__} + + @classmethod + def from_hello(cls, mapping: Mapping[bytes | str, bytes | str]) -> Fingerprint | None: + decoded = {_text(key): _text(value) for key, value in mapping.items()} + try: + return cls(**{name: decoded[name] for name in cls.__struct_fields__}) + except KeyError: + return None + + +def _text(value: bytes | str) -> str: + return value.decode("utf-8", errors="replace") if isinstance(value, bytes) else value + + +def hello_field(mapping: Mapping[bytes | str, bytes | str], name: str) -> str: + for key, value in mapping.items(): + if _text(key) == name: + return _text(value) + return "" + + +def hello_schemes(mapping: Mapping[bytes | str, bytes | str]) -> str: + return hello_field(mapping, "schemes") + + +def resolve_namespace(fingerprint: Fingerprint, override: str | None) -> str: + override = (override or "").strip() + return override if override else fingerprint.namespace() + + +class Keys(msgspec.Struct, frozen=True): + prefix: str + + @classmethod + def for_namespace(cls, namespace: str) -> Keys: + return cls(prefix=f"{KEY_ROOT}:{namespace}") + + @property + def hello(self) -> str: + return f"{self.prefix}:hello" + + @property + def jobs(self) -> str: + return f"{self.prefix}:jobs" + + def result(self, job_id: str) -> str: + return f"{self.prefix}:res:{job_id}" + + +class JobItem(msgspec.Struct): + modality: str + url: str + + +class Job(msgspec.Struct): + v: int + job_id: str + request_id: str + fingerprint: Fingerprint + prompt_token_ids: list[int] + prompt: str | None + items: list[JobItem] + enqueued_ms: int + deadline_ms: int + + +class Placeholder(msgspec.Struct): + offset: int + length: int + is_embed: list[bool] | None = None + + +class Timing(msgspec.Struct): + queue_ms: int = 0 + fetch_ms: int = 0 + process_ms: int = 0 + + +class JobResult(msgspec.Struct): + v: int + job_id: str + ok: bool + fingerprint: Fingerprint | None = None + prompt_token_ids: list[int] = msgspec.field(default_factory=list) + mm_hashes: dict[str, list[str]] = msgspec.field(default_factory=dict) + mm_placeholders: dict[str, list[Placeholder]] = msgspec.field(default_factory=dict) + # Per item: one vLLM MsgpackEncoder buffer of a MultiModalKwargsItem. + mm_kwargs: dict[str, list[bytes]] = msgspec.field(default_factory=dict) + code: str = "" + message: str = "" + timing: Timing | None = None + cache: str = "" + + +def failure(job_id: str, code: str, message: str) -> JobResult: + return JobResult(v=SCHEMA_VERSION, job_id=job_id, ok=False, code=code, message=message) + + +_job_encoder = msgspec.msgpack.Encoder() +_job_decoder = msgspec.msgpack.Decoder(type=Job) +_result_encoder = msgspec.msgpack.Encoder() +_result_decoder = msgspec.msgpack.Decoder(type=JobResult) + + +def encode_job(job: Job) -> bytes: + return _job_encoder.encode(job) + + +def decode_job(raw: bytes) -> Job: + return _job_decoder.decode(raw) + + +def encode_result(result: JobResult) -> bytes: + return _result_encoder.encode(result) + + +def decode_result(raw: bytes) -> JobResult: + return _result_decoder.decode(raw) diff --git a/grpc_servicer/smg_grpc_servicer/vllm/mm_processor.py b/grpc_servicer/smg_grpc_servicer/vllm/mm_processor.py index f14efc1b12..2549319fb2 100644 --- a/grpc_servicer/smg_grpc_servicer/vllm/mm_processor.py +++ b/grpc_servicer/smg_grpc_servicer/vllm/mm_processor.py @@ -9,12 +9,43 @@ from __future__ import annotations import asyncio +import dataclasses import inspect +import json import logging import os +import time +import uuid from collections.abc import Awaitable, Mapping, Sequence from typing import Any +from smg_grpc_servicer.mm_sidecar_protocol import ( + CLIENT_ERROR_CODES, + DEFAULT_MAX_QUEUE, + DEFAULT_REDIS_URL, + DEFAULT_TIMEOUT_MS, + ENV_MAX_QUEUE, + ENV_NAMESPACE, + ENV_REDIS_URL, + ENV_TIMEOUT_MS, + SCHEMA_VERSION, + Fingerprint, + Job, + JobItem, + JobResult, + Keys, + decode_result, + encode_job, + hello_field, + hello_schemes, + resolve_namespace, +) +from smg_grpc_servicer.vllm.media_refs import ( + BASE_SCHEMES, + advertised_schemes, + parse_scheme_list, +) + logger = logging.getLogger(__name__) ENV_PROCESSOR = "SMG_VLLM_MM_PROCESSOR" @@ -138,8 +169,6 @@ def __init__( from vllm.multimodal.media.connector import MEDIA_CONNECTOR_REGISTRY from vllm.transformers_utils.processor import get_video_processor_cls_name - from smg_grpc_servicer.vllm.media_refs import advertised_schemes, parse_scheme_list - model_config = engine.model_config mm_config = model_config.get_multimodal_config() self._engine = engine @@ -173,6 +202,8 @@ async def process( prompt_text: str | None, items: Sequence[Any], arrival_time: float, + *, + request_id: str = "", ): enforce_item_count(items, self._max_items) enforce_item_bytes(items, self._max_item_bytes) @@ -215,6 +246,294 @@ async def _fetch(self, index: int, item): raise ValueError(f"unsupported media modality {item.modality!r}") +# A modality name vLLM never configures: its resolved count is the unset default. +_UNSET_MODALITY = "__unset__" + + +def _limit_options(options) -> dict[str, Any]: + """A validated per-modality limit as a comparable dict, unset fields dropped. + + Unknown shapes are hashed by their attributes, never reduced to a count: + the two ends must fail closed rather than falsely match. + """ + if dataclasses.is_dataclass(options) and not isinstance(options, type): + data = dataclasses.asdict(options) + elif isinstance(options, Mapping): + data = dict(options) + elif isinstance(options, int): # legacy count-only form + data = {"count": options} + elif hasattr(options, "model_dump"): # a pydantic model + data = dict(options.model_dump()) + elif hasattr(options, "__dict__"): + logger.warning("limit_per_prompt value of type %s: hashing its attributes", type(options)) + data = dict(vars(options)) + else: + raise TypeError(f"unsupported limit_per_prompt value: {options!r}") + return {key: value for key, value in data.items() if value is not None} + + +def resolved_mm_limits(mm_config) -> dict[str, dict[str, Any]]: + """Per-modality prompt limits as vLLM validated them: the unset default under + "*", then each configured modality's effective count (get_limit_per_prompt) + plus its sibling options, omitting entries that only spell out the default, + so equivalent --limit-mm-per-prompt spellings match and a real skew fails + closed.""" + unset = {"count": mm_config.get_limit_per_prompt(_UNSET_MODALITY)} + limits: dict[str, dict[str, Any]] = {"*": unset} + configured = getattr(mm_config, "limit_per_prompt", None) or {} + for modality in sorted(configured): + extra = _limit_options(configured[modality]) + extra.pop("count", None) + entry = {"count": mm_config.get_limit_per_prompt(modality), **extra} + if entry != unset: + limits[modality] = entry + return limits + + +def fingerprint_from_model_config(model_config) -> Fingerprint: + """The one derivation the worker and the sidecar share.""" + import vllm + from vllm import envs + + mm_config = model_config.get_multimodal_config() + return Fingerprint( + model=model_config.model, + vllm_version=vllm.__version__, + dtype=str(model_config.dtype), + video_backend=envs.VLLM_VIDEO_LOADER_BACKEND, + media_io_kwargs=json.dumps(mm_config.media_io_kwargs or {}, sort_keys=True, default=str), + mm_processor_kwargs=json.dumps( + mm_config.mm_processor_kwargs or {}, sort_keys=True, default=str + ), + limit_per_prompt=json.dumps(resolved_mm_limits(mm_config), sort_keys=True, default=str), + ) + + +def engine_fingerprint(engine) -> Fingerprint: + """What the sidecar must match to process media for this engine.""" + return fingerprint_from_model_config(engine.model_config) + + +def _cast_floats(data, dtype): + """Cast floating tensors to the model dtype, as the HF processor path does.""" + import torch + + if torch.is_tensor(data): + return data.to(dtype=dtype) if data.is_floating_point() else data + if isinstance(data, list): + return [_cast_floats(part, dtype) for part in data] + return data + + +class RedisMediaProcessor: + """Hand jobs to a media-processing sidecar over Redis lists. + + The sidecar fetches and runs vLLM's processor; the result carries full + tensors, rebuilt here into the same pre-rendered engine input the + preprocessed path uses. + """ + + name = MODE_REDIS + + def __init__( + self, + engine, + fingerprint: Fingerprint, + *, + redis_url: str = DEFAULT_REDIS_URL, + timeout_ms: int = DEFAULT_TIMEOUT_MS, + max_queue: int = DEFAULT_MAX_QUEUE, + namespace: str | None = None, + max_item_bytes: int = DEFAULT_MAX_ITEM_BYTES, + max_inflight: int = DEFAULT_MAX_INFLIGHT, + max_items: int = DEFAULT_MAX_ITEMS, + client=None, + ) -> None: + self._engine = engine + self._fingerprint = fingerprint + self._timeout_ms = timeout_ms + self._max_queue = max_queue + self._max_item_bytes = max_item_bytes + self._max_items = max_items + self.max_inflight = max_inflight + self._keys = Keys.for_namespace(resolve_namespace(fingerprint, namespace)) + self._client = client if client is not None else _redis_client(redis_url) + self._probe_logged = False + # Until the sidecar says otherwise, assume the default fetch schemes. + self.schemes = ",".join(BASE_SCHEMES) + self.accepted_schemes = parse_scheme_list(self.schemes) + logger.info("Redis media sidecar keys under %s", self._keys.prefix) + + async def probe(self) -> bool: + """Whether a sidecar with a matching fingerprint is alive.""" + try: + hello = await self._client.hgetall(self._keys.hello) + except Exception as e: # noqa: BLE001 - any transport failure means "not advertised" + self._log_probe_once("redis unreachable: %s", e) + return False + schema = hello_field(hello, "schema") if hello else "" + if hello and schema != str(SCHEMA_VERSION): + self._log_probe_once( + "sidecar speaks protocol v%s, worker speaks v%s", schema or "?", SCHEMA_VERSION + ) + return False + remote = Fingerprint.from_hello(hello) if hello else None + if remote is None: + # A sidecar built for any other fingerprint lives under another + # namespace, so this is the message every disagreement produces. + self._log_probe_once( + "no sidecar hello at %s (worker fingerprint=%s)", + self._keys.hello, + self._fingerprint.to_hello(), + ) + return False + mismatches = self._fingerprint.mismatches(remote) + if mismatches: + self._log_probe_once( + "sidecar fingerprint mismatch on %s (worker=%s sidecar=%s)", + ",".join(mismatches), + self._fingerprint.to_hello(), + remote.to_hello(), + ) + return False + schemes = hello_schemes(hello) + if schemes: + self.schemes = schemes + self.accepted_schemes = parse_scheme_list(schemes) + self._probe_logged = False + return True + + def _log_probe_once(self, message: str, *args) -> None: + if not self._probe_logged: + logger.error("Media sidecar not advertised: " + message, *args) + self._probe_logged = True + + async def process( + self, + prompt_token_ids: list[int], + prompt_text: str | None, + items: Sequence[Any], + arrival_time: float, + *, + request_id: str = "", + ): + enforce_item_count(items, self._max_items) + enforce_item_bytes(items, self._max_item_bytes) + now_ms = int(time.time() * 1000) + job = Job( + v=SCHEMA_VERSION, + job_id=uuid.uuid4().hex, + request_id=request_id, + fingerprint=self._fingerprint, + prompt_token_ids=list(prompt_token_ids), + prompt=prompt_text, + items=[JobItem(modality=item.modality, url=item.url) for item in items], + enqueued_ms=now_ms, + deadline_ms=now_ms + self._timeout_ms, + ) + result = await self._submit_and_wait(job) + return self._rebuild(result, prompt_text, arrival_time) + + async def _submit_and_wait(self, job: Job) -> JobResult: + """Transport only: queue the job and wait for its result.""" + try: + depth = await self._client.llen(self._keys.jobs) + if depth >= self._max_queue: + raise MmProcessorUnavailable( + f"sidecar_overloaded: {depth} jobs queued (cap {self._max_queue})" + ) + await self._client.lpush(self._keys.jobs, encode_job(job)) + popped = await self._client.brpop( + self._keys.result(job.job_id), timeout=self._timeout_ms / 1000 + ) + except MmProcessorUnavailable: + raise + except Exception as e: # noqa: BLE001 - redis transport failures are retryable + raise MmProcessorUnavailable(f"sidecar_unavailable: {e}") from e + if popped is None: + raise MmProcessorUnavailable( + f"sidecar_timeout: no result for job {job.job_id} within {self._timeout_ms} ms" + ) + _, raw = popped + try: + result = decode_result(raw) + except Exception as e: # noqa: BLE001 - an undecodable result means "try another worker" + raise MmProcessorUnavailable(f"sidecar_protocol: undecodable result: {e}") from e + if result.v != SCHEMA_VERSION: + raise MmProcessorUnavailable( + f"sidecar_protocol: result schema v{result.v}, worker speaks v{SCHEMA_VERSION}" + ) + if result.job_id != job.job_id: + raise MmProcessorUnavailable( + f"sidecar_protocol: result for job {result.job_id} on key of {job.job_id}" + ) + if not result.ok: + if result.code in CLIENT_ERROR_CODES: + raise ValueError(f"media processing failed ({result.code}): {result.message}") + raise MmProcessorUnavailable(f"{result.code or 'processor_error'}: {result.message}") + if result.fingerprint is not None: + mismatches = self._fingerprint.mismatches(result.fingerprint) + if mismatches: + raise MmProcessorUnavailable(f"fingerprint_mismatch: {','.join(mismatches)}") + return result + + def _rebuild(self, result: JobResult, prompt_text: str | None, arrival_time: float): + """Turn a sidecar result into the pre-rendered engine input.""" + import torch + from vllm.inputs.engine import mm_input + from vllm.multimodal.inputs import ( + MultiModalKwargsItem, + MultiModalKwargsItems, + PlaceholderRange, + ) + from vllm.v1.serial_utils import MsgpackDecoder + + decoder = MsgpackDecoder(t=MultiModalKwargsItem) + dtype = self._engine.model_config.dtype + mm_kwargs: dict[str, list[MultiModalKwargsItem]] = {} + for modality, blobs in result.mm_kwargs.items(): + items = [] + for blob in blobs: + item = decoder.decode(blob) + for elem in item.values(): + elem.data = _cast_floats(elem.data, dtype) + items.append(item) + mm_kwargs[modality] = items + + mm_placeholders = { + modality: [ + PlaceholderRange( + offset=p.offset, + length=p.length, + is_embed=torch.tensor(p.is_embed, dtype=torch.bool) if p.is_embed else None, + ) + for p in ranges + ] + for modality, ranges in result.mm_placeholders.items() + } + prompt = mm_input( + prompt_token_ids=list(result.prompt_token_ids), + mm_kwargs=MultiModalKwargsItems(mm_kwargs), + mm_hashes={modality: list(hashes) for modality, hashes in result.mm_hashes.items()}, + mm_placeholders=mm_placeholders, + prompt=prompt_text, + ) + prompt["arrival_time"] = arrival_time + return prompt + + +def _redis_client(redis_url: str): + try: + import redis.asyncio as redis_asyncio + except ImportError as e: + raise ValueError( + f"{ENV_PROCESSOR}=redis requires the redis client: " + "pip install smg-grpc-servicer[vllm,vllm-redis]" + ) from e + + return redis_asyncio.from_url(redis_url, decode_responses=False, socket_connect_timeout=1.0) + + def build_mm_processor(engine, *, env: Mapping[str, str] = os.environ): """Construct the configured backend, or None when worker-side processing is off.""" mode = resolve_mm_processor_mode(env) @@ -231,4 +550,14 @@ def build_mm_processor(engine, *, env: Mapping[str, str] = os.environ): return InProcessMediaProcessor( engine, max_item_bytes=max_item_bytes, max_inflight=max_inflight, max_items=max_items ) - raise ValueError(f"{ENV_PROCESSOR}={mode} is not available in this servicer build") + return RedisMediaProcessor( + engine, + engine_fingerprint(engine), + redis_url=env.get(ENV_REDIS_URL) or DEFAULT_REDIS_URL, + timeout_ms=env_int(env, ENV_TIMEOUT_MS, DEFAULT_TIMEOUT_MS), + max_queue=env_int(env, ENV_MAX_QUEUE, DEFAULT_MAX_QUEUE), + namespace=env.get(ENV_NAMESPACE), + max_item_bytes=max_item_bytes, + max_inflight=max_inflight, + max_items=max_items, + ) diff --git a/grpc_servicer/smg_grpc_servicer/vllm/mm_sidecar.py b/grpc_servicer/smg_grpc_servicer/vllm/mm_sidecar.py new file mode 100644 index 0000000000..a01448a6b4 --- /dev/null +++ b/grpc_servicer/smg_grpc_servicer/vllm/mm_sidecar.py @@ -0,0 +1,290 @@ +"""Reference media-processing sidecar for vLLM gRPC workers (GPU-free). + +Pops jobs from Redis, fetches the media with vLLM's MediaConnector, runs vLLM's +own multimodal processor over the unexpanded prompt, and pushes full-tensor +results back. Built the way ``vllm launch render`` builds its renderer, so the +tensors match what the worker would have produced itself. + +Usage: + python -m smg_grpc_servicer.vllm.mm_sidecar --model \\ + --redis-url redis://127.0.0.1:6379/0 [--namespace NS] [--concurrency N] \\ + [AsyncEngineArgs flags: --dtype --allowed-media-domains --media-io-kwargs ...] +""" + +from __future__ import annotations + +import argparse +import asyncio +import logging +import time + +from smg_grpc_servicer.mm_sidecar_protocol import ( + DEFAULT_REDIS_URL, + HELLO_REFRESH_S, + HELLO_TTL_S, + RESULT_TTL_S, + SCHEMA_VERSION, + Fingerprint, + Job, + JobResult, + Keys, + Placeholder, + Timing, + decode_job, + encode_result, + failure, + resolve_namespace, +) +from smg_grpc_servicer.vllm.media_refs import advertised_schemes, parse_scheme_list, url_scheme +from smg_grpc_servicer.vllm.mm_processor import fingerprint_from_model_config + +logger = logging.getLogger("smg_grpc_servicer.vllm.mm_sidecar") + + +def build_config(args: argparse.Namespace): + """vLLM config for preprocessing only: no quantized kernels, no KV cache.""" + from vllm import envs + from vllm.config import VllmConfig + from vllm.engine.arg_utils import AsyncEngineArgs + + engine_args = AsyncEngineArgs.from_cli_args(args) + model_config = engine_args.create_model_config() + model_config.quantization = None + envs.VLLM_CPU_KVCACHE_SPACE = 0 + return VllmConfig(model_config=model_config) + + +def config_fingerprint(vllm_config) -> Fingerprint: + return fingerprint_from_model_config(vllm_config.model_config) + + +# Codes come from exception types where vLLM offers one. The two substring +# checks pin message wording from vLLM 0.27: connector.py's allowed-domain +# rejection ("... domain ...") and processing/context.py's per-prompt limit +# ("At most N ..."); test_vllm_mm_sidecar.py holds the wording. +def classify_fetch_error(exc: BaseException) -> str: + if isinstance(exc, ValueError) and "domain" in str(exc).lower(): + return "domain_not_allowed" + return "fetch_failed" + + +def classify_process_error(exc: BaseException) -> str: + if isinstance(exc, RuntimeError): + return "placeholder_mismatch" + if isinstance(exc, ValueError): + return "limit_exceeded" if "at most" in str(exc).lower() else "decode_failed" + return "processor_error" + + +class Sidecar: + def __init__(self, vllm_config, renderer, client, *, namespace: str | None, concurrency: int): + from vllm import TokensPrompt, envs + from vllm.multimodal.media.connector import MEDIA_CONNECTOR_REGISTRY + from vllm.transformers_utils.processor import get_video_processor_cls_name + from vllm.v1.serial_utils import MsgpackEncoder + + model_config = vllm_config.model_config + mm_config = model_config.get_multimodal_config() + self._renderer = renderer + self._tokens_prompt = TokensPrompt + # All tensors inline: one buffer per kwargs item. + self._encoder = MsgpackEncoder(size_threshold=2**62) + self._client = client + self._concurrency = max(1, concurrency) + self._fingerprint = config_fingerprint(vllm_config) + self._keys = Keys.for_namespace(resolve_namespace(self._fingerprint, namespace)) + self._connector = MEDIA_CONNECTOR_REGISTRY.load( + envs.VLLM_MEDIA_CONNECTOR, + media_io_kwargs=mm_config.media_io_kwargs, + allowed_local_media_path=model_config.allowed_local_media_path, + allowed_media_domains=model_config.allowed_media_domains, + ) + self._video_processor = get_video_processor_cls_name(model_config) + self._schemes = advertised_schemes(model_config.allowed_local_media_path) + self._accepted = parse_scheme_list(self._schemes) + self._started_at = time.time() + if not model_config.allowed_media_domains: + logger.warning( + "no --allowed-media-domains: this sidecar will fetch media from any host" + ) + + async def run(self) -> None: + logger.info( + "media sidecar serving under %s (concurrency=%d) fingerprint=%s", + self._keys.prefix, + self._concurrency, + self._fingerprint.to_hello(), + ) + tasks = [asyncio.create_task(self._heartbeat())] + tasks += [asyncio.create_task(self._worker(i)) for i in range(self._concurrency)] + try: + await asyncio.gather(*tasks) + finally: + for task in tasks: + task.cancel() + + async def _heartbeat(self) -> None: + mapping = { + **self._fingerprint.to_hello(), + "schema": str(SCHEMA_VERSION), + "schemes": self._schemes, + "started_at": str(int(self._started_at)), + } + while True: + try: + # One transaction: a hello can never outlive its TTL. + pipe = self._client.pipeline(transaction=True) + pipe.hset(self._keys.hello, mapping=mapping) + pipe.expire(self._keys.hello, HELLO_TTL_S) + await pipe.execute() + except Exception as e: # noqa: BLE001 - keep advertising through redis blips + logger.warning("hello refresh failed: %s", e) + await asyncio.sleep(HELLO_REFRESH_S) + + async def _worker(self, index: int) -> None: + while True: + try: + popped = await self._client.brpop(self._keys.jobs, timeout=5) + except Exception as e: # noqa: BLE001 - reconnect on the next iteration + logger.warning("worker %d: brpop failed: %s", index, e) + await asyncio.sleep(1) + continue + if popped is None: + continue + _, raw = popped + try: + job = decode_job(raw) + except Exception as e: # noqa: BLE001 - a malformed job has no result key to answer on + logger.error("worker %d: undecodable job dropped: %s", index, e) + continue + try: + result = await self.handle(job) + except Exception as e: # noqa: BLE001 - one bad job must not kill the worker + logger.exception("worker %d: job %s failed", index, job.job_id) + result = failure(job.job_id, "processor_error", repr(e)) + try: + key = self._keys.result(job.job_id) + pipe = self._client.pipeline(transaction=True) + pipe.lpush(key, encode_result(result)) + pipe.expire(key, RESULT_TTL_S) + await pipe.execute() + except Exception as e: # noqa: BLE001 - the servicer times out and retries + logger.warning("worker %d: result push failed for %s: %s", index, job.job_id, e) + + async def handle(self, job: Job) -> JobResult: + started = time.time() + now_ms = int(started * 1000) + if now_ms > job.deadline_ms: + return failure(job.job_id, "expired", "job deadline passed before processing") + mismatches = self._fingerprint.mismatches(job.fingerprint) + if mismatches: + return failure(job.job_id, "fingerprint_mismatch", ",".join(mismatches)) + + multi_modal_data: dict[str, list] = {} + for index, item in enumerate(job.items): + scheme = url_scheme(item.url) + if scheme not in self._accepted: + return failure( + job.job_id, + "scheme_not_accepted", + f"item {index}: scheme '{scheme or 'none'}' not in {self._schemes}", + ) + try: + media = await self._fetch(item.modality, item.url) + except Exception as e: # noqa: BLE001 - fetch failures are client-visible codes + return failure(job.job_id, classify_fetch_error(e), f"item {index}: {e}") + multi_modal_data.setdefault(item.modality, []).append(media) + fetched = time.time() + + prompt = self._tokens_prompt( + prompt_token_ids=list(job.prompt_token_ids), multi_modal_data=multi_modal_data + ) + if job.prompt: + prompt["prompt"] = job.prompt + try: + engine_input = await self._renderer.process_for_engine_async( + prompt, arrival_time=0.0, skip_mm_cache=True + ) + except Exception as e: # noqa: BLE001 - classified into a result code + code = classify_process_error(e) + if code == "processor_error": + logger.exception("processing failed for job %s", job.job_id) + return failure(job.job_id, code, str(e)) + processed = time.time() + + encoder = self._encoder + mm_kwargs: dict[str, list[bytes]] = {} + for modality, items in engine_input["mm_kwargs"].items(): + blobs = [] + for item in items: + bufs = encoder.encode(item) + if len(bufs) != 1: + return failure(job.job_id, "processor_error", "tensor not inlined") + blobs.append(bytes(bufs[0])) + mm_kwargs[modality] = blobs + mm_placeholders = { + modality: [ + Placeholder( + offset=r.offset, + length=r.length, + is_embed=r.is_embed.tolist() if r.is_embed is not None else None, + ) + for r in ranges + ] + for modality, ranges in engine_input["mm_placeholders"].items() + } + return JobResult( + v=SCHEMA_VERSION, + job_id=job.job_id, + ok=True, + fingerprint=self._fingerprint, + prompt_token_ids=list(engine_input["prompt_token_ids"]), + mm_hashes={m: list(h) for m, h in engine_input["mm_hashes"].items()}, + mm_placeholders=mm_placeholders, + mm_kwargs=mm_kwargs, + timing=Timing( + queue_ms=max(0, now_ms - job.enqueued_ms), + fetch_ms=int((fetched - started) * 1000), + process_ms=int((processed - fetched) * 1000), + ), + ) + + async def _fetch(self, modality: str, url: str): + if modality == "image": + return await self._connector.fetch_image_async(url) + if modality == "video": + return await self._connector.fetch_video_async( + url, video_processor=self._video_processor + ) + raise ValueError(f"unsupported media modality {modality!r}") + + +async def serve(args: argparse.Namespace) -> None: + import redis.asyncio as redis_asyncio + from vllm.renderers.registry import renderer_from_config + + vllm_config = build_config(args) + renderer = renderer_from_config(vllm_config) + client = redis_asyncio.from_url(args.redis_url, decode_responses=False) + sidecar = Sidecar( + vllm_config, renderer, client, namespace=args.namespace, concurrency=args.concurrency + ) + await sidecar.run() + + +def main() -> None: + from vllm.engine.arg_utils import AsyncEngineArgs + from vllm.utils.argparse_utils import FlexibleArgumentParser + + logging.basicConfig(level=logging.INFO) + parser = FlexibleArgumentParser(description="smg media-processing sidecar for vLLM") + parser.add_argument("--redis-url", default=DEFAULT_REDIS_URL) + parser.add_argument("--namespace", default=None, help="override the derived key namespace") + parser.add_argument("--concurrency", type=int, default=2) + parser = AsyncEngineArgs.add_cli_args(parser) + args = parser.parse_args() + asyncio.run(serve(args)) + + +if __name__ == "__main__": + main() diff --git a/grpc_servicer/smg_grpc_servicer/vllm/servicer.py b/grpc_servicer/smg_grpc_servicer/vllm/servicer.py index 4fac4659f8..a9adfec4ea 100755 --- a/grpc_servicer/smg_grpc_servicer/vllm/servicer.py +++ b/grpc_servicer/smg_grpc_servicer/vllm/servicer.py @@ -249,6 +249,7 @@ async def Generate( request.tokenized.original_text or None, items, arrival_time, + request_id=request_id, ) elif has_preprocessed_mm and input_type == "tokenized": # A pixel-less payload (PD decode leg) is only decodable with diff --git a/grpc_servicer/tests/test_vllm_mm_processor.py b/grpc_servicer/tests/test_vllm_mm_processor.py index b9304fb381..869407b245 100644 --- a/grpc_servicer/tests/test_vllm_mm_processor.py +++ b/grpc_servicer/tests/test_vllm_mm_processor.py @@ -105,11 +105,6 @@ def test_non_multimodal_model_returns_none(self): env = {"SMG_VLLM_MM_PROCESSOR": "inprocess"} assert mm_processor.build_mm_processor(self._Engine(multimodal=False), env=env) is None - def test_redis_not_available_yet(self): - env = {"SMG_VLLM_MM_PROCESSOR": "redis"} - with pytest.raises(ValueError, match="not available"): - mm_processor.build_mm_processor(self._Engine(), env=env) - def test_invalid_item_cap_is_rejected_before_construction(self): env = {"SMG_VLLM_MM_PROCESSOR": "inprocess", "SMG_VLLM_MM_MAX_ITEM_BYTES": "0"} with pytest.raises(ValueError, match="SMG_VLLM_MM_MAX_ITEM_BYTES"): diff --git a/grpc_servicer/tests/test_vllm_mm_sidecar.py b/grpc_servicer/tests/test_vllm_mm_sidecar.py new file mode 100644 index 0000000000..978d1120a1 --- /dev/null +++ b/grpc_servicer/tests/test_vllm_mm_sidecar.py @@ -0,0 +1,347 @@ +"""Unit tests for the reference media sidecar loop (engine-free, fake Redis). + +Run with: pytest grpc_servicer/tests/test_vllm_mm_sidecar.py +""" + +import asyncio +import sys +import time +import types +from dataclasses import dataclass + +import pytest + +pytest.importorskip("msgspec") +from smg_grpc_servicer import mm_sidecar_protocol as proto # noqa: E402 +from smg_grpc_servicer.vllm import mm_processor, mm_sidecar # noqa: E402 + + +class FakePipeline: + def __init__(self, store): + self.store = store + self.ops = [] + + def hset(self, key, mapping=None): + self.ops.append(("hset", key, dict(mapping or {}))) + + def lpush(self, key, value): + self.ops.append(("lpush", key, value)) + + def expire(self, key, seconds): + self.ops.append(("expire", key, seconds)) + + async def execute(self): + self.store.executed.append(list(self.ops)) + if self.store.execute_fail is not None: + raise self.store.execute_fail + return [True] * len(self.ops) + + +class FakeRedis: + """Records transactions; brpop yields queued jobs then stops the loop.""" + + def __init__(self, jobs=()): + self.jobs = list(jobs) + self.executed = [] + self.execute_fail = None + + def pipeline(self, transaction=True): + assert transaction + return FakePipeline(self) + + async def brpop(self, key, timeout=0): + if self.jobs: + return (key.encode(), proto.encode_job(self.jobs.pop(0))) + raise asyncio.CancelledError + + +def fingerprint(): + return proto.Fingerprint( + model="m", + vllm_version="0.27.1", + dtype="torch.bfloat16", + video_backend="opencv", + media_io_kwargs="{}", + mm_processor_kwargs="{}", + limit_per_prompt="{}", + ) + + +# Shaped like vLLM's validated limit_per_prompt values (pydantic dataclasses). +@dataclass +class _BaseOptions: + count: int = 999 + + +@dataclass +class _ImageOptions(_BaseOptions): + width: int | None = None + height: int | None = None + + +@dataclass +class _VideoOptions(_BaseOptions): + num_frames: int | None = None + + +class _MmConfig: + def __init__(self, limit_per_prompt): + self.limit_per_prompt = limit_per_prompt + self.media_io_kwargs = {"video": {"num_frames": 8}} + self.mm_processor_kwargs = None + + def get_limit_per_prompt(self, modality): + options = self.limit_per_prompt.get(modality) + return 999 if options is None else options.count + + +class _ModelConfig: + def __init__(self, limit_per_prompt): + self.model = "m" + self.dtype = "torch.bfloat16" + self._mm = _MmConfig(limit_per_prompt) + + def get_multimodal_config(self): + return self._mm + + +def _fake_vllm(monkeypatch): + fake = types.ModuleType("vllm") + fake.__version__ = "0.27.1" + fake.envs = types.SimpleNamespace(VLLM_VIDEO_LOADER_BACKEND="opencv") + monkeypatch.setitem(sys.modules, "vllm", fake) + + +class TestFingerprintDerivation: + def test_worker_and_sidecar_derive_the_same_fingerprint(self, monkeypatch): + _fake_vllm(monkeypatch) + limits = {"image": _ImageOptions(count=8)} + engine = types.SimpleNamespace(model_config=_ModelConfig(limits)) + vllm_config = types.SimpleNamespace(model_config=_ModelConfig(limits)) + worker = mm_processor.engine_fingerprint(engine) + assert worker == mm_sidecar.config_fingerprint(vllm_config) + assert worker.limit_per_prompt == '{"*": {"count": 999}, "image": {"count": 8}}' + + def test_equivalent_limit_spellings_share_a_namespace(self, monkeypatch): + _fake_vllm(monkeypatch) + implicit = mm_processor.fingerprint_from_model_config(_ModelConfig({})) + for spelled_out in ( + {"image": _ImageOptions(count=999)}, + {"point_cloud": _BaseOptions(count=999)}, + {"video": _VideoOptions(count=999), "audio": _BaseOptions()}, + ): + assert mm_processor.fingerprint_from_model_config(_ModelConfig(spelled_out)) == implicit + for skewed in ( + {"image": _ImageOptions(count=8)}, + {"image": _ImageOptions(count=999, width=4)}, + {"point_cloud": _BaseOptions(count=2)}, + ): + fp = mm_processor.fingerprint_from_model_config(_ModelConfig(skewed)) + assert fp.namespace() != implicit.namespace(), skewed + + def test_sibling_options_and_extra_modalities_stay_in_the_fingerprint(self): + limits = { + "video": _VideoOptions(count=2, num_frames=16), + "image": _ImageOptions(count=8, width=4), + "point_cloud": _BaseOptions(count=1), + } + assert mm_processor.resolved_mm_limits(_MmConfig(limits)) == { + "*": {"count": 999}, + "image": {"count": 8, "width": 4}, + "point_cloud": {"count": 1}, + "video": {"count": 2, "num_frames": 16}, + } + + def test_unknown_limit_shapes_are_loud_not_lossy(self): + class _Model: + def __init__(self): + self.count = 2 + self.num_frames = 16 + + def model_dump(self): + return {"count": self.count, "num_frames": self.num_frames} + + assert mm_processor._limit_options(_Model()) == {"count": 2, "num_frames": 16} + plain = types.SimpleNamespace(count=2, num_frames=16, width=None) + assert mm_processor._limit_options(plain) == {"count": 2, "num_frames": 16} + assert mm_processor._limit_options(7) == {"count": 7} + assert mm_processor._limit_options({"count": 3, "length": 9}) == {"count": 3, "length": 9} + with pytest.raises(TypeError, match="unsupported limit_per_prompt value"): + mm_processor._limit_options(object()) + + def test_pydantic_dataclass_options_resolve_like_vllm(self): + # vLLM's *DummyOptions are pydantic dataclasses; mirror the decorator so the + # is_dataclass/asdict path runs without vLLM installed. + pydantic = pytest.importorskip("pydantic") + from pydantic.dataclasses import dataclass as pydantic_dataclass + + @pydantic_dataclass + class _Base: + count: int = pydantic.Field(999, ge=0) + + @pydantic_dataclass(config=pydantic.ConfigDict(extra="forbid")) + class _Video(_Base): + num_frames: int | None = pydantic.Field(None, gt=0) + width: int | None = pydantic.Field(None, gt=0) + + cfg = _MmConfig({"video": _Video(count=2, num_frames=16), "image": _Base(count=999)}) + assert mm_processor.resolved_mm_limits(cfg) == { + "*": {"count": 999}, + "video": {"count": 2, "num_frames": 16}, + } + + def test_resolved_limits_match_vllm(self): + pytest.importorskip("vllm") + from vllm.config import MultiModalConfig + + # The derivation drops entries equal to one probed unset default, which + # is only right while that default is uniform across the options classes + # and an unknown key resolves to it instead of raising. + cfg = MultiModalConfig() + modalities = ("image", "video", "audio") + assert {m: cfg.get_limit_per_prompt(m) for m in modalities} == dict.fromkeys( + modalities, 999 + ) + assert cfg.get_limit_per_prompt(mm_processor._UNSET_MODALITY) == 999 + unset = mm_processor.resolved_mm_limits(cfg) + assert unset == {"*": {"count": 999}} + for modality in modalities: + for spelled_out in (999, {}): + explicit = MultiModalConfig(limit_per_prompt={modality: spelled_out}) + assert mm_processor.resolved_mm_limits(explicit) == unset, (modality, spelled_out) + resolved = mm_processor.resolved_mm_limits( + MultiModalConfig(limit_per_prompt={"image": 8, "video": {"count": 2, "num_frames": 16}}) + ) + assert resolved["image"] == {"count": 8} + assert resolved["video"] == {"count": 2, "num_frames": 16} + + +def sidecar(client): + s = mm_sidecar.Sidecar.__new__(mm_sidecar.Sidecar) + s._client = client + s._fingerprint = fingerprint() + s._keys = proto.Keys.for_namespace("t") + s._schemes = "http,https,data" + s._accepted = {"http", "https", "data"} + s._started_at = time.time() + s._concurrency = 1 + return s + + +def job(job_id="j1"): + return proto.Job( + v=proto.SCHEMA_VERSION, + job_id=job_id, + request_id="r1", + fingerprint=fingerprint(), + prompt_token_ids=[1, 2, 3], + prompt=None, + items=[proto.JobItem(modality="image", url="https://a/1.png")], + enqueued_ms=0, + deadline_ms=int(time.time() * 1000) + 60_000, + ) + + +def run(coro): + return asyncio.new_event_loop().run_until_complete(coro) + + +class TestClassification: + def test_fetch_errors(self): + assert mm_sidecar.classify_fetch_error(ValueError("domain not allowed")) == ( + "domain_not_allowed" + ) + assert mm_sidecar.classify_fetch_error(ValueError("bad base64")) == "fetch_failed" + assert mm_sidecar.classify_fetch_error(TimeoutError("slow")) == "fetch_failed" + + def test_process_errors(self): + assert mm_sidecar.classify_process_error(RuntimeError("placeholder")) == ( + "placeholder_mismatch" + ) + # vLLM 0.27 wording: "At most 1 image(s) may be provided in one prompt." + assert ( + mm_sidecar.classify_process_error( + ValueError("At most 1 image(s) may be provided in one prompt.") + ) + == "limit_exceeded" + ) + assert mm_sidecar.classify_process_error(ValueError("cannot identify image")) == ( + "decode_failed" + ) + assert mm_sidecar.classify_process_error(KeyError("mm_kwargs")) == "processor_error" + + +class TestWorkerLoop: + def test_a_failing_handle_answers_and_keeps_the_loop_alive(self, monkeypatch): + client = FakeRedis(jobs=[job("j1"), job("j2")]) + s = sidecar(client) + calls = [] + + async def handle(j): + calls.append(j.job_id) + if j.job_id == "j1": + raise KeyError("mm_kwargs") + return proto.JobResult(v=1, job_id=j.job_id, ok=True) + + monkeypatch.setattr(s, "handle", handle) + with pytest.raises(asyncio.CancelledError): + run(s._worker(0)) + assert calls == ["j1", "j2"], "the second job was still served" + first, second = client.executed + assert first[0][0] == "lpush" and first[0][1] == s._keys.result("j1") + answered = proto.decode_result(first[0][2]) + assert not answered.ok and answered.code == "processor_error" + assert first[1] == ("expire", s._keys.result("j1"), proto.RESULT_TTL_S) + assert proto.decode_result(second[0][2]).ok + + def test_result_push_is_one_transaction(self, monkeypatch): + client = FakeRedis(jobs=[job("j3")]) + s = sidecar(client) + + async def handle(j): + return proto.JobResult(v=1, job_id=j.job_id, ok=True) + + monkeypatch.setattr(s, "handle", handle) + with pytest.raises(asyncio.CancelledError): + run(s._worker(0)) + (ops,) = client.executed + assert [op[0] for op in ops] == ["lpush", "expire"] + + +class TestHeartbeat: + def test_hello_and_ttl_are_one_transaction(self, monkeypatch): + client = FakeRedis() + s = sidecar(client) + monkeypatch.setattr(mm_sidecar, "HELLO_REFRESH_S", 0) + + async def stop_after_two(): + task = asyncio.ensure_future(s._heartbeat()) + while len(client.executed) < 2: + await asyncio.sleep(0) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + run(stop_after_two()) + ops = client.executed[0] + assert ops[0][0] == "hset" and ops[0][1] == s._keys.hello + assert ops[0][2]["schemes"] == "http,https,data" + assert ops[0][2]["model"] == "m" + assert ops[1] == ("expire", s._keys.hello, proto.HELLO_TTL_S) + + def test_heartbeat_survives_a_failed_refresh(self, monkeypatch): + client = FakeRedis() + client.execute_fail = ConnectionError("redis down") + s = sidecar(client) + monkeypatch.setattr(mm_sidecar, "HELLO_REFRESH_S", 0) + + async def stop_after_two(): + task = asyncio.ensure_future(s._heartbeat()) + while len(client.executed) < 2: + await asyncio.sleep(0) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + run(stop_after_two()) + assert len(client.executed) >= 2 diff --git a/grpc_servicer/tests/test_vllm_mm_sidecar_protocol.py b/grpc_servicer/tests/test_vllm_mm_sidecar_protocol.py new file mode 100644 index 0000000000..738718905c --- /dev/null +++ b/grpc_servicer/tests/test_vllm_mm_sidecar_protocol.py @@ -0,0 +1,115 @@ +"""Unit tests for the Redis media-sidecar protocol (engine-free, needs msgspec). + +Run with: pytest grpc_servicer/tests/test_vllm_mm_sidecar_protocol.py +""" + +import pytest + +pytest.importorskip("msgspec") +from smg_grpc_servicer import mm_sidecar_protocol as proto # noqa: E402 + + +def fingerprint(**overrides) -> proto.Fingerprint: + values = { + "model": "Qwen/Qwen3-VL-8B-Instruct", + "vllm_version": "0.27.1", + "dtype": "torch.bfloat16", + "video_backend": "opencv", + "media_io_kwargs": "{}", + "mm_processor_kwargs": "{}", + "limit_per_prompt": "{}", + } + values.update(overrides) + return proto.Fingerprint(**values) + + +class TestFingerprint: + def test_namespace_covers_every_fingerprint_field(self): + a = fingerprint().namespace() + assert a == fingerprint().namespace() + assert len(a) == 16 + # Fleets that would fail the fingerprint check never share keys. + assert a != fingerprint(dtype="torch.float16").namespace() + assert a != fingerprint(video_backend="decord").namespace() + assert a != fingerprint(media_io_kwargs='{"video":{"num_frames":8}}').namespace() + assert a != fingerprint(mm_processor_kwargs='{"max_pixels":1}').namespace() + assert a != fingerprint(limit_per_prompt='{"image":8}').namespace() + + def test_override_wins(self): + assert proto.resolve_namespace(fingerprint(), " prod-a ") == "prod-a" + assert proto.resolve_namespace(fingerprint(), "") == fingerprint().namespace() + + def test_mismatches_name_fields(self): + assert fingerprint().mismatches(fingerprint()) == [] + assert fingerprint().mismatches(fingerprint(vllm_version="0.28.0", dtype="x")) == [ + "vllm_version", + "dtype", + ] + + def test_hello_roundtrip_with_bytes_mapping(self): + hello = {k.encode(): v.encode() for k, v in fingerprint().to_hello().items()} + hello[b"schemes"] = b"http,https,data" + hello[b"schema"] = b"1" + assert proto.Fingerprint.from_hello(hello) == fingerprint() + assert proto.hello_schemes(hello) == "http,https,data" + + def test_hello_missing_field_is_none(self): + hello = {b"model": b"m"} + assert proto.Fingerprint.from_hello(hello) is None + assert proto.hello_schemes(hello) == "" + + +class TestKeys: + def test_key_layout(self): + keys = proto.Keys.for_namespace("abc") + assert keys.hello == "smg:mm:v1:abc:hello" + assert keys.jobs == "smg:mm:v1:abc:jobs" + assert keys.result("j1") == "smg:mm:v1:abc:res:j1" + + +class TestMessages: + def test_job_roundtrip(self): + job = proto.Job( + v=proto.SCHEMA_VERSION, + job_id="j1", + request_id="r1", + fingerprint=fingerprint(), + prompt_token_ids=[1, 2, 3], + prompt="hi <|image_pad|>", + items=[proto.JobItem(modality="image", url="https://a/1.png")], + enqueued_ms=100, + deadline_ms=30100, + ) + assert proto.decode_job(proto.encode_job(job)) == job + + def test_result_roundtrip_keeps_blobs_and_masks(self): + result = proto.JobResult( + v=proto.SCHEMA_VERSION, + job_id="j1", + ok=True, + fingerprint=fingerprint(), + prompt_token_ids=[1, 5, 5, 5, 3], + mm_hashes={"image": ["abc"]}, + mm_placeholders={ + "image": [proto.Placeholder(offset=1, length=3, is_embed=[True, False, True])] + }, + mm_kwargs={"image": [b"\x93\x01\x02\x03"]}, + timing=proto.Timing(queue_ms=1, fetch_ms=2, process_ms=3), + cache="miss", + ) + parsed = proto.decode_result(proto.encode_result(result)) + assert parsed == result + assert parsed.mm_kwargs["image"][0] == b"\x93\x01\x02\x03" + assert parsed.mm_placeholders["image"][0].is_embed == [True, False, True] + + def test_failure_defaults(self): + result = proto.failure("j2", "expired", "too late") + assert not result.ok + assert result.code == "expired" + assert result.mm_kwargs == {} + assert proto.decode_result(proto.encode_result(result)) == result + + def test_error_code_sets_are_disjoint(self): + assert not (proto.CLIENT_ERROR_CODES & proto.RETRYABLE_ERROR_CODES) + assert "placeholder_mismatch" in proto.CLIENT_ERROR_CODES + assert "fingerprint_mismatch" in proto.RETRYABLE_ERROR_CODES diff --git a/grpc_servicer/tests/test_vllm_redis_media_processor.py b/grpc_servicer/tests/test_vllm_redis_media_processor.py new file mode 100644 index 0000000000..9a82e94ca4 --- /dev/null +++ b/grpc_servicer/tests/test_vllm_redis_media_processor.py @@ -0,0 +1,277 @@ +"""Unit tests for RedisMediaProcessor transport semantics (engine-free, fake Redis). + +Run with: pytest grpc_servicer/tests/test_vllm_redis_media_processor.py +""" + +import asyncio +import importlib.util +import sys +from dataclasses import dataclass +from pathlib import Path + +import pytest + +pytest.importorskip("msgspec") +from smg_grpc_servicer import mm_sidecar_protocol as proto # noqa: E402 + +# Import the module directly to avoid pulling vllm via the package __init__ +_MODULE_PATH = Path(__file__).parents[1] / "smg_grpc_servicer" / "vllm" / "mm_processor.py" +_spec = importlib.util.spec_from_file_location("mm_processor", _MODULE_PATH) +mm_processor = importlib.util.module_from_spec(_spec) +sys.modules[_spec.name] = mm_processor +_spec.loader.exec_module(mm_processor) + + +@dataclass +class _Item: + modality: str + url: str + + +class _ModelConfig: + dtype = "bf16" + is_multimodal_model = True + allowed_local_media_path = "" + + +class _Engine: + model_config = _ModelConfig() + + +class FakeRedis: + """Just enough of redis.asyncio for the servicer side of the protocol.""" + + def __init__(self, *, hello=None, queue_depth=0, responder=None, fail=None): + self.hello = hello or {} + self.queue_depth = queue_depth + self.responder = responder + self.fail = fail + self.pushed = [] + self.expired = {} + + async def hgetall(self, key): + if self.fail: + raise self.fail + return dict(self.hello) + + async def llen(self, key): + if self.fail: + raise self.fail + return self.queue_depth + + async def lpush(self, key, value): + self.pushed.append((key, value)) + + async def brpop(self, key, timeout=0): + if self.responder is None: + return None + job = proto.decode_job(self.pushed[-1][1]) + result = self.responder(job) + payload = result if isinstance(result, bytes) else proto.encode_result(result) + return (key.encode(), payload) + + async def expire(self, key, seconds): + self.expired[key] = seconds + + +def fingerprint(**overrides): + values = { + "model": "m", + "vllm_version": "0.27.1", + "dtype": "torch.bfloat16", + "video_backend": "opencv", + "media_io_kwargs": "{}", + "mm_processor_kwargs": "{}", + "limit_per_prompt": "{}", + } + values.update(overrides) + return proto.Fingerprint(**values) + + +def processor(client, **kwargs): + return mm_processor.RedisMediaProcessor( + _Engine(), fingerprint(), client=client, timeout_ms=50, max_queue=4, **kwargs + ) + + +def run(coro): + return asyncio.new_event_loop().run_until_complete(coro) + + +def ok_result(job, **extra): + return proto.JobResult( + v=extra.pop("v", proto.SCHEMA_VERSION), + job_id=job.job_id, + ok=True, + fingerprint=extra.pop("fingerprint", fingerprint()), + prompt_token_ids=[1, 5, 5, 3], + mm_hashes={"image": ["h1"]}, + mm_placeholders={"image": [proto.Placeholder(offset=1, length=2)]}, + mm_kwargs={"image": [b"\x80"]}, + **extra, + ) + + +class TestProbe: + def test_no_hello_is_not_advertised(self): + assert run(processor(FakeRedis()).probe()) is False + + def test_redis_error_is_not_advertised(self): + assert run(processor(FakeRedis(fail=ConnectionError("down"))).probe()) is False + + def test_fingerprint_mismatch_is_not_advertised(self): + hello = {k.encode(): v.encode() for k, v in fingerprint(dtype="x").to_hello().items()} + hello[b"schema"] = str(proto.SCHEMA_VERSION).encode() + assert run(processor(FakeRedis(hello=hello)).probe()) is False + + def test_protocol_schema_skew_is_not_advertised(self): + hello = {k.encode(): v.encode() for k, v in fingerprint().to_hello().items()} + hello[b"schema"] = str(proto.SCHEMA_VERSION + 1).encode() + assert run(processor(FakeRedis(hello=hello)).probe()) is False + + def test_hello_without_schema_is_not_advertised(self): + hello = {k.encode(): v.encode() for k, v in fingerprint().to_hello().items()} + assert run(processor(FakeRedis(hello=hello)).probe()) is False + + def test_matching_hello_advertises_and_adopts_schemes(self): + hello = {k.encode(): v.encode() for k, v in fingerprint().to_hello().items()} + hello[b"schema"] = str(proto.SCHEMA_VERSION).encode() + hello[b"schemes"] = b"http,https,data,file" + p = processor(FakeRedis(hello=hello)) + assert run(p.probe()) is True + assert p.schemes == "http,https,data,file" + assert "file" in p.accepted_schemes + + +class TestSubmitAndWait: + def job(self, job_id="j1"): + return proto.Job( + v=proto.SCHEMA_VERSION, + job_id=job_id, + request_id="r1", + fingerprint=fingerprint(), + prompt_token_ids=[1, 2, 3], + prompt=None, + items=[proto.JobItem(modality="image", url="https://a/1.png")], + enqueued_ms=0, + deadline_ms=50, + ) + + def test_queue_over_cap_fails_fast(self): + client = FakeRedis(queue_depth=4) + with pytest.raises(mm_processor.MmProcessorUnavailable, match="sidecar_overloaded"): + run(processor(client)._submit_and_wait(self.job())) + assert client.pushed == [] + + def test_timeout_is_retryable(self): + client = FakeRedis() + with pytest.raises(mm_processor.MmProcessorUnavailable, match="sidecar_timeout"): + run(processor(client)._submit_and_wait(self.job())) + key, raw = client.pushed[0] + assert key.endswith(":jobs") + assert proto.decode_job(raw).job_id == "j1" + + def test_transport_error_is_retryable(self): + client = FakeRedis(fail=ConnectionError("refused")) + with pytest.raises(mm_processor.MmProcessorUnavailable, match="sidecar_unavailable"): + run(processor(client)._submit_and_wait(self.job())) + + def test_client_error_codes_become_value_errors(self): + client = FakeRedis( + responder=lambda job: proto.failure(job.job_id, "domain_not_allowed", "x") + ) + with pytest.raises(ValueError, match="domain_not_allowed"): + run(processor(client)._submit_and_wait(self.job())) + + def test_retryable_error_codes_stay_unavailable(self): + client = FakeRedis(responder=lambda job: proto.failure(job.job_id, "expired", "late")) + with pytest.raises(mm_processor.MmProcessorUnavailable, match="expired"): + run(processor(client)._submit_and_wait(self.job())) + + def test_undecodable_result_is_retryable(self): + client = FakeRedis(responder=lambda job: b"\xc1not-a-result") + with pytest.raises(mm_processor.MmProcessorUnavailable, match="undecodable result"): + run(processor(client)._submit_and_wait(self.job())) + + def test_result_schema_version_mismatch_is_retryable(self): + client = FakeRedis(responder=lambda job: ok_result(job, v=proto.SCHEMA_VERSION + 1)) + with pytest.raises(mm_processor.MmProcessorUnavailable, match="result schema v2"): + run(processor(client)._submit_and_wait(self.job())) + + def test_result_fingerprint_mismatch_is_unavailable(self): + client = FakeRedis( + responder=lambda job: ok_result(job, fingerprint=fingerprint(vllm_version="0.1")) + ) + with pytest.raises(mm_processor.MmProcessorUnavailable, match="fingerprint_mismatch"): + run(processor(client)._submit_and_wait(self.job())) + + def test_success_returns_full_result(self): + client = FakeRedis(responder=ok_result) + result = run(processor(client)._submit_and_wait(self.job())) + assert result.ok + assert result.prompt_token_ids == [1, 5, 5, 3] + assert result.mm_kwargs["image"] == [b"\x80"] + + +class TestProcess: + def test_process_mints_a_fresh_job_per_attempt(self, monkeypatch): + seen = [] + + def responder(job): + seen.append(job) + return ok_result(job) + + client = FakeRedis(responder=responder) + p = processor(client) + monkeypatch.setattr(p, "_rebuild", lambda result, text, arrival: ("built", result.job_id)) + items = [_Item("image", "https://a/1.png")] + first = run(p.process([1, 2, 3], "text", items, 1.0, request_id="req")) + second = run(p.process([1, 2, 3], "text", items, 2.0, request_id="req")) + assert first[0] == "built" and second[0] == "built" + assert first[1] != second[1], "each attempt gets its own job id" + assert [job.request_id for job in seen] == ["req", "req"] + assert seen[0].items[0].url == "https://a/1.png" + assert seen[0].deadline_ms - seen[0].enqueued_ms == 50 + + def test_process_caps_inline_payloads_before_queueing(self): + client = FakeRedis(responder=ok_result) + p = processor(client, max_item_bytes=4) + items = [_Item("image", "data:image/png;base64,AAAAAAAAAAAA")] + with pytest.raises(ValueError, match="above the 4-byte cap"): + run(p.process([1], None, items, 0.0)) + assert client.pushed == [] + + +class TestItemCap: + def test_process_caps_item_count_before_queueing(self): + client = FakeRedis(responder=ok_result) + p = processor(client, max_items=1) + items = [_Item("image", "https://a/1.png"), _Item("image", "https://a/2.png")] + with pytest.raises(ValueError, match="SMG_VLLM_MM_MAX_ITEMS"): + run(p.process([1, 2, 3], None, items, 0.0)) + assert client.pushed == [] + + +class TestRedisClientImport: + def test_missing_client_names_the_extra(self, monkeypatch): + monkeypatch.setitem(sys.modules, "redis", None) + monkeypatch.setitem(sys.modules, "redis.asyncio", None) + with pytest.raises(ValueError, match=r"vllm,vllm-redis"): + mm_processor._redis_client("redis://127.0.0.1:6379/0") + + +class TestBuild: + def test_redis_mode_builds_processor(self, monkeypatch): + monkeypatch.setattr(mm_processor, "engine_fingerprint", lambda engine: fingerprint()) + monkeypatch.setattr(mm_processor, "_redis_client", lambda url: FakeRedis()) + env = { + "SMG_VLLM_MM_PROCESSOR": "redis", + "SMG_VLLM_MM_SIDECAR_TIMEOUT_MS": "1000", + "SMG_VLLM_MM_SIDECAR_MAX_QUEUE": "8", + "SMG_VLLM_MM_SIDECAR_NAMESPACE": "ns-1", + } + p = mm_processor.build_mm_processor(_Engine(), env=env) + assert p.name == "redis" + assert p._timeout_ms == 1000 + assert p._max_queue == 8 + assert p._keys.prefix == "smg:mm:v1:ns-1"