[test]: CPU regression tests for dynamic batching scheduler and signature fixes (#1453)
Covers the review's adversarial probes: expired-deadline greedy drain coalesces jobs queued behind an in-flight generation (delay>0 no longer degrades to sequential), submit-after-stop raises, stop() fails still-queued futures without hanging, a dispatch exception fans out to every future in the group, a crashed _run task fails in-flight and queued futures and rejects later submits, the pipeline capability gate rejects non-opted-in pipelines (DMD/causal always vetoed), pre-set latents are never merged, and distinct opaque objects are conservatively incompatible. Verified the deadline and gate tests fail on the pre-fix scheduler.
This commit is contained in:
SolitaryThinker
@@ -0,0 +1,226 @@
|
||||
# SPDX-License-Identifier: Apache-2.0
|
||||
"""CPU regression tests for the dynamic VideoBatchScheduler (no GPU needed)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import threading
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
|
||||
from fastvideo.configs.pipelines.base import PipelineConfig
|
||||
from fastvideo.configs.pipelines.wan import (
|
||||
FastWan2_1_T2V_480P_Config,
|
||||
SelfForcingWanT2V480PConfig,
|
||||
WanT2V480PConfig,
|
||||
)
|
||||
from fastvideo.entrypoints.openai.batching import VideoBatchScheduler
|
||||
|
||||
|
||||
class _FakeBatchGenerator:
|
||||
|
||||
def __init__(self):
|
||||
self.calls: list[list[dict]] = []
|
||||
|
||||
def generate_video_batch(self, request_kwargs):
|
||||
self.calls.append([dict(item) for item in request_kwargs])
|
||||
return [{"prompts": item["prompt"], "video_path": item["output_path"]} for item in request_kwargs]
|
||||
|
||||
|
||||
class _BlockingFirstCallGenerator(_FakeBatchGenerator):
|
||||
"""Blocks the first generate_video_batch call until released.
|
||||
|
||||
Lets a test deterministically hold one generation "in flight" while more
|
||||
requests queue up behind it.
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
self.started = threading.Event()
|
||||
self.release = threading.Event()
|
||||
self._first_call = True
|
||||
|
||||
def generate_video_batch(self, request_kwargs):
|
||||
if self._first_call:
|
||||
self._first_call = False
|
||||
self.started.set()
|
||||
assert self.release.wait(timeout=10)
|
||||
return super().generate_video_batch(request_kwargs)
|
||||
|
||||
|
||||
def _batch_scheduler_args(**overrides):
|
||||
defaults = dict(
|
||||
model_path="test-model",
|
||||
batching_mode="dynamic",
|
||||
batching_max_size=2,
|
||||
batching_delay_ms=25.0,
|
||||
enable_batching_metrics=False,
|
||||
pipeline_config=PipelineConfig(supports_dynamic_batching=True),
|
||||
)
|
||||
defaults.update(overrides)
|
||||
return SimpleNamespace(**defaults)
|
||||
|
||||
|
||||
def _job_kwargs(tmp_path, index: int) -> dict:
|
||||
return {
|
||||
"prompt": f"p{index}",
|
||||
"height": 256,
|
||||
"width": 256,
|
||||
"num_frames": 1,
|
||||
"num_inference_steps": 2,
|
||||
"seed": index,
|
||||
"output_path": str(tmp_path / f"video_{index}.mp4"),
|
||||
"save_video": False,
|
||||
}
|
||||
|
||||
|
||||
def test_expired_deadline_coalesces_jobs_queued_behind_inflight_generation(tmp_path):
|
||||
"""Jobs whose delay deadline expired while a generation was in flight must
|
||||
still coalesce (greedy drain), not dispatch as batches of 1."""
|
||||
|
||||
async def run():
|
||||
generator = _BlockingFirstCallGenerator()
|
||||
scheduler = VideoBatchScheduler(generator, _batch_scheduler_args(batching_max_size=4, batching_delay_ms=25.0))
|
||||
await scheduler.start()
|
||||
try:
|
||||
loop = asyncio.get_running_loop()
|
||||
first = asyncio.ensure_future(scheduler.submit("req-1", _job_kwargs(tmp_path, 1)))
|
||||
await asyncio.wait_for(loop.run_in_executor(None, generator.started.wait, 5), timeout=10)
|
||||
rest = [asyncio.ensure_future(scheduler.submit(f"req-{i}", _job_kwargs(tmp_path, i))) for i in (2, 3, 4)]
|
||||
# Let the in-queue jobs outlive the 25 ms batching window while the
|
||||
# first generation is still blocking the scheduler.
|
||||
await asyncio.sleep(0.1)
|
||||
generator.release.set()
|
||||
await asyncio.wait_for(asyncio.gather(first, *rest), timeout=10)
|
||||
finally:
|
||||
await scheduler.stop()
|
||||
return [[item["prompt"] for item in call] for call in generator.calls]
|
||||
|
||||
assert asyncio.run(run()) == [["p1"], ["p2", "p3", "p4"]]
|
||||
|
||||
|
||||
def test_submit_after_stop_raises(tmp_path):
|
||||
|
||||
async def run():
|
||||
scheduler = VideoBatchScheduler(_FakeBatchGenerator(), _batch_scheduler_args())
|
||||
await scheduler.start()
|
||||
await scheduler.stop()
|
||||
with pytest.raises(RuntimeError, match="stopped"):
|
||||
await scheduler.submit("req-1", _job_kwargs(tmp_path, 1))
|
||||
|
||||
asyncio.run(run())
|
||||
|
||||
|
||||
def test_stop_fails_jobs_still_queued_without_hanging(tmp_path):
|
||||
"""stop() during an in-flight generation must fail the still-queued jobs
|
||||
instead of stranding their futures forever."""
|
||||
|
||||
async def run():
|
||||
generator = _BlockingFirstCallGenerator()
|
||||
scheduler = VideoBatchScheduler(generator, _batch_scheduler_args())
|
||||
await scheduler.start()
|
||||
loop = asyncio.get_running_loop()
|
||||
first = asyncio.ensure_future(scheduler.submit("req-1", _job_kwargs(tmp_path, 1)))
|
||||
await asyncio.wait_for(loop.run_in_executor(None, generator.started.wait, 5), timeout=10)
|
||||
queued = [asyncio.ensure_future(scheduler.submit(f"req-{i}", _job_kwargs(tmp_path, i))) for i in (2, 3)]
|
||||
await asyncio.sleep(0) # let the queued submits enqueue
|
||||
stop_task = asyncio.ensure_future(scheduler.stop())
|
||||
await asyncio.sleep(0) # let stop() mark the scheduler stopped
|
||||
generator.release.set()
|
||||
|
||||
result = await asyncio.wait_for(first, timeout=10)
|
||||
assert result["prompts"] == "p1"
|
||||
for task in queued:
|
||||
with pytest.raises(RuntimeError, match="stopped before dispatch"):
|
||||
await asyncio.wait_for(task, timeout=10)
|
||||
await asyncio.wait_for(stop_task, timeout=10)
|
||||
|
||||
asyncio.run(run())
|
||||
|
||||
|
||||
def test_dispatch_exception_fails_all_group_futures(tmp_path):
|
||||
|
||||
class _ExplodingGenerator:
|
||||
|
||||
def generate_video_batch(self, request_kwargs):
|
||||
raise ValueError("generation exploded")
|
||||
|
||||
async def run():
|
||||
scheduler = VideoBatchScheduler(_ExplodingGenerator(), _batch_scheduler_args())
|
||||
await scheduler.start()
|
||||
try:
|
||||
return await asyncio.wait_for(
|
||||
asyncio.gather(
|
||||
scheduler.submit("req-1", _job_kwargs(tmp_path, 1)),
|
||||
scheduler.submit("req-2", _job_kwargs(tmp_path, 2)),
|
||||
return_exceptions=True,
|
||||
),
|
||||
timeout=10,
|
||||
)
|
||||
finally:
|
||||
await scheduler.stop()
|
||||
|
||||
results = asyncio.run(run())
|
||||
|
||||
assert len(results) == 2
|
||||
for result in results:
|
||||
assert isinstance(result, ValueError)
|
||||
assert "generation exploded" in str(result)
|
||||
|
||||
|
||||
def test_run_crash_fails_inflight_and_queued_futures(tmp_path):
|
||||
"""If the _run task ever dies, every waiting future must fail instead of
|
||||
hanging, and later submits must be rejected."""
|
||||
|
||||
async def run():
|
||||
scheduler = VideoBatchScheduler(_FakeBatchGenerator(), _batch_scheduler_args())
|
||||
|
||||
async def boom(first):
|
||||
raise RuntimeError("kaboom")
|
||||
|
||||
scheduler._collect_batch = boom # type: ignore[method-assign]
|
||||
await scheduler.start()
|
||||
results = await asyncio.wait_for(
|
||||
asyncio.gather(
|
||||
scheduler.submit("req-1", _job_kwargs(tmp_path, 1)),
|
||||
scheduler.submit("req-2", _job_kwargs(tmp_path, 2)),
|
||||
return_exceptions=True,
|
||||
),
|
||||
timeout=10,
|
||||
)
|
||||
with pytest.raises(RuntimeError, match="stopped"):
|
||||
await scheduler.submit("req-3", _job_kwargs(tmp_path, 3))
|
||||
return results
|
||||
|
||||
results = asyncio.run(run())
|
||||
|
||||
assert len(results) == 2
|
||||
for result in results:
|
||||
assert isinstance(result, RuntimeError)
|
||||
assert "crashed" in str(result)
|
||||
|
||||
|
||||
def test_scheduler_start_rejects_pipeline_without_capability(tmp_path):
|
||||
|
||||
async def run():
|
||||
scheduler = VideoBatchScheduler(
|
||||
_FakeBatchGenerator(),
|
||||
_batch_scheduler_args(pipeline_config=PipelineConfig()),
|
||||
)
|
||||
with pytest.raises(RuntimeError, match="does not support dynamic request batching"):
|
||||
await scheduler.start()
|
||||
assert scheduler._task is None
|
||||
|
||||
asyncio.run(run())
|
||||
|
||||
|
||||
def test_dynamic_batching_capability_gate_defaults():
|
||||
# Default: every pipeline is excluded until its family opts in.
|
||||
assert PipelineConfig().dynamic_batching_supported() is False
|
||||
# The verified generic Wan denoising path opts in.
|
||||
assert WanT2V480PConfig().dynamic_batching_supported() is True
|
||||
# DMD and causal denoising variants are always vetoed (shared-RNG risk),
|
||||
# even though they inherit the Wan opt-in flag.
|
||||
assert FastWan2_1_T2V_480P_Config().dynamic_batching_supported() is False
|
||||
assert SelfForcingWanT2V480PConfig().dynamic_batching_supported() is False
|
||||
@@ -53,5 +53,31 @@ def test_can_dynamic_batch_rejects_image_conditioning() -> None:
|
||||
assert result.reason == "image_path"
|
||||
|
||||
|
||||
def test_can_dynamic_batch_rejects_preset_latents() -> None:
|
||||
first = _request()
|
||||
second = _request(latents=object())
|
||||
|
||||
result = can_dynamic_batch(first, second)
|
||||
|
||||
assert result.can_batch is False
|
||||
assert result.reason == "latents"
|
||||
|
||||
|
||||
def test_signature_treats_distinct_opaque_objects_as_incompatible() -> None:
|
||||
|
||||
class Opaque:
|
||||
pass
|
||||
|
||||
first = _request("first")
|
||||
second = _request("second")
|
||||
|
||||
distinct = can_dynamic_batch(first, second, base_extra={"foo": Opaque()}, candidate_extra={"foo": Opaque()})
|
||||
assert distinct.can_batch is False
|
||||
|
||||
shared = Opaque()
|
||||
same_object = can_dynamic_batch(first, second, base_extra={"foo": shared}, candidate_extra={"foo": shared})
|
||||
assert same_object.can_batch is True
|
||||
|
||||
|
||||
def test_resolution_key_uses_generation_shape() -> None:
|
||||
assert resolution_key(_request(height=720, width=1280, num_frames=81)) == "720x1280x81"
|
||||
|
||||
Reference in New Issue
Block a user