feat(multimodal): graph Qwen2.5-VL custom encoder

benchmarks/multimodal/jsonl/generate_images.py

@@ -3,14 +3,20 @@
 
 """Utilities for generating and sampling image pools."""
 
+import hashlib
+import io
 import json
 import random
 import uuid as _uuid
 from pathlib import Path
+from typing import Any
 
 import numpy as np
 from PIL import Image
 
+JPEG_TARGET_MIN_BYTES = 50 * 1024
+JPEG_TARGET_MAX_BYTES = 60 * 1024
+
 
 def compute_image_uuid(ref: str) -> str:
     """Stable UUID for an image reference (path or URL).
@@ -48,6 +54,136 @@ def generate_image_pool_base64(
     return pool
 
 
+def _encode_resampled_noise_jpeg(
+    noise: Image.Image,
+    texture_side: int,
+    image_size: tuple[int, int],
+    quality: int,
+) -> bytes:
+    """Encode deterministic, compressible noise at a fixed JPEG quality."""
+    image = noise.resize(
+        (texture_side, texture_side), Image.Resampling.BILINEAR
+    ).resize(image_size, Image.Resampling.BICUBIC)
+    encoded = io.BytesIO()
+    image.save(
+        encoded,
+        format="JPEG",
+        quality=quality,
+        optimize=True,
+        subsampling=2,
+    )
+    return encoded.getvalue()
+
+
+def generate_target_sized_jpeg(
+    np_rng: np.random.Generator,
+    path: Path,
+    image_size: tuple[int, int] = (500, 500),
+    quality: int = 85,
+    min_bytes: int = JPEG_TARGET_MIN_BYTES,
+    max_bytes: int = JPEG_TARGET_MAX_BYTES,
+) -> dict[str, Any]:
+    """Write one deterministic JPEG whose encoded size is within a byte range.
+
+    JPEG quality remains fixed. The generator changes only the resolution of a
+    seeded noise texture, which controls compressibility without changing the
+    decoded image dimensions. A 180-pixel texture normally lands near 55 KiB;
+    binary search is used only when an encoder/version produces a result outside
+    the requested range.
+    """
+    if min_bytes <= 0 or max_bytes < min_bytes:
+        raise ValueError("expected 0 < min_bytes <= max_bytes")
+    if not 1 <= quality <= 100:
+        raise ValueError("quality must be between 1 and 100")
+
+    width, height = image_size
+    pixels = np_rng.integers(0, 256, (height, width, 3), dtype=np.uint8)
+    noise = Image.fromarray(pixels)
+    target_bytes = (min_bytes + max_bytes) // 2
+
+    candidates: list[tuple[int, bytes]] = []
+
+    def encode(texture_side: int) -> bytes:
+        payload = _encode_resampled_noise_jpeg(noise, texture_side, image_size, quality)
+        candidates.append((texture_side, payload))
+        return payload
+
+    payload = encode(min(180, width, height))
+    if not min_bytes <= len(payload) <= max_bytes:
+        lower = 8
+        upper = min(width, height)
+        while lower <= upper:
+            texture_side = (lower + upper) // 2
+            payload = encode(texture_side)
+            if min_bytes <= len(payload) <= max_bytes:
+                break
+            if len(payload) < min_bytes:
+                lower = texture_side + 1
+            else:
+                upper = texture_side - 1
+
+    texture_side, payload = min(
+        candidates, key=lambda candidate: abs(len(candidate[1]) - target_bytes)
+    )
+    if not min_bytes <= len(payload) <= max_bytes:
+        raise RuntimeError(
+            f"could not generate JPEG in [{min_bytes}, {max_bytes}] bytes; "
+            f"closest was {len(payload)} bytes"
+        )
+
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_bytes(payload)
+    with Image.open(io.BytesIO(payload)) as encoded:
+        decoded = encoded.convert("RGB")
+        decoded_hash = hashlib.sha256(decoded.tobytes()).hexdigest()
+
+    return {
+        "path": str(path.resolve()),
+        "width": width,
+        "height": height,
+        "size_bytes": len(payload),
+        "jpeg_quality": quality,
+        "texture_side": texture_side,
+        "encoded_sha256": hashlib.sha256(payload).hexdigest(),
+        "decoded_rgb_sha256": decoded_hash,
+    }
+
+
+def generate_target_sized_jpeg_pool(
+    pool_size: int,
+    image_dir: Path,
+    seed: int,
+    image_size: tuple[int, int] = (500, 500),
+    quality: int = 85,
+    min_bytes: int = JPEG_TARGET_MIN_BYTES,
+    max_bytes: int = JPEG_TARGET_MAX_BYTES,
+    start_index: int = 0,
+) -> list[dict[str, Any]]:
+    """Generate a deterministic pool of unique target-sized JPEGs."""
+    image_dir.mkdir(parents=True, exist_ok=True)
+    records: list[dict[str, Any]] = []
+    encoded_hashes: set[str] = set()
+    decoded_hashes: set[str] = set()
+    for offset in range(pool_size):
+        index = start_index + offset
+        record = generate_target_sized_jpeg(
+            np.random.default_rng(seed + index),
+            image_dir / f"image_{index:04d}_{image_size[0]}x{image_size[1]}.jpg",
+            image_size=image_size,
+            quality=quality,
+            min_bytes=min_bytes,
+            max_bytes=max_bytes,
+        )
+        if record["encoded_sha256"] in encoded_hashes:
+            raise RuntimeError(f"duplicate encoded JPEG at {record['path']}")
+        if record["decoded_rgb_sha256"] in decoded_hashes:
+            raise RuntimeError(f"duplicate decoded RGB image at {record['path']}")
+        encoded_hashes.add(record["encoded_sha256"])
+        decoded_hashes.add(record["decoded_rgb_sha256"])
+        records.append(record)
+    return records
+
+
 def generate_image_pool_http(
     py_rng: random.Random,
     pool_size: int,

benchmarks/multimodal/sweep/README.md

@@ -41,6 +41,14 @@ output_dir: benchmarks/multimodal/sweep/results/vllm_serve
 env:
   ENABLE_ENCODER_CACHE: "0"
 
+# Optional arguments appended to every aiperf invocation. Values are converted
+# to strings, so numeric YAML values are accepted.
+aiperf_extra_args:
+  - --random-seed
+  - 42
+  - --workers-max
+  - 20
+
 # JSONL files produced by benchmarks/multimodal/jsonl/
 input_files:
   - benchmarks/multimodal/jsonl/1000req_1img_200pool_400word_http.jsonl

benchmarks/multimodal/sweep/config.py

@@ -42,6 +42,7 @@ class SweepConfig:
     skip_plots: bool = False
     restart_server_every_benchmark: bool = True
     env: Dict[str, str] = field(default_factory=dict)
+    aiperf_extra_args: List[str] = field(default_factory=list)
 
     @property
     def sweep_mode(self) -> str:
@@ -137,6 +138,7 @@ def load_config(
         skip_plots=raw.get("skip_plots", False),
         restart_server_every_benchmark=raw.get("restart_server_every_benchmark", True),
         env=raw.get("env", {}),
+        aiperf_extra_args=[str(a) for a in raw.get("aiperf_extra_args", [])],
     )
 
     if cli_overrides:

benchmarks/multimodal/sweep/orchestrator.py

@@ -156,7 +156,7 @@ def _run_config(
     try:
         for input_file, file_tag, value, artifact_dir, conversation_num in pending_runs:
             _print_banner(
-                f"[{file_tag}] Config: {bench_cfg.label}  " f"{sweep_mode}={value}",
+                f"[{file_tag}] Config: {bench_cfg.label}  {sweep_mode}={value}",
                 char="-",
             )
 
@@ -179,6 +179,7 @@ def _run_config(
                     input_file=input_file,
                     osl=config.osl,
                     artifact_dir=artifact_dir,
+                    extra_args=config.aiperf_extra_args,
                 )
             finally:
                 if config.restart_server_every_benchmark:

benchmarks/multimodal/sweep/runner.py

@@ -3,9 +3,10 @@
 
 from __future__ import annotations
 
+import shlex
 import subprocess
 from pathlib import Path
-from typing import List
+from typing import List, Optional
 
 
 def _build_aiperf_cmd(
@@ -18,13 +19,14 @@ def _build_aiperf_cmd(
     input_file: str,
     osl: int,
     artifact_dir: Path,
+    extra_args: Optional[List[str]] = None,
 ) -> List[str]:
     if sweep_mode == "concurrency":
         sweep_flag = "--concurrency"
     else:
         sweep_flag = "--request-rate"
 
-    return [
+    cmd = [
         "aiperf",
         "profile",
         "-m",
@@ -56,6 +58,9 @@ def _build_aiperf_cmd(
         "none",
         "--no-server-metrics",
     ]
+    if extra_args:
+        cmd.extend(extra_args)
+    return cmd
 
 
 def run_aiperf_single(
@@ -68,6 +73,7 @@ def run_aiperf_single(
     input_file: str,
     osl: int,
     artifact_dir: Path,
+    extra_args: Optional[List[str]] = None,
 ) -> None:
     """Run a single aiperf profile invocation."""
     artifact_dir.mkdir(parents=True, exist_ok=True)
@@ -81,10 +87,15 @@ def run_aiperf_single(
         input_file=input_file,
         osl=osl,
         artifact_dir=artifact_dir,
+        extra_args=extra_args,
     )
 
+    (artifact_dir / "command.txt").write_text(shlex.join(cmd) + "\n", encoding="utf-8")
+
     print(f"  aiperf {sweep_mode}={sweep_value} -> {artifact_dir}", flush=True)
     proc = subprocess.run(cmd, capture_output=True, text=True)
+    (artifact_dir / "aiperf.stdout.log").write_text(proc.stdout, encoding="utf-8")
+    (artifact_dir / "aiperf.stderr.log").write_text(proc.stderr, encoding="utf-8")
 
     if proc.returncode != 0:
         print(f"  aiperf FAILED (exit {proc.returncode})", flush=True)
@@ -109,6 +120,7 @@ def run_sweep(
     input_file: str,
     osl: int,
     output_dir: Path,
+    extra_args: Optional[List[str]] = None,
 ) -> None:
     """Run aiperf across all sweep values, writing results under output_dir/{mode}{N}/."""
     output_dir.mkdir(parents=True, exist_ok=True)
@@ -124,6 +136,7 @@ def run_sweep(
             input_file=input_file,
             osl=osl,
             artifact_dir=output_dir / f"{sweep_mode}{value}",
+            extra_args=extra_args,
         )
 
     print(f"Sweep complete. Results in {output_dir}", flush=True)

benchmarks/multimodal/sweep/server.py

@@ -3,6 +3,7 @@
 
 from __future__ import annotations
 
+import json
 import os
 import signal
 import subprocess
@@ -61,7 +62,7 @@ def start(
         self.wait_for_ready(model)
 
     def wait_for_ready(self, model: str) -> None:
-        """Poll /v1/models until the expected model name appears."""
+        """Wait for model registration, then require one real inference."""
         import urllib.error
         import urllib.request
 
@@ -85,15 +86,45 @@ def wait_for_ready(self, model: str) -> None:
                 with urllib.request.urlopen(req, timeout=5) as resp:
                     body = resp.read().decode()
                     if model in body:
-                        print("Server is ready (model registered).", flush=True)
-                        return
+                        if self._chat_probe(model):
+                            print(
+                                "Server is ready (chat inference succeeded).",
+                                flush=True,
+                            )
+                            return
             except (urllib.error.URLError, OSError, TimeoutError):
                 pass
             time.sleep(5)
 
         self.stop()
         raise TimeoutError(f"Server did not become ready within {self.timeout}s")
 
+    def _chat_probe(self, model: str) -> bool:
+        """Return true only after a minimal non-streaming generation succeeds."""
+        import urllib.error
+        import urllib.request
+
+        body = json.dumps(
+            {
+                "model": model,
+                "messages": [{"role": "user", "content": "Reply ready."}],
+                "max_tokens": 1,
+                "temperature": 0,
+                "stream": False,
+            }
+        ).encode()
+        request = urllib.request.Request(
+            f"http://localhost:{self.port}/v1/chat/completions",
+            data=body,
+            headers={"Content-Type": "application/json"},
+            method="POST",
+        )
+        try:
+            with urllib.request.urlopen(request, timeout=30) as response:
+                return 200 <= response.status < 300
+        except (urllib.error.URLError, OSError, TimeoutError):
+            return False
+
     def stop(self) -> None:
         """Stop the server by killing its process group."""
         if self._process is None:

components/src/dynamo/vllm/handlers.py

@@ -1089,18 +1089,32 @@ def _load_custom_encoder(self, config: Config) -> None:
                 f"VisionEncoderBackend subclass, got {backend_cls!r}."
             )
         # The author writes the VisionEncoderBackend; Dynamo wraps it in the
-        # AsyncVisionEncoder glue, which owns the preprocess pool and actor
-        # thread. load() runs backend.build() on the actor thread
+        # AsyncVisionEncoder glue, which owns the preprocess pool and
+        # ThreadedMicroBatcher actor thread. load() runs backend.build() there
         # (the backend picks its own device) and cleans that thread up on failure.
-        encoder = AsyncVisionEncoder(backend_cls())
+        queue_wait_raw = os.environ.get("DYN_CUSTOM_ENCODER_QUEUE_WAIT_MS", "0")
+        try:
+            queue_wait_ms = float(queue_wait_raw)
+        except ValueError as exc:
+            raise ValueError(
+                "DYN_CUSTOM_ENCODER_QUEUE_WAIT_MS must be finite and nonnegative, "
+                f"got {queue_wait_raw!r}"
+            ) from exc
+        if not math.isfinite(queue_wait_ms) or queue_wait_ms < 0:
+            raise ValueError(
+                "DYN_CUSTOM_ENCODER_QUEUE_WAIT_MS must be finite and nonnegative, "
+                f"got {queue_wait_raw!r}"
+            )
+        encoder = AsyncVisionEncoder(backend_cls(), queue_wait_ms=queue_wait_ms)
         encoder.load(config.model)
         # Assign only after a successful load so a failed load (which already shut
         # its own thread down) leaves _custom_encoder None.
         self._custom_encoder = encoder
         logger.info(
-            "Loaded CustomEncoder %s from %s",
+            "Loaded CustomEncoder %s from %s (queue_wait_ms=%.3f)",
             custom_encoder_class,
             config.model,
+            queue_wait_ms,
         )
 
     def _shutdown_on_engine_dead(self, e: EngineDeadError) -> NoReturn:
@@ -2925,8 +2939,8 @@ async def _assemble_custom_encoder_prompt(
         # failure becomes a structured request error instead of escaping the
         # request coroutine and tearing down the stream.
         try:
-            # encode() preprocesses off-thread and serializes forwards on one
-            # dedicated actor thread.
+            # AsyncVisionEncoder preprocesses off-thread; its ThreadedMicroBatcher
+            # coalesces concurrent calls onto one dedicated actor thread.
             img_tensors: list[torch.Tensor] = await self._custom_encoder.encode(
                 image_urls
             )