diff --git a/benchmarks/README.md b/benchmarks/README.md index 0a8fb8058..182bba02d 100644 --- a/benchmarks/README.md +++ b/benchmarks/README.md @@ -4,6 +4,11 @@ This directory contains benchmark suites for evaluating different model families ## Benchmark families +### [Qwen2.5-Omni](qwen2_5_omni/README.md) — Embedding Wrapper Latency + +CPU-observed A/B timings for redundant text embedding removal, with real checkpoint +embeddings, warmup, paired blocks, and embedding call-count validation. + ### [TTS](tts/README.md) — Text-to-Speech Model-agnostic serving benchmarks for TTS models, including Qwen3-TTS and VoxCPM2. diff --git a/benchmarks/qwen2_5_omni/README.md b/benchmarks/qwen2_5_omni/README.md new file mode 100644 index 000000000..c8aea8980 --- /dev/null +++ b/benchmarks/qwen2_5_omni/README.md @@ -0,0 +1,66 @@ +# Qwen2.5-Omni embedding microbenchmark + +Measure the host-observed latency of `embed_input_ids` before and after the +redundant text embedding removal in PR #7477. Both variants use the same real +checkpoint embedding table, dependencies, inputs and batch shapes. The baseline +wrapper is extracted from the specified Git commit; the fixed wrapper comes from +the current checkout. No full model is initialized. + +## Run + +Use Linux, a CUDA GPU, the project dependencies, and a local +`Qwen/Qwen2.5-Omni-7B` checkpoint with its safetensors index and weights. +From the repository root: + +```bash +git fetch origin 01a2f93256975c7ff9565c5414b0fe0225bc4765 +PYTHONPATH="$PWD" OMP_NUM_THREADS=1 python benchmarks/qwen2_5_omni/benchmark_embeddings.py \ + --model /path/to/Qwen2.5-Omni-7B \ + --baseline-ref 01a2f93256975c7ff9565c5414b0fe0225bc4765 \ + --cpu-core 2 --lengths 1172 --warmup 50 --blocks 20 --iterations 20 \ + --output-dir /tmp/qwen25-embedding-results +``` + +Choose a CPU core in your allowed affinity set. For the original CUDA length +sweep, use `--lengths 512 1172 4096 16384`. + +## Method + +- BF16 checkpoint embeddings, TP=1, CUDA int32 token IDs, CPU multimodal masks. +- One pinned CPU core, one PyTorch thread, garbage collection disabled. +- 50 warmups per variant, then 20 paired blocks of 20 calls with balanced, + randomized variant order. Summary: median of block medians. +- Host latency includes synchronization inside the wrapper. Explicit GPU + synchronization before and after the timed call is excluded from host latency. + Calling-thread CPU time and synchronized completion time are recorded separately. +- CPU frequency is sampled at block boundaries; clocks are not locked. +- Validation checks one text embedding call for the fixed wrapper, two for the + multimodal baseline, and `torch.equal` for identical inputs and batch shapes. + This does not establish batch invariance or full-model output equality. + +Interleaved audio/video uses a valid synthetic multimodal mask. This benchmark +measures wrapper latency, not E2E throughput. + +## Reference results + +Original environment: H100 80GB; Python 3.12.3; vLLM 0.29.0+cu129; +PyTorch 2.13.0+cu129; Transformers 5.14.0; Qwen2.5-Omni-7B BF16. +At 1172 tokens, the September 13 measurements were: + +| Input | Before (μs) | After (μs) | Host latency reduction | +| --- | ---: | ---: | ---: | +| Text | 18.81 | 18.66 | 0.80% | +| Audio | 104.32 | 85.30 | 18.23% | +| Image | 104.72 | 86.27 | 17.62% | +| Video | 104.73 | 85.76 | 18.11% | +| Audio + image + video | 567.92 | 549.54 | 3.24% | +| Interleaved audio/video | 622.76 | 597.84 | 4.00% | + +Results vary between runs and environments. These local savings did not produce +a consistent E2E throughput improvement in the separate repeated E2E experiment. + +## Output + +The output directory contains `metadata.json`, `micro-raw.json`, +`validation.json`, and `summary.json`. Raw records include per-call timings, +block timestamps and CPU frequency samples. Timer overhead is measured separately. diff --git a/benchmarks/qwen2_5_omni/benchmark_embeddings.py b/benchmarks/qwen2_5_omni/benchmark_embeddings.py new file mode 100644 index 000000000..1a5b32435 --- /dev/null +++ b/benchmarks/qwen2_5_omni/benchmark_embeddings.py @@ -0,0 +1,306 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project +"""Reproduce PR #7477 host-observed embedding-wrapper latency on one CUDA GPU. + +Run from the PR checkout with PYTHONPATH=$PWD. Baseline and fixed methods share +weights and inputs. Only the historical wrapper is loaded from --baseline-ref. +""" + +import argparse +import ast +import gc +import json +import os +import random +import statistics +import subprocess +import textwrap +import time +from pathlib import Path + +import torch +from safetensors import safe_open +from torch import nn +from transformers import Qwen2_5OmniThinkerConfig +from vllm.model_executor.layers.vocab_parallel_embedding import VocabParallelEmbedding +from vllm.multimodal.utils import set_mm_embedding_modality + +from vllm_omni.model_executor.models.qwen2_5_omni.qwen2_5_omni_thinker import ( + Qwen2_5OmniThinkerForConditionalGeneration as Thinker, +) +from vllm_omni.platforms import current_omni_platform + + +def baseline_method(cls, ref): + path = "vllm_omni/model_executor/models/qwen2_5_omni/qwen2_5_omni_thinker.py" + source = subprocess.check_output( + ["git", "-C", str(Path(__file__).resolve().parents[2]), "show", f"{ref}:{path}"], text=True + ) + tree = ast.parse(source) + owner = next(n for n in tree.body if isinstance(n, ast.ClassDef) and n.name == cls.__name__) + node = next(n for n in owner.body if isinstance(n, ast.FunctionDef) and n.name == "embed_input_ids") + method = textwrap.dedent(ast.get_source_segment(source, node)) + # Preserve zero-argument super() against the real model class and MRO. + factory = "from __future__ import annotations\ndef make_method(cls):\n __class__ = cls\n" + factory += textwrap.indent(method, " ") + "\n return embed_input_ids\n" + namespace = dict(cls.embed_input_ids.__globals__) + exec(compile(factory, f"{ref}:{path}", "exec"), namespace) + return namespace["make_method"](cls) + + +parser = argparse.ArgumentParser(description=__doc__) +parser.add_argument("--model", type=Path, required=True) +parser.add_argument("--baseline-ref", default="01a2f93256975c7ff9565c5414b0fe0225bc4765") +parser.add_argument("--cpu-core", type=int, default=sorted(os.sched_getaffinity(0))[0]) +parser.add_argument("--lengths", type=int, nargs="+", default=[1172]) +parser.add_argument("--warmup", type=int, default=50) +parser.add_argument("--blocks", type=int, default=20) +parser.add_argument("--iterations", type=int, default=20) +parser.add_argument("--output-dir", type=Path, default=Path("micro-results")) +args = parser.parse_args() +if min(args.lengths) < 300 or min(args.blocks, args.iterations) < 1 or args.warmup < 0: + parser.error("Require lengths >= 300, positive blocks/iterations, and nonnegative warmup") +if args.cpu_core not in os.sched_getaffinity(0): + parser.error("CPU core is outside the allowed affinity set") +ROOT, MODEL, CORE = args.output_dir, args.model, args.cpu_core +ROOT.mkdir(parents=True, exist_ok=True) +CONFIG = json.loads((MODEL / "config.json").read_text())["thinker_config"] +HIDDEN = CONFIG["text_config"]["hidden_size"] +VOCAB = CONFIG["text_config"]["vocab_size"] +IDS = {m: CONFIG[m + "_token_index"] for m in ("audio", "image", "video")} +METHODS = {"baseline": baseline_method(Thinker, args.baseline_ref), "fixed": Thinker.embed_input_ids} +RESULTS, VALIDATIONS = [], [] +os.sched_setaffinity(0, {CORE}) +torch.set_num_threads(1) +torch.set_num_interop_threads(1) +gc.disable() + + +def cpu_frequency(): + path = Path(f"/sys/devices/system/cpu/cpu{CORE}/cpufreq/scaling_cur_freq") + if path.exists(): + return float(path.read_text()) / 1000 + for section in Path("/proc/cpuinfo").read_text().split("\n\n"): + fields = dict(line.split(":", 1) for line in section.splitlines() if ":" in line) + fields = {k.strip(): v.strip() for k, v in fields.items()} + if fields.get("processor") == str(CORE): + return float(fields.get("cpu MHz", "nan")) + + +def payload(case, length, device): + generator = torch.Generator().manual_seed(7451 + length) + ids = torch.randint(2, 10000, (length,), generator=generator) + specs = { + "text": [], + "audio": [("audio", 100)], + "image": [("image", 16)], + "video": [("video", 128)], + "mixed": [("audio", 100), ("image", 16), ("video", 128)], + "interleaved": [("video", 64), ("audio", 50), ("video", 64), ("audio", 50)], + }[case] + offset = 44 + counts = {} + for modality, count in specs: + ids[offset : offset + count] = IDS[modality] + counts[modality] = counts.get(modality, 0) + count + offset += count + (0 if case == "interleaved" else 2) + mask = torch.zeros(length, dtype=torch.bool) + for token in IDS.values(): + mask |= ids == token + mm = [ + set_mm_embedding_modality(torch.full((count, HIDDEN), i + 1, device=device, dtype=torch.bfloat16), modality) + for i, (modality, count) in enumerate(counts.items()) + ] + return ids.to(device=device, dtype=torch.int32), None if case == "text" else mm, None if case == "text" else mask + + +def once(fn, cuda): + # Drain preceding work before the host timer; the production function's + # own D2H synchronizations remain inside the timed call. + if cuda: + current_omni_platform.synchronize() + w0 = time.perf_counter_ns() + c0 = time.thread_time_ns() + value = fn() + c1 = time.thread_time_ns() + w1 = time.perf_counter_ns() + if cuda: + current_omni_platform.synchronize() + w2 = time.perf_counter_ns() if cuda else w1 + del value + return [(w1 - w0) / 1000, (c1 - c0) / 1000, (w2 - w0) / 1000] + + +def measure(functions, device, case, length, scope, rounds=20, iterations=20, warmups=50): + cuda = device == "cuda" + for fn in functions.values(): + for _ in range(warmups): + once(fn, cuda) + order_rng = random.Random(7451 + length + len(case) + len(scope)) + directions = [i % 2 for i in range(rounds)] + order_rng.shuffle(directions) + names = list(functions) + for block, direction in enumerate(directions): + order = names if direction == 0 else names[::-1] + for name in order: + frequency_start = cpu_frequency() + wall_start = time.time() + samples = [once(functions[name], cuda) for _ in range(iterations)] + wall_end = time.time() + RESULTS.append( + dict( + device=device, + case=case, + tokens=length, + scope=scope, + variant=name, + block=block, + iterations=iterations, + samples_us=samples, + wall_start=wall_start, + wall_end=wall_end, + cpu_mhz_start=frequency_start, + cpu_mhz_end=cpu_frequency(), + ) + ) + (ROOT / "micro-raw.json").write_text(json.dumps(RESULTS)) + current = [r for r in RESULTS if (r["device"], r["case"], r["tokens"], r["scope"]) == (device, case, length, scope)] + print( + "RESULT", + json.dumps( + { + "device": device, + "case": case, + "tokens": length, + "scope": scope, + "host_median_us": { + v: statistics.median(t[0] for r in current if r["variant"] == v for t in r["samples_us"]) + for v in names + }, + } + ), + flush=True, + ) + + +def build_model(weights, device): + model = Thinker.__new__(Thinker) + nn.Module.__init__(model) + model.config = Qwen2_5OmniThinkerConfig(**CONFIG) + model._has_oov_mm_tokens = False + table = VocabParallelEmbedding(VOCAB, HIDDEN, params_dtype=torch.bfloat16, disable_tp=True) + table.weight = nn.Parameter(weights.to(device), requires_grad=False) + + class LM(nn.Module): + def __init__(self): + super().__init__() + self.embed_tokens = table + + def embed_input_ids(self, ids): + return self.embed_tokens(ids) + + model.language_model = LM() + return model + + +def validate(model, args, case, length, device): + ids, mm, mask = args + lm = model.get_language_model() + original = lm.embed_input_ids + outputs, counts = {}, {} + for variant, method in METHODS.items(): + calls = [] + + def counted(tokens): + calls.append(tokens.numel()) + return original(tokens) + + lm.embed_input_ids = counted + try: + outputs[variant] = method(model, ids, mm, is_multimodal=mask) + finally: + lm.embed_input_ids = original + counts[variant] = len(calls) + assert torch.equal(outputs["baseline"], outputs["fixed"]) + assert counts == {"baseline": 1 if case == "text" else 2, "fixed": 1} + VALIDATIONS.append(dict(case=case, tokens=length, device=device, calls=counts, same_input_wrapper_torch_equal=True)) + (ROOT / "validation.json").write_text(json.dumps(VALIDATIONS, indent=2)) + + +@torch.inference_mode() +def main(): + import transformers + import vllm + + metadata = dict( + baseline_ref=args.baseline_ref, + checkout=subprocess.check_output(["git", "rev-parse", "HEAD"], text=True).strip(), + torch=torch.__version__, + vllm=vllm.__version__, + transformers=transformers.__version__, + gpu=torch.cuda.get_device_name(), + cpu_core=CORE, + torch_cpu_threads=1, + warmup=args.warmup, + blocks=args.blocks, + iterations=args.iterations, + lengths=args.lengths, + embedding_shape=[VOCAB, HIDDEN], + dtype="bfloat16", + input_ids_dtype="int32", + mask_device="cpu", + start_time=time.time(), + metric_order=["host_observed_us", "calling_thread_cpu_us", "synchronized_completion_us"], + ) + index = json.loads((MODEL / "model.safetensors.index.json").read_text())["weight_map"] + key = "thinker.model.embed_tokens.weight" + with safe_open(MODEL / index[key], framework="pt", device="cpu") as f: + weights = f.get_tensor(key) + assert weights.shape == (VOCAB, HIDDEN) and weights.dtype == torch.bfloat16 + model = build_model(weights, "cuda") + measure({"timer_floor": lambda: None}, "cuda", "timer", 0, "overhead", rounds=10, iterations=100) + for length in args.lengths: + for case in ("text", "audio", "image", "video", "mixed", "interleaved"): + ids, mm, mask = payload(case, length, "cuda") + validate(model, (ids, mm, mask), case, length, "cuda") + functions = { + variant: (lambda method=method: method(model, ids, mm, is_multimodal=mask)) + for variant, method in METHODS.items() + } + measure( + functions, + "cuda", + case, + length, + "whole_wrapper", + rounds=args.blocks, + iterations=args.iterations, + warmups=args.warmup, + ) + summary = [] + for length in args.lengths: + for case in ("text", "audio", "image", "video", "mixed", "interleaved"): + medians = { + variant: statistics.median( + statistics.median(sample[0] for sample in record["samples_us"]) + for record in RESULTS + if record["tokens"] == length and record["case"] == case and record["variant"] == variant + ) + for variant in METHODS + } + summary.append( + dict( + tokens=length, + case=case, + host_us=medians, + reduction_pct=100 * (1 - medians["fixed"] / medians["baseline"]), + ) + ) + metadata["end_time"] = time.time() + (ROOT / "metadata.json").write_text(json.dumps(metadata, indent=2)) + (ROOT / "summary.json").write_text(json.dumps(summary, indent=2)) + print(json.dumps(summary, indent=2)) + + +if __name__ == "__main__": + main() diff --git a/tests/model_executor/models/qwen2_5_omni/test_qwen2_5_omni_embed.py b/tests/model_executor/models/qwen2_5_omni/test_qwen2_5_omni_embed.py index 13899d47d..7b624ba11 100644 --- a/tests/model_executor/models/qwen2_5_omni/test_qwen2_5_omni_embed.py +++ b/tests/model_executor/models/qwen2_5_omni/test_qwen2_5_omni_embed.py @@ -11,6 +11,7 @@ Regression test for: https://github.com/vllm-project/vllm/issues/34506 """ import functools +from types import SimpleNamespace import pytest import torch @@ -134,16 +135,15 @@ def make_mock_model(mocker: MockerFixture, hidden: int = 8, thinker_cls=None): that has enough structure to run embed_input_ids. ``thinker_cls`` selects whose ``embed_input_ids`` the mock dispatches to. - It defaults to upstream's. Only the interleaved case passes omni's override: - omni's non-interleaved path falls through to a zero-arg ``super()`` call, - which raises TypeError when ``self`` is a Mock rather than an instance. + It defaults to upstream's. Use the selected class as the mock spec so + zero-argument ``super()`` in omni's override resolves correctly. """ m = _qwen2_5_omni_thinker_mod() Qwen2_5OmniThinkerForConditionalGeneration = m.Qwen2_5OmniThinkerForConditionalGeneration if thinker_cls is None: thinker_cls = Qwen2_5OmniThinkerForConditionalGeneration - model = mocker.Mock(spec=Qwen2_5OmniThinkerForConditionalGeneration) + model = mocker.Mock(spec=thinker_cls) # Config with token IDs cfg = mocker.Mock() @@ -157,8 +157,7 @@ def make_mock_model(mocker: MockerFixture, hidden: int = 8, thinker_cls=None): # view with shared memory, which masked_scatter_ cannot handle). return ids.float().unsqueeze(-1).expand(-1, hidden).clone() - lang_model = mocker.Mock() - lang_model.embed_input_ids = fake_lm_embed + lang_model = SimpleNamespace(embed_input_ids=fake_lm_embed) model.get_language_model = mocker.Mock(return_value=lang_model) from vllm.model_executor.models.interfaces import SupportsMultiModal @@ -271,6 +270,27 @@ class TestEmbedInputIds: "Text positions should keep text embeddings" ) + @pytest.mark.parametrize( + "audio_n,image_n,video_n", + [(5, 0, 0), (0, 4, 0), (0, 0, 6), (5, 4, 6)], + ids=["audio", "image", "video", "mixed"], + ) + def test_non_interleaved_multimodal_single_embedding(self, mocker: MockerFixture, audio_n, image_n, video_n): + """Omni's regular path embeds once through super() and preserves placement.""" + hidden = 8 + input_ids, is_multimodal = make_token_seq(audio_n, image_n, video_n) + mm_embeds = build_mm_embeds(audio_n, image_n, video_n, hidden) + model, _ = make_mock_model(mocker, hidden, thinker_cls=_omni_qwen2_5_thinker_cls()) + embed = mocker.spy(model.get_language_model(), "embed_input_ids") + + result = model.embed_input_ids(input_ids, mm_embeds, is_multimodal=is_multimodal) + + embed.assert_called_once() + expected = input_ids.float().unsqueeze(-1).expand(-1, hidden).clone() + for token_id, value in ((AUDIO_TOKEN_ID, 10.0), (IMAGE_TOKEN_ID, 20.0), (VIDEO_TOKEN_ID, 30.0)): + expected[input_ids == token_id] = value + assert torch.equal(result, expected) + def test_interleaved_use_audio_in_video(self, mocker: MockerFixture): """ Interleaved (use_audio_in_video): video chunks interleaved with audio. @@ -296,7 +316,9 @@ class TestEmbedInputIds: ] model, _ = make_mock_model(mocker, hidden, thinker_cls=_omni_qwen2_5_thinker_cls()) + embed = mocker.spy(model.get_language_model(), "embed_input_ids") result = model.embed_input_ids(input_ids, mm_embeds, is_multimodal=is_multimodal) + embed.assert_called_once() video_pos = (input_ids == VIDEO_TOKEN_ID).nonzero(as_tuple=True)[0] audio_pos = (input_ids == AUDIO_TOKEN_ID).nonzero(as_tuple=True)[0] diff --git a/vllm_omni/model_executor/models/qwen2_5_omni/qwen2_5_omni_thinker.py b/vllm_omni/model_executor/models/qwen2_5_omni/qwen2_5_omni_thinker.py index 6bbefceab..263ffa2ab 100644 --- a/vllm_omni/model_executor/models/qwen2_5_omni/qwen2_5_omni_thinker.py +++ b/vllm_omni/model_executor/models/qwen2_5_omni/qwen2_5_omni_thinker.py @@ -1326,14 +1326,12 @@ class Qwen2_5OmniThinkerForConditionalGeneration( if multimodal_embeddings is None or is_multimodal is None: return super().embed_input_ids(input_ids) - inputs_embeds = self._embed_text_input_ids( - input_ids, - self.get_language_model().embed_input_ids, - is_multimodal=is_multimodal, - ) - if len(multimodal_embeddings) == 0: - return inputs_embeds + return self._embed_text_input_ids( + input_ids, + self.get_language_model().embed_input_ids, + is_multimodal=is_multimodal, + ) # Check for audio-in-video: interleaved video and audio tokens # in the multimodal region. Only use the interleaved path when @@ -1341,6 +1339,8 @@ class Qwen2_5OmniThinkerForConditionalGeneration( video_token_id = self.config.video_token_index audio_token_id = self.config.audio_token_index + # The interleaving check reads scalar positions in a Python loop. + # Keep its masks on CPU to avoid repeated CUDA synchronization. input_ids_cpu = input_ids.cpu() is_video = is_multimodal & (input_ids_cpu == video_token_id) is_audio = is_multimodal & (input_ids_cpu == audio_token_id)