diff --git a/.github/ci_model_cache.yaml b/.github/ci_model_cache.yaml index 1cad1bf0f..050f84faf 100644 --- a/.github/ci_model_cache.yaml +++ b/.github/ci_model_cache.yaml @@ -2,7 +2,7 @@ # # Adding/removing/changing an entry here changes the cache key and will trigger # a fresh download + cache save on the next push to main. Total uncompressed -# size of all entries must stay under 8 GiB (GHA cache limit is 10 GiB). +# size of all entries must stay under 10 GiB (GHA cache hard limit). models: - repo: ibm-ai-platform/micro-g3.3-8b-instruct-1b @@ -13,3 +13,8 @@ models: revision: cf74d8acd4f198de950bf004b262e6accfed5d2c - repo: cross-encoder/stsb-roberta-large revision: 2b12c2c0088918e76151fd5937b7bba986ef1f98 + # Random-init CI fixture (~10 MB) that stands in for granite-vision-3.2-2b + # in the async-MM-encoder e2e tests. See tests/e2e/conftest.py — the + # revision must match NANO_GV_REVISION there. + - repo: joerunde/nano-gv + revision: c9470d9e54b023dd9ab8a8a98057489fdb18ba03 diff --git a/docs/contributing/parallel_vision_encode_plan.md b/docs/contributing/parallel_vision_encode_plan.md new file mode 100644 index 000000000..c8ca39c10 --- /dev/null +++ b/docs/contributing/parallel_vision_encode_plan.md @@ -0,0 +1,107 @@ +# Parallel Vision Encoder Execution In Single Instance vLLM + +## Background + +Multimodal models on Spyre compute the vision encoder on CPU (rank 0 only) and broadcast embeddings to other ranks via POSIX shared memory. Today this encoding runs serially, once per request, at the start of that request's first prefill step. MM encoding is expensive operation and in current implementation its blocking, so no other operation like prefill and decode of other requests can run in parallel affecting overall performance. + +## Goal + +Overlap CPU / NNPA vision encoding with AIU prefill/decode by running the encoder in a separate subprocess. Embeddings are written to POSIX shared memory and all TP workers read them independently — no rank-0 broadcast of large tensors. The scheduler gates MM request prefill on encoding readiness, so a request only enters prefill once its embedding is available. + +## Evolution Path + +**Phase 1 and 2:** Combined in current implementation. Vision encoding runs in a dedicated non-daemon subprocess (`mm-encoder`) managed by `SpyreMultiprocExecutor`. The encoder subprocess loads only the vision model via `get_model(..., vision_only=True)`. +The scheduler submits MM requests for encoding on every step and gates prefill on encoding. + +**Phase 3 (future):** Enable vision encoder batching within the encoder subprocess. This will further improve the performance by handling all pending MM requests in single batch. This requires FMS changes to stack same-resolution images instead of +concatenating. See [Phase 3](#phase-3-add-vision-encoder-batching) below. + +### Current Flow + +```text +Scheduler picks 1 MM request + → execute_model(): + encode(request) # CPU, rank 0, single request + broadcast embeddings # SHM + dist.broadcast + prefill chunk on Spyre + (repeat for each chunk) + → decode steps +``` + +### Implemented Flow + +```text +Encoder subprocess starts AFTER warmup completes + (SpyreMultiprocExecutor hooks on collective_rpc("compile_or_warm_up_model")) + +Scheduler emits unsubmitted waiting MM requests on EVERY step (prefill and decode). +Scheduler gates MM prefill on _mm_encoding_ready + (only applies when SENDNN_INFERENCE_ASYNC_MM_ENCODER=1). + +SpyreMultiprocExecutor.execute_model() on every step: + 1. Submit new _spyre_mm_encode_requests → job_queue # non-blocking put_nowait + 2. Drain result_queue (non-blocking) # collect completed encodings + if results: + collective_rpc("store_mm_embeddings") # all TP workers read SHM + cleanup SHM blocks + set scheduler_output._spyre_newly_encoded_req_ids + 3. super().execute_model() → workers run AIU forward # concurrent with encoder subprocess encoder process runs in parallel + +scheduler.update_from_output(): + _mm_encoding_ready.update(_spyre_newly_encoded_req_ids) + +Next schedule() call: request now in _mm_encoding_ready → scheduled for prefill + add_new_request(): cached_mm_embeddings = pending_mm_embeddings.pop(req_id) + _prepare_chunked_prefill(): uses cached embeddings, skips inline encoding +``` + +--- + +## Changes Summary + +| File | Change | +|---|---| +| `sendnn_inference/platform.py` | Register `SpyreMultiprocExecutor` when `SENDNN_INFERENCE_ASYNC_MM_ENCODER=1` and TP > 1 | +| `sendnn_inference/v1/executor/spyre_executor.py` | `SpyreMultiprocExecutor`: override `execute_model` to submit encode jobs, collect results, call `store_mm_embeddings` on workers | +| `sendnn_inference/v1/worker/mm_encoder_process.py` | `VisionEncoderRunner` + `encoder_process_main`: load vision-only model, serve encode jobs, write embeddings to SHM | +| `sendnn_inference/v1/worker/spyre_worker.py` | Add `store_mm_embeddings` — delegates to model runner | +| `sendnn_inference/v1/worker/spyre_model_runner.py` | Add `pending_mm_embeddings` dict, `store_mm_embeddings` (reads from SHM), `_compute_and_cache_mm_embeddings` as inline fallback for warmup; consume in `add_new_request` | +| `sendnn_inference/v1/core/scheduler.py` | Add `MMEncodeRequest` dataclass; emit encode jobs every step; track `_mm_encoding_submitted` / `_mm_encoding_ready`; gate MM prefill on encoding readiness (async mode only); update state in `update_from_output` and `finish_requests` | +| `sendnn_inference/model_executor/model_loader/spyre.py` | Extract `cast_params_for_spyre` as module-level function reusable by encoder subprocess | +| `sendnn_inference/envs.py` | Add `SENDNN_INFERENCE_ASYNC_MM_ENCODER` env var (default 0) | + +Non-MM requests, the warmup path, chunked prefill logic, and TP broadcast are unaffected. + +## Alternatives considered + +### Threading (abandoned) + +**What we tried:** Start a `threading.Thread` in the worker model runner. The thread uses the already-loaded `fms_model` directly (no copy) and encodes waiting MM requests in the background while the AIU runs. + +**Why it failed:** Spyre operations and vision encoding both are blocking operations. The background thread cannot make any progress during AIU execution. Encoding only runs in tiny Python gaps between AIU calls and prefill / decode gets impacted by encoding operations. + +**Verdict:** No benefit. Reverted. + +### Subprocess from worker (abandoned) + +**What we tried:** Start a `multiprocessing.Process` from inside the worker's `load_model` or +`complete_warmup`. + +**Why it failed:** vLLM spawns worker processes as **daemon processes** +(`multiprocessing.Process(daemon=True)`). Python forbids daemon processes from spawning children (`AssertionError: daemonic processes are not allowed to have children`). + +**Verdict:** Architecturally impossible from a worker process. + +### Subprocess from MultiprocExecutor (**implemented**) + +**The idea:** vLLM's `MultiprocExecutor` runs in the **main (non-daemon) process**. Any process it spawns is also non-daemon. By subclassing `MultiprocExecutor` as `SpyreMultiprocExecutor`, we can start the encoder process at the executor level. + +**Model weight loading:** FMS now supports `get_model(..., vision_only=True)`, which loads only vision tower + projector + text embedding from the checkpoint, skipping the LLM decoder. The encoder subprocess calls this directly. + +**SHM-based result delivery:** The encoder process writes completed embeddings to POSIX SHM and puts only `(req_id, shape, dtype)` metadata on the result queue (no large tensors in the queue). The executor calls `collective_rpc("store_mm_embeddings", metadata)` so all TP workers read from SHM independently — no rank-0 to others tensor broadcast. + +**Scheduler-level encoding readiness gate:** The scheduler tracks `_mm_encoding_submitted` and `_mm_encoding_ready` sets. MM requests are only eligible for prefill when their encoding is confirmed complete. Text-only requests are completely unaffected. The scheduler submits encoding jobs on every step (prefill AND decode) so the encoder stays ahead of the prefill queue. + +## Phase 3: Add Vision Encoder Batching + +For N same-resolution images, the vision transformer can runs once on `[N, P, D]` instead of N times on `[1, P, D]`. CPU / NNPA matmul efficiently scales with batch size, so the single batched call should be significantly faster than N sequential calls — particularly for large images where the `P²` self-attention dominates. diff --git a/sendnn_inference/envs.py b/sendnn_inference/envs.py index ad6415a2d..0cbcde5c3 100644 --- a/sendnn_inference/envs.py +++ b/sendnn_inference/envs.py @@ -26,6 +26,7 @@ SENDNN_INFERENCE_MODEL_CONFIG_FILE: str | None = None SENDNN_INFERENCE_CPU_MM_DTYPE: torch.dtype = torch.float16 SENDNN_INFERENCE_MM_DEVICE: str = "auto" + SENDNN_INFERENCE_ASYNC_MM_ENCODER: bool = True SENDNN_INFERENCE_TP_MM_SHARING: bool = True SENDNN_INFERENCE_LONG_OUT_PRIO: bool = False SENDNN_INFERENCE_PAUSING_ENABLED: bool = True @@ -175,6 +176,21 @@ def clear_env_cache(): "SENDNN_INFERENCE_MM_DEVICE": lambda: parse_mm_device( os.getenv("SENDNN_INFERENCE_MM_DEVICE", "auto") ), + # Enable the async vision encoder subprocess. + # When set to 1, SpyreMultiprocExecutor spawns a separate process that loads + # only the vision model via get_model(..., vision_only=True) and pre-encodes + # MM requests in parallel with AIU prefill/decode. The scheduler gates MM + # request prefill on encoding readiness so a request only starts prefill once + # its embedding is available. Only effective for decoder models with TP > 1. + # Defaults to 0 (disabled) — uses the Phase 1 blocking encode path. + "SENDNN_INFERENCE_ASYNC_MM_ENCODER": lambda: bool( + int( + os.getenv( + "SENDNN_INFERENCE_ASYNC_MM_ENCODER", + "0" if platform.machine() == "ppc64le" else "1", + ) + ) + ), # When "1" (default), rank 0 runs the vision encoder and shares the result # with other TP ranks via POSIX shared memory (one encoder call instead of # world_size calls). Set to "0" to fall back to every TP rank running the diff --git a/sendnn_inference/model_executor/model_loader/spyre.py b/sendnn_inference/model_executor/model_loader/spyre.py index 3d1cde74a..0c0897e4e 100644 --- a/sendnn_inference/model_executor/model_loader/spyre.py +++ b/sendnn_inference/model_executor/model_loader/spyre.py @@ -44,6 +44,67 @@ class SpyreAttentionMetadata: is_prefill: bool +def cast_params_for_spyre( + fms_model: nn.Module, + mm_parameter_prefixes: tuple[str, ...], + is_fp8_model: bool = False, +) -> str: + """Cast model params for Spyre execution and return the resolved mm_device. + + Places MM submodules (vision_tower / multi_modal_projector) onto + SENDNN_INFERENCE_MM_DEVICE with SENDNN_INFERENCE_CPU_MM_DTYPE, and casts + all other submodules to fp16 for Spyre. For FP8 models only bf16 + params/buffers are converted to fp16 — fp8 weights and fp32 scales are + left untouched. + + Callable from both SpyreCausalLM._cast_params_for_spyre (full model) and + VisionEncoderRunner (vision-only encoder subprocess). + """ + cpu_mm_dtype = envs_spyre.SENDNN_INFERENCE_CPU_MM_DTYPE + mm_device = envs_spyre.SENDNN_INFERENCE_MM_DEVICE + mm_module_names = {p.rstrip(".") for p in mm_parameter_prefixes} + + if mm_device == "nnpa" and mm_module_names and not utils_spyre.ensure_nnpa_registered(): + raise RuntimeError( + "SENDNN_INFERENCE_MM_DEVICE resolved to nnpa (torch_nnpa is installed) " + "but the nnpa device could not be initialized. Refusing to fall back to " + "CPU; a broken nnpa backend must not be silently masked. Set " + "SENDNN_INFERENCE_MM_DEVICE=cpu to run the vision tower on CPU." + ) + + # Cast the (non-mm) model to fp16 for Spyre, and place the multimodal + # submodules on their configured device/dtype. The mm submodules are + # moved wholesale with Module.to(...) (required because + # nn.Parameter.set_data can't swap the CPU->nnpa backend; Module._apply + # rebuilds the Parameters, and buffers move too). Their descendants must + # be skipped so the non-mm branch doesn't re-cast them back to fp16 + # after placement (named_modules yields parents before children). + mm_prefixes_tuple = tuple(mm_parameter_prefixes) + for module_name, module in fms_model.named_modules(): + if module_name in mm_module_names: + logger.debug( + "Placing %s submodule on device=%s dtype=%s.", + module_name, + mm_device, + cpu_mm_dtype, + ) + module.to(device=mm_device, dtype=cpu_mm_dtype) + elif module_name.startswith(mm_prefixes_tuple): + # Descendant of an mm submodule; already placed with its ancestor. + continue + elif is_fp8_model: + # Per-param cast restricted to bf16: leaves fp8 weights and + # fp32 scales alone. recurse=False so each param is visited + # once via the outer named_modules() walk. + for param in module.parameters(recurse=False): + if param.dtype == torch.bfloat16: + param.data = param.data.to(dtype=torch.float16) + else: + module.to(dtype=torch.float16) + + return mm_device + + class SpyreCausalLM(nn.Module): def __init__( self, @@ -295,49 +356,10 @@ def _cast_params_for_spyre(self): For quantized (e.g. FP8) models we only convert bf16 params/buffers to fp16 — fp8 weights and fp32 scales must be left untouched. """ - cpu_mm_dtype = envs_spyre.SENDNN_INFERENCE_CPU_MM_DTYPE - mm_device = envs_spyre.SENDNN_INFERENCE_MM_DEVICE mm_prefixes = self.mm_model_utils.mm_parameter_prefixes if self.mm_model_utils else () - mm_module_names = {p.rstrip(".") for p in mm_prefixes} - - if mm_device == "nnpa" and mm_module_names and not utils_spyre.ensure_nnpa_registered(): - raise RuntimeError( - "SENDNN_INFERENCE_MM_DEVICE resolved to nnpa (torch_nnpa is installed) " - "but the nnpa device could not be initialized. Refusing to fall back to " - "CPU; a broken nnpa backend must not be silently masked. Set " - "SENDNN_INFERENCE_MM_DEVICE=cpu to run the vision tower on CPU." - ) - self.mm_device = mm_device - - # Cast the (non-mm) model to fp16 for Spyre, and place the multimodal - # submodules on their configured device/dtype. The mm submodules are - # moved wholesale with Module.to(...) (required because - # nn.Parameter.set_data can't swap the CPU->nnpa backend; Module._apply - # rebuilds the Parameters, and buffers move too). Their descendants must - # be skipped so the non-mm branch doesn't re-cast them back to fp16 - # after placement (named_modules yields parents before children). - mm_prefixes_tuple = tuple(mm_prefixes) - for module_name, module in self.fms_model.named_modules(): - if module_name in mm_module_names: - logger.debug( - "Placing %s submodule on device=%s dtype=%s.", - module_name, - mm_device, - cpu_mm_dtype, - ) - module.to(device=mm_device, dtype=cpu_mm_dtype) - elif module_name.startswith(mm_prefixes_tuple): - # Descendant of an mm submodule; already placed with its ancestor. - continue - elif self.is_fp8_model: - # Per-param cast restricted to bf16: leaves fp8 weights and - # fp32 scales alone. recurse=False so each param is visited - # once via the outer named_modules() walk. - for param in module.parameters(recurse=False): - if param.dtype == torch.bfloat16: - param.data = param.data.to(dtype=torch.float16) - else: - module.to(dtype=torch.float16) + self.mm_device = cast_params_for_spyre( + self.fms_model, mm_prefixes, is_fp8_model=self.is_fp8_model + ) def _cast_to_f32(self): """Cast model parameters to f32.""" diff --git a/sendnn_inference/multimodal/mm_mappings/llava_next.py b/sendnn_inference/multimodal/mm_mappings/llava_next.py index 94ef22456..85c881ee8 100644 --- a/sendnn_inference/multimodal/mm_mappings/llava_next.py +++ b/sendnn_inference/multimodal/mm_mappings/llava_next.py @@ -53,13 +53,30 @@ def unwrap_mm_kv_cache_opts(self): @staticmethod def get_mm_specific_load_overrides(hf_config: PretrainedConfig): """Get any overrides needed for initializing the FMS model from the - transformers config. For this model, we need to fix the head_dim, which - currently surfaces as a problem for all 2b variants of granite 3.x LLMs - when running through FMS. - - TODO: If additional variants of granite vision are added, or broader - llava next support is added in FMS, handle it properly here. + transformers config. For granite-vision-3.2-2b in HF format, the + text_config doesn't declare `head_dim` explicitly, and the default + fallback (`hidden_size / num_attention_heads` = 2048/32 = 64) is + wrong for the FMS Granite implementation, which expects 128. We + patch it in here. + + A config that already sets `head_dim` correctly (whether to 128 or + any other legal value) needs no override — returning a partial + `text_config` dict here with `override_hf_pretrained_config=True` + would obliterate every other field FMS built from the HF config + (emb_dim, nheads, etc.) because FMS's override merges are + whole-key replacements, not deep merges. """ + text_cfg = hf_config.text_config + explicit_head_dim = getattr(text_cfg, "head_dim", None) + if explicit_head_dim is not None: + # HF config knows its own head_dim — trust it. + return {} + + implicit_head_dim = text_cfg.hidden_size // text_cfg.num_attention_heads + if implicit_head_dim == 128: + # Implicit head_dim already matches what FMS wants; no override. + return {} + return { "override_hf_pretrained_config": True, "text_config": {"head_dim": 128}, diff --git a/sendnn_inference/platform.py b/sendnn_inference/platform.py index 9d54718e6..5adf3c7f1 100644 --- a/sendnn_inference/platform.py +++ b/sendnn_inference/platform.py @@ -258,6 +258,28 @@ def check_and_update_config(cls, vllm_config: VllmConfig) -> None: if parallel_config.worker_cls == "auto": parallel_config.worker_cls = "sendnn_inference.v1.worker.spyre_worker.SpyreWorker" + # Use SpyreMultiprocExecutor when async MM encoding is enabled via + # SENDNN_INFERENCE_ASYNC_MM_ENCODER=1. The executor manages a separate + # vision encoder subprocess that runs in parallel with AIU inference. + # Pass the class object directly — Executor.get_class handles + # isinstance(backend, type) before string-based dispatch, which avoids + # Pydantic's Literal validator silently dropping a string class path. + if ( + is_decoder + and model_config.is_multimodal_model + and parallel_config.world_size > 1 + and envs_spyre.SENDNN_INFERENCE_ASYNC_MM_ENCODER + ): + from sendnn_inference.v1.executor.spyre_executor import SpyreMultiprocExecutor + + parallel_config.distributed_executor_backend = SpyreMultiprocExecutor + logger.info( + "Using SpyreMultiprocExecutor with async MM encoder subprocess " + "(world_size=%d, model=%s)", + parallel_config.world_size, + model_config.model, + ) + cls._check_threading_config(parallel_config.world_size) # set env vars based on the model @@ -670,7 +692,7 @@ def _check_threading_config(cls, worker_count: int): # NOTE: math.ceil can output a number for each worker that sums # to a total greater than cpu_count. - if is_multimodal: + if is_multimodal and not envs_spyre.SENDNN_INFERENCE_ASYNC_MM_ENCODER: if cpu_count is None: cpus_per_worker = None elif platform.machine() == "ppc64le": diff --git a/sendnn_inference/v1/core/scheduler.py b/sendnn_inference/v1/core/scheduler.py index cebf233dc..50c353c15 100644 --- a/sendnn_inference/v1/core/scheduler.py +++ b/sendnn_inference/v1/core/scheduler.py @@ -2,8 +2,8 @@ import math from collections import deque +from dataclasses import dataclass, field from typing import TYPE_CHECKING, Iterable, Union -from dataclasses import dataclass from collections import defaultdict @@ -23,6 +23,17 @@ logger = init_logger(__name__) + +@dataclass +class MMEncodeRequest: + """Lightweight descriptor for a waiting MM request that should be + pre-encoded before its Spyre prefill step begins.""" + + request_id: str + prompt_token_ids: list[int] + mm_features: list = field(default_factory=list) + + # Ensure that block_size is 64 # This ensures the rounding function is correct assert SpyrePlatform.get_block_size() == 64 @@ -212,6 +223,16 @@ def __init__(self, *args, **kwargs) -> None: self.tkv = 0 self.block_size = SpyrePlatform.get_block_size() + + # Async MM encoding state. + # _mm_encoding_submitted: requests whose encode job has been dispatched to + # the encoder subprocess but whose result has not yet been received. + # _mm_encoding_ready: requests whose embeddings are ready in + # pending_mm_embeddings (confirmed via _spyre_newly_encoded_req_ids in + # the model runner output). Only MM requests in this set are eligible + # for prefill scheduling. + self._mm_encoding_submitted: set[str] = set() + self._mm_encoding_ready: set[str] = set() self.max_batch_tkv_limit = SpyrePlatform.get_max_batch_tkv_limit() assert self.max_batch_tkv_limit != -1, ( @@ -231,6 +252,24 @@ def update_from_output(self, scheduler_output, model_runner_output): "Expecting an instance of CPSpyreModelRunnerOutput when doing chunked prefill." ) + # Update async MM encoding state: move newly encoded requests from + # "submitted" to "ready" so they become eligible for prefill. + # Read from scheduler_output (set by SpyreMultiprocExecutor.execute_model) + # rather than model_runner_output — the executor uses non_block=True which + # returns a Future, so attributes set on the Future never reach the resolved + # ModelRunnerOutput. scheduler_output is the same object in both places. + for req_id in getattr(scheduler_output, "_spyre_newly_encoded_req_ids", []): + self._mm_encoding_submitted.discard(req_id) + # Only promote to ready if the request is still known to the scheduler. + # If it was aborted while encoding was in-flight, finish_requests already + # removed it — skip to avoid a stale _mm_encoding_ready entry. + if req_id in self.requests: + self._mm_encoding_ready.add(req_id) + # Abort any request whose encode job failed — no retries. + for req_id in getattr(scheduler_output, "_spyre_failed_encode_req_ids", []): + logger.error("MM encode failed for req '%s' — aborting request", req_id) + self.finish_requests([req_id], RequestStatus.FINISHED_ABORTED) + # Remove completed prefills self.ongoing_prefills = [ req for req in self.ongoing_prefills if req.num_computed_tokens < req.num_prompt_tokens @@ -350,16 +389,23 @@ def schedule(self, throttle_prefills: bool = False) -> "SchedulerOutput": # req_id -> cached_blocks, new_blocks required_blocks = dict[str, tuple[int, int]]() - # Check if new requests can be scheduled for prefill + # Check if new requests can be scheduled for prefill. + # Per-request ineligibility (MM encoding not ready, shape mismatch, …) + # is collected in skipped_requests and restored to holdback_queue after + # the loop so they remain available for future schedule() calls. + # Block-count capacity is the only FIFO-correct reason to stop early: + # if the front request exceeds available blocks, later requests are + # unlikely to fit either. available_blocks = self._get_free_blocks() - self.total_reserved_blocks + skipped_requests: list[Request] = [] while holdback_queue: - new_request = holdback_queue[0] + new_request = holdback_queue.popleft() cached, blocks = self._get_required_blocks(new_request, True) if blocks > available_blocks: + holdback_queue.appendleft(new_request) break if self.can_schedule_prefill(new_request): - holdback_queue.popleft() required_blocks[new_request.request_id] = (cached, blocks) available_blocks -= blocks @@ -372,9 +418,14 @@ def schedule(self, throttle_prefills: bool = False) -> "SchedulerOutput": # Add request to the waiting queue self.waiting.append(new_request) else: - # Otherwise, we simply stop here so that the scheduler - # can work with the batch we have - break + # Per-request reason (e.g. MM encoding still in-flight): skip + # this request and keep checking later ones. + skipped_requests.append(new_request) + + # Restore skipped requests at the front of holdback_queue so they + # are returned to self.waiting (line below) in their original order. + for req in reversed(skipped_requests): + holdback_queue.appendleft(req) assert len(self.ongoing_prefills) <= 1, ( "Only one request can be prefilled at a time, but got %d" % len(self.ongoing_prefills) @@ -444,6 +495,30 @@ def schedule(self, throttle_prefills: bool = False) -> "SchedulerOutput": if not self.step_is_prefill: self._handle_decode_requests_pausing() + # Collect MM encode requests for ALL waiting multimodal requests that + # have not yet been submitted to the encoder subprocess. Emitted on + # every schedule() call (prefill AND decode steps) so the encoder can + # stay ahead of the prefill queue. The executor submits each request + # exactly once (tracked here via _mm_encoding_submitted). + mm_encode_requests: list[MMEncodeRequest] = [] + for req in holdback_queue: + if not getattr(req, "mm_features", None): + continue + if req.request_id in self._mm_encoding_submitted: + continue + if req.request_id in self._mm_encoding_ready: + continue + mm_encode_requests.append( + MMEncodeRequest( + request_id=req.request_id, + prompt_token_ids=list(req.prompt_token_ids or []), + mm_features=req.mm_features, + ) + ) + self._mm_encoding_submitted.add(req.request_id) + if len(mm_encode_requests) >= self.max_num_running_reqs: + break + # Cap chunk-0 token count to chunk_size - left_padding so the upstream KV # cache manager doesn't allocate a real blocks for the left-padding region. # Only matters at chunk 0; later chunks land on natural chunk boundaries. @@ -474,6 +549,8 @@ def schedule(self, throttle_prefills: bool = False) -> "SchedulerOutput": ): logger.debug("Scheduled tokens in this step: %s", outputs.num_scheduled_tokens) + outputs._spyre_mm_encode_requests = mm_encode_requests # type: ignore[attr-defined] + # Collect grammar bitmask synchronously for structured outputs. # NOTE: This is done here because vllm-spyre currently combines token sampling # in model_executor.execute_model() rather than implementing sample_tokens() @@ -511,6 +588,16 @@ def schedule(self, throttle_prefills: bool = False) -> "SchedulerOutput": return outputs def can_schedule_prefill(self, request: Request) -> bool: + # MM requests must wait until their vision embedding is ready. + # Only applies in async encoder mode; in non-async mode nothing ever + # populates _mm_encoding_ready so the gate would block all MM requests. + # Text-only requests are completely unaffected by this check. + if getattr(request, "mm_features", None) and ( + envs_spyre.SENDNN_INFERENCE_ASYNC_MM_ENCODER + and request.request_id not in self._mm_encoding_ready + ): + return False + # running and waiting queues are both empty, we can start a new batch # which can always be scheduled if len(self.running) + len(self.waiting) == 0: @@ -824,11 +911,33 @@ def finish_requests( ) # request_ids None means all requests are finished - self.ongoing_prefills = ( - [] - if request_ids is None - else [r for r in self.ongoing_prefills if r.request_id not in request_ids] - ) + if request_ids is None: + self.ongoing_prefills = [] + self._mm_encoding_submitted.clear() + self._mm_encoding_ready.clear() + else: + self.ongoing_prefills = [ + r for r in self.ongoing_prefills if r.request_id not in request_ids + ] + for rid in request_ids: + # If the encode job is queued but not yet started, send a cancel + # token so the encoder subprocess skips it rather than running a + # full (expensive) vision-tower forward for a dead request. + if rid in self._mm_encoding_submitted: + from sendnn_inference.v1.executor.spyre_executor import SpyreMultiprocExecutor + + cq = SpyreMultiprocExecutor.get_mm_cancel_queue() + if cq is not None: + try: + cq.put_nowait(rid) + except Exception as exc: + logger.debug( + "scheduler: failed to send cancel for req '%s': %s", + rid, + exc, + ) + self._mm_encoding_submitted.discard(rid) + self._mm_encoding_ready.discard(rid) # Also remove from paused_decoding_requests self.paused_decoding_requests = ( diff --git a/sendnn_inference/v1/executor/__init__.py b/sendnn_inference/v1/executor/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/sendnn_inference/v1/executor/spyre_executor.py b/sendnn_inference/v1/executor/spyre_executor.py new file mode 100644 index 000000000..9bb9754fc --- /dev/null +++ b/sendnn_inference/v1/executor/spyre_executor.py @@ -0,0 +1,289 @@ +"""SpyreMultiprocExecutor — extends vLLM's MultiprocExecutor with an +async MM encoder subprocess. + +vLLM spawns worker processes as daemon processes, which means workers +cannot themselves spawn child processes. By starting the encoder process +here (from the non-daemon executor), we sidestep that restriction. + +The executor owns the job/result queues: + +- submit jobs from ``_spyre_mm_encode_requests`` in each execute_model call +- collect completed (req_id, shape, dtype) metadata from the result queue +- broadcast metadata to all TP workers via ``collective_rpc("store_mm_embeddings")`` + so each worker reads the embedding from SHM independently — no rank-0 + tensor broadcast needed +- unlink SHM blocks after collective_rpc returns (all workers have read) +""" + +import multiprocessing +import multiprocessing.process +import multiprocessing.synchronize +import queue as queue_mod +from typing import Any, Callable + +from vllm.logger import init_logger +from vllm.v1.executor.multiproc_executor import MultiprocExecutor + +from sendnn_inference.v1.worker.mm_shared_memory import cleanup_embeddings_by_name + +logger = init_logger(__name__) + + +class SpyreMultiprocExecutor(MultiprocExecutor): + """MultiprocExecutor subclass that manages a non-daemon vision encoder + subprocess for async MM pre-encoding.""" + + # Process-global handle to the encoder job queue. Published after the + # encoder starts so the scheduler (same EngineCore process) can send + # cancellations via the dedicated cancel queue. + _shared_mm_job_queue: "multiprocessing.Queue | None" = None + # Dedicated cancel queue: scheduler puts req_id strings here when a request + # is aborted. Kept separate from the job queue so the job queue carries + # only MMEncodeRequest objects — no type-tagged mixed messages. + _shared_mm_cancel_queue: "multiprocessing.Queue | None" = None + + @classmethod + def get_mm_job_queue(cls) -> "multiprocessing.Queue | None": + return cls._shared_mm_job_queue + + @classmethod + def get_mm_cancel_queue(cls) -> "multiprocessing.Queue | None": + return cls._shared_mm_cancel_queue + + def _init_executor(self) -> None: + logger.info("SpyreMultiprocExecutor._init_executor: custom executor active") + self._mm_encoder_proc: multiprocessing.process.BaseProcess | None = None + self._mm_job_queue: multiprocessing.Queue | None = None + self._mm_cancel_queue: multiprocessing.Queue | None = None + self._mm_result_queue: multiprocessing.Queue | None = None + self._mm_stop_event: multiprocessing.synchronize.Event | None = None + # Number of encode jobs submitted but not yet collected. + self._mm_in_flight: int = 0 + super()._init_executor() + # Do NOT start the encoder process here — warmup runs separately via + # collective_rpc("compile_or_warm_up_model") and spawning a subprocess + # while distributed collectives are active corrupts SHM broadcasts. + # _try_start_mm_encoder is called from collective_rpc below instead. + # Currently, the warmup step doesn't consume MM encoder process. + + def collective_rpc( # ty: ignore[invalid-method-override] + self, + method: str | Callable, + timeout: float | None = None, + args: tuple = (), + kwargs: dict | None = None, + **extra_kwargs, + ) -> Any: + result = super().collective_rpc( + method, timeout=timeout, args=args, kwargs=kwargs, **extra_kwargs + ) + # Start the encoder process after warmup completes — all distributed + # collectives are done at this point so the subprocess spawn is safe. + if method == "compile_or_warm_up_model" and self._mm_encoder_proc is None: + logger.info("SpyreMultiprocExecutor: warmup complete, starting encoder process") + self._try_start_mm_encoder() + return result + + def execute_model(self, scheduler_output: Any, non_block: bool = False) -> Any: + """Submit encode jobs, collect completed results, then run the model step. + + Per-step flow: + 1. Submit new MM encode jobs to the encoder subprocess (non-blocking). + 2. Collect any completed results from the result queue (non-blocking). + 3. For each completed result: tell all TP workers to read from SHM, + then unlink the SHM block. + 4. Run the normal model step via super().execute_model(). + 5. Attach newly_encoded_req_ids to the output for scheduler feedback. + """ + newly_encoded_req_ids: list[str] = [] + + if self._mm_encoder_proc is not None and not self._mm_encoder_proc.is_alive(): + # TODO: instead of killing the server, restart the encoder subprocess + # and fall back to inline encoding for in-flight MM requests + # during the vision model reload window. + raise RuntimeError( + f"MM encoder process died unexpectedly " + f"(exit code {self._mm_encoder_proc.exitcode}) — " + "restart the server to restore MM encoding" + ) + + failed_encode_req_ids: list[str] = [] + if self._mm_job_queue is not None: + # Submit new encode jobs. The scheduler ensures each request + # appears in _spyre_mm_encode_requests exactly once (tracked via + # _mm_encoding_submitted), so no dedup is needed here. + mm_encode_reqs = getattr(scheduler_output, "_spyre_mm_encode_requests", []) + for req in mm_encode_reqs: + try: + self._mm_job_queue.put_nowait(req) + self._mm_in_flight += 1 + logger.debug("Submitted MM encode job for req '%s'", req.request_id) + except Exception as exc: + # put_nowait failed (BrokenPipeError, queue.Full, PicklingError, …). + # The scheduler has already recorded this req_id in + # _mm_encoding_submitted, so silently swallowing the error + # would strand the request forever with no result and no error. + # Surface it as a failed encode so the scheduler aborts it cleanly. + logger.warning( + "MM job queue submission failed for req '%s': %s — aborting request", + req.request_id, + exc, + ) + failed_encode_req_ids.append(req.request_id) + + if self._mm_result_queue is not None and self._mm_in_flight > 0: + # Collect completed results (non-blocking drain). + newly_encoded_metadata: list[tuple] = [] + while True: + try: + req_id, shape, dtype = self._mm_result_queue.get_nowait() + self._mm_in_flight -= 1 + if shape is not None and dtype is not None: + newly_encoded_metadata.append((req_id, shape, dtype)) + else: + # Encoder failed for this request. + logger.warning( + "Encoder process returned error for req '%s'", + req_id, + ) + failed_encode_req_ids.append(req_id) + except queue_mod.Empty: + break + + if newly_encoded_metadata: + # Tell all TP workers to read embeddings from SHM independently. + # collective_rpc is synchronous — when it returns, all workers + # have finished reading, so the SHM blocks can be safely unlinked. + self.collective_rpc( + "store_mm_embeddings", + args=(newly_encoded_metadata,), + ) + for req_id, _, _ in newly_encoded_metadata: + cleanup_embeddings_by_name(req_id) + newly_encoded_req_ids = [r for r, _, _ in newly_encoded_metadata] + logger.debug( + "Stored async MM embeddings for %d request(s): %s", + len(newly_encoded_req_ids), + newly_encoded_req_ids, + ) + + # Attach newly_encoded_req_ids to scheduler_output (not to the model + # runner output) because execute_model(non_block=True) returns a Future, + # not the resolved ModelRunnerOutput. The same scheduler_output object + # is passed to scheduler.update_from_output(), so setting it here means + # the scheduler will see it regardless of the async/sync execution path. + if newly_encoded_req_ids: + scheduler_output._spyre_newly_encoded_req_ids = newly_encoded_req_ids + if failed_encode_req_ids: + scheduler_output._spyre_failed_encode_req_ids = failed_encode_req_ids + + # Clear _spyre_mm_encode_requests before dispatching to workers. + # The async encoder owns all MM encoding jobs + scheduler_output._spyre_mm_encode_requests = [] + + return super().execute_model(scheduler_output, non_block=non_block) + + def _try_start_mm_encoder(self) -> None: + """Start the vision encoder subprocess. + + The encoder process loads vision weights directly from disk via + ``get_model(..., vision_only=True)`` — no collective_rpc state-dict + extraction from rank 0 is required. + + Skipped silently for non-MM models. + """ + # SpyreMultiprocExecutor is only registered (via platform.py) when + # SENDNN_INFERENCE_ASYNC_MM_ENCODER=1, so _try_start_mm_encoder is only + # called for configurations where async encoding is explicitly enabled. + # No model-type detection needed here — the env var is the gate. + import sendnn_inference.envs as envs_spyre + + if not envs_spyre.SENDNN_INFERENCE_ASYNC_MM_ENCODER: + logger.info( + "SpyreMultiprocExecutor: SENDNN_INFERENCE_ASYNC_MM_ENCODER not set, " + "skipping encoder process" + ) + return + + from sendnn_inference.v1.worker.mm_encoder_process import encoder_process_main + + try: + # Plain multiprocessing queues and event, + # automatically closed when the executor (their owner) exits. + ctx_q = multiprocessing.get_context("spawn") + self._mm_job_queue = ctx_q.Queue() + # Dedicated cancel queue: carries req_id strings for aborted requests. + # The encoder drains this before processing each job so it can skip + # cancelled requests without running the expensive vision-tower forward. + self._mm_cancel_queue = ctx_q.Queue() + self._mm_result_queue = ctx_q.Queue() + self._mm_stop_event = ctx_q.Event() + + # Spawn as daemon so the process is automatically killed when the + # executor (its parent) exits — including unclean server termination. + # Using daemon=True is safe here because the encoder process never + # spawns children of its own. + # Note: workers are daemon too. + ctx = multiprocessing.get_context("spawn") + self._mm_encoder_proc = ctx.Process( + target=encoder_process_main, + args=( + self.vllm_config, + self._mm_job_queue, + self._mm_result_queue, + self._mm_stop_event, + self._mm_cancel_queue, + ), + daemon=True, + name="mm-encoder", + ) + self._mm_encoder_proc.start() + logger.info( + "SpyreMultiprocExecutor: encoder process started (pid=%d), " + "waiting for vision model to load...", + self._mm_encoder_proc.pid, + ) + + # Block until the encoder process has finished loading the vision model. + signal = self._mm_result_queue.get(timeout=300) + if signal != "READY": + raise RuntimeError(f"Encoder process startup failed: {signal}") + + logger.info("SpyreMultiprocExecutor: encoder process ready") + SpyreMultiprocExecutor._shared_mm_job_queue = self._mm_job_queue + SpyreMultiprocExecutor._shared_mm_cancel_queue = self._mm_cancel_queue + + except Exception as exc: + logger.exception( + "SpyreMultiprocExecutor: failed to start encoder process (%s: %s) — " + "restarting the server is required to restore MM encoding.", + type(exc).__name__, + exc, + ) + self._cleanup_encoder() + raise + + def _cleanup_encoder(self) -> None: + if self._mm_stop_event is not None: + self._mm_stop_event.set() + if self._mm_encoder_proc and self._mm_encoder_proc.is_alive(): + self._mm_encoder_proc.join(timeout=5) + if self._mm_encoder_proc.is_alive(): + self._mm_encoder_proc.terminate() + self._mm_encoder_proc = None + self._mm_stop_event = None + self._mm_job_queue = None + self._mm_cancel_queue = None + self._mm_result_queue = None + self._mm_in_flight = 0 + SpyreMultiprocExecutor._shared_mm_job_queue = None + SpyreMultiprocExecutor._shared_mm_cancel_queue = None + + def shutdown(self) -> None: + if self._mm_stop_event is not None: + self._mm_stop_event.set() + if self._mm_encoder_proc and self._mm_encoder_proc.is_alive(): + self._mm_encoder_proc.join(timeout=10) + if self._mm_encoder_proc.is_alive(): + self._mm_encoder_proc.terminate() + super().shutdown() diff --git a/sendnn_inference/v1/worker/mm_encoder_process.py b/sendnn_inference/v1/worker/mm_encoder_process.py new file mode 100644 index 000000000..b9e887a5b --- /dev/null +++ b/sendnn_inference/v1/worker/mm_encoder_process.py @@ -0,0 +1,322 @@ +"""Async vision encoder subprocess for MM pre-encoding. + +The encoder process loads only the vision components of the multimodal model +(vision_tower + multi_modal_projector + text_embedding) using FMS's +``get_model(..., vision_only=True)``, which selectively loads vision +weights from the checkpoint — skipping the LLM decoder entirely. + +This process is non-daemon (started by the non-daemon SpyreMultiprocExecutor) +so it runs truly parallel to AIU forward passes. Results are written to POSIX +shared memory; only a small metadata tuple is sent back through the result queue, +so all TP workers can read the embedding independently without a rank-0 broadcast +of the full tensor. +""" + +import logging +import math +import os +import platform +import queue as queue_mod +import time + +import torch +from vllm.config import VllmConfig + +import sendnn_inference.envs as envs_spyre +from sendnn_inference.model_executor.model_loader.spyre import SpyreCausalLM, cast_params_for_spyre +from sendnn_inference.platform import SpyrePlatform, THREADING_ENVS +from sendnn_inference.v1.worker.mm_shared_memory import write_embeddings + +logger = logging.getLogger(__name__) + + +def _resolve_mm_utils_cls(hf_config): + """Return the MMUtils class for *hf_config*. + + Callers should first pass *hf_config* through + ``SpyreCausalLM.resolve_hf_config()`` so that format-specific conversions + (e.g. Mistral-format pixtral → Mistral3Config) are applied before the + registry lookup. The model_type scan below handles any remaining cases + where Pydantic serialization loses the specific subclass. + """ + from sendnn_inference.multimodal import MM_HF_CFG_REGISTRY + + utils_cls = MM_HF_CFG_REGISTRY.get(type(hf_config)) + if utils_cls is not None: + return utils_cls + + # Fallback: scan by model_type string for when hf_config is still a base + # PretrainedConfig after Pydantic deserialization (e.g. HF-format Mistral3 + # whose class was lost in transit to the encoder subprocess). + model_type = getattr(hf_config, "model_type", "") + for cfg_cls, cls in MM_HF_CFG_REGISTRY.items(): + if getattr(cfg_cls, "model_type", None) == model_type: + return cls + + raise ValueError( + f"encoder_process: no MMUtils found for hf_config type={type(hf_config).__name__!r} " + f"model_type={model_type!r}; known: {[c.__name__ for c in MM_HF_CFG_REGISTRY]}" + ) + + +# ── VisionEncoderRunner ─────────────────────────────────────────────────────── + + +class VisionEncoderRunner: + """Loads the vision-only FMS model and encodes MMEncodeRequest jobs. + + Uses ``get_model("hf_pretrained", model_path=..., vision_only=True)`` so + only the vision tower, projector, and text embedding are loaded from disk. + Model loading happens in ``__init__`` so construction raises on failure. + """ + + def __init__(self, vllm_config: VllmConfig) -> None: + from fms.models import get_model + + # Spyre always compiles the LLM decoder in float16 (see SpyreCausalLM.get_dtype()). + # NNPA may return float32 embeddings even when model weights are float16; + # cast to float16 before writing to SHM so the decoder sees the compiled dtype. + self._decoder_dtype = torch.float16 + + model_config = vllm_config.model_config + + # Must be called before any NNPA or model operations + SpyrePlatform.maybe_ensure_sendnn_configured(model_config) + + model_name = model_config.model + # FMS hf_pretrained + variant resolves from HF cache without a separate + # download step — the workers already downloaded the weights, so FMS finds + # them in the local HF snapshot cache. Use model_path instead when the + # model is already a local directory (avoids an unnecessary cache lookup). + is_local = os.path.isdir(model_name) + fms_kwargs: dict = {"model_path": model_name} if is_local else {"variant": model_name} + + logger.info( + "encoder_process: loading vision-only model %r " + "(mm_device=%s, mm_dtype=%s, output_dtype=%s)", + model_name, + envs_spyre.SENDNN_INFERENCE_MM_DEVICE, + envs_spyre.SENDNN_INFERENCE_CPU_MM_DTYPE, + self._decoder_dtype, + ) + t0 = time.time() + self.fms_model = get_model( + "hf_pretrained", + vision_only=True, + # Required for AIU/NNPA: fused QKV is not handled efficiently by + # NNPA hardware; unfused weights use the optimised NNPA path. + # Workers also pass fused_weights=False (see spyre.py load_weights). + fused_weights=False, + **fms_kwargs, + ) + + # resolve_hf_config normalises format-specific configs (e.g. Mistral-format + # pixtral → Mistral3Config) so the MM_HF_CFG_REGISTRY lookup in + # _resolve_mm_utils_cls succeeds directly via class-type match. + normalized_hf_config = SpyreCausalLM.resolve_hf_config(vllm_config) + self.mm_utils_cls = _resolve_mm_utils_cls(normalized_hf_config) + + self.fms_model.eval() + self.mm_device = cast_params_for_spyre( + self.fms_model, + self.mm_utils_cls.mm_parameter_prefixes, + is_fp8_model=False, + ) + logger.info("encoder_process: mm_utils=%s", self.mm_utils_cls.__name__) + torch.set_grad_enabled(False) + logger.info("encoder_process: vision model loaded in %.2fs", time.time() - t0) + + def execute_model(self, request) -> torch.Tensor: + """Encode a single MMEncodeRequest and return a CPU-contiguous tensor.""" + input_ids = torch.tensor(request.prompt_token_ids, dtype=torch.int64).unsqueeze(0) + with torch.inference_mode(): + embeds = self.mm_utils_cls.get_maybe_mm_embeddings( + self.fms_model, + input_ids, + request.mm_features, + is_decode=False, + mm_device=self.mm_device, + ) + return embeds.to(dtype=self._decoder_dtype).cpu().contiguous() + + +# ── Process entry point ─────────────────────────────────────────────────────── + + +def _configure_encoder_threads() -> None: + """Give the encoder subprocess the full cpu_count thread budget. + + Workers each receive ``cpu_count / num_workers`` threads (set by + SpyrePlatform.configure_thread_settings). The encoder subprocess inherits + that reduced count, but vision encoding is a serial CPU/NNPA workload that + benefits from all available cores. We restore the full thread count here + so OMP, DT_PARALLEL_THREADS, torch inter-op, and intra-op thread pools all + see the right value when NNPA is initialised. + + Worker thread count is reduced proportionally so the total thread budget + across all processes stays at cpu_count: + workers: cpu_count / num_workers (unchanged — set by platform.py) + encoder: cpu_count + """ + cpu_count, _ = SpyrePlatform.get_cpu_count() + + if cpu_count is None: + encoder_cpu_count = None + elif platform.machine() == "ppc64le": + encoder_cpu_count = min(cpu_count, 36.0) + else: + encoder_cpu_count = math.ceil(cpu_count) + + if encoder_cpu_count is None: + logger.warning( + "encoder_process: could not determine encoder_cpu_count; " + "thread configuration unchanged (inherited from parent)" + ) + return + + # The encoder gets the full cpu_count. + encoder_threads = max(1, math.ceil(encoder_cpu_count)) + + for env in THREADING_ENVS: + os.environ[env] = str(encoder_threads) + + # torch intra-op thread pool — can be changed at any time. + torch.set_num_threads(encoder_threads) + # torch inter-op thread pool — can only be set before parallel work starts; + # ignore if it's too late (e.g. in unit tests where torch is already warm). + try: + torch.set_num_interop_threads(encoder_threads) + except RuntimeError as e: + logger.warning( + "encoder_process: could not set inter-op threads to %d: %s", + encoder_threads, + e, + ) + + logger.info( + "encoder_process: thread config — encoder=%d,(encoder_cpu_count=%.1f)", + encoder_threads, + encoder_cpu_count, + ) + + +def encoder_process_main( + vllm_config: VllmConfig, + job_queue, + result_queue, + stop_event, + cancel_queue=None, +) -> None: + """Entry point for the vision encoder subprocess. + + Loads the vision-only model, signals READY, then serves execute_model jobs. + Results are written to POSIX SHM; only ``(req_id, shape, dtype)`` metadata + is put on the result queue so all TP workers can read the embedding + independently without a rank-0 tensor broadcast. + + ``stop_event`` is a ``multiprocessing.Event`` set by the executor on + shutdown. The job loop polls it via a timeout on ``job_queue.get`` so + the process exits cleanly on both graceful and abrupt server termination. + + ``cancel_queue`` carries req_id strings for aborted requests. The encoder + drains it after dequeuing each job so cancelled jobs are skipped before the + expensive vision-tower forward pass begins. + + Job loop: + get(MMEncodeRequest) → drain cancel_queue → skip if cancelled + → execute_model → write SHM → put (req_id, shape, dtype) + Exits when stop_event is set or None sentinel is received. + """ + logger.info("encoder_process: starting") + + # ── Thread configuration ────────────────────────────────────────────────── + # The encoder subprocess inherits the per-worker thread count that the parent + # set (cpu_count / num_workers). We override it here to use the full cpu_count + # so the vision encoder gets maximum CPU/NNPA parallelism. + # + # This MUST happen before maybe_ensure_sendnn_configured() because the NNPA + # backend captures thread-pool settings at import time. + _configure_encoder_threads() + + try: + runner = VisionEncoderRunner(vllm_config) + except Exception as exc: + logger.exception("encoder_process: failed to load vision model: %s", exc) + result_queue.put(f"ERROR: {exc}") + return + + result_queue.put("READY") + logger.info( + "encoder_process: ready, waiting for jobs " + "(torch_num_threads=%d, OMP_NUM_THREADS=%s, DT_PARALLEL_THREADS=%s)", + torch.get_num_threads(), + os.environ.get("OMP_NUM_THREADS", "unset"), + os.environ.get("DT_PARALLEL_THREADS", "unset"), + ) + + # skip_ids: req_ids drained from cancel_queue that have not yet been dequeued + # as a job — the encode will be skipped when the job arrives. + # processed_ids: tombstones for completed encodes — if a cancel arrives after + # the encode is done, discard the tombstone instead of adding to skip_ids. + skip_ids: set[str] = set() + processed_ids: set[str] = set() + + while not stop_event.is_set(): + try: + job = job_queue.get(timeout=1.0) + except KeyboardInterrupt: + logger.info("encoder_process: interrupted, exiting") + break + except Exception: + # queue.Empty from timeout — loop back and re-check stop_event. + continue + if job is None: + logger.info("encoder_process: shutdown received") + break + + # Drain the cancel queue before processing this job so that any + # cancellations that arrived while we were waiting are captured. + if cancel_queue is not None: + while True: + try: + rid = cancel_queue.get_nowait() + if rid in processed_ids: + processed_ids.discard(rid) + logger.debug( + "encoder_process: late cancel for req '%s' (already done)", rid + ) + else: + skip_ids.add(rid) + logger.debug("encoder_process: pre-cancel for req '%s'", rid) + except queue_mod.Empty: + break + + req_id = job.request_id + + # Skip encode for cancelled requests; send abort result so the scheduler + # can clean up promptly instead of waiting for a timeout. + if req_id in skip_ids: + skip_ids.discard(req_id) + result_queue.put((req_id, None, None)) + logger.debug("encoder_process: skipped encode for cancelled req '%s'", req_id) + continue + + t0 = time.time() + try: + embeds = runner.execute_model(job) + + # Write embedding to POSIX SHM; close our handle without unlinking + # so all TP workers can still open it by name. Rank 0 will unlink + # the block after all workers have read (via _collect_async_mm_results). + shm = write_embeddings(embeds, req_id) + shm.close() + + t_elapsed = time.time() - t0 + result_queue.put((req_id, tuple(embeds.shape), embeds.dtype)) + # Tombstone: a late cancel may still arrive on cancel_queue for this req_id. + processed_ids.add(req_id) + logger.info("maybe_mm_embedding processing time: %.2fms", t_elapsed * 1000) + except Exception as exc: + logger.exception("encoder_process: failed to execute_model '%s': %s", req_id, exc) + result_queue.put((req_id, None, None)) + processed_ids.add(req_id) diff --git a/sendnn_inference/v1/worker/mm_shared_memory.py b/sendnn_inference/v1/worker/mm_shared_memory.py index 73176eb71..a4b88a76f 100644 --- a/sendnn_inference/v1/worker/mm_shared_memory.py +++ b/sendnn_inference/v1/worker/mm_shared_memory.py @@ -6,6 +6,7 @@ """ import hashlib +import math from multiprocessing.shared_memory import SharedMemory import torch @@ -76,7 +77,15 @@ def write_embeddings(tensor: torch.Tensor, req_id: str) -> SharedMemory: assert tensor.dtype in _DTYPE_TO_IDX, f"Unsupported dtype for SHM transfer: {tensor.dtype}" data_shm = SharedMemory(create=True, size=tensor.nbytes, name=_shm_name(req_id)) - torch.frombuffer(data_shm.buf, dtype=tensor.dtype).reshape(tensor.shape).copy_(tensor) + assert data_shm.buf is not None + # POSIX shared memory rounds the segment up to a page boundary (16 KiB + # on macOS ARM64, 4 KiB on Linux x86_64), so data_shm.buf is generally + # larger than tensor.nbytes. Slice to the exact byte count before + # reshaping, otherwise the trailing padding produces an off-by-page-size + # reshape error. + torch.frombuffer(data_shm.buf[: tensor.nbytes], dtype=tensor.dtype).reshape(tensor.shape).copy_( + tensor + ) logger.debug( "Wrote MM embeddings to SHM for req '%s': shape=%s dtype=%s bytes=%d", @@ -101,8 +110,12 @@ def read_embeddings( Opens and closes the shared-memory handle internally. """ data_shm = SharedMemory(name=_shm_name(req_id)) + assert data_shm.buf is not None + # Match the byte count slice from write_embeddings — the SHM buf is + # page-padded, so passing the full buf would fail .reshape(shape). + nbytes = math.prod(shape) * torch.empty((), dtype=dtype).element_size() # .clone() detaches the tensor from the SHM buffer so the handle can be closed. - result = torch.frombuffer(data_shm.buf, dtype=dtype).reshape(shape).clone() + result = torch.frombuffer(data_shm.buf[:nbytes], dtype=dtype).reshape(shape).clone() data_shm.close() logger.debug( @@ -128,3 +141,19 @@ def cleanup_embeddings(data_shm: SharedMemory) -> None: data_shm.close() except Exception as exc: logger.debug("SHM close skipped (%s): %s", data_shm.name, exc) + + +def cleanup_embeddings_by_name(req_id: str) -> None: + """Unlink the shared-memory block for *req_id* by name. + + Used when the caller does not hold the original ``SharedMemory`` handle + (e.g. after all TP ranks have independently read the block and rank 0 + needs to unlink it). Safe to call even if the block was already removed. + """ + name = _shm_name(req_id) + try: + shm = SharedMemory(name=name) + shm.unlink() + shm.close() + except Exception as exc: + logger.debug("SHM cleanup_by_name skipped (%s): %s", name, exc) diff --git a/sendnn_inference/v1/worker/spyre_model_runner.py b/sendnn_inference/v1/worker/spyre_model_runner.py index da60b984d..71f9f6022 100644 --- a/sendnn_inference/v1/worker/spyre_model_runner.py +++ b/sendnn_inference/v1/worker/spyre_model_runner.py @@ -404,10 +404,10 @@ def vocab_size(self) -> int: # self.model here is probably a transformers model class if self.model_config.architecture in FMS_POOLING_MODEL_LIST: assert isinstance(self.model.config.src_vocab_size, int) - return self.model.config.src_vocab_size # ty: ignore[invalid-return-type] + return self.model.config.src_vocab_size else: assert isinstance(self.model.config.vocab_size, int) - return self.model.config.vocab_size # ty: ignore[invalid-return-type] + return self.model.config.vocab_size def _prepare_pad_input_ids( self, @@ -772,6 +772,17 @@ def __init__( # Initialize performance metric logger for tracking embedding times self.perf_logger = create_perf_metric_logger(rank=rank) + # Pre-computed MM embeddings for waiting requests (keyed by request_id). + # Populated by store_mm_embeddings() (async path via executor). + # Consumed and removed by add_new_request() when the request begins prefill. + self.pending_mm_embeddings: dict[str, torch.Tensor] = {} + + # Request IDs that finished (aborted/completed) while their encode job + # was still in-flight. store_mm_embeddings() checks this set and discards + # late-arriving results rather than letting them leak in pending_mm_embeddings. + # Entries are removed once the late result arrives (self-cleaning). + self._finished_encode_req_ids: set[str] = set() + def load_model(self) -> None: self._model = SpyreCausalLM( vllm_config=self.vllm_config, @@ -794,6 +805,33 @@ def prompt_len(request: NewRequestData | Request) -> int: assert request.prompt_token_ids is not None, "prompt token ids are required" return len(request.prompt_token_ids) + def store_mm_embeddings(self, results: list[tuple]) -> None: + """Read completed async MM embeddings from POSIX SHM and cache them. + + Called on all TP ranks via collective_rpc by SpyreMultiprocExecutor + after the encoder subprocess has written embeddings to SHM. Each rank + reads independently — no rank-0 tensor broadcast is needed. + + ``results`` is a list of ``(req_id, shape, dtype)`` tuples. The SHM + blocks are kept alive by the executor until this call returns. + """ + for req_id, shape, dtype in results: + if req_id in self._finished_encode_req_ids: + # Request finished while its encode was in-flight; discard the + # late result and clean up the tombstone entry. + self._finished_encode_req_ids.discard(req_id) + logger.debug("Discarding late async MM embeddings for finished req '%s'", req_id) + continue + embeds = read_embeddings(req_id, shape, dtype) + self.pending_mm_embeddings[req_id] = embeds + logger.debug( + "Stored async MM embeddings for req '%s' (rank %d): shape=%s dtype=%s", + req_id, + self.rank, + shape, + dtype, + ) + def get_supported_tasks(self) -> tuple[SupportedTask, ...]: tasks = list[SupportedTask]() @@ -1018,6 +1056,11 @@ def _compute_and_cache_mm_embeddings( mm_features=mm_features, is_decode=False, ) + # Ensure embeddings are on CPU before passing to the Spyre decoder. + # When the vision tower runs on NNPA, get_maybe_mm_embeddings returns + # an NNPA tensor; slicing it into a CPU input_embeds in + # _prepare_chunked_prefill would fail without this explicit move. + full_embeds = full_embeds.cpu().contiguous() t_elapsed = time.time() - t0 logger.info("maybe_mm_embedding processing time: %.2fms", (t_elapsed * 1000)) self.perf_logger.log( @@ -1235,6 +1278,7 @@ def _prepare_chunked_prefill(self, req_id: str) -> SamplingForwardInputs: full_input_tokens = torch.tensor( prompt_token_ids, dtype=torch.int64, device=self.device ).unsqueeze(0) + self._compute_and_cache_mm_embeddings(request, req_id, full_input_tokens, mm_features) # Slice the cached embeddings for this chunk @@ -1493,6 +1537,17 @@ def add_new_request(self, request: NewRequestData): self.requests[req_id] = req_state + # Consume pre-computed embeddings if they were encoded in advance by + # pre_encode_mm_requests(). With cached_mm_embeddings already set, + # _prepare_chunked_prefill skips its encoding block entirely. + if req_id in self.pending_mm_embeddings: + req_state.cached_mm_embeddings = self.pending_mm_embeddings.pop(req_id) + logger.debug( + "Consumed pre-encoded MM embeddings for request '%s' (rank %d)", + req_id, + self.rank, + ) + # Add only to prefill batch, it will be added later to the input batch # once if is fully prefilled self.prefill_batch.add_request(req_state) @@ -1653,6 +1708,11 @@ def _update_batch(self, scheduler_output: SchedulerOutput): if scheduler_output.finished_req_ids: for req_id in scheduler_output.finished_req_ids: self.input_batch.remove_request(req_id) + # If the embedding was already delivered, discard it now. + # If the encode is still in-flight, mark the req_id so that + # store_mm_embeddings() discards the late result when it arrives. + if self.pending_mm_embeddings.pop(req_id, None) is None: + self._finished_encode_req_ids.add(req_id) self.requests.pop(req_id, None) # Clean up paused tracking for finished requests self.paused_req_ids.discard(req_id) @@ -1722,7 +1782,7 @@ def execute_model( # Return empty ModelRunnerOutput if there's no work to do. return self.get_empty_output() - # Initialize internal request states if this is the first chunk of a very new prefill + # Initialize internal request states if this is the first chunk of a new prefill self.maybe_setup_new_prefill(scheduler_output) model_input = self.prepare_model_input(scheduler_output) diff --git a/sendnn_inference/v1/worker/spyre_worker.py b/sendnn_inference/v1/worker/spyre_worker.py index 76d9f24a0..9cb23aa9a 100644 --- a/sendnn_inference/v1/worker/spyre_worker.py +++ b/sendnn_inference/v1/worker/spyre_worker.py @@ -803,6 +803,19 @@ def execute_model( output = self.model_runner.execute_model(scheduler_output) return output if self.is_driver_worker else None + def store_mm_embeddings(self, results: list[tuple]) -> None: + """Read completed MM embeddings from SHM and cache them for prefill. + + Called on all TP ranks via collective_rpc by SpyreMultiprocExecutor + after the encoder subprocess has written embeddings to POSIX SHM. + Each worker reads independently — no rank-0 tensor broadcast needed. + + ``results`` is a list of ``(req_id, shape, dtype)`` tuples identifying + the SHM blocks written by the encoder process. + """ + if isinstance(self.model_runner, ChunkedPrefillModelRunner): + self.model_runner.store_mm_embeddings(results) + def _get_num_tokens(self, r: NewRequestData) -> int: assert r.prompt_token_ids is not None, "requests should have tokens!" return len(r.prompt_token_ids) diff --git a/tests/e2e/test_async_mm_encoder.py b/tests/e2e/test_async_mm_encoder.py new file mode 100644 index 000000000..36472bfde --- /dev/null +++ b/tests/e2e/test_async_mm_encoder.py @@ -0,0 +1,213 @@ +"""PR #1015 e2e test — async MM encoder wiring through SpyreMultiprocExecutor. + +Boots a real vLLM `LLM` with the async MM encoder enabled and the +nano-gv model (a random-init ~16 MB granite-vision variant built by +`tools/build_nano_gv.py`) then drives one MM request through the full +stack: + + request → scheduler emits encode job → SpyreMultiprocExecutor submits + to encoder subprocess → encoder writes embedding to SHM → workers read + SHM via collective_rpc → scheduler unblocks prefill → decode completes. + +This is the only test that exercises the executor → encoder-process → +worker handshake end-to-end, with all four real processes (engine, two TP +workers, encoder). The unit tests for findings 1-3 stub at the executor +boundary; the wiring (queue creation, READY handshake, scheduler / +executor binding, collective_rpc store_mm_embeddings) only runs here. + +Runs on CPU in eager mode against the nano-gv fixture (see +`REFERENCE_MODELS["joerunde/nano-gv"]` in tests/spyre_util.py), so +warmup takes ~2 s instead of ~2 min. No Spyre hardware needed. +""" + +from __future__ import annotations + +import base64 +import io + +import pytest +from PIL import Image + +# Ensure the llava_next mm mapping is imported. FMS serialization +# utilities are patched at import time and the patching is not idempotent — +# the existing tests/e2e/test_spyre_mm.py carries the same note. +import sendnn_inference.multimodal.mm_mappings.llava_next # noqa: F401 +from spyre_util import REFERENCE_MODELS + +pytestmark = [pytest.mark.multimodal, pytest.mark.cpu, pytest.mark.e2e] + +NANO_GV_MODEL = REFERENCE_MODELS["joerunde/nano-gv"] +MAX_TOKENS = 4 # keep CPU work small +NANO_IMAGE_SIZE = 112 # matches the nano-gv vision_config.image_size + + +# Env required for the async MM encoder path. See the `llm` fixture body for +# the actual env-setting — it's done there so a module-scoped fixture can use +# it (a function-scoped `monkeypatch` fixture can't be promoted to module). +# +# `SENDNN_INFERENCE_ASYNC_MM_ENCODER=1` + `tensor_parallel_size > 1` are +# required for `SpyrePlatform.check_and_update_config` to swap in +# `SpyreMultiprocExecutor` (platform.py:266-273). +# +# Notable knobs: +# - VLLM_WORKER_MULTIPROC_METHOD=spawn — vLLM defaults to fork, which +# SIGSEGVs on macOS when the MM stack pulls in tvm_ffi (via xgrammar). +# - SENDNN_INFERENCE_TP_MM_SHARING=0 — disables the rank-0 → all-ranks +# SHM-broadcast embedding share path. Its torch.distributed.broadcast +# collective hangs during warmup with TP > 1 on macOS (pre-existing +# bug, not in scope for this test). The async encoder uses a different +# SHM path (collective_rpc store_mm_embeddings), so disabling sharing +# here only affects warmup encoding — the async encoder takes over +# after warmup completes. +# - VLLM_ENABLE_V1_MULTIPROCESSING=0 from _local_envs_for_test.sh is left +# intact. It controls the engine-core IPC (in-process vs out-of-process), +# NOT worker TP — the existing TP=2 tests rely on it staying off. + + +@pytest.fixture(scope="module") +def llm(): + """Real vLLM LLM with TP=2, eager backend, async MM encoder enabled. + + Module-scoped because: + 1. Startup is the dominant cost (spawn 2 workers + spawn encoder + + warmup ~= 5 s with nano-gv, but each fresh LLM binds MASTER_PORT + via torch.distributed rendezvous). + 2. `MASTER_PORT=12345` is fixed in `_local_envs_for_test.sh`, so a + function-scoped LLM hits `EADDRINUSE` when the second test tries + to spin up a fresh rendezvous before the OS has released the + port from the previous test. + + Neither test in this module mutates encoder state in a way that would + leak into the next. + """ + # `async_encoder_env` is function-scoped (monkeypatch lifetime). Promote + # by passing through — pytest accepts a module-scoped fixture depending + # on a function-scoped one ONLY if the inner one is converted to be + # module-scoped too. We can't easily do that with monkeypatch, so set + # env vars directly inside the fixture body instead. + import os + + os.environ["SENDNN_INFERENCE_ASYNC_MM_ENCODER"] = "1" + os.environ["SENDNN_INFERENCE_DYNAMO_BACKEND"] = "eager" + os.environ["VLLM_WORKER_MULTIPROC_METHOD"] = "spawn" + os.environ["SENDNN_INFERENCE_TP_MM_SHARING"] = "0" + # Keep the EngineCore in-process so test_async_encoder_subprocess_is_running + # can reach llm.llm_engine.model_executor (set by V1 LLMEngine only when + # multiprocess_mode=False). With VLLM_ENABLE_V1_MULTIPROCESSING=1 the + # engine_core is a SyncMPClient and the executor is not reachable from + # the test process. + os.environ["VLLM_ENABLE_V1_MULTIPROCESSING"] = "0" + # Use a distinct MASTER_PORT so back-to-back runs with + # test_mm_cancel_storm.py don't collide on port 12345 (torch.distributed + # holds it in TIME_WAIT for ~60s after teardown). + os.environ["MASTER_PORT"] = "12346" + + from vllm import LLM + + return LLM( + model=NANO_GV_MODEL.name, + revision=NANO_GV_MODEL.revision, + tensor_parallel_size=2, + enforce_eager=True, + max_num_seqs=2, + # nano-gv image_size is 112, which produces ~64 vision patches + + # chat template overhead — 1024 is more than enough. + max_model_len=1024, + # Disable prefix caching so each request goes through the full + # encode-then-prefill path even if we send the same image twice. + enable_prefix_caching=False, + ) + + +def _build_mm_prompt() -> list[dict]: + """OpenAI-style chat messages with a small synthetic image. + + The image content is meaningless (nano-gv has random weights); we + just need something the processor can tokenize into an image-token + span so the async encoder path fires. + """ + img = Image.new("RGB", (NANO_IMAGE_SIZE, NANO_IMAGE_SIZE), color=(120, 80, 200)) + buf = io.BytesIO() + img.save(buf, format="PNG") + url = "data:image/png;base64," + base64.b64encode(buf.getvalue()).decode("ascii") + + return [ + { + "role": "user", + "content": [ + {"type": "image_url", "image_url": {"url": url}}, + {"type": "text", "text": "Describe this image."}, + ], + } + ] + + +# --------------------------------------------------------------------------- +# Happy path: a single MM request completes through the async encoder. +# --------------------------------------------------------------------------- + + +def test_mm_request_completes_through_async_encoder(llm): + """Submit one MM request, generate a few tokens, assert completion. + + This is the minimum bar for the PR: the executor spawned an encoder + subprocess, the encoder loaded vision-only weights, the scheduler + gated the request on encoding, the executor relayed embeddings via + SHM + `collective_rpc("store_mm_embeddings")`, and the workers + consumed `pending_mm_embeddings` during prefill. + + We deliberately don't assert on output text — nano-gv has random + weights and its outputs are garbage; the boolean "did it complete" + is what matters for the wiring check. + """ + from vllm import SamplingParams + + sampling_params = SamplingParams( + max_tokens=MAX_TOKENS, + temperature=0.0, + ) + + output = llm.chat(_build_mm_prompt(), sampling_params)[0] + + # Completion is the assertion. If the async encoder wiring is broken, + # the request will hang on the scheduler's MM gate (request stays in + # `_mm_encoding_submitted` forever) and pytest's per-test timeout + # will fire instead — that is the failure mode this test surfaces. + assert output.outputs, "no completion returned — request hung on the MM gate?" + assert output.outputs[0].text or output.outputs[0].token_ids, ( + "request completed but produced no tokens — the prefill path saw an " + "empty / malformed embedding from SHM" + ) + + +# --------------------------------------------------------------------------- +# Executor health: encoder subprocess actually started. +# --------------------------------------------------------------------------- + + +def test_async_encoder_subprocess_is_running(llm): + """Confirm the executor's encoder subprocess is up after warmup. + + Diagnostic for the wiring path in `SpyreMultiprocExecutor.collective_rpc` + that starts the encoder on the first `compile_or_warm_up_model` call. + If this assertion fails the executor was selected but the encoder + never started — every subsequent MM test in this module will hang on + the scheduler's gate, so this fails fast with a clear reason. + """ + from sendnn_inference.v1.executor.spyre_executor import SpyreMultiprocExecutor + + # model_executor is set as a direct shortcut by V1 LLMEngine.__init__ + # when multiprocess_mode=False (VLLM_ENABLE_V1_MULTIPROCESSING=0). + executor = llm.llm_engine.model_executor # type: ignore[attr-defined] + assert isinstance(executor, SpyreMultiprocExecutor), ( + f"expected SpyreMultiprocExecutor, got {type(executor).__name__}; " + "platform.py did not swap in the async-MM executor — check that " + "SENDNN_INFERENCE_ASYNC_MM_ENCODER=1 and tensor_parallel_size > 1." + ) + assert executor._mm_encoder_proc is not None, ( + "executor did not start the encoder subprocess on warmup" + ) + assert executor._mm_encoder_proc.is_alive(), ( + "encoder subprocess died after startup — check the ERROR sentinel " + "in the encoder process log" + ) diff --git a/tests/e2e/test_mm_cancel_storm.py b/tests/e2e/test_mm_cancel_storm.py new file mode 100644 index 000000000..02b75f7b3 --- /dev/null +++ b/tests/e2e/test_mm_cancel_storm.py @@ -0,0 +1,357 @@ +"""PR #1015 e2e cancellation-storm test. + +Boots the real OpenAI-compatible `vllm serve` with the async MM encoder +enabled, fires a burst of MM requests, cancels them mid-flight, and then +asserts the server is still responsive. + +The failure scenarios this catches: + + - Finding 1: cancelled requests still encode. If the encoder subprocess + is busy chewing through abandoned large-image jobs, the post-burst + legitimate request times out and this test fails. + + - Deadlock: if a put_nowait / scheduler-state mismatch leaves the + encoder or scheduler in a wedged state, /health or the follow-up + request hangs. + +Both are exactly the failure modes the unit tests for findings 1 and 3 +sketch in isolation; this test runs the whole stack so we can see what +actually happens under load. + +Runs on CPU eager mode against the nano-gv fixture (see +`REFERENCE_MODELS["joerunde/nano-gv"]` in tests/spyre_util.py). No Spyre +hardware needed. Tagged `e2e` because it spawns a full server process +and is slow to start. +""" + +from __future__ import annotations + +import asyncio +import base64 +import io +import time + +import httpx +import openai +import pytest +from PIL import Image + +# Ensure the llava_next mm mapping is imported. FMS serialization +# utilities are patched at import time and the patching is not idempotent — +# same caveat as tests/e2e/test_spyre_mm.py and test_pr1015_async_mm_encoder.py. +import sendnn_inference.multimodal.mm_mappings.llava_next # noqa: F401 +from spyre_util import REFERENCE_MODELS, RemoteOpenAIServer + +pytestmark = [pytest.mark.multimodal, pytest.mark.cpu, pytest.mark.e2e] + +NANO_GV_MODEL = REFERENCE_MODELS["joerunde/nano-gv"] +NANO_IMAGE_SIZE = 112 # matches the nano-gv vision_config.image_size + +# Tuning knobs — sized to make regressions in the cancel path externally +# visible against a ~6 ms/encode nano-gv model. +# +# `N_CANCELLED` is the storm depth. 100 concurrent cancels is enough to +# thoroughly exercise the scheduler → cancel_queue → encoder drain → +# aborted-result plumbing on every path. Going higher hits diminishing +# returns because the test harness (asyncio + httpx + vLLM request queue) +# becomes the bottleneck long before the encoder does. +# +# `POST_BURST_TIMEOUT` is the ceiling for the follow-up legitimate +# request. With nano-gv it should return in ~100 ms; a completely broken +# cancel path that grinds through all N abandoned encodes serially would +# still fit under a second. 5 s is comfortably above CI noise but +# tight enough that any actual DoS regression fails the test. +N_CANCELLED = 100 # MM requests to fire and immediately cancel +CANCEL_AFTER_SECONDS = 0.5 # how long to let them run before cancelling +POST_BURST_TIMEOUT = 5 # max wait for the fresh request after the storm + + +# --------------------------------------------------------------------------- +# Server fixture +# --------------------------------------------------------------------------- + + +@pytest.fixture(scope="module") +def server(): + """Real `vllm serve` with TP=2, eager backend, async MM encoder on, + running the nano-gv model. + + Module-scoped because server startup (worker spawn + encoder spawn + + warmup) is still the dominant cost and both tests in this module can + share a single server — they don't mutate executor state in ways that + need a clean restart between cases. + + `RemoteOpenAIServer` takes a `ModelInfo` directly and auto-adds + `--revision ` to the vllm serve args, so we get the pinned + revision plumbed through without touching `vllm_serve_args` here. + """ + # RemoteOpenAIServer merges env_dict over os.environ; we only need to + # pass the overrides here. + env_overrides = { + "SENDNN_INFERENCE_ASYNC_MM_ENCODER": "1", + "SENDNN_INFERENCE_DYNAMO_BACKEND": "eager", + # vLLM defaults VLLM_WORKER_MULTIPROC_METHOD=fork, which segfaults + # on macOS when the MM stack pulls in tvm_ffi (via xgrammar for + # structured outputs). spawn works on Linux too. + "VLLM_WORKER_MULTIPROC_METHOD": "spawn", + # Disable the synchronous rank-0 → all-ranks SHM-broadcast embedding + # share path — pre-existing bug that hangs during warmup with + # TP > 1 on macOS (torch.distributed.broadcast collective doesn't + # complete). The async encoder uses a different SHM path + # (collective_rpc store_mm_embeddings), so this only affects the + # inline warmup encoding. + "SENDNN_INFERENCE_TP_MM_SHARING": "0", + # Use a distinct MASTER_PORT so back-to-back runs with + # test_async_mm_encoder.py don't collide on port 12345 (torch.distributed + # holds it in TIME_WAIT for ~60s after teardown). + "MASTER_PORT": "12347", + } + + with RemoteOpenAIServer( + NANO_GV_MODEL, + vllm_serve_args=[ + "--tensor-parallel-size", + "2", + "--enforce-eager", + "--max-num-seqs", + "4", + # nano-gv image_size is 112, which yields ~64 vision patches + # + chat template overhead — 1024 is plenty. + "--max-model-len", + "1024", + "--no-enable-prefix-caching", + # served-model-name is what clients pass in the OpenAI API's + # `model` field. Use a short stable alias so tests don't need + # to know the tmp-path where nano-gv was built. + "--served-model-name", + "nano-gv", + ], + env_dict=env_overrides, + max_wait_seconds=600, + ) as srv: + yield srv + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _make_image_data_url(size_px: int = NANO_IMAGE_SIZE) -> str: + """Build a base64 data: URL for a synthetic image sized for nano-gv. + + Nano-gv expects 112×112 images (config.vision_config.image_size). + Image content is meaningless — random weights make output garbage. + """ + img = Image.new("RGB", (size_px, size_px), color=(120, 80, 200)) + buf = io.BytesIO() + img.save(buf, format="PNG") + return "data:image/png;base64," + base64.b64encode(buf.getvalue()).decode("ascii") + + +def _build_chat_messages(question: str = "Describe this image.") -> list[dict]: + return [ + { + "role": "user", + "content": [ + {"type": "image_url", "image_url": {"url": _make_image_data_url()}}, + {"type": "text", "text": question}, + ], + } + ] + + +async def _submit_and_cancel( + base_url: str, api_key: str, idx: int, cancel_after: float, stream: bool = True +) -> str: + """Submit one chat completion, then cancel it after `cancel_after` seconds. + + When ``stream=True`` (default) the request uses SSE so that closing the + client connection mid-stream causes a write failure on the server side. + That write failure propagates an HTTP disconnect to vLLM's + ``listen_for_disconnect`` coroutine which then calls abort_request — + populating the cancel queue so the encoder subprocess can skip the + abandoned encode jobs. This is the reliable cancel path. + + When ``stream=False`` the request is a plain POST. In vLLM ≥ v0.24.0 + (non-streaming harmony refactor) the server's ``abort()`` runs as + background cleanup after the handler task is cancelled; cancel signals + reach the encoder on a best-effort basis. The encoder may process some + or all jobs before the signals arrive, but nano-gv jobs are short enough + (~6 ms each) that the server drains within ``POST_BURST_TIMEOUT`` anyway. + + Returns a short status tag for logging: "cancelled" if we cancelled + cleanly, "completed_early" if the response beat us to the punch, + "errored:" if the server returned an error. + """ + payload = { + "model": "nano-gv", + "messages": _build_chat_messages(f"What is in image #{idx}?"), + "max_tokens": 64, + "temperature": 0.0, + "stream": stream, + } + headers = {"Authorization": f"Bearer {api_key}"} + + async def _do_request() -> str: + async with httpx.AsyncClient(timeout=httpx.Timeout(120)) as client: + if stream: + async with client.stream( + "POST", f"{base_url}/chat/completions", json=payload, headers=headers + ) as resp: + if resp.status_code != 200: + body = await resp.aread() + return f"errored:{resp.status_code}:{body[:120]!r}" + async for _ in resp.aiter_bytes(): + pass + return "completed_early" + else: + resp = await client.post( + f"{base_url}/chat/completions", json=payload, headers=headers + ) + return f"completed_early:{resp.status_code}" + + task = asyncio.create_task(_do_request()) + await asyncio.sleep(cancel_after) + if task.done(): + try: + return task.result() + except Exception as exc: + return f"errored:{type(exc).__name__}" + # Cancel the task. For stream=True this triggers aclose() on the httpx + # stream context manager, sending a TCP FIN/RST that uvicorn converts into + # an http.disconnect ASGI event. For stream=False, httpx abandons the + # connection; abort() runs async in the server as background cleanup. + task.cancel() + try: + await task + except asyncio.CancelledError: + return "cancelled" + except Exception as exc: + return f"errored:{type(exc).__name__}" + + +async def _fire_cancel_storm(base_url: str, api_key: str, n: int, stream: bool = True) -> list[str]: + """Fire N requests concurrently and cancel each after CANCEL_AFTER_SECONDS.""" + statuses = await asyncio.gather( + *( + _submit_and_cancel(base_url, api_key, i, CANCEL_AFTER_SECONDS, stream=stream) + for i in range(n) + ) + ) + return list(statuses) + + +# --------------------------------------------------------------------------- +# Test 1: server stays responsive after a cancellation storm +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize("stream", [True, False], ids=["streaming", "non_streaming"]) +def test_server_health_survives_cancel_storm(server: RemoteOpenAIServer, stream: bool): + """Hammer the server with MM requests, cancel them, then hit /health. + + Even if the encoder is wedged on abandoned jobs, the HTTP server + thread should remain responsive — this is the cheapest tripwire. + A non-200 health check post-storm is unambiguous evidence the engine + deadlocked. + + Parametrized over streaming and non-streaming cancels: both paths must + leave the server healthy. The encoder cancel-skip mechanism is exercised + more reliably by streaming (where TCP FIN triggers an immediate disconnect + event), but non-streaming aborts (best-effort in vLLM ≥ v0.24.0) must + not deadlock the engine either. + """ + asyncio.run( + _fire_cancel_storm(server.url_for("v1"), server.DUMMY_API_KEY, N_CANCELLED, stream=stream) + ) + + # The health endpoint should answer immediately. Give it a tiny grace + # period in case the in-flight cancellations are still draining. + deadline = time.time() + 30 + last_status = None + while time.time() < deadline: + try: + r = httpx.get(server.url_for("health"), timeout=5) + last_status = r.status_code + if r.status_code == 200: + return + except Exception as exc: + last_status = f"exception: {exc!r}" + time.sleep(1) + + pytest.fail( + f"server /health did not return 200 within 30s after a " + f"{N_CANCELLED}-request cancellation storm (last status: {last_status}). " + "Symptom of deadlock from finding 2/3 or encoder-pinning DoS from finding 1." + ) + + +# --------------------------------------------------------------------------- +# Test 2: a fresh MM request completes after the storm +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize("stream", [True, False], ids=["streaming", "non_streaming"]) +def test_fresh_request_completes_after_cancel_storm(server: RemoteOpenAIServer, stream: bool): + """The real bar: after the burst, can a legitimate MM request still + get through within a reasonable time? + + If the encoder is busy chewing through abandoned jobs (finding 1), this + request waits behind them and exceeds POST_BURST_TIMEOUT. If the + scheduler is wedged from put-failure stranding (finding 3), it never + leaves _mm_encoding_submitted and the request hangs forever. + + Either failure mode shows up as the OpenAI client timing out. + + Parametrized over streaming vs non-streaming cancels: + + - ``stream=True``: TCP FIN on cancel triggers an immediate disconnect + event; abort() fires before the encoder processes most jobs. The + encoder-skip path is exercised reliably — the fresh request should + return well under POST_BURST_TIMEOUT (soft threshold: 0.7×). + + - ``stream=False``: In vLLM ≥ v0.24.0, abort() runs as background + cleanup after task cancellation; cancel signals reach the encoder on + a best-effort basis. The encoder may process some or all of the 100 + abandoned jobs, but nano-gv jobs are short (~6 ms each) so the server + drains within POST_BURST_TIMEOUT regardless. Only the hard ceiling + is asserted here. + """ + asyncio.run( + _fire_cancel_storm(server.url_for("v1"), server.DUMMY_API_KEY, N_CANCELLED, stream=stream) + ) + + client = server.get_client(timeout=POST_BURST_TIMEOUT) + t0 = time.time() + try: + response = client.chat.completions.create( + model="nano-gv", + messages=_build_chat_messages("Describe this image."), + max_tokens=4, + temperature=0.0, + ) + except openai.APITimeoutError: + elapsed = time.time() - t0 + pytest.fail( + f"fresh MM request timed out after {elapsed:.1f}s following a " + f"{N_CANCELLED}-request {'streaming' if stream else 'non-streaming'} " + "cancellation storm. Most likely cause: " + "encoder subprocess is still encoding cancelled requests " + "(finding 1), or scheduler has a stranded request from a " + "swallowed put_nowait failure (finding 3)." + ) + + elapsed = time.time() - t0 + assert response.choices, f"empty response after {elapsed:.1f}s" + assert response.choices[0].message.content is not None, "no content in response" + + if stream and elapsed > POST_BURST_TIMEOUT * 0.7: + # With streaming cancels the encoder-skip path is reliable: TCP FIN + # triggers an immediate abort, so the encoder should skip most of the + # 100 abandoned jobs. A slow result here means the skip path is broken. + pytest.fail( + f"fresh MM request took {elapsed:.1f}s after streaming cancellation storm — " + f"under the {POST_BURST_TIMEOUT}s ceiling but in the danger zone. " + "Encoder is probably still draining abandoned jobs (finding 1)." + ) diff --git a/tests/multimodal/test_llava_next.py b/tests/multimodal/test_llava_next.py index e93e7badb..ed7d39f04 100644 --- a/tests/multimodal/test_llava_next.py +++ b/tests/multimodal/test_llava_next.py @@ -2,6 +2,12 @@ (Non e2e) tests related to Llava Next (Granite Vision); these tests primarily verify the correctness of some of the helper utils, especially with respect to the creation of warmup features. + +Runs against the nano-gv fixture — same LlavaNext + granite + siglip +architecture as the real granite-vision-3.2-2b but with tiny dims, so +`AutoConfig` / `AutoProcessor` loads are near-instant and no multi-GB +download is required. Only config / util shape logic is checked here; +model weights aren't loaded. """ import copy @@ -17,7 +23,7 @@ import sendnn_inference.multimodal as spyre_mm from spyre_util import REFERENCE_MODELS -GVISION_MODEL = REFERENCE_MODELS["ibm-granite/granite-vision-3.2-2b"] +NANO_GV_MODEL = REFERENCE_MODELS["joerunde/nano-gv"] # Marks all tests in this file as multimodal and CPU to match # multimodal wf; tests in this file should be very fast. pytestmark = [pytest.mark.multimodal, pytest.mark.cpu] @@ -26,25 +32,29 @@ # NOTE: --forked forks after module scoped fixtures @pytest.fixture(scope="module") def hf_config(): - """Get a transformers config for granite vision.""" + """Get a transformers config for the nano-gv llava-next variant.""" return AutoConfig.from_pretrained( - GVISION_MODEL.name, - revision=GVISION_MODEL.revision, + NANO_GV_MODEL.name, + revision=NANO_GV_MODEL.revision, ) @pytest.fixture(scope="module") def fms_config(hf_config): - """Get the FMS config corresponding to the above.""" + """Get the FMS config corresponding to the above. + + nano-gv already declares `head_dim` explicitly in its text_config, so + no manual override is needed — `build_llava_next_params` reads it + straight from the HF config. + """ config_params = build_llava_next_params(hf_config) - config_params["text_config"].head_dim = 128 return LlavaNextConfig(**config_params) @pytest.fixture(scope="module") def llava_next_mm_utils(fms_config, hf_config): return spyre_mm.maybe_get_mm_utils( - model_path=GVISION_MODEL.name, + model_path=NANO_GV_MODEL.name, fms_config=fms_config, hf_config=hf_config, ) @@ -138,7 +148,7 @@ def test_warmup_feature_correctness(llava_next_mm_utils): num_expanded_mm_ids = torch.sum(warmup_toks == image_token_id).item() - processor = AutoProcessor.from_pretrained(GVISION_MODEL.name, revision=GVISION_MODEL.revision) + processor = AutoProcessor.from_pretrained(NANO_GV_MODEL.name, revision=NANO_GV_MODEL.revision) # Create a random PIL Image that matches the size of the hardcoded # inputs and run it through the processor to check the feature sizes. image_dims = warmup_mm_features.data["image_sizes"].data diff --git a/tests/spyre_util.py b/tests/spyre_util.py index c177f5d57..ec0f47802 100644 --- a/tests/spyre_util.py +++ b/tests/spyre_util.py @@ -297,9 +297,16 @@ def register_model_info(name: str, revision: str, is_quantized: bool = False): revision="4b5990b8d402a75febe0086abbf1e490af494e3d", ) ### Multimodal +# nano-gv: random-init ~10 MB CI fixture built by tools/build_nano_gv.py, +# a shape-compatible stand-in for granite-vision-3.2-2b. The full model +# (~5.5 GB) is too large for the GHA cache and too slow for CPU eager +# warmup, and none of our multimodal tests actually depend on its +# trained weights — they only exercise config / wiring paths. +# When rebuilding, bump the revision here AND in +# .github/ci_model_cache.yaml so the CI cache stays in sync. register_model_info( - name="ibm-granite/granite-vision-3.2-2b", - revision="2818ae5b93cb750b099df1b65f7864e4a0401271", + name="joerunde/nano-gv", + revision="c9470d9e54b023dd9ab8a8a98057489fdb18ba03", ) register_model_info( name="mistralai/Mistral-Small-3.1-24B-Instruct-2503", @@ -330,9 +337,7 @@ def _default_test_models( return [pytest.param(model, marks=[pytest.mark.scoring], id=model.name)] if isMultimodal: - # NOTE: use 3.2 instead of 3.3 here since it's minimal case currently - # has fewer image tokens (1 tile + base patch instead of 2). - model = REFERENCE_MODELS["ibm-granite/granite-vision-3.2-2b"] + model = REFERENCE_MODELS["joerunde/nano-gv"] return [pytest.param(model, marks=[pytest.mark.multimodal], id=model.name)] # Decoders diff --git a/tests/v1/core/test_scheduler_mm_encoding.py b/tests/v1/core/test_scheduler_mm_encoding.py new file mode 100644 index 000000000..7e6a80d26 --- /dev/null +++ b/tests/v1/core/test_scheduler_mm_encoding.py @@ -0,0 +1,416 @@ +""" +Covers: + - Text-only requests are never gated by the MM encoding check + - Unencoded MM requests are not promoted to prefill until encoding is ready + - Text-only requests behind an unencoded MM request are still scheduled (no break) + - update_from_output promotes requests from submitted → ready + - Encoder failures abort the request immediately (no hang) + - finish_requests cleans up encoding state +""" + +import pytest +from unittest.mock import MagicMock, Mock, patch + +from vllm.sampling_params import SamplingParams +from vllm.v1.request import Request, RequestStatus + +from sendnn_inference.v1.core.scheduler import ChunkedPrefillSpyreScheduler + +pytestmark = [pytest.mark.multimodal, pytest.mark.cpu] + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +def _make_request(req_id, mm_features=None, num_tokens=10): + """Build a minimal Request for scheduler unit tests.""" + params = SamplingParams(max_tokens=20, temperature=0.0) + req = Request( + request_id=req_id, + sampling_params=params, + prompt_token_ids=list(range(num_tokens)), + arrival_time=0, + lora_request=None, + pooling_params=None, + ) + if mm_features is not None: + req.mm_features = mm_features + return req + + +@pytest.fixture +def scheduler(): + """Minimal ChunkedPrefillSpyreScheduler with heavy infra mocked out.""" + from vllm.v1.core.sched.output import CachedRequestData + from vllm.v1.core.sched.request_queue import FCFSRequestQueue + + with patch.object(ChunkedPrefillSpyreScheduler, "__init__", lambda self, *a, **k: None): + sched = ChunkedPrefillSpyreScheduler() + + sched.waiting = FCFSRequestQueue() + sched.skipped_waiting = FCFSRequestQueue() + sched.running = [] + sched.ongoing_prefills = [] + sched.chunk_size = 128 + sched.do_interleaving = False + sched.previous_step_was_prefill = False + sched.max_num_running_reqs = 8 + sched.tkv = 0 + sched.block_size = 64 + sched.total_reserved_blocks = 0 + sched.reserved_blocks = {} + sched.max_batch_tkv_limit = "8192" + sched.requests = {} # vLLM base Scheduler attribute: req_id → Request + sched._mm_encoding_submitted = set() + sched._mm_encoding_ready = set() + sched.paused_decoding_requests = [] + sched.request_last_decode_step = {} + sched.long_output_prio = False + sched.pause_events = 0 + sched.resume_events = 0 + sched._get_required_blocks = lambda req, *a, **k: (0, 0) + sched._get_free_blocks = lambda: 100 + sched.kv_cache_manager = Mock() + sched.kv_cache_manager.get_computed_blocks.return_value = (None, 0) + sched.kv_cache_manager.block_pool = Mock() + # scheduler_config needed by _current_chunk_token_threshold inside schedule() + sched.scheduler_config = Mock() + sched.scheduler_config.long_prefill_token_threshold = 128 + + mock_output = Mock() + mock_output.num_scheduled_tokens = {} + mock_output.scheduled_new_reqs = [] + mock_output.scheduled_cached_reqs = CachedRequestData.make_empty() + + with ( + patch("vllm.v1.core.sched.scheduler.Scheduler.schedule", return_value=mock_output), + patch("vllm.v1.core.sched.scheduler.Scheduler.finish_requests", return_value=[]), + patch("vllm.v1.core.sched.scheduler.Scheduler.update_from_output", return_value=None), + ): + yield sched + + +# --------------------------------------------------------------------------- +# can_schedule_prefill: MM encoding gate +# --------------------------------------------------------------------------- + + +class TestCanSchedulePrefill: + def test_text_request_never_gated(self, scheduler): + """Text-only request (no mm_features) must not be affected by encoding state.""" + req = _make_request("text-1") + # Even with async encoder enabled and nothing in _mm_encoding_ready: + # Stub remaining checks so only the MM gate matters + with ( + patch("sendnn_inference.envs.SENDNN_INFERENCE_ASYNC_MM_ENCODER", True), + patch.object(scheduler, "_has_scheduling_priority", return_value=True), + patch.object(scheduler, "_satisfies_constraints", return_value=True), + ): + scheduler.running = [] + scheduler.waiting.append(Mock()) # non-empty so len check triggers + assert scheduler.can_schedule_prefill(req) is True + + def test_mm_request_gated_when_encoding_not_ready(self, scheduler): + """MM request must return False when async mode on and req not in _mm_encoding_ready.""" + req = _make_request("mm-1", mm_features=[Mock()]) + scheduler._mm_encoding_ready = set() + + with patch("sendnn_inference.envs.SENDNN_INFERENCE_ASYNC_MM_ENCODER", True): + assert scheduler.can_schedule_prefill(req) is False + + def test_mm_request_passes_when_encoding_ready(self, scheduler): + """MM request must pass the gate once its req_id is in _mm_encoding_ready.""" + req = _make_request("mm-ready", mm_features=[Mock()]) + scheduler._mm_encoding_ready = {"mm-ready"} + + with ( + patch("sendnn_inference.envs.SENDNN_INFERENCE_ASYNC_MM_ENCODER", True), + patch.object(scheduler, "_has_scheduling_priority", return_value=True), + patch.object(scheduler, "_satisfies_constraints", return_value=True), + ): + scheduler.running = [] + assert scheduler.can_schedule_prefill(req) is True + + def test_mm_gate_inactive_without_async_encoder(self, scheduler): + """In non-async mode the gate must not block MM requests.""" + req = _make_request("mm-sync", mm_features=[Mock()]) + scheduler._mm_encoding_ready = set() # nothing ready + + with ( + patch("sendnn_inference.envs.SENDNN_INFERENCE_ASYNC_MM_ENCODER", False), + patch.object(scheduler, "_has_scheduling_priority", return_value=True), + patch.object(scheduler, "_satisfies_constraints", return_value=True), + ): + scheduler.running = [] + assert scheduler.can_schedule_prefill(req) is True + + +# --------------------------------------------------------------------------- +# schedule(): text requests not blocked by unencoded MM at queue head +# --------------------------------------------------------------------------- + + +class TestScheduleMixedQueue: + def _capture_prefill_candidates(self, scheduler, *args, **kwargs): + """Side effect for Scheduler.schedule: capture self.waiting at call time. + + Requests in self.waiting when super().schedule() is called are the ones + promoted for prefill this step (line ~426: self.waiting.append(new_request)). + Requests added by the post-loop drain (line ~546) are for future steps. + """ + self._waiting_at_super_call = list(scheduler.waiting) + return self._mock_output + + def test_text_request_scheduled_behind_unencoded_mm(self, scheduler): + """With an unencoded MM request at the head, a text-only request + further back must still be promoted to prefill in the same step.""" + mm_req = _make_request("mm-not-ready", mm_features=[Mock()]) + text_req = _make_request("text-behind") + + scheduler.waiting.append(mm_req) + scheduler.waiting.append(text_req) + scheduler._mm_encoding_ready = set() + + self._waiting_at_super_call = [] + self._mock_output = Mock() + self._mock_output.num_scheduled_tokens = {} + self._mock_output.scheduled_new_reqs = [] + from vllm.v1.core.sched.output import CachedRequestData + + self._mock_output.scheduled_cached_reqs = CachedRequestData.make_empty() + + with ( + patch("sendnn_inference.envs.SENDNN_INFERENCE_ASYNC_MM_ENCODER", True), + patch.object(scheduler, "_has_scheduling_priority", return_value=True), + patch.object(scheduler, "_satisfies_constraints", return_value=True), + patch( + "vllm.v1.core.sched.scheduler.Scheduler.schedule", + side_effect=lambda **kwargs: self._capture_prefill_candidates(scheduler), + ), + ): + scheduler.schedule() + + scheduled_ids = {r.request_id for r in self._waiting_at_super_call} + assert "text-behind" in scheduled_ids, "text-only request must be promoted to prefill" + assert "mm-not-ready" not in scheduled_ids, "unencoded MM must not be promoted this step" + + def test_unencoded_mm_not_promoted_to_waiting(self, scheduler): + """An MM request whose encoding isn't ready must not be promoted for prefill.""" + mm_req = _make_request("mm-held", mm_features=[Mock()]) + scheduler.waiting.append(mm_req) + scheduler._mm_encoding_ready = set() + + self._waiting_at_super_call = [] + self._mock_output = Mock() + self._mock_output.num_scheduled_tokens = {} + self._mock_output.scheduled_new_reqs = [] + from vllm.v1.core.sched.output import CachedRequestData + + self._mock_output.scheduled_cached_reqs = CachedRequestData.make_empty() + + with ( + patch("sendnn_inference.envs.SENDNN_INFERENCE_ASYNC_MM_ENCODER", True), + patch.object(scheduler, "_has_scheduling_priority", return_value=True), + patch.object(scheduler, "_satisfies_constraints", return_value=True), + patch( + "vllm.v1.core.sched.scheduler.Scheduler.schedule", + side_effect=lambda **kwargs: self._capture_prefill_candidates(scheduler), + ), + ): + scheduler.schedule() + + scheduled_ids = {r.request_id for r in self._waiting_at_super_call} + assert "mm-held" not in scheduled_ids + + def test_mm_encode_requests_emitted_for_unencoded_mm(self, scheduler): + """schedule() must emit an MMEncodeRequest for each unencoded MM request.""" + mm_req = _make_request("mm-enc", mm_features=[Mock()]) + scheduler.waiting.append(mm_req) + scheduler._mm_encoding_ready = set() + scheduler._mm_encoding_submitted = set() + + with ( + patch("sendnn_inference.envs.SENDNN_INFERENCE_ASYNC_MM_ENCODER", True), + patch.object(scheduler, "_has_scheduling_priority", return_value=True), + patch.object(scheduler, "_satisfies_constraints", return_value=True), + ): + output = scheduler.schedule() + + encode_reqs = getattr(output, "_spyre_mm_encode_requests", []) + assert len(encode_reqs) == 1 + assert encode_reqs[0].request_id == "mm-enc" + assert "mm-enc" in scheduler._mm_encoding_submitted + + def test_encode_not_re_submitted_when_already_submitted(self, scheduler): + """A request already in _mm_encoding_submitted must not be re-emitted.""" + mm_req = _make_request("mm-dup", mm_features=[Mock()]) + scheduler.waiting.append(mm_req) + scheduler._mm_encoding_submitted = {"mm-dup"} # already submitted + scheduler._mm_encoding_ready = set() + + with ( + patch("sendnn_inference.envs.SENDNN_INFERENCE_ASYNC_MM_ENCODER", True), + patch.object(scheduler, "_has_scheduling_priority", return_value=True), + patch.object(scheduler, "_satisfies_constraints", return_value=True), + ): + output = scheduler.schedule() + + encode_reqs = getattr(output, "_spyre_mm_encode_requests", []) + assert len(encode_reqs) == 0 + + +# --------------------------------------------------------------------------- +# update_from_output: encoding state transitions +# --------------------------------------------------------------------------- + + +class TestUpdateFromOutput: + def _make_model_output(self): + from sendnn_inference.v1.worker.spyre_model_runner import SpyreModelRunnerOutput + + out = SpyreModelRunnerOutput.__new__(SpyreModelRunnerOutput) + out.req_ids = [] + out.req_id_to_index = {} + out.sampled_token_ids = [] + out.spec_token_ids = None + out.logprobs = None + out.prompt_logprobs_dict = {} + out.pooler_output = [] + out.tkv = 0 + out.left_padding = {} + out.prefix_cache_hit_len = {} + return out + + def _make_sched_out(self, newly_encoded=None, failed=None): + """Build a minimal scheduler_output mock with explicit list attributes.""" + out = Mock() + out._spyre_newly_encoded_req_ids = newly_encoded or [] + out._spyre_failed_encode_req_ids = failed or [] + return out + + def test_newly_encoded_moves_to_ready(self, scheduler): + """_spyre_newly_encoded_req_ids must move req_ids to _mm_encoding_ready + when the request is still live (present in scheduler.requests).""" + scheduler._mm_encoding_submitted = {"req-A"} + scheduler._mm_encoding_ready = set() + scheduler.finished_req_ids = set() + scheduler.requests["req-A"] = Mock() # request still alive + + scheduler.update_from_output( + self._make_sched_out(newly_encoded=["req-A"]), + self._make_model_output(), + ) + + assert "req-A" in scheduler._mm_encoding_ready + assert "req-A" not in scheduler._mm_encoding_submitted + + def test_newly_encoded_skips_ready_for_aborted_request(self, scheduler): + """If a request was aborted while encoding, its late result must NOT + add a stale entry to _mm_encoding_ready.""" + scheduler._mm_encoding_submitted = {"req-aborted"} + scheduler._mm_encoding_ready = set() + scheduler.finished_req_ids = set() + # req-aborted is NOT in scheduler.requests (already removed by finish_requests) + + scheduler.update_from_output( + self._make_sched_out(newly_encoded=["req-aborted"]), + self._make_model_output(), + ) + + assert "req-aborted" not in scheduler._mm_encoding_ready + assert "req-aborted" not in scheduler._mm_encoding_submitted + + def test_failed_encode_aborts_request(self, scheduler): + """_spyre_failed_encode_req_ids must abort the request immediately.""" + scheduler._mm_encoding_submitted = {"req-fail"} + scheduler._mm_encoding_ready = set() + scheduler.finished_req_ids = set() + + with patch.object(scheduler, "finish_requests") as mock_finish: + scheduler.update_from_output( + self._make_sched_out(failed=["req-fail"]), + self._make_model_output(), + ) + + mock_finish.assert_called_once_with(["req-fail"], RequestStatus.FINISHED_ABORTED) + + +# --------------------------------------------------------------------------- +# finish_requests: encoding state cleanup +# --------------------------------------------------------------------------- + + +class TestFinishRequests: + def test_finish_clears_encoding_state(self, scheduler): + """finish_requests must clean up _mm_encoding_submitted and _mm_encoding_ready.""" + scheduler._mm_encoding_submitted = {"req-1", "req-2"} + scheduler._mm_encoding_ready = {"req-2"} + + with patch("vllm.v1.core.sched.scheduler.Scheduler.finish_requests", return_value=[]): + scheduler.finish_requests(["req-1", "req-2"], RequestStatus.FINISHED_STOPPED) + + assert "req-1" not in scheduler._mm_encoding_submitted + assert "req-2" not in scheduler._mm_encoding_submitted + assert "req-2" not in scheduler._mm_encoding_ready + + def test_finish_puts_submitted_req_on_cancel_queue(self, scheduler): + """finish_requests must put the req_id on the cancel queue for any + request that is in _mm_encoding_submitted so the encoder can skip it.""" + from sendnn_inference.v1.executor.spyre_executor import SpyreMultiprocExecutor + + scheduler._mm_encoding_submitted = {"req-cancel"} + scheduler._mm_encoding_ready = set() + + mock_cq = MagicMock() + with ( + patch("vllm.v1.core.sched.scheduler.Scheduler.finish_requests", return_value=[]), + patch.object(SpyreMultiprocExecutor, "get_mm_cancel_queue", return_value=mock_cq), + ): + scheduler.finish_requests(["req-cancel"], RequestStatus.FINISHED_ABORTED) + + mock_cq.put_nowait.assert_called_with("req-cancel") + assert "req-cancel" not in scheduler._mm_encoding_submitted + + def test_scheduler_finish_requests_notifies_executor_of_cancellation(self, scheduler): + """ChunkedPrefillSpyreScheduler.finish_requests must, in addition to its + local _mm_encoding_* cleanup, tell the bound executor to cancel any + in-flight encode jobs. + + Without this, the encode job is orphaned in the queue and still consumed + by the encoder — wasting CPU/NNPA on a dead request (DoS vector for large + images: submit N, cancel N, encoder encodes all N serially). + + The notification is sent via the cancel queue exposed by + SpyreMultiprocExecutor.get_mm_cancel_queue(); the encoder subprocess drains + it before each job and skips any req_id it finds there. + """ + from sendnn_inference.v1.executor.spyre_executor import SpyreMultiprocExecutor + + scheduler._mm_encoding_submitted = {"req-aborted"} + scheduler._mm_encoding_ready = set() + + mock_cq = MagicMock() + with ( + patch("vllm.v1.core.sched.scheduler.Scheduler.finish_requests", return_value=[]), + patch.object(SpyreMultiprocExecutor, "get_mm_cancel_queue", return_value=mock_cq), + ): + scheduler.finish_requests(["req-aborted"], RequestStatus.FINISHED_ABORTED) + + assert mock_cq.put_nowait.called, ( + "scheduler.finish_requests did not notify the encoder of the cancelled " + "request; the encoder will still process it (DoS via cancelled large images)." + ) + mock_cq.put_nowait.assert_called_with("req-aborted") + + def test_finish_all_clears_everything(self, scheduler): + """finish_requests(None, …) must clear all encoding sets.""" + scheduler._mm_encoding_submitted = {"req-A", "req-B"} + scheduler._mm_encoding_ready = {"req-C"} + + with patch("vllm.v1.core.sched.scheduler.Scheduler.finish_requests", return_value=[]): + scheduler.finish_requests(None, RequestStatus.FINISHED_STOPPED) + + assert len(scheduler._mm_encoding_submitted) == 0 + assert len(scheduler._mm_encoding_ready) == 0 diff --git a/tests/v1/executor/test_spyre_executor.py b/tests/v1/executor/test_spyre_executor.py new file mode 100644 index 000000000..589c07f6a --- /dev/null +++ b/tests/v1/executor/test_spyre_executor.py @@ -0,0 +1,326 @@ +""" +Tests cover: + - execute_model: job submission, result draining, collective_rpc, SHM cleanup, + scheduler_output annotations, error result handling, no-op when + no queue is present + - collective_rpc: triggers _try_start_mm_encoder only after warmup + - _try_start_mm_encoder: env flag gate, startup failure → _cleanup_encoder + - _cleanup_encoder: resets all queue/process state +""" + +import queue as queue_mod +from unittest.mock import MagicMock, patch + +import pytest +import torch + +from vllm.v1.executor.multiproc_executor import MultiprocExecutor + +from sendnn_inference.v1.executor.spyre_executor import SpyreMultiprocExecutor + +pytestmark = [pytest.mark.multimodal, pytest.mark.cpu] + + +# --------------------------------------------------------------------------- +# Fixture: executor with parent class mocked out +# --------------------------------------------------------------------------- + + +@pytest.fixture +def executor(): + """SpyreMultiprocExecutor instance with MultiprocExecutor internals mocked. + + _init_executor, execute_model, collective_rpc, and shutdown are all replaced + so no real workers are started. The fixture yields the instance with patches + still active so methods can be called safely during the test. + """ + parent_execute = MagicMock(return_value=MagicMock()) + parent_collective_rpc = MagicMock(return_value=None) + + with ( + patch.object(MultiprocExecutor, "_init_executor", return_value=None), + patch.object(MultiprocExecutor, "execute_model", parent_execute), + patch.object(MultiprocExecutor, "collective_rpc", parent_collective_rpc), + patch.object(MultiprocExecutor, "shutdown", return_value=None), + ): + exc = SpyreMultiprocExecutor.__new__(SpyreMultiprocExecutor) + exc.vllm_config = MagicMock() + exc._init_executor() + # expose parent mocks for assertion + exc._parent_execute = parent_execute + exc._parent_collective_rpc = parent_collective_rpc + yield exc + + +def _install_queues(exc, result_items=None): + """Install mock queues on the executor. + + Uses MagicMock rather than real multiprocessing.Queue so get_nowait() + returns preset items deterministically without IPC timing concerns. + result_items: list of items get_nowait() should return before raising Empty. + """ + job_q = MagicMock() + result_q = MagicMock() + side = list(result_items or []) + [queue_mod.Empty()] + result_q.get_nowait.side_effect = side + exc._mm_job_queue = job_q + exc._mm_result_queue = result_q + exc._mm_in_flight = 0 + + +def _make_encode_req(req_id="req-1"): + from sendnn_inference.v1.core.scheduler import MMEncodeRequest + + return MMEncodeRequest(request_id=req_id, prompt_token_ids=[1, 2, 3], mm_features=[]) + + +def _make_scheduler_output(encode_reqs=None): + out = MagicMock() + out._spyre_mm_encode_requests = encode_reqs or [] + return out + + +# --------------------------------------------------------------------------- +# execute_model +# --------------------------------------------------------------------------- + + +class TestExecuteModel: + def test_no_op_when_queue_not_initialised(self, executor): + """When _mm_job_queue is None, execute_model must just delegate to super().""" + sched = _make_scheduler_output() + executor.execute_model(sched) + executor._parent_execute.assert_called_once() + + def test_jobs_submitted_to_queue(self, executor): + _install_queues(executor) + r1 = _make_encode_req("r1") + r2 = _make_encode_req("r2") + sched = _make_scheduler_output([r1, r2]) + + executor.execute_model(sched) + + assert executor._mm_in_flight == 2 + put_calls = executor._mm_job_queue.put_nowait.call_args_list + assert len(put_calls) == 2 + assert put_calls[0][0][0] is r1 + assert put_calls[1][0][0] is r2 + + def test_encode_requests_cleared_before_super(self, executor): + """_spyre_mm_encode_requests must be emptied before super().execute_model.""" + _install_queues(executor) + sched = _make_scheduler_output([_make_encode_req()]) + executor.execute_model(sched) + assert sched._spyre_mm_encode_requests == [] + + def test_successful_result_triggers_store_and_cleanup(self, executor): + """When a result is drained, collective_rpc + cleanup must fire.""" + shape = (1, 4, 8) + dtype = torch.float16 + _install_queues(executor, result_items=[("req-done", shape, dtype)]) + executor._mm_in_flight = 1 + + sched = _make_scheduler_output() + + with patch( + "sendnn_inference.v1.executor.spyre_executor.cleanup_embeddings_by_name" + ) as mock_cleanup: + executor.execute_model(sched) + + # super().collective_rpc() is called with method + forwarded kwargs + executor._parent_collective_rpc.assert_called_once() + rpc_call = executor._parent_collective_rpc.call_args + assert rpc_call[0][0] == "store_mm_embeddings" + # args is passed as a tuple wrapping the metadata list: args=([...],) + assert rpc_call[1]["args"][0] == [("req-done", shape, dtype)] + + mock_cleanup.assert_called_once_with("req-done") + assert sched._spyre_newly_encoded_req_ids == ["req-done"] + assert executor._mm_in_flight == 0 + + def test_error_result_sets_failed_req_ids_for_scheduler_retry(self, executor): + """(req_id, None, None) must be collected into _spyre_failed_encode_req_ids + so the scheduler can clear _mm_encoding_submitted and allow retry.""" + _install_queues(executor, result_items=[("req-err", None, None)]) + executor._mm_in_flight = 1 + + sched = _make_scheduler_output() + + with patch("sendnn_inference.v1.executor.spyre_executor.cleanup_embeddings_by_name"): + executor.execute_model(sched) + + # collective_rpc must NOT be called (no valid metadata) + executor._parent_collective_rpc.assert_not_called() + # Failed req_id must be surfaced so scheduler can clear submitted state + assert sched._spyre_failed_encode_req_ids == ["req-err"] + + def test_put_nowait_failure_aborts_request_via_failed_req_ids(self, executor): + """PR finding: put_nowait failure must surface as _spyre_failed_encode_req_ids. + + If put_nowait raises (BrokenPipeError, queue.Full, PicklingError, …) and the + exception is silently swallowed, the req_id stays in _mm_encoding_submitted + forever — the scheduler gates prefill on _mm_encoding_ready that is never + populated, stranding the client indefinitely with no error. + """ + _install_queues(executor) + executor._mm_job_queue.put_nowait.side_effect = BrokenPipeError("encoder dead") + + r1 = _make_encode_req("req-broken") + sched = _make_scheduler_output([r1]) + + executor.execute_model(sched) + + # in_flight must NOT be incremented (job was never submitted) + assert executor._mm_in_flight == 0 + # failed req_id must be surfaced so scheduler aborts it + assert sched._spyre_failed_encode_req_ids == ["req-broken"] + + def test_in_flight_not_incremented_on_put_failure(self, executor): + """Regression guard: the bookkeeping counter must stay consistent with + what is actually in the queue. If put_nowait raised, the job is NOT in + flight and the counter must stay where it was. + + Today's behaviour is correct (increment follows the call), but this test + locks it in so a refactor that moves the increment above the call is caught. + """ + _install_queues(executor) + executor._mm_in_flight = 3 # simulate existing in-flight jobs + executor._mm_job_queue.put_nowait.side_effect = RuntimeError("queue broken") + + sched = _make_scheduler_output([_make_encode_req("req-fail")]) + executor.execute_model(sched) + + assert executor._mm_in_flight == 3 + + def test_queue_full_during_put_surfaces_to_scheduler(self, executor): + """Defense-in-depth: once the encoder subprocess dies, multiprocessing.Queue + will eventually raise queue.Full (the parent feeder buffer fills because + nothing consumes it). The executor must surface this the same way it surfaces + BrokenPipeError — otherwise silent stranding becomes the dominant failure mode + for a dead encoder. + """ + _install_queues(executor) + executor._mm_job_queue.put_nowait.side_effect = queue_mod.Full() + + sched = _make_scheduler_output([_make_encode_req("req-full")]) + executor.execute_model(sched) + + assert executor._mm_in_flight == 0 + assert sched._spyre_failed_encode_req_ids == ["req-full"] + + def test_in_flight_zero_skips_result_drain(self, executor): + """When _mm_in_flight == 0, the result queue must not be polled.""" + # Even though the queue conceptually has an item, _mm_in_flight==0 + # should prevent any get_nowait() call. + _install_queues(executor, result_items=[("req-sneaky", (1, 4, 8), torch.float16)]) + executor._mm_in_flight = 0 + + sched = _make_scheduler_output() + executor.execute_model(sched) + + executor._mm_result_queue.get_nowait.assert_not_called() + executor._parent_collective_rpc.assert_not_called() + + +# --------------------------------------------------------------------------- +# collective_rpc +# --------------------------------------------------------------------------- + + +class TestCollectiveRpc: + def test_triggers_encoder_start_after_warmup(self, executor): + """collective_rpc("compile_or_warm_up_model") must call _try_start_mm_encoder.""" + with patch.object(executor, "_try_start_mm_encoder") as mock_start: + executor.collective_rpc("compile_or_warm_up_model") + mock_start.assert_called_once() + + def test_does_not_trigger_for_other_methods(self, executor): + """Other collective_rpc calls must not start the encoder.""" + with patch.object(executor, "_try_start_mm_encoder") as mock_start: + executor.collective_rpc("store_mm_embeddings", args=([],)) + executor.collective_rpc("load_model") + mock_start.assert_not_called() + + def test_does_not_trigger_if_encoder_already_running(self, executor): + """If _mm_encoder_proc is already set, encoder must not be re-started.""" + executor._mm_encoder_proc = MagicMock() # simulate already running + with patch.object(executor, "_try_start_mm_encoder") as mock_start: + executor.collective_rpc("compile_or_warm_up_model") + mock_start.assert_not_called() + + +# --------------------------------------------------------------------------- +# _try_start_mm_encoder +# --------------------------------------------------------------------------- + + +class TestTryStartMmEncoder: + def test_skips_when_env_not_set(self, executor): + """SENDNN_INFERENCE_ASYNC_MM_ENCODER=False → no process started.""" + with patch("sendnn_inference.envs.SENDNN_INFERENCE_ASYNC_MM_ENCODER", False): + executor._try_start_mm_encoder() + + assert executor._mm_encoder_proc is None + assert executor._mm_job_queue is None + + def test_raises_on_startup_failure(self, executor): + """PR #1015 finding 2: encoder startup failure must raise, not silently fall back. + + A silent fallback leaves MM scheduling permanently broken — the scheduler + keeps gating MM requests on _mm_encoding_ready which is never populated, + so every MM request hangs indefinitely. Raising here lets the supervisor + restart the process with a clear error instead of masking the failure. + """ + fake_result_q = MagicMock() + fake_result_q.get.return_value = "ERROR: vision load failed" + + fake_ctx = MagicMock() + fake_ctx.Queue.side_effect = [MagicMock(), MagicMock(), fake_result_q] + fake_ctx.Event.return_value = MagicMock() + fake_ctx.Process.return_value = MagicMock() + + with ( + patch("sendnn_inference.envs.SENDNN_INFERENCE_ASYNC_MM_ENCODER", True), + patch("multiprocessing.get_context", return_value=fake_ctx), + patch("sendnn_inference.v1.worker.mm_encoder_process.encoder_process_main"), + pytest.raises(RuntimeError, match="Encoder process startup failed"), + ): + executor._try_start_mm_encoder() + + +# --------------------------------------------------------------------------- +# _cleanup_encoder +# --------------------------------------------------------------------------- + + +class TestCleanupEncoder: + def test_resets_all_state(self, executor): + """_cleanup_encoder must nil out all queue/process references.""" + executor._mm_stop_event = MagicMock() + executor._mm_encoder_proc = MagicMock() + executor._mm_encoder_proc.is_alive.return_value = False + executor._mm_job_queue = MagicMock() + executor._mm_result_queue = MagicMock() + executor._mm_in_flight = 5 + + executor._cleanup_encoder() + + assert executor._mm_encoder_proc is None + assert executor._mm_job_queue is None + assert executor._mm_result_queue is None + assert executor._mm_stop_event is None + assert executor._mm_in_flight == 0 + + def test_terminates_live_process(self, executor): + """A still-alive encoder process must be terminated if join times out.""" + stop = MagicMock() + proc = MagicMock() + proc.is_alive.side_effect = [True, True] # alive before and after join + executor._mm_stop_event = stop + executor._mm_encoder_proc = proc + + executor._cleanup_encoder() + + stop.set.assert_called_once() + proc.join.assert_called_once() + proc.terminate.assert_called_once() diff --git a/tests/v1/worker/test_mm_encoder_process.py b/tests/v1/worker/test_mm_encoder_process.py new file mode 100644 index 000000000..0db4c3dd0 --- /dev/null +++ b/tests/v1/worker/test_mm_encoder_process.py @@ -0,0 +1,397 @@ +""" +Tests cover: + - _resolve_mm_utils_cls: exact registry match, model_type scan fallback, unknown raises + - VisionEncoderRunner.__init__: local path → model_path kwarg, HF ID → variant kwarg + - VisionEncoderRunner.execute_model: dtype cast, CPU output + - encoder_process_main: READY signal, ERROR signal, job processed, stop_event, encode error +""" + +import multiprocessing +from unittest.mock import MagicMock, patch + +import pytest +import torch + +pytestmark = [pytest.mark.multimodal, pytest.mark.cpu] + + +# --------------------------------------------------------------------------- +# helpers +# --------------------------------------------------------------------------- + + +def _make_vllm_config(model="fake-model"): + cfg = MagicMock() + cfg.model_config.model = model + cfg.model_config.revision = None + return cfg + + +def _make_mm_encode_request(req_id="req-1"): + from sendnn_inference.v1.core.scheduler import MMEncodeRequest + + return MMEncodeRequest(request_id=req_id, prompt_token_ids=[1, 2, 3], mm_features=[]) + + +# --------------------------------------------------------------------------- +# _resolve_mm_utils_cls +# --------------------------------------------------------------------------- + + +class TestResolveMmUtilsCls: + def test_model_type_scan_fallback_for_base_pretrained_config(self): + """When hf_config is a base PretrainedConfig (Pydantic serialization loss), + the fallback scan by model_type string must still find the right class.""" + from transformers import PretrainedConfig + + from sendnn_inference.v1.worker.mm_encoder_process import _resolve_mm_utils_cls + + class KnownHfConfig(PretrainedConfig): + model_type = "known_scan" + + KnownUtils = type("KnownUtils", (), {}) + + # base PretrainedConfig carrying model_type only (class info lost) + base_cfg = PretrainedConfig() + base_cfg.model_type = "known_scan" + + with patch("sendnn_inference.multimodal.MM_HF_CFG_REGISTRY", {KnownHfConfig: KnownUtils}): + result = _resolve_mm_utils_cls(base_cfg) + + assert result is KnownUtils + + def test_unknown_model_type_raises_value_error(self): + from transformers import PretrainedConfig + + from sendnn_inference.v1.worker.mm_encoder_process import _resolve_mm_utils_cls + + unknown_cfg = PretrainedConfig() + unknown_cfg.model_type = "totally_unknown_xyz" + + with ( + patch("sendnn_inference.multimodal.MM_HF_CFG_REGISTRY", {}), + pytest.raises(ValueError, match="no MMUtils found"), + ): + _resolve_mm_utils_cls(unknown_cfg) + + +# --------------------------------------------------------------------------- +# VisionEncoderRunner — __init__ (model loading path selection) +# --------------------------------------------------------------------------- + + +def _make_runner(vllm_config): + """Construct a VisionEncoderRunner with all heavy deps mocked. + Returns (runner, get_model_mock).""" + from sendnn_inference.platform import SpyrePlatform + from sendnn_inference.v1.worker.mm_encoder_process import VisionEncoderRunner + + fake_fms_model = MagicMock() + fake_utils_cls = MagicMock() + fake_utils_cls.__name__ = "FakeUtils" + fake_utils_cls.mm_parameter_prefixes = ("vision_tower.",) + get_model_mock = MagicMock(return_value=fake_fms_model) + + with ( + patch("fms.models.get_model", get_model_mock), + patch.object(SpyrePlatform, "maybe_ensure_sendnn_configured"), + patch( + "sendnn_inference.v1.worker.mm_encoder_process.SpyreCausalLM.resolve_hf_config", + return_value=MagicMock(), + ), + patch( + "sendnn_inference.v1.worker.mm_encoder_process._resolve_mm_utils_cls", + return_value=fake_utils_cls, + ), + patch( + "sendnn_inference.v1.worker.mm_encoder_process.cast_params_for_spyre", + return_value="cpu", + ), + ): + runner = VisionEncoderRunner(vllm_config) + + return runner, get_model_mock + + +class TestVisionEncoderRunnerInit: + def test_local_path_passes_model_path_kwarg(self, tmp_path): + """When model is a local directory, get_model must receive model_path=.""" + cfg = _make_vllm_config(model=str(tmp_path)) + _, get_model_mock = _make_runner(cfg) + + _, kwargs = get_model_mock.call_args + assert "model_path" in kwargs + assert kwargs["model_path"] == str(tmp_path) + assert "variant" not in kwargs + + def test_hf_id_passes_variant_kwarg(self): + """When model is a non-local HF ID, get_model must receive variant=.""" + cfg = _make_vllm_config(model="org/model-name") + + with patch("os.path.isdir", return_value=False): + _, get_model_mock = _make_runner(cfg) + + _, kwargs = get_model_mock.call_args + assert "variant" in kwargs + assert kwargs["variant"] == "org/model-name" + assert "model_path" not in kwargs + + def test_vision_only_and_fused_weights_always_set(self, tmp_path): + """vision_only=True and fused_weights=False must always be passed.""" + cfg = _make_vllm_config(model=str(tmp_path)) + _, get_model_mock = _make_runner(cfg) + + _, kwargs = get_model_mock.call_args + assert kwargs.get("vision_only") is True + assert kwargs.get("fused_weights") is False + + +# --------------------------------------------------------------------------- +# VisionEncoderRunner — execute_model +# --------------------------------------------------------------------------- + + +class TestVisionEncoderRunnerExecuteModel: + def _make_runner_direct(self): + """Build a VisionEncoderRunner instance bypassing __init__.""" + from sendnn_inference.v1.worker.mm_encoder_process import VisionEncoderRunner + + runner = VisionEncoderRunner.__new__(VisionEncoderRunner) + runner._decoder_dtype = torch.float16 + runner.mm_device = "cpu" + runner.mm_utils_cls = MagicMock() + runner.fms_model = MagicMock() + return runner + + def test_output_is_float16_cpu_contiguous(self): + """execute_model must cast to _decoder_dtype and return a CPU tensor.""" + runner = self._make_runner_direct() + # Simulate vision encoder returning float32 + raw_embeds = torch.ones(1, 8, 16, dtype=torch.float32) + runner.mm_utils_cls.get_maybe_mm_embeddings.return_value = raw_embeds + + job = _make_mm_encode_request() + result = runner.execute_model(job) + + assert result.dtype == torch.float16 + assert result.device.type == "cpu" + assert result.is_contiguous() + + def test_input_ids_built_from_prompt_token_ids(self): + """execute_model must pass the job's prompt_token_ids as input_ids.""" + runner = self._make_runner_direct() + runner.mm_utils_cls.get_maybe_mm_embeddings.return_value = torch.zeros( + 1, 4, 8, dtype=torch.float16 + ) + job = _make_mm_encode_request() + job.prompt_token_ids = [10, 20, 30] + + runner.execute_model(job) + + call_kwargs = runner.mm_utils_cls.get_maybe_mm_embeddings.call_args + input_ids = call_kwargs[0][1] # second positional arg + assert input_ids.shape == (1, 3) + assert input_ids.tolist() == [[10, 20, 30]] + + +# --------------------------------------------------------------------------- +# encoder_process_main +# --------------------------------------------------------------------------- + + +class TestEncoderProcessMain: + """Tests for the encoder subprocess entry point. + + encoder_process_main is called directly (not via subprocess) with mocked + VisionEncoderRunner so no model weights are loaded. + """ + + def test_ready_signal_on_success(self): + from sendnn_inference.v1.worker.mm_encoder_process import encoder_process_main + + jq = multiprocessing.Queue() + rq = multiprocessing.Queue() + stop = multiprocessing.Event() + jq.put(None) # sentinel → loop exits immediately after READY + + with patch("sendnn_inference.v1.worker.mm_encoder_process.VisionEncoderRunner"): + encoder_process_main(_make_vllm_config(), jq, rq, stop) + + assert rq.get(timeout=2) == "READY" + assert rq.empty() + + def test_error_signal_on_load_failure(self): + from sendnn_inference.v1.worker.mm_encoder_process import encoder_process_main + + jq = multiprocessing.Queue() + rq = multiprocessing.Queue() + stop = multiprocessing.Event() + + with patch( + "sendnn_inference.v1.worker.mm_encoder_process.VisionEncoderRunner", + side_effect=RuntimeError("load failed"), + ): + encoder_process_main(_make_vllm_config(), jq, rq, stop) + + # multiprocessing.Queue uses a background writer thread; use a short + # timeout instead of get_nowait() to avoid a race on early-exit paths. + signal = rq.get(timeout=2) + assert signal.startswith("ERROR:") + assert "load failed" in signal + + def test_job_processed_result_on_queue(self): + from sendnn_inference.v1.worker.mm_encoder_process import encoder_process_main + + jq = multiprocessing.Queue() + rq = multiprocessing.Queue() + stop = multiprocessing.Event() + + fake_embeds = torch.zeros(1, 4, 8, dtype=torch.float16) + job = _make_mm_encode_request("req-job") + jq.put(job) + jq.put(None) # terminate after job + + mock_runner = MagicMock() + mock_runner.execute_model.return_value = fake_embeds + fake_shm = MagicMock() + + with ( + patch( + "sendnn_inference.v1.worker.mm_encoder_process.VisionEncoderRunner", + return_value=mock_runner, + ), + patch( + "sendnn_inference.v1.worker.mm_encoder_process.write_embeddings", + return_value=fake_shm, + ), + ): + encoder_process_main(_make_vllm_config(), jq, rq, stop) + + assert rq.get(timeout=2) == "READY" + req_id, shape, dtype = rq.get(timeout=2) + assert req_id == "req-job" + assert shape == tuple(fake_embeds.shape) + assert dtype == fake_embeds.dtype + + def test_encode_failure_puts_none_metadata(self): + """When execute_model raises, (req_id, None, None) must be put on result queue.""" + from sendnn_inference.v1.worker.mm_encoder_process import encoder_process_main + + jq = multiprocessing.Queue() + rq = multiprocessing.Queue() + stop = multiprocessing.Event() + + job = _make_mm_encode_request("req-fail") + jq.put(job) + jq.put(None) + + mock_runner = MagicMock() + mock_runner.execute_model.side_effect = RuntimeError("encode error") + + with patch( + "sendnn_inference.v1.worker.mm_encoder_process.VisionEncoderRunner", + return_value=mock_runner, + ): + encoder_process_main(_make_vllm_config(), jq, rq, stop) + + assert rq.get(timeout=2) == "READY" + req_id, shape, dtype = rq.get(timeout=2) + assert req_id == "req-fail" + assert shape is None + assert dtype is None + + def test_cancel_queue_skips_job_before_encode(self): + """When a req_id is on the cancel_queue before its job is dequeued, + execute_model must not be called and (req_id, None, None) is returned.""" + from sendnn_inference.v1.worker.mm_encoder_process import encoder_process_main + + jq = multiprocessing.Queue() + rq = multiprocessing.Queue() + cq = multiprocessing.Queue() + stop = multiprocessing.Event() + + cq.put("req-cancel") # cancel token arrives before the job + job = _make_mm_encode_request("req-cancel") + jq.put(job) + jq.put(None) + + mock_runner = MagicMock() + + with ( + patch( + "sendnn_inference.v1.worker.mm_encoder_process.VisionEncoderRunner", + return_value=mock_runner, + ), + patch("sendnn_inference.v1.worker.mm_encoder_process.write_embeddings"), + ): + encoder_process_main(_make_vllm_config(), jq, rq, stop, cq) + + assert rq.get(timeout=2) == "READY" + assert rq.get(timeout=2) == ("req-cancel", None, None) + assert not mock_runner.execute_model.called + + def test_resubmitted_request_encodes_after_cancel_consumed(self): + """Scenario: req_id_1 cancelled, re-request with same req_id arrives. + + Safe path: the cancel token is consumed when the original job is + skipped, so skip_ids is cleared before the re-request arrives. + The re-request must be encoded normally. + + cancel_queue: [req_id_1] + job_queue: [job(req_id_1), job(req_id_1-retry), None] + Expected: first job skipped, retry encoded normally. + + Note: in practice vLLM assigns a unique UUID per request so req_ids + are not reused; this tests the skip_ids cleanup invariant. + """ + from sendnn_inference.v1.worker.mm_encoder_process import encoder_process_main + + jq = multiprocessing.Queue() + rq = multiprocessing.Queue() + cq = multiprocessing.Queue() + stop = multiprocessing.Event() + + jq.put(_make_mm_encode_request("req-1")) # original job already queued + cq.put("req-1") # user cancels after job was submitted + jq.put(_make_mm_encode_request("req-1")) # re-request with same req_id + jq.put(None) + + fake_embeds = torch.zeros(1, 4, 8, dtype=torch.float16) + mock_runner = MagicMock() + mock_runner.execute_model.return_value = fake_embeds + mock_shm = MagicMock() + + with ( + patch( + "sendnn_inference.v1.worker.mm_encoder_process.VisionEncoderRunner", + return_value=mock_runner, + ), + patch( + "sendnn_inference.v1.worker.mm_encoder_process.write_embeddings", + return_value=mock_shm, + ), + ): + encoder_process_main(_make_vllm_config(), jq, rq, stop, cq) + + assert rq.get(timeout=2) == "READY" + # First job: cancelled → abort result + assert rq.get(timeout=2) == ("req-1", None, None) + # Re-request: skip_ids cleared → encoded normally + req_id, shape, dtype = rq.get(timeout=2) + assert req_id == "req-1" + assert shape is not None + mock_runner.execute_model.assert_called_once() + + def test_stop_event_terminates_loop(self): + from sendnn_inference.v1.worker.mm_encoder_process import encoder_process_main + + jq = multiprocessing.Queue() + rq = multiprocessing.Queue() + stop = multiprocessing.Event() + stop.set() # pre-set: loop exits on first iteration check + + with patch("sendnn_inference.v1.worker.mm_encoder_process.VisionEncoderRunner"): + encoder_process_main(_make_vllm_config(), jq, rq, stop) + + assert rq.get(timeout=2) == "READY" + assert rq.empty() diff --git a/tests/v1/worker/test_mm_shared_memory.py b/tests/v1/worker/test_mm_shared_memory.py index 042d9817e..c840f1e38 100644 --- a/tests/v1/worker/test_mm_shared_memory.py +++ b/tests/v1/worker/test_mm_shared_memory.py @@ -326,3 +326,89 @@ def test_all_zeros_preserved(self): _cleanup_if_exists(req_id) assert torch.equal(result, t) + + +# --------------------------------------------------------------------------- +# store_mm_embeddings: aborted-request guard +# --------------------------------------------------------------------------- + + +class TestStoreMmEmbeddings: + """Tests for ChunkedPrefillModelRunner.store_mm_embeddings.""" + + def _make_runner(self): + """Build a ChunkedPrefillModelRunner instance bypassing __init__.""" + from sendnn_inference.v1.worker.spyre_model_runner import ChunkedPrefillModelRunner + + runner = ChunkedPrefillModelRunner.__new__(ChunkedPrefillModelRunner) + runner.rank = 0 + runner.requests = {} + runner.pending_mm_embeddings = {} + runner._finished_encode_req_ids = set() + return runner + + def test_stores_embedding_for_waiting_request(self): + """Embedding for a request not yet in self.requests (still waiting in + scheduler queue) must be written to pending_mm_embeddings so that + add_new_request can consume it when the request begins prefill. + + self.requests only contains requests currently in prefill/decode. + A request waiting in the scheduler has not called add_new_request yet + and must not be treated as aborted. + """ + runner = self._make_runner() + runner.requests = {} # not yet scheduled — waiting in scheduler queue + + req_id = "waiting-req" + shape = (1, 4, 8) + dtype = torch.float16 + t = torch.zeros(shape, dtype=dtype) + shm = write_embeddings(t, req_id) + try: + runner.store_mm_embeddings([(req_id, shape, dtype)]) + finally: + cleanup_embeddings(shm) + _cleanup_if_exists(req_id) + + assert req_id in runner.pending_mm_embeddings + + def test_stores_embedding_for_active_request(self): + """Embedding for an active request must be written to pending_mm_embeddings.""" + from unittest.mock import MagicMock + + runner = self._make_runner() + runner.requests = {"active-req": MagicMock()} + + req_id = "active-req" + shape = (1, 4, 8) + dtype = torch.float16 + t = torch.ones(shape, dtype=dtype) + shm = write_embeddings(t, req_id) + try: + runner.store_mm_embeddings([(req_id, shape, dtype)]) + finally: + cleanup_embeddings(shm) + _cleanup_if_exists(req_id) + + assert req_id in runner.pending_mm_embeddings + assert runner.pending_mm_embeddings[req_id].shape == torch.Size(shape) + + def test_discards_late_result_for_finished_request(self): + """If a request finishes while its encode is in-flight, the late-arriving + result must be discarded and the tombstone entry removed.""" + runner = self._make_runner() + runner._finished_encode_req_ids = {"late-req"} # marked finished by _update_batch + + req_id = "late-req" + shape = (1, 4, 8) + dtype = torch.float16 + t = torch.zeros(shape, dtype=dtype) + shm = write_embeddings(t, req_id) + try: + runner.store_mm_embeddings([(req_id, shape, dtype)]) + finally: + cleanup_embeddings(shm) + _cleanup_if_exists(req_id) + + assert req_id not in runner.pending_mm_embeddings + assert req_id not in runner._finished_encode_req_ids # self-cleaned diff --git a/tools/build_nano_gv.py b/tools/build_nano_gv.py new file mode 100644 index 000000000..18b417050 --- /dev/null +++ b/tools/build_nano_gv.py @@ -0,0 +1,310 @@ +#!/usr/bin/env python3 +"""Build a nano-scale random-init granite-vision variant for CI tests. + +Produces a `LlavaNextForConditionalGeneration` snapshot with: + + - `model_type: llava_next` (passes ``LlavaNextMMUtils._validate_configs``) + - `text_config.model_type: granite` (same) + - `vision_config.model_type: siglip_vision_model` (only backbone FMS supports) + - Tiny dims (text hidden 64 × 2 layers; vision hidden 64 × 4 layers) + - Random init — outputs are meaningless, only wiring matters + - Tokenizer + processor copied from a source snapshot so + ``image_token_index`` and chat template stay valid + +Usage: + + # Build locally + python3 tools/build_nano_gv.py --out /tmp/nano-gv + + # Build and push to HF Hub (requires `huggingface-cli login` first) + python3 tools/build_nano_gv.py --out /tmp/nano-gv --publish joerunde/nano-gv + +The output directory can then be passed to ``LLM(model=...)`` or +``vllm serve`` as a local model path (or via the HF id after --publish). + +Model on disk is ~16 MB — fits comfortably alongside the other CI cache +entries. +""" + +from __future__ import annotations + +import argparse +import shutil +from pathlib import Path + +import torch +from transformers import AutoConfig, AutoProcessor, LlavaNextForConditionalGeneration + + +# Files carried over from the source snapshot (tokenizer + preprocessor state). +# We keep the full tokenizer so `image_token_index=49155` and the chat template +# stay valid — shrinking the tokenizer would break the whole processor pipeline +# and gives us at most a few MB back. +_TOKENIZER_FILES = [ + "added_tokens.json", + "chat_template.json", + "merges.txt", + "preprocessor_config.json", + "processor_config.json", + "special_tokens_map.json", + "tokenizer.json", + "tokenizer_config.json", + "vocab.json", +] + + +def _make_nano_config(source_config): + """Return a shrunken copy of the source LlavaNextConfig. + + Preserved fields the plugin / FMS looks at: + - top-level: model_type, image_token_index, image_grid_pinpoints (kept + as a single entry matching image_size), vision_feature_layer, + vision_feature_select_strategy, tie_word_embeddings, + use_image_newline_parameter + - text_config: model_type=granite plus all the granite-specific + multipliers (attention_multiplier, embedding_multiplier, + logits_scaling, residual_multiplier); shrunk dims + - vision_config: model_type=siglip_vision_model plus shrunk dims; + image_size aligned to patch_size + + The `vision_feature_layer=[-4,-3,-2,-1]` pattern is preserved so the + multi-modal projector input dim (hidden * 4) matches what the FMS + LlavaNext implementation expects. That forces `num_hidden_layers >= 4` + on the vision tower. + """ + cfg = source_config.to_dict() + + # ── vision tower ────────────────────────────────────────────────────── + # 4 layers so `vision_feature_layer=[-4,-3,-2,-1]` remains valid. + # image_size=112 with patch_size=14 → 8×8=64 patches. Tiny but still a + # real image tensor. + cfg["vision_config"] = { + **cfg["vision_config"], + "hidden_size": 64, + "intermediate_size": 128, + "num_hidden_layers": 4, + "num_attention_heads": 4, + "image_size": 112, + "patch_size": 14, + } + # image_grid_pinpoints must include at least the base image_size so the + # processor can tile a plain 112×112 input. + cfg["image_grid_pinpoints"] = [[112, 112], [112, 224], [224, 112]] + # `vision_feature_layer` in the source is [-24, -20, -12, -1] — indices + # into a 27-layer vision tower. With num_hidden_layers=4 those are out + # of range and produce IndexError in fms/models/llava_next.py:348. + # Keep four features (so the multi_modal_projector input dim stays at + # `vision_hidden * 4 = 256` and matches the checkpoint), but reference + # every layer in the 4-layer stack. + cfg["vision_feature_layer"] = [-4, -3, -2, -1] + + # ── text backbone ───────────────────────────────────────────────────── + # `num_attention_heads=2` is the minimum that satisfies vLLM's TP=2 + # divisibility check. hidden_size=64 with head_dim=32 keeps Q/K/V/O + # projections symmetric (64→64 in and out) so FMS's weight-expansion + # logic doesn't kick in. + # + # `head_dim` is set explicitly so that + # `LlavaNextMMUtils.get_mm_specific_load_overrides` sees an explicit + # value and skips its `head_dim=128` fallback (which would otherwise + # clobber the entire text_config with a partial dict). See the plugin + # file for the details. + cfg["text_config"] = { + **cfg["text_config"], + "hidden_size": 64, + "intermediate_size": 128, + "num_hidden_layers": 2, + "num_attention_heads": 2, + "num_key_value_heads": 2, + "head_dim": 32, + # vocab_size unchanged — must match the tokenizer we copy over. + } + + # Preserve `tie_word_embeddings=True` on the text config too — the FMS + # loader reads `config.tie_word_embeddings` and the safetensors index + # for the tied case only stores one copy. + cfg["text_config"]["tie_word_embeddings"] = True + + from transformers import LlavaNextConfig + + return LlavaNextConfig(**cfg) + + +def build(source_model: str, out_dir: Path) -> None: + out_dir.mkdir(parents=True, exist_ok=True) + + print(f"[nano-gv] loading source config from {source_model!r}") + source_config = AutoConfig.from_pretrained(source_model) + + print("[nano-gv] shrinking config") + nano_config = _make_nano_config(source_config) + + print("[nano-gv] instantiating LlavaNextForConditionalGeneration with random weights") + # dtype=bfloat16 matches the source model's `torch_dtype` — keeps the + # plugin's fp16 cast path deterministic (bf16 params → fp16 for Spyre). + torch.manual_seed(0) + model = LlavaNextForConditionalGeneration(nano_config).to(dtype=torch.bfloat16) + + print(f"[nano-gv] saving model to {out_dir}") + model.save_pretrained(out_dir, safe_serialization=True) + + # Copy tokenizer + processor state from source, then re-save so all + # image-preprocessor dimensions match our nano vision_config. The + # source processor's crop_size / size / image_grid_pinpoints all + # reference the source image_size (384) — if we don't update them, + # vLLM's dummy-input generator will build 384×384 pixel tensors that + # produce 27²=729 patches, and the nano vision tower (expecting 8²=64 + # patches from a 112×112 image) will fail at forward time. + print("[nano-gv] copying tokenizer + processor from source") + processor = AutoProcessor.from_pretrained(source_model) + + img_size = nano_config.vision_config.image_size + if hasattr(processor, "image_processor"): + proc = processor.image_processor + if hasattr(proc, "crop_size"): + proc.crop_size = {"height": img_size, "width": img_size} + if hasattr(proc, "size"): + proc.size = {"height": img_size, "width": img_size} + # image_grid_pinpoints must be a subset (or match) of the top-level + # config's image_grid_pinpoints. Use the same tiny list. + if hasattr(proc, "image_grid_pinpoints"): + proc.image_grid_pinpoints = list(nano_config.image_grid_pinpoints) + + processor.save_pretrained(out_dir) + + # AutoProcessor.save_pretrained handles most of _TOKENIZER_FILES, but not + # merges.txt / vocab.json for BPE tokenizers — copy manually if missing. + src_snapshot = ( + Path(processor.tokenizer.vocab_file).parent + if getattr(processor.tokenizer, "vocab_file", None) + else None + ) + if src_snapshot is not None: + for fname in ("merges.txt", "vocab.json"): + src = src_snapshot / fname + dst = out_dir / fname + if src.is_file() and not dst.exists(): + shutil.copy2(src, dst) + print(f"[nano-gv] copied {fname}") + + _write_readme(out_dir, source_model) + + total = sum(f.stat().st_size for f in out_dir.iterdir() if f.is_file()) + print(f"[nano-gv] done — {total / 1024**2:.1f} MB in {out_dir}") + + +def _write_readme(out_dir: Path, source_model: str) -> None: + """Drop a short model card next to the weights. + + HF Hub's model-card linter warns on repos without a README, and it's + the right place to record that this is a CI fixture (random weights) + rather than something anyone should use for inference. + """ + readme = out_dir / "README.md" + readme.write_text( + f"""--- +license: apache-2.0 +tags: + - test + - fixture +--- + +# nano-gv + +Random-init ~16 MB CI fixture for the [sendnn-inference](https://github.com/torch-spyre/sendnn-inference) +plugin's async multimodal-encoder tests. Generated by `tools/build_nano_gv.py` +from [`{source_model}`](https://huggingface.co/{source_model})'s config. + +**Not a functional model — outputs are meaningless.** Do not use for +inference. The architecture identifiers (`LlavaNextForConditionalGeneration` ++ granite text + siglip vision) match the source, but every dim is +shrunk to the minimum that still exercises the plugin's MM code paths: + +| component | source | nano | +|-----------|------------------|---------------------| +| text hidden | 2048 | 64 | +| text layers | 40 | 2 | +| text heads | 32 | 2 | +| text head_dim | 64 | 32 | +| vision hidden | 1152 | 64 | +| vision layers | 27 | 4 | +| image size | 384 | 112 | + +The tokenizer + processor are copied from the source verbatim (with +image-size fields rescaled to 112) so `image_token_index` and the chat +template stay valid. +""" + ) + print("[nano-gv] wrote README.md") + + +def _publish(local_dir: Path, repo_id: str, private: bool, message: str) -> None: + """Upload the local snapshot to HF Hub at *repo_id*. + + Creates the repo if it doesn't exist. Requires an HF write token — + run `huggingface-cli login` first, or set HF_TOKEN in the environment. + """ + from huggingface_hub import HfApi + + api = HfApi() + print(f"[nano-gv] publishing to https://huggingface.co/{repo_id}") + + api.create_repo( + repo_id=repo_id, + repo_type="model", + private=private, + exist_ok=True, + ) + + api.upload_folder( + repo_id=repo_id, + folder_path=str(local_dir), + repo_type="model", + commit_message=message, + ) + print(f"[nano-gv] published: https://huggingface.co/{repo_id}") + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--source", + default="ibm-granite/granite-vision-3.2-2b", + help="HF model id to derive the config + tokenizer from", + ) + parser.add_argument( + "--out", + type=Path, + required=True, + help="Output directory for the nano model snapshot", + ) + parser.add_argument( + "--publish", + metavar="REPO_ID", + default=None, + help=( + "After building, upload the snapshot to HF Hub at REPO_ID " + "(e.g. 'joerunde/nano-gv'). Requires `huggingface-cli login` " + "or HF_TOKEN in the environment." + ), + ) + parser.add_argument( + "--private", + action="store_true", + help="If publishing, create the repo as private.", + ) + parser.add_argument( + "--commit-message", + default="Update nano-gv snapshot", + help="Commit message for the HF Hub upload.", + ) + args = parser.parse_args() + + build(args.source, args.out) + + if args.publish: + _publish(args.out, args.publish, args.private, args.commit_message) + + +if __name__ == "__main__": + main() diff --git a/uv.lock b/uv.lock index db5379324..2c1608eb5 100644 --- a/uv.lock +++ b/uv.lock @@ -1365,7 +1365,7 @@ wheels = [ [[package]] name = "ibm-fms" -version = "1.12" +version = "1.12.1" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "torch", version = "2.11.0", source = { registry = "https://download.pytorch.org/whl/cpu" }, marker = "platform_machine != 'aarch64' and sys_platform == 'darwin'" }, @@ -1373,7 +1373,7 @@ dependencies = [ { name = "transformers" }, ] wheels = [ - { url = "https://files.pythonhosted.org/packages/db/a4/8a89579e1c3b47eedad6527e43b4af78f1e2f7f4224a45117b5df42f64d4/ibm_fms-1.12-py3-none-any.whl", hash = "sha256:ccc7224f6761f8281ce7e82ea9cca6b4692d6534d22cfd352c8ad8df7bbc3795", size = 258330, upload-time = "2026-06-25T21:28:02.553Z" }, + { url = "https://files.pythonhosted.org/packages/37/d8/d15d4280780c236499dc9e991c8f0b65d5be5811659cbac1bcb98ddf0520/ibm_fms-1.12.1-py3-none-any.whl", hash = "sha256:fc9214c98b245703f9fa828988c617c5eb16a3afa16fbbf4b7b34a5a8c5fe927", size = 258422, upload-time = "2026-07-01T16:52:31.722Z" }, ] [[package]]