[Refactor][WIP] Add missing test for modelrunner (G1/N) (#5333)

Signed-off-by: Taichang Zhou <tzhouam@connect.ust.hk>
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Zhou TaichangandClaude Opus 4.8 authored and GitHub committed 2026-07-27 16:24:47 +08:00
1 parent d688aa82fc
commit 776424f0de
9 files changed
+1196

No files matched your search

@@ -0,0 +1,79 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
"""Static coupling check for the ``has_preprocess`` capture/replay buffers.
For models with ``has_preprocess``, CUDA-graph *capture* (``_dummy_run``) and
*replay* (``_preprocess``) MUST read from the same pre-allocated buffers
(``self.input_ids.gpu`` + ``self.inputs_embeds.gpu``); otherwise replay reads a
buffer capture never wrote and produces garbage. This is a source-level check
(no GPU) that catches a one-sided edit to either branch.
The check is AST-scoped: it inspects ONLY the body of the ``has_preprocess``
guarded branch (not a line window), so a buffer read removed from that branch
cannot be masked by an adjacent ``else``/``elif`` that happens to read it. It
matches by attribute chain (``self.input_ids.gpu``), so local variable renames
do not break it.
Scope: the BASE runner (``OmniGPUModelRunner``) only. The generation runner is
KNOWN to lack the ``has_preprocess`` branch; extend this
check to the generation file when it adds the branch.
"""
import ast
import inspect
import textwrap
import pytest
from vllm_omni.worker.gpu_model_runner import OmniGPUModelRunner
pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
_REQUIRED_BUFFERS = {"self.input_ids.gpu", "self.inputs_embeds.gpu"}
def _attr_chain(node: ast.AST) -> str | None:
"""Reconstruct a dotted ``self.x.y`` chain from an ast.Attribute node."""
parts: list[str] = []
cur = node
while isinstance(cur, ast.Attribute):
parts.append(cur.attr)
cur = cur.value
if isinstance(cur, ast.Name):
parts.append(cur.id)
return ".".join(reversed(parts))
return None
def _attr_chains_in(nodes: list[ast.stmt]) -> set[str]:
chains: set[str] = set()
for stmt in nodes:
for inner in ast.walk(stmt):
if isinstance(inner, ast.Attribute):
chain = _attr_chain(inner)
if chain is not None:
chains.add(chain)
return chains
def _has_preprocess_branch_reads_both_buffers(method) -> bool:
"""True iff SOME ``if has_preprocess:`` branch body reads BOTH buffers.
Only the guarded branch's own ``body`` is inspected (elif/else excluded),
which is what makes a one-sided buffer removal fail.
"""
src = textwrap.dedent(inspect.getsource(method))
tree = ast.parse(src)
guarded = [node for node in ast.walk(tree) if isinstance(node, ast.If) and "has_preprocess" in ast.dump(node.test)]
assert guarded, f"{method.__qualname__}: has_preprocess branch removed entirely"
return any(_REQUIRED_BUFFERS <= _attr_chains_in(node.body) for node in guarded)
@pytest.mark.parametrize("method_name", ["_dummy_run", "_preprocess"])
def test_has_preprocess_branch_reads_shared_buffers(method_name):
method = getattr(OmniGPUModelRunner, method_name)
assert _has_preprocess_branch_reads_both_buffers(method), (
f"{OmniGPUModelRunner.__name__}.{method_name}: the has_preprocess branch no "
"longer reads BOTH self.input_ids.gpu and self.inputs_embeds.gpu — capture "
"and replay would diverge."
)
@@ -0,0 +1,85 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
"""Characterization of ``OmniGPUModelRunner.initialize_metadata_builders``.
Upstream ``FlashAttentionMetadataBuilder`` pre-allocates ``scheduler_metadata``
for only ``max_num_seqs + 1`` entries, but FA3 with split scheduling
(``max_num_splits > 1``) may need ``max_num_seqs * max_num_splits + 1`` during
CUDA-graph capture. ``OmniGPUModelRunner`` re-sizes the buffer up to that
requirement after upstream init — this test pins that Omni-specific workaround
exactly.
It drives the Omni runner's OWN method (with the upstream ``super()`` call
stubbed and a fake builder standing in), so a change to the resize is caught and
a failure is unambiguously an Omni-runner regression. Pure CPU — no model, no
CUDA, no flashinfer.
"""
from types import SimpleNamespace
import pytest
import torch
from vllm_omni.worker.gpu_model_runner import OmniGPUModelRunner
pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
# The class ``super().initialize_metadata_builders`` dispatches to.
_UPSTREAM_RUNNER = OmniGPUModelRunner.__mro__[1]
def _builder(*, prealloc: int | None, max_num_splits: int) -> SimpleNamespace:
sm = None if prealloc is None else torch.zeros(prealloc, dtype=torch.int32)
return SimpleNamespace(scheduler_metadata=sm, max_num_splits=max_num_splits)
def _make_runner(monkeypatch, *, max_num_seqs: int, builders: list) -> OmniGPUModelRunner:
# Stub the upstream initialize_metadata_builders; our fake runner already
# carries the attn_groups the Omni override iterates over.
monkeypatch.setattr(_UPSTREAM_RUNNER, "initialize_metadata_builders", lambda self, *a, **k: None)
runner = object.__new__(OmniGPUModelRunner)
runner.scheduler_config = SimpleNamespace(max_num_seqs=max_num_seqs)
runner.cache_config = SimpleNamespace(enable_prefix_caching=False) # skip omni_prefix_cache branch
runner.attn_groups = [[SimpleNamespace(metadata_builders=builders)]]
return runner
def test_resizes_up_when_split_scheduling_needs_more(monkeypatch):
max_num_seqs, splits, prealloc = 32, 4, 33 # 33 == upstream default (max_num_seqs + 1)
builder = _builder(prealloc=prealloc, max_num_splits=splits)
runner = _make_runner(monkeypatch, max_num_seqs=max_num_seqs, builders=[builder])
OmniGPUModelRunner.initialize_metadata_builders(runner, None, None)
assert builder.scheduler_metadata.shape[0] == max_num_seqs * splits + 1
assert builder.scheduler_metadata.dtype == torch.int32
def test_no_resize_when_not_split_scheduling(monkeypatch):
builder = _builder(prealloc=33, max_num_splits=1) # max_num_splits <= 1 -> workaround skipped
runner = _make_runner(monkeypatch, max_num_seqs=32, builders=[builder])
original = builder.scheduler_metadata
OmniGPUModelRunner.initialize_metadata_builders(runner, None, None)
assert builder.scheduler_metadata is original # untouched
def test_no_resize_when_prealloc_already_sufficient(monkeypatch):
max_num_seqs, splits = 32, 4
builder = _builder(prealloc=max_num_seqs * splits + 1, max_num_splits=splits) # already big enough
runner = _make_runner(monkeypatch, max_num_seqs=max_num_seqs, builders=[builder])
original = builder.scheduler_metadata
OmniGPUModelRunner.initialize_metadata_builders(runner, None, None)
assert builder.scheduler_metadata is original
def test_no_crash_when_builder_has_no_scheduler_metadata(monkeypatch):
builder = _builder(prealloc=None, max_num_splits=4) # scheduler_metadata is None
runner = _make_runner(monkeypatch, max_num_seqs=32, builders=[builder])
OmniGPUModelRunner.initialize_metadata_builders(runner, None, None)
assert builder.scheduler_metadata is None
+92
View File
@@ -802,3 +802,95 @@ def test_build_omni_output_falls_back_to_mm_cpu_without_prefix_merge(monkeypatch
assert torch.equal(output.inter_stage_outputs[0]["codes.audio"], codes[0:1])
assert torch.equal(output.inter_stage_outputs[1]["codes.audio"], codes[1:2])
assert output.multimodal_outputs is None
# --- builder gap-fill: contract corners not covered by the tests above ---
def test_build_omni_output_never_leaks_internal_pooler_output_on_wire(monkeypatch):
"""The internal pooler_output feeds full-payload accumulation only; the wire
object's ``pooler_output`` is always None (builder NOTE at gpu_ar:1738)."""
runner = _make_async_output_runner()
monkeypatch.setattr(
GPUARModelRunner,
"_resolve_pooler_payload_req_ids",
lambda self, req_ids: ("audio", req_ids),
)
monkeypatch.setattr(GPUARModelRunner, "_should_accumulate_full_payload_output", lambda self: True)
monkeypatch.setattr(
GPUARModelRunner,
"accumulate_full_payload_output",
lambda self, rid, payload, request: None,
)
monkeypatch.setattr(GPUARModelRunner, "get_omni_connector_output", lambda self: None)
output = GPUARModelRunner._build_omni_model_runner_output_from_snapshot(
runner,
scheduler_output=SimpleNamespace(
total_num_scheduled_tokens=3,
num_scheduled_tokens={"r1": 1, "r2": 2},
),
hidden_states=torch.tensor([[1.0], [2.0], [3.0]]),
staged_hidden_states_cpu=None,
multimodal_outputs={"foo": torch.tensor([10.0, 20.0, 30.0])},
req_ids_output_copy=["r1", "r2"],
req_id_to_index_output_copy={"r1": 0, "r2": 1},
valid_sampled_token_ids=[[101], [102]],
logprobs_lists=None,
prompt_logprobs_dict={},
num_nans_in_logits=None,
kv_connector_output=None,
ec_connector_output=None,
cudagraph_stats=None,
kv_extracted_req_ids=None,
num_scheduled_tokens_np=np.array([1, 2], dtype=np.int32),
query_start_loc_cpu=torch.tensor([0, 1], dtype=torch.long),
)
assert output.pooler_output is None
def test_build_omni_output_splits_client_mm_from_inter_stage_keys(monkeypatch):
"""The inter-stage vs client-mm partition: a client-root key ("audio") lands in
``multimodal_outputs`` while inter-stage keys ("hidden") stay in
``inter_stage_outputs`` (partition_flat_payload / _CLIENT_MM_ROOT_KEYS)."""
runner = _make_async_output_runner()
monkeypatch.setattr(
GPUARModelRunner,
"_resolve_pooler_payload_req_ids",
lambda self, req_ids: ("audio", req_ids),
)
monkeypatch.setattr(GPUARModelRunner, "_should_accumulate_full_payload_output", lambda self: False)
monkeypatch.setattr(GPUARModelRunner, "get_omni_connector_output", lambda self: None)
output = GPUARModelRunner._build_omni_model_runner_output_from_snapshot(
runner,
scheduler_output=SimpleNamespace(
total_num_scheduled_tokens=3,
num_scheduled_tokens={"r1": 1, "r2": 2},
),
hidden_states=torch.tensor([[1.0], [2.0], [3.0]]),
staged_hidden_states_cpu=None,
# "audio" root is a client mm key; "codes.audio" root ("codes") is inter-stage.
multimodal_outputs={"audio": torch.tensor([10.0, 20.0, 30.0])},
req_ids_output_copy=["r1", "r2"],
req_id_to_index_output_copy={"r1": 0, "r2": 1},
valid_sampled_token_ids=[[101], [102]],
logprobs_lists=None,
prompt_logprobs_dict={},
num_nans_in_logits=None,
kv_connector_output=None,
ec_connector_output=None,
cudagraph_stats=None,
kv_extracted_req_ids=None,
num_scheduled_tokens_np=np.array([1, 2], dtype=np.int32),
query_start_loc_cpu=torch.tensor([0, 1], dtype=torch.long),
)
# Client mm key routed to the wire multimodal_outputs, per request.
assert output.multimodal_outputs is not None
assert "audio" in output.multimodal_outputs[0]
# Inter-stage hidden stayed on inter_stage_outputs and did NOT leak to client.
assert output.inter_stage_outputs is not None
assert "hidden" in output.inter_stage_outputs[0]
assert "audio" not in output.inter_stage_outputs[0]
+71
View File
@@ -0,0 +1,71 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
"""Characterization tests for ``vllm_omni.worker.memory_utils``.
``request_memory_tolerant`` is still live in both workers
(``gpu_ar_worker.py`` / ``gpu_generation_worker.py``); a later change makes its cap loud
but must not change the numeric contract. These tests pin that contract with a
mocked ``MemorySnapshot`` — pure CPU, no GPU / NVML.
"""
import math
from types import SimpleNamespace
import pytest
from vllm_omni.worker.memory_utils import request_memory_tolerant
pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
GIB = 1024**3
def _snapshot(total: int, free: int) -> SimpleNamespace:
# Only the fields request_memory_tolerant reads (incl. device_ for the log).
return SimpleNamespace(total_memory=total, free_memory=free, device_="cuda:0")
def _cache_config(util: float) -> SimpleNamespace:
return SimpleNamespace(gpu_memory_utilization=util)
def test_passes_through_requested_when_free_suffices():
snap = _snapshot(total=40 * GIB, free=40 * GIB)
cfg = _cache_config(0.5)
out = request_memory_tolerant(snap, cfg)
assert out == math.ceil(40 * GIB * 0.5)
def test_caps_to_free_when_insufficient():
cfg = _cache_config(0.9)
requested = math.ceil(40 * GIB * 0.9)
snap = _snapshot(total=40 * GIB, free=10 * GIB) # free < requested
out = request_memory_tolerant(snap, cfg)
assert out == 10 * GIB
assert out < requested
def test_boundary_free_equals_requested_is_not_capped():
util = 0.5
total = 40 * GIB
requested = math.ceil(total * util)
snap = _snapshot(total=total, free=requested) # strict `<`, so no cap at equality
out = request_memory_tolerant(snap, _cache_config(util))
assert out == requested
def test_requested_uses_ceil_of_total_times_util():
# 7 GiB * 0.3333... rounds UP (math.ceil), not down.
total = 7 * GIB
util = 1 / 3
snap = _snapshot(total=total, free=total)
out = request_memory_tolerant(snap, _cache_config(util))
assert out == math.ceil(total * util)
+166
View File
@@ -0,0 +1,166 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
"""Characterization tests for ``vllm_omni.worker.payload_span``.
These pin the CURRENT behaviour of the thinker-decode span helpers so the later
worker refactors that touch span
plumbing have a regression net. Pure CPU — no model, no CUDA.
"""
import pytest
import torch
from vllm_omni.worker.payload_span import (
get_tensor_span,
get_tensor_span_row,
merge_tensor_spans,
)
pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
def _rows(n: int, *, start_val: int = 0, hidden: int = 1) -> torch.Tensor:
"""A ``[n, hidden]`` tensor whose first column counts up from ``start_val``.
``[n, hidden]`` mirrors the connector async-chunk shape (tokens x hidden).
"""
base = torch.arange(start_val, start_val + n, dtype=torch.float32).unsqueeze(1)
return base.repeat(1, hidden)
# --------------------------------------------------------------------------- #
# get_tensor_span #
# --------------------------------------------------------------------------- #
def test_get_tensor_span_valid_returns_triplet():
tensor = _rows(3, hidden=4)
payload = {"emb": tensor, "start": 5, "end": 8}
span = get_tensor_span(payload, tensor_key="emb", start_key="start", end_key="end")
assert span is not None
got_tensor, start, end = span
assert got_tensor is tensor
assert (start, end) == (5, 8)
def test_get_tensor_span_non_tensor_returns_none():
payload = {"emb": None, "start": 0, "end": 0}
assert get_tensor_span(payload, tensor_key="emb", start_key="start", end_key="end") is None
def test_get_tensor_span_non_int_bounds_returns_none():
tensor = _rows(2)
# start is a float, not an int -> rejected.
payload = {"emb": tensor, "start": 0.0, "end": 2}
assert get_tensor_span(payload, tensor_key="emb", start_key="start", end_key="end") is None
def test_get_tensor_span_negative_start_returns_none():
tensor = _rows(1)
payload = {"emb": tensor, "start": -1, "end": 0}
assert get_tensor_span(payload, tensor_key="emb", start_key="start", end_key="end") is None
def test_get_tensor_span_end_before_start_returns_none():
tensor = _rows(2)
payload = {"emb": tensor, "start": 5, "end": 3}
assert get_tensor_span(payload, tensor_key="emb", start_key="start", end_key="end") is None
def test_get_tensor_span_length_mismatch_returns_none():
tensor = _rows(3) # tensor.shape[0] == 3
payload = {"emb": tensor, "start": 0, "end": 2} # end - start == 2
assert get_tensor_span(payload, tensor_key="emb", start_key="start", end_key="end") is None
# --------------------------------------------------------------------------- #
# merge_tensor_spans #
# --------------------------------------------------------------------------- #
def test_merge_none_operand_returns_none():
span = (_rows(2), 0, 2)
assert merge_tensor_spans(None, span) is None
assert merge_tensor_spans(span, None) is None
def test_merge_adjacent_spans_concatenates():
existing = (_rows(3, start_val=0), 0, 3)
incoming = (_rows(2, start_val=3), 3, 5)
merged = merge_tensor_spans(existing, incoming)
assert merged is not None
tensor, start, end = merged
assert (start, end) == (0, 5)
assert tensor.shape[0] == 5
assert tensor[:, 0].tolist() == [0.0, 1.0, 2.0, 3.0, 4.0]
def test_merge_overlapping_spans_trims_incoming():
# existing covers [0, 3); incoming covers [2, 5) -> 1 row of overlap.
existing = (_rows(3, start_val=0), 0, 3)
incoming = (_rows(3, start_val=100), 2, 5)
merged = merge_tensor_spans(existing, incoming)
assert merged is not None
tensor, start, end = merged
assert (start, end) == (0, 5)
# First 3 rows from existing, then incoming's tail (overlap=1 row trimmed).
assert tensor[:, 0].tolist() == [0.0, 1.0, 2.0, 101.0, 102.0]
def test_merge_fully_overlapping_incoming_keeps_existing():
# incoming [1, 3) is fully covered by existing [0, 3): overlap >= incoming len.
existing = (_rows(3, start_val=0), 0, 3)
incoming = (_rows(2, start_val=100), 1, 3)
merged = merge_tensor_spans(existing, incoming)
assert merged is not None
tensor, start, end = merged
assert (start, end) == (0, 3)
assert tensor is existing[0] # returned unchanged
def test_merge_non_contiguous_spans_returns_none():
# gap between existing end (3) and incoming start (5).
existing = (_rows(3, start_val=0), 0, 3)
incoming = (_rows(2, start_val=100), 5, 7)
assert merge_tensor_spans(existing, incoming) is None
def test_merge_casts_incoming_to_existing_dtype_and_device():
existing = (_rows(2, start_val=0).to(torch.float32), 0, 2)
incoming = (torch.arange(2, 4, dtype=torch.int64).unsqueeze(1), 2, 4)
merged = merge_tensor_spans(existing, incoming)
assert merged is not None
tensor, _, _ = merged
assert tensor.dtype == torch.float32
assert tensor.device == existing[0].device
# --------------------------------------------------------------------------- #
# get_tensor_span_row #
# --------------------------------------------------------------------------- #
def test_get_tensor_span_row_in_range():
span = (_rows(3, start_val=10), 5, 8) # rows map to indices 5, 6, 7
row = get_tensor_span_row(span, 6)
assert row is not None
assert row[0].item() == 11.0 # second row (start_val 10 + 1)
def test_get_tensor_span_row_none_span_returns_none():
assert get_tensor_span_row(None, 0) is None
def test_get_tensor_span_row_below_start_returns_none():
span = (_rows(3), 5, 8)
assert get_tensor_span_row(span, 4) is None
def test_get_tensor_span_row_at_or_above_end_returns_none():
span = (_rows(3), 5, 8)
assert get_tensor_span_row(span, 8) is None
@@ -0,0 +1,142 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
"""Golden characterization of ``_build_model_sampler_output_token_ids``.
The base runner (``OmniGPUModelRunner``) and the AR runner
(``GPUARModelRunner``) implement the SAME async-placeholder backfill, but the
AR copy adds a trailing crop step: after backfill it truncates every history at
the first remaining ``-1``, while the base copy KEEPS residual ``-1``s.
A later change collapses these two copies into one; these goldens pin today's divergence
exactly (crop vs keep) so that merge is provably behaviour-preserving. Pure CPU.
"""
from types import SimpleNamespace
import pytest
import torch
from vllm_omni.worker.gpu_ar_model_runner import GPUARModelRunner
from vllm_omni.worker.gpu_model_runner import OmniGPUModelRunner
pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
def _runner(
cls,
*,
req_ids,
req_output_token_ids,
sampled_token_ids_cpu,
prev_req_id_to_index,
):
"""Build a fake runner exposing only the ``input_batch`` fields the method reads."""
runner = object.__new__(cls)
event = SimpleNamespace(synchronize=lambda: None)
runner.input_batch = SimpleNamespace(
req_output_token_ids=req_output_token_ids,
req_ids=req_ids,
sampled_token_ids_cpu=(None if sampled_token_ids_cpu is None else torch.tensor(sampled_token_ids_cpu)),
async_copy_ready_event=event,
prev_req_id_to_index=prev_req_id_to_index,
)
return runner
def _build(cls, **kwargs):
return cls._build_model_sampler_output_token_ids(_runner(cls, **kwargs))
# --------------------------------------------------------------------------- #
# Shared behaviour — base and AR agree #
# --------------------------------------------------------------------------- #
@pytest.mark.parametrize("cls", [OmniGPUModelRunner, GPUARModelRunner])
def test_full_backfill_no_residual_placeholder(cls):
# One trailing placeholder, one sampled token -> fully resolved for both.
out = _build(
cls,
req_ids=["r0"],
req_output_token_ids=[[10, 11, -1]],
sampled_token_ids_cpu=[[42]],
prev_req_id_to_index={"r0": 0},
)
assert out == [[10, 11, 42]]
@pytest.mark.parametrize("cls", [OmniGPUModelRunner, GPUARModelRunner])
def test_spec_decode_trailing_pad_in_sampled_is_trimmed(cls):
# sampled row has a trailing -1 pad; only the real prefix [42, 43] is used.
out = _build(
cls,
req_ids=["r0"],
req_output_token_ids=[[10, -1, -1]],
sampled_token_ids_cpu=[[42, 43, -1]],
prev_req_id_to_index={"r0": 0},
)
assert out == [[10, 42, 43]]
@pytest.mark.parametrize("cls", [OmniGPUModelRunner, GPUARModelRunner])
def test_no_trailing_placeholder_is_untouched(cls):
out = _build(
cls,
req_ids=["r0"],
req_output_token_ids=[[10, 11, 12]],
sampled_token_ids_cpu=[[42]],
prev_req_id_to_index={"r0": 0},
)
assert out == [[10, 11, 12]]
@pytest.mark.parametrize("cls", [OmniGPUModelRunner, GPUARModelRunner])
def test_early_return_when_sampled_absent_keeps_placeholders(cls):
# sampled_token_ids_cpu is None -> both copies return BEFORE the crop step,
# so even the AR copy keeps residual -1 here.
out = _build(
cls,
req_ids=["r0"],
req_output_token_ids=[[10, -1]],
sampled_token_ids_cpu=None,
prev_req_id_to_index={"r0": 0},
)
assert out == [[10, -1]]
# --------------------------------------------------------------------------- #
# Divergence — base KEEPS residual -1, AR CROPS at the first -1 #
# --------------------------------------------------------------------------- #
_RESIDUAL_CASE = dict(
req_ids=["r0"],
req_output_token_ids=[[10, -1, -1]], # two placeholders
sampled_token_ids_cpu=[[42]], # only one sampled token -> one -1 remains
prev_req_id_to_index={"r0": 0},
)
def test_base_keeps_residual_placeholder():
out = _build(OmniGPUModelRunner, **_RESIDUAL_CASE)
assert out == [[10, 42, -1]]
def test_ar_crops_residual_placeholder():
out = _build(GPUARModelRunner, **_RESIDUAL_CASE)
assert out == [[10, 42]]
_ABSENT_FROM_PREV_CASE = dict(
req_ids=["r0"],
req_output_token_ids=[[10, -1]],
sampled_token_ids_cpu=[[42]],
prev_req_id_to_index={}, # r0 not present last step -> no backfill happens
)
def test_base_keeps_placeholder_when_request_absent_from_prev_step():
out = _build(OmniGPUModelRunner, **_ABSENT_FROM_PREV_CASE)
assert out == [[10, -1]]
def test_ar_crops_placeholder_when_request_absent_from_prev_step():
# No backfill occurred, but the AR crop pass still truncates at the -1.
out = _build(GPUARModelRunner, **_ABSENT_FROM_PREV_CASE)
assert out == [[10]]
+237
View File
@@ -0,0 +1,237 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
"""Characterization tests for ``vllm_omni.worker.base.OmniGPUWorkerBase``.
Pins the CURRENT behaviour of ``determine_available_memory`` (BOTH the NVML /
process-scoped arm AND the profiling-fallback arm — this base still carries the
NVML branch), plus the
memory-pool / sleep / wake-up plumbing that a later change may touch.
Workers are constructed via ``object.__new__`` with only the attributes each
method reads; module-level collaborators are monkeypatched. Pure CPU.
"""
from contextlib import contextmanager, nullcontext
from types import SimpleNamespace
import pytest
import torch
import vllm_omni.worker.base as base
from vllm_omni.worker.base import OmniGPUWorkerBase
pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
GIB = 1024**3
def _fake_memory_profiling(*, non_torch: int, torch_peak: int):
"""A stand-in for ``vllm.utils.mem_utils.memory_profiling`` context manager."""
@contextmanager
def _mp(snapshot, weights_memory): # noqa: ARG001 - signature parity only
yield SimpleNamespace(
non_torch_increase=non_torch,
torch_peak_increase=torch_peak,
)
return _mp
def _make_worker(*, requested_memory: int, kv_cache_memory_bytes: int = 0, model_memory_usage: int = 0):
worker = object.__new__(OmniGPUWorkerBase)
worker.cache_config = SimpleNamespace(kv_cache_memory_bytes=kv_cache_memory_bytes)
worker.model_runner = SimpleNamespace(
profile_run=lambda: None,
model_memory_usage=model_memory_usage,
)
worker.init_snapshot = object() # unused once memory_profiling is faked
worker.requested_memory = requested_memory
worker.local_rank = 0
return worker
# --------------------------------------------------------------------------- #
# determine_available_memory #
# --------------------------------------------------------------------------- #
def test_determine_available_memory_nvml_arm(monkeypatch):
"""NVML available -> available = requested - per-process memory."""
worker = _make_worker(requested_memory=30 * GIB, model_memory_usage=10 * GIB)
monkeypatch.setattr(base, "memory_profiling", _fake_memory_profiling(non_torch=1 * GIB, torch_peak=2 * GIB))
monkeypatch.setattr(base, "is_process_scoped_memory_available", lambda: True)
monkeypatch.setattr(base, "detect_pid_host", lambda: True)
monkeypatch.setattr(base, "get_process_gpu_memory", lambda local_rank: 8 * GIB)
out = OmniGPUWorkerBase.determine_available_memory(worker)
# NVML arm ignores profiled usage; uses process_memory directly.
assert out == 30 * GIB - 8 * GIB
def test_determine_available_memory_profiling_fallback_arm(monkeypatch):
"""NVML unavailable -> available = requested - (weights + peak + non_torch)."""
worker = _make_worker(requested_memory=30 * GIB, model_memory_usage=10 * GIB)
monkeypatch.setattr(base, "memory_profiling", _fake_memory_profiling(non_torch=1 * GIB, torch_peak=2 * GIB))
monkeypatch.setattr(base, "is_process_scoped_memory_available", lambda: False)
# detect_pid_host is short-circuited by the `and`, but pin it anyway.
monkeypatch.setattr(base, "detect_pid_host", lambda: True)
out = OmniGPUWorkerBase.determine_available_memory(worker)
profiled = 10 * GIB + 2 * GIB + 1 * GIB
assert out == 30 * GIB - profiled
def test_determine_available_memory_falls_back_when_not_pid_host(monkeypatch):
"""detect_pid_host False also routes to the profiling fallback."""
worker = _make_worker(requested_memory=20 * GIB, model_memory_usage=5 * GIB)
monkeypatch.setattr(base, "memory_profiling", _fake_memory_profiling(non_torch=0, torch_peak=0))
monkeypatch.setattr(base, "is_process_scoped_memory_available", lambda: True)
monkeypatch.setattr(base, "detect_pid_host", lambda: False)
out = OmniGPUWorkerBase.determine_available_memory(worker)
assert out == 20 * GIB - 5 * GIB
def test_determine_available_memory_clamps_to_zero(monkeypatch):
"""Over-subscription clamps the KV budget to 0, never negative."""
worker = _make_worker(requested_memory=5 * GIB, model_memory_usage=0)
monkeypatch.setattr(base, "memory_profiling", _fake_memory_profiling(non_torch=0, torch_peak=0))
monkeypatch.setattr(base, "is_process_scoped_memory_available", lambda: True)
monkeypatch.setattr(base, "detect_pid_host", lambda: True)
monkeypatch.setattr(base, "get_process_gpu_memory", lambda local_rank: 8 * GIB)
out = OmniGPUWorkerBase.determine_available_memory(worker)
assert out == 0
def test_determine_available_memory_kv_cache_short_circuit(monkeypatch):
"""A pre-set kv_cache_memory_bytes short-circuits profiling and is returned."""
worker = _make_worker(requested_memory=30 * GIB, kv_cache_memory_bytes=7 * GIB)
profile_calls = []
worker.model_runner.profile_run = lambda: profile_calls.append(True)
# is_rocm gates only an extra synchronize on the short-circuit path.
monkeypatch.setattr(base, "current_omni_platform", SimpleNamespace(is_rocm=lambda: False))
out = OmniGPUWorkerBase.determine_available_memory(worker)
assert out == 7 * GIB
assert profile_calls == [True] # profile_run still runs before returning
def test_determine_available_memory_kv_short_circuit_syncs_on_rocm(monkeypatch):
"""On ROCm the short-circuit adds an accelerator synchronize (a later change may remove it)."""
worker = _make_worker(requested_memory=30 * GIB, kv_cache_memory_bytes=7 * GIB)
monkeypatch.setattr(base, "current_omni_platform", SimpleNamespace(is_rocm=lambda: True))
synced = []
monkeypatch.setattr(torch.accelerator, "synchronize", lambda: synced.append(True))
out = OmniGPUWorkerBase.determine_available_memory(worker)
assert out == 7 * GIB
assert synced == [True] # ROCm path synchronizes; non-ROCm path (above) does not
# --------------------------------------------------------------------------- #
# _maybe_get_memory_pool_context #
# --------------------------------------------------------------------------- #
def test_memory_pool_context_disabled_returns_nullcontext():
worker = object.__new__(OmniGPUWorkerBase)
worker.rank = 0
worker.cache_config = SimpleNamespace(enable_sleep_mode=False)
# No vllm_config attribute -> v1_config_enabled stays False.
ctx = OmniGPUWorkerBase._maybe_get_memory_pool_context(worker, "weights")
assert isinstance(ctx, type(nullcontext()))
def test_memory_pool_context_enabled_uses_cumem_pool_with_tag(monkeypatch):
import vllm.device_allocator.cumem as cumem_mod
worker = object.__new__(OmniGPUWorkerBase)
worker.rank = 0
worker.cache_config = SimpleNamespace(enable_sleep_mode=True)
monkeypatch.setattr(base, "current_omni_platform", SimpleNamespace(synchronize=lambda: None))
recorded = {}
class FakeAllocator:
@classmethod
def get_instance(cls):
return cls()
def use_memory_pool(self, tag):
recorded["tag"] = tag
return "POOL_CTX"
monkeypatch.setattr(cumem_mod, "CuMemAllocator", FakeAllocator)
ctx = OmniGPUWorkerBase._maybe_get_memory_pool_context(worker, "weights")
assert ctx == "POOL_CTX"
assert recorded["tag"] == "weights"
# --------------------------------------------------------------------------- #
# sleep / wake_up #
# --------------------------------------------------------------------------- #
def _fake_platform():
return SimpleNamespace(
get_current_memory_usage=lambda device: 0,
empty_cache=lambda: None,
synchronize=lambda: None,
)
@pytest.mark.parametrize(
("level", "expected_tags"),
[(1, ("weights",)), (2, tuple())],
)
def test_sleep_offload_tags_by_level(monkeypatch, level, expected_tags):
import vllm.device_allocator.cumem as cumem_mod
worker = object.__new__(OmniGPUWorkerBase)
worker.rank = 0
worker.device = "cuda:0"
monkeypatch.setattr(base, "current_omni_platform", _fake_platform())
calls = {}
class FakeAllocator:
@classmethod
def get_instance(cls):
return cls()
def sleep(self, offload_tags):
calls["offload_tags"] = offload_tags
monkeypatch.setattr(cumem_mod, "CuMemAllocator", FakeAllocator)
assert OmniGPUWorkerBase.sleep(worker, level=level) is True
assert calls["offload_tags"] == expected_tags
def test_wake_up_forwards_tags_to_allocator(monkeypatch):
import vllm.device_allocator.cumem as cumem_mod
worker = object.__new__(OmniGPUWorkerBase)
worker.rank = 0
monkeypatch.setattr(base, "current_omni_platform", SimpleNamespace(synchronize=lambda: None))
calls = {}
class FakeAllocator:
@classmethod
def get_instance(cls):
return cls()
def wake_up(self, tags):
calls["tags"] = tags
monkeypatch.setattr(cumem_mod, "CuMemAllocator", FakeAllocator)
assert OmniGPUWorkerBase.wake_up(worker, tags=["weights"]) is True
assert calls["tags"] == ["weights"]
+276
View File
@@ -0,0 +1,276 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM project
"""Record / verify a deterministic text+audio output hash baseline for an Omni model.
This is the cheap "output is bit-identical" gate for the behaviour-preserving
worker/model-runner changes: record a baseline on the bench box
BEFORE any code moves, then re-run with ``--check`` after each refactor.
Determinism: **batch size 1** (one prompt per ``generate`` call) + **fixed seed**.
Thinker and code2wav run greedy (temperature 0); the talker's speech sampler needs
temperature to emit valid codec tokens, so it runs at its natural sampling with a
fixed seed (reproducible via seed + batch-1, not pure-greedy). 3 fixed prompts, run
with async output OFF and ON (6 entries total). Hashes are only meaningful WITHIN a single recorded
environment (driver/cuda/torch/gpu), so the baseline JSON records the env and
``--check`` refuses to compare across a changed env unless ``--force-env``.
Baseline JSON schema (results keyed by BOTH prompt_id and async_mode so the
matrix is self-identifying)::
{
"model": "...",
"commit": "<git sha>",
"env": {"driver": "<nvidia driver>", "cuda": "<cuda build>", "torch": "...", "gpu": "..."},
"results": [
{"prompt_id": 0, "async_mode": false, "text_sha": "...", "wav_sha": "..."},
... # exactly len(FIXED_PROMPTS) * 2 entries
]
}
The file content is JSON, but committed baselines use a NON-``.json`` extension
(``.baseline``) because the repo ``.gitignore`` excludes ``*.json``; ``--check``
loads by content, so the extension does not matter to the script.
Usage::
# record (run twice, confirm the two runs are identical before committing)
python tools/baselines/omni_hash_baseline.py <model> --yaml <deploy.yaml> \
--out tools/baselines/<model>_hash.baseline
# verify after a refactor (same bench box)
python tools/baselines/omni_hash_baseline.py <model> --yaml <deploy.yaml> \
--check tools/baselines/<model>_hash.baseline
"""
from __future__ import annotations
import argparse
import hashlib
import json
import subprocess
import sys
import torch
from vllm.sampling_params import SamplingParams
from vllm_omni.entrypoints.omni import Omni
SEED = 42
DEFAULT_SYSTEM = (
"You are Qwen, a virtual human developed by the Qwen Team, Alibaba Group, "
"capable of perceiving auditory and visual inputs, as well as generating "
"text and speech."
)
# 3 fixed prompts — do NOT edit once a baseline is recorded (prompt_id is positional).
FIXED_PROMPTS = [
"Explain the system architecture for a scalable audio generation pipeline. Answer in 15 words.",
"Describe how ocean tides work. Answer in 15 words.",
"List three benefits of unit tests. Answer in 15 words.",
]
ASYNC_MODES = (False, True)
def _text_prompt(question: str) -> dict:
return {
"prompt": (
f"<|im_start|>system\n{DEFAULT_SYSTEM}<|im_end|>\n"
f"<|im_start|>user\n{question}<|im_end|>\n"
f"<|im_start|>assistant\n"
)
}
def _sha(data: bytes) -> str:
return hashlib.sha256(data).hexdigest()
def _git_commit() -> str:
try:
return subprocess.check_output(["git", "rev-parse", "HEAD"], text=True).strip()
except Exception:
return "unknown"
def _driver_version() -> str:
"""The actual NVIDIA driver version (NOT torch.version.cuda, which is the build)."""
try:
out = subprocess.check_output(["nvidia-smi", "--query-gpu=driver_version", "--format=csv,noheader"], text=True)
return out.strip().splitlines()[0].strip()
except Exception:
return "unknown"
def _env_fingerprint() -> dict:
gpu = torch.cuda.get_device_name(0) if torch.cuda.is_available() else "cpu"
return {
"driver": _driver_version(),
"cuda": str(torch.version.cuda),
"torch": torch.__version__,
"gpu": gpu,
}
def _sampling_params() -> list[SamplingParams]:
# thinker + code2wav: greedy. talker: seeded (temperature>0) — its speech sampler
# needs temperature to emit valid codec tokens; the fixed seed keeps it reproducible.
thinker = SamplingParams(
temperature=0.0, top_p=1.0, top_k=-1, max_tokens=128, seed=SEED, detokenize=True, repetition_penalty=1.1
)
talker = SamplingParams(
temperature=0.9, top_p=0.8, top_k=40, max_tokens=256, seed=SEED, detokenize=True, repetition_penalty=1.05
)
code2wav = SamplingParams(
temperature=0.0, top_p=1.0, top_k=-1, max_tokens=256, seed=SEED, detokenize=True, repetition_penalty=1.1
)
return [thinker, talker, code2wav]
def _audio_bytes(audio) -> bytes:
"""Deterministic bytes for one request's audio, handling chunked (list) output.
async-chunk mode can emit a list of audio chunks instead of a single tensor;
concatenate them along the last axis in emit order.
"""
if isinstance(audio, (list, tuple)):
chunks = [a for a in audio if a is not None]
if not chunks:
return b""
audio = torch.cat([torch.as_tensor(a).detach().to(torch.float32).cpu() for a in chunks], dim=-1)
if audio is None:
return b""
return torch.as_tensor(audio).detach().to(torch.float32).cpu().numpy().tobytes()
def _run_once(model: str, yaml_path: str, *, async_mode: bool) -> list[dict]:
"""One deterministic pass, batch size 1 per prompt; returns per-prompt hashes."""
# ``async_chunk`` is the repo's documented async-output toggle for streaming
# (see tests/e2e/accuracy/qwen3_omni). Confirm the exact knob at bench time.
omni = Omni(
model,
stage_configs_path=yaml_path,
stage_init_timeout=1800,
init_timeout=3600,
async_chunk=async_mode,
)
results: list[dict] = []
try:
for prompt_id, question in enumerate(FIXED_PROMPTS):
text: str | None = None
wav: bytes = b""
# Batch size 1: submit a single prompt so scheduling is deterministic.
for step in omni.generate([_text_prompt(question)], _sampling_params(), py_generator=False):
out = step.request_output
if step.final_output_type == "text":
text = out.outputs[0].text # exact text, no strip -> truly bit-identical
elif step.final_output_type == "audio":
wav = _audio_bytes(out.outputs[0].multimodal_output["audio"])
if text is None:
raise RuntimeError(f"prompt_id={prompt_id} async={async_mode}: produced no text output")
if not wav:
raise RuntimeError(f"prompt_id={prompt_id} async={async_mode}: produced no/empty audio output")
results.append(
{
"prompt_id": prompt_id,
"async_mode": async_mode,
"text_sha": _sha(text.encode("utf-8")),
"wav_sha": _sha(wav),
}
)
finally:
omni.close()
return results
def _record(model: str, yaml_path: str) -> dict:
results: list[dict] = []
for async_mode in ASYNC_MODES:
results.extend(_run_once(model, yaml_path, async_mode=async_mode))
return {"model": model, "commit": _git_commit(), "env": _env_fingerprint(), "results": results}
def _key(record: dict):
return (record["prompt_id"], record["async_mode"])
def _expected_keys() -> set:
return {(pid, mode) for pid in range(len(FIXED_PROMPTS)) for mode in ASYNC_MODES}
def _check(model: str, yaml_path: str, baseline_path: str, *, force_env: bool) -> int:
with open(baseline_path) as fh:
baseline = json.load(fh)
now = _record(model, yaml_path)
if not force_env and baseline.get("env") != now["env"]:
print(
f"ENV MISMATCH: baseline={baseline.get('env')} now={now['env']} — hashes are "
"env-specific. Re-record on this box or pass --force-env.",
file=sys.stderr,
)
return 2
problems: list[str] = []
def _index_unique(records: list[dict], label: str) -> dict:
by_key: dict = {}
for record in records:
key = _key(record)
if key in by_key:
problems.append(f"{label}: duplicate entry for (prompt_id={key[0]}, async={key[1]})")
by_key[key] = record
return by_key
base_by_key = _index_unique(baseline["results"], "baseline")
now_by_key = _index_unique(now["results"], "current-run")
expected = _expected_keys()
# Guard against duplicate-collapse: the record count must match the key count.
if len(baseline["results"]) != len(expected):
problems.append(f"baseline has {len(baseline['results'])} records, expected exactly {len(expected)}")
if set(now_by_key) != expected:
problems.append(f"current-run key set {sorted(set(now_by_key))} != expected {sorted(expected)}")
if set(base_by_key) != set(now_by_key):
missing = sorted(set(base_by_key) - set(now_by_key))
extra = sorted(set(now_by_key) - set(base_by_key))
problems.append(f"baseline/run key-set mismatch: missing-from-run={missing} extra-in-run={extra}")
for key in sorted(set(base_by_key) & set(now_by_key)):
b, r = base_by_key[key], now_by_key[key]
if (b["text_sha"], b["wav_sha"]) != (r["text_sha"], r["wav_sha"]):
problems.append(
f"HASH MISMATCH (prompt_id={key[0]}, async={key[1]}): "
f"text {b['text_sha'][:8]}!={r['text_sha'][:8]} / wav {b['wav_sha'][:8]}!={r['wav_sha'][:8]}"
)
if problems:
for problem in problems:
print(problem, file=sys.stderr)
return 1
print(f"OK: all {len(expected)} (prompt_id, async_mode) entries match the baseline.")
return 0
def main() -> int:
p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
p.add_argument("model")
p.add_argument("--yaml", required=True, help="Stage deploy config path")
p.add_argument("--out", help="Write a new baseline JSON to this path")
p.add_argument("--check", metavar="BASELINE_JSON", help="Verify current output against this baseline")
p.add_argument("--force-env", action="store_true", help="Compare hashes even if the recorded env differs")
args = p.parse_args()
if args.check:
return _check(args.model, args.yaml, args.check, force_env=args.force_env)
baseline = _record(args.model, args.yaml)
payload = json.dumps(baseline, indent=2)
if args.out:
with open(args.out, "w") as fh:
fh.write(payload)
print(f"Wrote baseline: {args.out}")
else:
print(payload)
return 0
if __name__ == "__main__":
raise SystemExit(main())
+48
View File
@@ -0,0 +1,48 @@
{
"model": "Qwen/Qwen3-Omni-30B-A3B-Instruct",
"commit": "24eb4ab89cc2ad3f8cdd83f57b66ce7786a0d65c",
"env": {
"driver": "570.133.20",
"cuda": "13.0",
"torch": "2.11.0+cu130",
"gpu": "NVIDIA L20X"
},
"results": [
{
"prompt_id": 0,
"async_mode": false,
"text_sha": "e4bf6ed906c0dada879874a9cc09f9ba3a4399149262a831a01fd53e6a8fd043",
"wav_sha": "20a203a149005f948afd485bbea780fa06c109f3689ebb6b3110f51b3237bd35"
},
{
"prompt_id": 1,
"async_mode": false,
"text_sha": "2e8a6b141a9747bd68aadc2931f278b89712ad7ad7ae12b25b39e193b241bf68",
"wav_sha": "1839a759bb5d8671b6352b2e3110dab5edfa718e38ff073d379176201f1eac8c"
},
{
"prompt_id": 2,
"async_mode": false,
"text_sha": "e0aa5a513a5162024f35afee3830aeb67f2e378381726adb9d3a66b984aa652d",
"wav_sha": "74338df3e8357bc9d0c05bf141b63c45251b12807de34495097edb07fc7b19f8"
},
{
"prompt_id": 0,
"async_mode": true,
"text_sha": "e4bf6ed906c0dada879874a9cc09f9ba3a4399149262a831a01fd53e6a8fd043",
"wav_sha": "e09f862f54b553cc7c64e1631b821a768167768812b3f2cd3d0c69d0da991ee7"
},
{
"prompt_id": 1,
"async_mode": true,
"text_sha": "2e8a6b141a9747bd68aadc2931f278b89712ad7ad7ae12b25b39e193b241bf68",
"wav_sha": "9115d481f62a4b771c4c3d8b89d5c5a01149f002096289ab1da3f034fa388a2b"
},
{
"prompt_id": 2,
"async_mode": true,
"text_sha": "e0aa5a513a5162024f35afee3830aeb67f2e378381726adb9d3a66b984aa652d",
"wav_sha": "6a5985c9d4ca1596518c291228cd29239e76fcaf310c1cdb5921276b7efce801"
}
]
}