perf: compare native EC H2D overlap

benchmarks/multimodal/jsonl/generate_images.py

@@ -36,10 +36,11 @@ def generate_image_pool_base64(
 ) -> list[str]:
     """Generate pool_size random PNG files and return their paths."""
     image_dir.mkdir(parents=True, exist_ok=True)
+    width, height = image_size
     pool: list[str] = []
     for idx in range(pool_size):
         path = image_dir / f"img_{idx:04d}.png"
-        pixels = np_rng.integers(0, 256, (*image_size, 3), dtype=np.uint8)
+        pixels = np_rng.integers(0, 256, (height, width, 3), dtype=np.uint8)
         Image.fromarray(pixels).save(path)
         pool.append(str(path.resolve()))
     print(

benchmarks/multimodal/jsonl/prepare_ec_h2d_overlap.py

@@ -0,0 +1,276 @@
+#!/usr/bin/env python3
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Prepare deterministic workloads for the native CPU EC overlap benchmark."""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+from collections.abc import Mapping, Sequence
+from pathlib import Path
+from typing import Any, Protocol
+
+import numpy as np
+
+try:
+    from transformers import AutoTokenizer
+except ModuleNotFoundError:  # Optional for tokenizer-independent unit tests.
+    AutoTokenizer = None
+
+from benchmarks.multimodal.jsonl.generate_images import (
+    compute_image_uuid,
+    generate_image_pool_base64,
+)
+
+DEFAULT_MODEL = "Qwen/Qwen3.5-122B-A10B-FP8"
+DEFAULT_OUTPUT_DIR = Path("/dynamo-tmp/data")
+DEFAULT_IMAGE_DIR = DEFAULT_OUTPUT_DIR / "ec_h2d_overlap_images_2400x1080_seed42"
+SYSTEM_PROMPT_TOKENS = 8000
+USER_TEXT = "Describe the newest image and summarize only its visible content."
+SYSTEM_CONTEXT = (
+    " The attached images may contain objects, text, diagrams, tables, labels,"
+    " quantities, and annotations. Consider visual details in context, distinguish"
+    " similar elements carefully, and report only information supported by the images."
+)
+
+
+class Tokenizer(Protocol):
+    chat_template: str | dict[str, str] | None
+    init_kwargs: dict[str, Any]
+
+    def encode(self, text: str, *, add_special_tokens: bool = False) -> list[int]:
+        ...
+
+    def decode(
+        self,
+        ids: Sequence[int],
+        *,
+        skip_special_tokens: bool = True,
+        clean_up_tokenization_spaces: bool = False,
+    ) -> str:
+        ...
+
+    def apply_chat_template(self, conversation: list[dict[str, str]], **kwargs: Any):
+        ...
+
+
+def _sha256(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as source:
+        for chunk in iter(lambda: source.read(1024 * 1024), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def count_tokens(tokenizer: Tokenizer, text: str) -> int:
+    return len(tokenizer.encode(text, add_special_tokens=False))
+
+
+def tokenized_length(tokenized: Any) -> int:
+    """Count input IDs returned as a sequence or tokenizer batch encoding."""
+    if isinstance(tokenized, Mapping):
+        tokenized = tokenized["input_ids"]
+    if tokenized and isinstance(tokenized[0], Sequence):
+        if len(tokenized) != 1:
+            raise ValueError("expected exactly one rendered conversation")
+        tokenized = tokenized[0]
+    return len(tokenized)
+
+
+def make_exact_text(tokenizer: Tokenizer, target_tokens: int) -> str:
+    """Return deterministic text with exactly ``target_tokens`` tokenizer tokens."""
+    if target_tokens <= 0:
+        raise ValueError("target_tokens must be positive")
+
+    unit_tokens = count_tokens(tokenizer, SYSTEM_CONTEXT)
+    repeats = max(2, target_tokens // unit_tokens + 2)
+    keep = target_tokens
+    for _ in range(64):
+        ids = tokenizer.encode(SYSTEM_CONTEXT * repeats, add_special_tokens=False)
+        while len(ids) < keep:
+            repeats *= 2
+            ids = tokenizer.encode(SYSTEM_CONTEXT * repeats, add_special_tokens=False)
+        text = tokenizer.decode(
+            ids[:keep],
+            skip_special_tokens=True,
+            clean_up_tokenization_spaces=False,
+        )
+        observed = count_tokens(tokenizer, text)
+        if observed == target_tokens:
+            return text
+        keep += target_tokens - observed
+    raise RuntimeError(f"could not construct exact {target_tokens}-token text")
+
+
+def build_chat_template(base_template: str, system_prompt: str) -> str:
+    """Prepend the benchmark system message without changing client payloads."""
+    prompt_literal = json.dumps(system_prompt, ensure_ascii=False)
+    prefix = (
+        "{%- if not messages or messages[0]['role'] != 'system' -%}"
+        "{%- set messages = [{'role': 'system', 'content': "
+        f"{prompt_literal}"
+        "}] + messages -%}"
+        "{%- endif -%}\n"
+    )
+    return prefix + base_template
+
+
+def write_sliding_dataset(
+    path: Path,
+    image_pool: Sequence[str],
+    *,
+    num_users: int,
+    turns_per_user: int,
+    window_size: int,
+    images_per_user: int,
+    user_text: str,
+) -> dict[str, int]:
+    """Write one turn-major sliding-window dataset from a shared image pool."""
+    required_images = num_users * images_per_user
+    if len(image_pool) < required_images:
+        raise ValueError(
+            f"image pool has {len(image_pool)} entries; {required_images} required"
+        )
+    if window_size + turns_per_user - 1 > images_per_user:
+        raise ValueError("images_per_user cannot cover every sliding window")
+
+    path.parent.mkdir(parents=True, exist_ok=True)
+    unique_refs: set[str] = set()
+    rows = 0
+    with path.open("w", encoding="utf-8") as output:
+        for turn_idx in range(turns_per_user):
+            for user_idx in range(num_users):
+                offset = user_idx * images_per_user + turn_idx
+                images = list(image_pool[offset : offset + window_size])
+                unique_refs.update(images)
+                row = {
+                    "session_id": f"user_{user_idx}",
+                    "text": user_text,
+                    "images": images,
+                    "image_uuids": [compute_image_uuid(ref) for ref in images],
+                }
+                output.write(json.dumps(row, separators=(",", ":")) + "\n")
+                rows += 1
+
+    total_slots = rows * window_size
+    return {
+        "rows": rows,
+        "unique_images": len(unique_refs),
+        "content_images": len(unique_refs),
+        "stripped_images": total_slots - len(unique_refs),
+    }
+
+
+def _parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--model", default=DEFAULT_MODEL)
+    parser.add_argument("--output-dir", type=Path, default=DEFAULT_OUTPUT_DIR)
+    parser.add_argument("--image-dir", type=Path, default=DEFAULT_IMAGE_DIR)
+    parser.add_argument("--seed", type=int, default=42)
+    parser.add_argument("--force", action="store_true")
+    return parser.parse_args()
+
+
+def main() -> None:
+    args = _parse_args()
+    args.output_dir.mkdir(parents=True, exist_ok=True)
+
+    prompt_path = args.output_dir / "qwen35_shared_system_8000.txt"
+    template_path = args.output_dir / "qwen35_shared_system_8000.jinja"
+    manifest_path = args.output_dir / "qwen35_ec_h2d_overlap_manifest.json"
+    dataset_paths = {
+        5: args.output_dir / "30u_8t_5w_shared8k_base64_uuid_seed42.jsonl",
+        10: args.output_dir / "30u_8t_10w_shared8k_base64_uuid_seed42.jsonl",
+    }
+    artifacts = [prompt_path, template_path, manifest_path, *dataset_paths.values()]
+    if not args.force and all(path.is_file() for path in artifacts):
+        print(f"All workload artifacts already exist; keeping {manifest_path}")
+        return
+
+    if AutoTokenizer is None:
+        raise RuntimeError("transformers is required to prepare EC workloads")
+    tokenizer = AutoTokenizer.from_pretrained(
+        args.model,
+        trust_remote_code=True,
+        local_files_only=True,
+    )
+    base_template = tokenizer.chat_template
+    if not isinstance(base_template, str):
+        raise TypeError("target tokenizer must expose one string chat_template")
+
+    system_prompt = make_exact_text(tokenizer, SYSTEM_PROMPT_TOKENS)
+    prompt_path.write_text(system_prompt, encoding="utf-8")
+    template = build_chat_template(base_template, system_prompt)
+    template_path.write_text(template, encoding="utf-8")
+
+    rendered_ids = tokenizer.apply_chat_template(
+        [{"role": "user", "content": USER_TEXT}],
+        chat_template=template,
+        tokenize=True,
+        add_generation_prompt=True,
+    )
+
+    num_users = 30
+    turns_per_user = 8
+    max_window_size = max(dataset_paths)
+    images_per_user = max_window_size + turns_per_user - 1
+    image_pool = generate_image_pool_base64(
+        np.random.default_rng(args.seed),
+        num_users * images_per_user,
+        args.image_dir,
+        (2400, 1080),
+    )
+
+    datasets: dict[str, dict[str, Any]] = {}
+    for window_size, path in dataset_paths.items():
+        counts = write_sliding_dataset(
+            path,
+            image_pool,
+            num_users=num_users,
+            turns_per_user=turns_per_user,
+            window_size=window_size,
+            images_per_user=images_per_user,
+            user_text=USER_TEXT,
+        )
+        datasets[str(window_size)] = {
+            "path": str(path),
+            "sha256": _sha256(path),
+            "window_size": window_size,
+            **counts,
+        }
+
+    manifest = {
+        "model": args.model,
+        "tokenizer_commit": tokenizer.init_kwargs.get("_commit_hash"),
+        "seed": args.seed,
+        "system_prompt": {
+            "path": str(prompt_path),
+            "tokens": count_tokens(tokenizer, system_prompt),
+            "sha256": _sha256(prompt_path),
+        },
+        "chat_template": {
+            "path": str(template_path),
+            "sha256": _sha256(template_path),
+        },
+        "user_text": USER_TEXT,
+        "user_text_tokens": count_tokens(tokenizer, USER_TEXT),
+        "rendered_text_prompt_tokens": tokenized_length(rendered_ids),
+        "images": {
+            "directory": str(args.image_dir),
+            "count": len(image_pool),
+            "width": 2400,
+            "height": 1080,
+        },
+        "datasets": datasets,
+    }
+    manifest_path.write_text(
+        json.dumps(manifest, indent=2, sort_keys=True) + "\n", encoding="utf-8"
+    )
+    print(json.dumps(manifest, indent=2, sort_keys=True))
+
+
+if __name__ == "__main__":
+    main()

benchmarks/multimodal/sweep/README.md

@@ -46,17 +46,42 @@ input_files:
   - benchmarks/multimodal/jsonl/1000req_1img_200pool_400word_http.jsonl
   - benchmarks/multimodal/jsonl/1000req_4img_200pool_400word_http.jsonl
 
-# Each config launches the workflow with its own extra_args
+# Each config launches the workflow with its own extra_args and optional env.
+# Per-arm env values expand variables from the harness process.
 configs:
   - label: cache-off
     workflow: benchmarks/multimodal/sweep/workflows/vllm_serve.sh
     extra_args: [--no-enable-prefix-caching, --multimodal-embedding-cache-capacity-gb, "0"]
 
   - label: cache-on
     workflow: benchmarks/multimodal/sweep/workflows/vllm_serve.sh
+    env:
+      PYTHONPATH: "${VLLM_PATCHED_PYTHONPATH}"
     extra_args: [--no-enable-prefix-caching, --multimodal-embedding-cache-capacity-gb, "10"]
 ```
 
+### vLLM workflow environment
+
+| Variable | Purpose |
+|---|---|
+| `DYN_DISABLE_NSYS` | Set to `0` to profile the vLLM server; defaults to `1`. |
+| `DYN_NSYS_BIN` | Nsight Systems executable path. |
+| `DYN_NSYS_DIR` / `DYN_NSYS_TMPDIR` | Final report and temporary capture directories. |
+| `DYN_NSYS_TRACE` | Nsight trace domains; defaults to `cuda,nvtx`. |
+| `DYN_NSYS_OUTPUT_PREFIX` | Report prefix; the orchestrator appends the arm label. |
+| `DYN_SERVER_TERMINATE_TIMEOUT` | Orchestrator shutdown timeout; defaults to 300 seconds with profiling and 15 otherwise. |
+| `DYN_SERVER_SHUTDOWN_GRACE_SECONDS` | Wrapper grace period before SIGKILL; defaults to 150 seconds with profiling and 10 otherwise. |
+| `DYN_PYTHON` | Python executable used by the repetition wrapper. |
+| `DYN_BENCHMARK_ORDER_SEED` | Seed for randomized per-repetition arm order; defaults to 42. |
+| `VLLM_SOURCE_REVISION` | Required tested-vLLM revision recorded by the repetition wrapper. |
+| `VLLM_BASELINE_SOURCE_REVISION` | Required baseline vLLM revision for the native-EC comparison. |
+| `VLLM_BASELINE_PYTHONPATH` / `VLLM_PATCHED_PYTHONPATH` | Source trees selected by the baseline and overlap arms. |
+| `CONTAINER_IMAGE` | Required runtime image reference recorded by the repetition wrapper. |
+| `CONTAINER_IMAGE_DIGEST` | Required immutable runtime image digest. |
+| `CONTAINER_IMAGE_FILE` | Required imported image/squashfs path used by the GPU run. |
+| `HARNESS_REVISION` | Required Dynamo benchmark-harness commit. |
+
+
 ## CLI Overrides
 
 Any top-level YAML field can be overridden from the command line:

benchmarks/multimodal/sweep/config.py

@@ -18,6 +18,7 @@ class BenchmarkConfig:
     label: str
     workflow: str
     extra_args: List[str] = field(default_factory=list)
+    env: Dict[str, str] = field(default_factory=dict)
 
 
 @dataclass
@@ -42,6 +43,8 @@ class SweepConfig:
     skip_plots: bool = False
     restart_server_every_benchmark: bool = True
     uuid_and_strip: bool = False
+    prompt_manifest: Optional[str] = None
+    prefix_cache_probe_min_cached_tokens: Optional[int] = None
     env: Dict[str, str] = field(default_factory=dict)
 
     @property
@@ -69,6 +72,16 @@ def validate(self, repo_root: Optional[Path] = None) -> None:
             if not Path(f).is_file():
                 raise FileNotFoundError(f"Input file not found: {f}")
 
+        if self.prompt_manifest and not Path(self.prompt_manifest).is_file():
+            raise FileNotFoundError(
+                f"Prompt manifest not found: {self.prompt_manifest}"
+            )
+        if (
+            self.prefix_cache_probe_min_cached_tokens is not None
+            and self.prefix_cache_probe_min_cached_tokens <= 0
+        ):
+            raise ValueError("prefix_cache_probe_min_cached_tokens must be positive")
+
         for cfg in self.configs:
             script = Path(cfg.workflow)
             if repo_root and not script.is_absolute():
@@ -96,6 +109,7 @@ def _parse_benchmark_config(raw: Dict[str, Any]) -> BenchmarkConfig:
         label=raw["label"],
         workflow=raw["workflow"],
         extra_args=[str(a) for a in raw.get("extra_args", [])],
+        env={str(k): str(v) for k, v in raw.get("env", {}).items()},
     )
 
 
@@ -138,6 +152,10 @@ def load_config(
         skip_plots=raw.get("skip_plots", False),
         restart_server_every_benchmark=raw.get("restart_server_every_benchmark", True),
         uuid_and_strip=raw.get("uuid_and_strip", False),
+        prompt_manifest=raw.get("prompt_manifest"),
+        prefix_cache_probe_min_cached_tokens=raw.get(
+            "prefix_cache_probe_min_cached_tokens"
+        ),
         env=raw.get("env", {}),
     )