anodyine/vllm-omni
1
0 Fork 0
Code Issues 4 Pull Requests 2 Actions Packages Projects Releases Wiki Activity

[CI][Bugfix] Improve e2e latency logging, update response classes to include detailed latency documentation and add startup time logging (#3246)

Browse Source 
Signed-off-by: wangyu <410167048@qq.com>
Signed-off-by: [Your Name] <your.email@example.com>
This commit is contained in:
wangyu authored and GitHub committed 2026-05-12 10:23:56 +08:00
1 parent 955fcff828
commit ac5fbed6c0
8 files changed
+179 -75
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

tests/e2e/online_serving/test_cosyvoice3_tts_expansion.py
+2 -2
View File
@@ -47,7 +47,7 @@ tts_server_params = [
OmniServerParams(
model=MODEL,
stage_config_path=get_stage_config(),
server_args=["--trust-remote-code", "--disable-log-stats", "--no-async-chunk"],
server_args=["--trust-remote-code", "--no-async-chunk"],
),
id="cosyvoice3",
)
@@ -58,7 +58,7 @@ tts_async_chunk_server_params = [
OmniServerParams(
model=MODEL,
stage_config_path=get_stage_config(),
server_args=["--trust-remote-code", "--disable-log-stats"],
server_args=["--trust-remote-code"],
),
id="cosyvoice3_async_chunk",
)
tests/e2e/online_serving/test_qwen3_tts_base.py
+1 -1
View File
@@ -47,7 +47,7 @@ tts_server_params = [
OmniServerParams(
model=MODEL,
stage_config_path=get_deploy_config_path("qwen3_tts.yaml"),
server_args=["--trust-remote-code", "--disable-log-stats"],
server_args=["--trust-remote-code"],
),
id="async_chunk",
)
tests/e2e/online_serving/test_qwen3_tts_base_expansion.py
+2 -2
View File
@@ -46,7 +46,7 @@ tts_server_params = [
OmniServerParams(
model=MODEL,
stage_config_path=get_deploy_config_path("qwen3_tts.yaml"),
server_args=["--trust-remote-code", "--disable-log-stats"],
server_args=["--trust-remote-code"],
),
id="async_chunk",
),
@@ -57,7 +57,7 @@ tts_server_params = [
OmniServerParams(
model=MODEL,
stage_config_path=get_deploy_config_path("qwen3_tts.yaml"),
server_args=["--trust-remote-code", "--disable-log-stats", "--no-async-chunk"],
server_args=["--trust-remote-code", "--no-async-chunk"],
),
id="no_async_chunk",
),
tests/e2e/online_serving/test_qwen3_tts_customvoice.py
+1 -1
View File
@@ -39,7 +39,7 @@ tts_server_params = [
OmniServerParams(
model=MODEL,
stage_config_path=get_deploy_config_path("qwen3_tts.yaml"),
server_args=["--trust-remote-code", "--disable-log-stats"],
server_args=["--trust-remote-code"],
),
id="async_chunk",
)
tests/e2e/online_serving/test_qwen3_tts_customvoice_expansion.py
+2 -2
View File
@@ -42,7 +42,7 @@ tts_server_params = [
OmniServerParams(
model=MODEL,
stage_config_path=get_deploy_config_path("qwen3_tts.yaml"),
server_args=["--trust-remote-code", "--disable-log-stats"],
server_args=["--trust-remote-code"],
),
id="async_chunk",
),
@@ -53,7 +53,7 @@ tts_server_params = [
OmniServerParams(
model=MODEL,
stage_config_path=get_deploy_config_path("qwen3_tts.yaml"),
server_args=["--trust-remote-code", "--disable-log-stats", "--no-async-chunk"],
server_args=["--trust-remote-code", "--no-async-chunk"],
),
id="no_async_chunk",
),
tests/helpers/assertions.py
-9
View File
@@ -437,9 +437,6 @@ def assert_omni_response(response: Any, request_config: dict[str, Any], run_leve
AssertionError: When the response does not meet validation criteria
"""
assert response.success, "The request failed."
e2e_latency = response.e2e_latency
if e2e_latency is not None:
print(f"the e2e latency is: {e2e_latency}")
modalities = request_config.get("modalities", ["text", "audio"])
@@ -555,9 +552,6 @@ def assert_audio_speech_response(response: Any, request_config: dict[str, Any],
return
assert response.success, "The request failed."
e2e_latency = getattr(response, "e2e_latency", None)
if e2e_latency is not None:
print(f"the avg e2e latency is: {e2e_latency}")
# Optional floor on decoded audio size (models with very short clips may use a lower value).
min_audio = request_config.get("min_audio_bytes")
@@ -595,9 +589,6 @@ def assert_audio_speech_response(response: Any, request_config: dict[str, Any],
def assert_diffusion_response(response: "DiffusionResponse", request_config: dict[str, Any], run_level: str = None):
assert response.success, "The request failed."
e2e_latency = getattr(response, "e2e_latency", None)
if e2e_latency is not None:
print(f"the avg e2e is: {e2e_latency}")
has_any_content = any(content is not None for content in (response.images, response.videos, response.audios))
assert has_any_content, "Response contains no images, videos, or audios"
if response.images is not None:
tests/helpers/fixtures/runtime.py
+14 -2
View File
@@ -51,7 +51,13 @@ def openai_client(request: pytest.FixtureRequest, run_level: str):
from tests.helpers.runtime import OpenAIClientHandler
server = request.getfixturevalue("omni_server")
return OpenAIClientHandler(host=server.host, port=server.port, api_key="EMPTY", run_level=run_level)
return OpenAIClientHandler(
host=server.host,
port=server.port,
api_key="EMPTY",
run_level=run_level,
log_stats=server.log_stats,
)
@pytest.fixture
@@ -60,7 +66,13 @@ def openai_client_function(request: pytest.FixtureRequest, run_level: str):
from tests.helpers.runtime import OpenAIClientHandler
server = request.getfixturevalue("omni_server_function")
return OpenAIClientHandler(host=server.host, port=server.port, api_key="EMPTY", run_level=run_level)
return OpenAIClientHandler(
host=server.host,
port=server.port,
api_key="EMPTY",
run_level=run_level,
log_stats=server.log_stats,
)
@pytest.fixture(scope="function")
tests/helpers/runtime.py
+157 -56
View File
@@ -205,7 +205,9 @@ class OmniServer:
run_post_test_cleanup()
cleanup_dist_env_and_memory()
self.model = model
self.serve_args = serve_args
args = list(serve_args)
self.serve_args = args
self.log_stats = "--disable-log-stats" not in args and "--log-stats" in args
self.env_dict = env_dict
self.use_omni = use_omni
self.proc: subprocess.Popen | None = None
@@ -234,6 +236,7 @@ class OmniServer:
cmd += self.serve_args
print(f"Launching OmniServer with: {' '.join(cmd)}")
startup_t0 = time.perf_counter()
self.proc = subprocess.Popen(
cmd,
env=env,
@@ -249,7 +252,12 @@ class OmniServer:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
sock.settimeout(1)
if sock.connect_ex((self.host, self.port)) == 0:
print(f"Server ready on {self.host}:{self.port}")
startup_s = time.perf_counter() - startup_t0
if self.log_stats:
print(
f"Server ready on {self.host}:{self.port} (OmniServer startup took {startup_s:.3f}s)",
flush=True,
)
return
time.sleep(2)
raise RuntimeError(f"Server failed to start within {max_wait} seconds")
@@ -527,6 +535,7 @@ class OmniServerStageCli(OmniServer):
)
def _start_server(self) -> None:
startup_t0 = time.perf_counter()
ordered_stage_ids = [0, *[stage_id for stage_id in self.stage_ids if stage_id != 0]]
self._launch_stage(0, headless=False, replica_id=0)
@@ -545,7 +554,13 @@ class OmniServerStageCli(OmniServer):
sock.settimeout(1)
result = sock.connect_ex((self.host, self.port))
if result == 0:
print(f"OmniServerStageCli ready on {self.host}:{self.port}")
startup_s = time.perf_counter() - startup_t0
if self.log_stats:
print(
f"OmniServerStageCli ready on {self.host}:{self.port} "
f"(stage-CLI startup took {startup_s:.3f}s)",
flush=True,
)
return
time.sleep(2)
@@ -607,6 +622,8 @@ class OmniResponse:
audio_content: str | None = None
audio_format: str | None = None
audio_bytes: bytes | None = None
#: End-to-end wall time in **seconds** (``perf_counter`` delta), from just before the
#: OpenAI client call through response parsing and local post-process (e.g. audio decode).
e2e_latency: float | None = None
success: bool = False
error_message: str | None = None
@@ -622,7 +639,9 @@ class DiffusionResponse:
text_content: str | None = None
images: list[Image.Image] | None = None
audios: list[Any] | None = None
videos: list[bytes] | None = None
videos: list[Any] | None = None
#: End-to-end wall time in **seconds** (``perf_counter`` delta), from just before
#: ``chat.completions.create`` through local image / audio decode.
e2e_latency: float | None = None
success: bool = False
error_message: str | None = None
@@ -643,16 +662,29 @@ def _merge_diffusion_responses(parts: list[DiffusionResponse]) -> DiffusionRespo
class OpenAIClientHandler:
def __init__(self, host: str = "127.0.0.1", port: int = None, api_key: str = "EMPTY", run_level: str = None):
def __init__(
self,
host: str = "127.0.0.1",
port: int = None,
api_key: str = "EMPTY",
run_level: str = None,
*,
log_stats: bool = True,
):
if port is None:
port = get_open_port()
self.base_url = f"http://{host}:{port}"
self.client = OpenAI(base_url=f"http://{host}:{port}/v1", api_key=api_key)
self.run_level = run_level
self.log_stats = log_stats
def _process_stream_omni_response(self, chat_completion) -> OmniResponse:
def _print_client_stat(self, message: str) -> None:
if self.log_stats:
print(message, flush=True)
def _process_stream_omni_response(self, chat_completion, *, wall_start: float) -> OmniResponse:
"""Wall clock from *before* ``chat.completions.create`` through stream drain + local decode."""
result = OmniResponse()
start_time = time.perf_counter()
try:
text_content = ""
audio_data = []
@@ -664,7 +696,6 @@ class OpenAIClientHandler:
audio_data.append(content)
elif modality == "text" and content:
text_content += content
result.e2e_latency = time.perf_counter() - start_time
audio_content = None
if audio_data:
merged_seg = _merge_base64_audio_to_segment(audio_data)
@@ -675,15 +706,16 @@ class OpenAIClientHandler:
result.text_content = text_content
result.audio_data = audio_data
result.audio_content = audio_content
result.e2e_latency = time.perf_counter() - wall_start
result.success = True
except Exception as e:
result.error_message = f"Stream processing error: {str(e)}"
print(f"Error: {result.error_message}")
return result
def _process_non_stream_omni_response(self, chat_completion) -> OmniResponse:
def _process_non_stream_omni_response(self, chat_completion, *, wall_start: float) -> OmniResponse:
"""Wall clock from *before* ``chat.completions.create`` through response parse + local decode."""
result = OmniResponse()
start_time = time.perf_counter()
try:
audio_data = None
text_content = None
@@ -698,13 +730,13 @@ class OpenAIClientHandler:
result.prompt_tokens = usage.prompt_tokens
if details := getattr(usage, "prompt_tokens_details", None):
result.cached_tokens = details.cached_tokens
result.e2e_latency = time.perf_counter() - start_time
audio_content = None
if audio_data:
result.audio_bytes = base64.b64decode(audio_data)
audio_content = convert_audio_bytes_to_text(result.audio_bytes)
result.text_content = text_content
result.audio_content = audio_content
result.e2e_latency = time.perf_counter() - wall_start
if chat_completion.choices and chat_completion.choices[0].logprobs is not None:
result.logprobs = chat_completion.choices[0].logprobs.content
result.success = True
@@ -713,9 +745,9 @@ class OpenAIClientHandler:
print(f"Error: {result.error_message}")
return result
def _process_diffusion_response(self, chat_completion) -> DiffusionResponse:
def _process_diffusion_response(self, chat_completion, *, wall_start: float) -> DiffusionResponse:
"""Wall clock from *before* ``chat.completions.create`` through image decode."""
result = DiffusionResponse()
start_time = time.perf_counter()
try:
images = []
audios = []
@@ -744,9 +776,9 @@ class OpenAIClientHandler:
"expires_at": getattr(audio_obj, "expires_at", None),
}
)
result.e2e_latency = time.perf_counter() - start_time
result.images = images if images else None
result.audios = audios if audios else None
result.e2e_latency = time.perf_counter() - wall_start
result.success = True
except Exception as e:
result.error_message = f"Diffusion response processing error: {str(e)}"
@@ -781,33 +813,46 @@ class OpenAIClientHandler:
create_kwargs["extra_body"] = extra_body
if request_num == 1:
wall_start = time.perf_counter()
chat_completion = self.client.chat.completions.create(**create_kwargs)
resp = (
self._process_stream_omni_response(chat_completion)
self._process_stream_omni_response(chat_completion, wall_start=wall_start)
if stream
else self._process_non_stream_omni_response(chat_completion)
else self._process_non_stream_omni_response(chat_completion, wall_start=wall_start)
)
assert_omni_response(resp, request_config, run_level=self.run_level)
if resp.e2e_latency is not None:
self._print_client_stat(f"[omni] request#1 success in {resp.e2e_latency:.3f}s")
else:
self._print_client_stat("[omni] request#1 completed")
responses.append(resp)
return responses
def _one():
wall_start = time.perf_counter()
chat_completion = self.client.chat.completions.create(**create_kwargs)
return (
self._process_stream_omni_response(chat_completion)
self._process_stream_omni_response(chat_completion, wall_start=wall_start)
if stream
else self._process_non_stream_omni_response(chat_completion)
else self._process_non_stream_omni_response(chat_completion, wall_start=wall_start)
)
with concurrent.futures.ThreadPoolExecutor(max_workers=request_num) as executor:
futures = [executor.submit(_one) for _ in range(request_num)]
futures = {executor.submit(_one): i + 1 for i in range(request_num)}
for future in concurrent.futures.as_completed(futures):
request_idx = futures[future]
resp = future.result()
assert_omni_response(resp, request_config, run_level=self.run_level)
if resp.e2e_latency is not None:
self._print_client_stat(f"[omni] request#{request_idx} success in {resp.e2e_latency:.3f}s")
else:
self._print_client_stat(f"[omni] request#{request_idx} completed")
responses.append(resp)
return responses
def _process_stream_audio_speech_response(self, response, *, response_format: str | None = None) -> OmniResponse:
def _process_stream_audio_speech_response(
self, response, *, response_format: str | None = None, wall_start: float
) -> OmniResponse:
"""
Process streaming /v1/audio/speech responses into an OmniResponse.
@@ -816,7 +861,6 @@ class OpenAIClientHandler:
from Whisper transcription.
"""
result = OmniResponse()
start_time = time.perf_counter()
try:
# Aggregate all audio bytes from the streaming response.
@@ -858,7 +902,7 @@ class OpenAIClientHandler:
# Populate OmniResponse.
result.audio_bytes = raw_bytes
result.audio_content = transcript
result.e2e_latency = time.perf_counter() - start_time
result.e2e_latency = time.perf_counter() - wall_start
result.success = True
result.audio_format = getattr(response, "response", None)
if result.audio_format is not None:
@@ -871,7 +915,7 @@ class OpenAIClientHandler:
return result
def _process_non_stream_audio_speech_response(
self, response, *, response_format: str | None = None
self, response, *, response_format: str | None = None, wall_start: float
) -> OmniResponse:
"""
Process non-streaming /v1/audio/speech responses into an OmniResponse.
@@ -880,7 +924,6 @@ class OpenAIClientHandler:
audio payload returned by audio.speech.create.
"""
result = OmniResponse()
start_time = time.perf_counter()
try:
# OpenAI non-streaming audio.speech.create returns HttpxBinaryResponseContent (.read() or .content)
@@ -898,7 +941,7 @@ class OpenAIClientHandler:
result.audio_bytes = raw_bytes
result.audio_content = transcript
result.e2e_latency = time.perf_counter() - start_time
result.e2e_latency = time.perf_counter() - wall_start
result.success = True
result.audio_format = getattr(response, "response", None)
if result.audio_format is not None:
@@ -963,12 +1006,12 @@ class OpenAIClientHandler:
)
if request_num == 1:
req_start = time.perf_counter()
if expect_error_handling:
# ``status`` and/or ``err_message`` requested: catch APIError (4xx) and (optionally) assert body text;
# HTTP 200 with JSON error body is handled in ``assert_audio_speech_response``.
try:
if stream:
wall_start = time.perf_counter()
with self.client.audio.speech.with_streaming_response.create(
model=model,
input=text_input,
@@ -977,8 +1020,11 @@ class OpenAIClientHandler:
timeout=timeout,
voice=voice,
) as resp:
omni_resp = self._process_stream_audio_speech_response(resp, response_format=speech_fmt)
omni_resp = self._process_stream_audio_speech_response(
resp, response_format=speech_fmt, wall_start=wall_start
)
else:
wall_start = time.perf_counter()
resp = self.client.audio.speech.create(
model=model,
input=text_input,
@@ -987,7 +1033,9 @@ class OpenAIClientHandler:
timeout=timeout,
voice=voice,
)
omni_resp = self._process_non_stream_audio_speech_response(resp, response_format=speech_fmt)
omni_resp = self._process_non_stream_audio_speech_response(
resp, response_format=speech_fmt, wall_start=wall_start
)
except APIError as e:
sc = getattr(e, "status_code", None)
if sc is None:
@@ -1002,6 +1050,7 @@ class OpenAIClientHandler:
omni_resp.status_code = 200
elif stream:
# Use streaming response helper.
wall_start = time.perf_counter()
with self.client.audio.speech.with_streaming_response.create(
model=model,
input=text_input,
@@ -1010,9 +1059,12 @@ class OpenAIClientHandler:
timeout=timeout,
voice=voice,
) as resp:
omni_resp = self._process_stream_audio_speech_response(resp, response_format=speech_fmt)
omni_resp = self._process_stream_audio_speech_response(
resp, response_format=speech_fmt, wall_start=wall_start
)
else:
# Non-streaming response.
wall_start = time.perf_counter()
resp = self.client.audio.speech.create(
model=model,
input=text_input,
@@ -1021,11 +1073,15 @@ class OpenAIClientHandler:
timeout=timeout,
voice=voice,
)
omni_resp = self._process_non_stream_audio_speech_response(resp, response_format=speech_fmt)
omni_resp = self._process_non_stream_audio_speech_response(
resp, response_format=speech_fmt, wall_start=wall_start
)
assert_audio_speech_response(omni_resp, request_config, run_level=self.run_level)
elapsed = time.perf_counter() - req_start
print(f"[audio.speech] request#1 success in {elapsed:.3f}s")
if omni_resp.e2e_latency is not None:
self._print_client_stat(f"[audio.speech] request#1 success in {omni_resp.e2e_latency:.3f}s")
else:
self._print_client_stat("[audio.speech] request#1 completed")
responses.append(omni_resp)
return responses
else:
@@ -1034,7 +1090,7 @@ class OpenAIClientHandler:
if stream:
def _stream_task(request_idx: int):
task_start = time.perf_counter()
wall_start = time.perf_counter()
with self.client.audio.speech.with_streaming_response.create(
model=model,
input=text_input,
@@ -1043,9 +1099,15 @@ class OpenAIClientHandler:
timeout=timeout,
voice=voice,
) as resp:
result = self._process_stream_audio_speech_response(resp, response_format=speech_fmt)
elapsed = time.perf_counter() - task_start
print(f"[audio.speech] request#{request_idx} success in {elapsed:.3f}s")
result = self._process_stream_audio_speech_response(
resp, response_format=speech_fmt, wall_start=wall_start
)
if result.e2e_latency is not None:
self._print_client_stat(
f"[audio.speech] request#{request_idx} success in {result.e2e_latency:.3f}s"
)
else:
self._print_client_stat(f"[audio.speech] request#{request_idx} completed")
return result
with concurrent.futures.ThreadPoolExecutor(max_workers=request_num) as executor:
@@ -1065,8 +1127,8 @@ class OpenAIClientHandler:
else:
def _non_stream_task(request_idx: int):
task_start = time.perf_counter()
resp = self.client.audio.speech.create(
wall_start = time.perf_counter()
r = self.client.audio.speech.create(
model=model,
input=text_input,
response_format=response_format,
@@ -1074,24 +1136,29 @@ class OpenAIClientHandler:
timeout=timeout,
voice=voice,
)
elapsed = time.perf_counter() - task_start
print(f"[audio.speech] request#{request_idx} success in {elapsed:.3f}s")
return resp
result = self._process_non_stream_audio_speech_response(
r, response_format=speech_fmt, wall_start=wall_start
)
if result.e2e_latency is not None:
self._print_client_stat(
f"[audio.speech] request#{request_idx} success in {result.e2e_latency:.3f}s"
)
else:
self._print_client_stat(f"[audio.speech] request#{request_idx} completed")
return result
with concurrent.futures.ThreadPoolExecutor(max_workers=request_num) as executor:
futures = {executor.submit(_non_stream_task, i + 1): i + 1 for i in range(request_num)}
for future in concurrent.futures.as_completed(futures):
request_idx = futures[future]
try:
resp = future.result()
omni_resp = future.result()
except Exception as e:
print(
f"[audio.speech] request#{request_idx} failed "
f"(stream={stream}, timeout={timeout:.1f}s): {e!r}"
)
raise
omni_resp = self._process_non_stream_audio_speech_response(resp, response_format=speech_fmt)
assert_audio_speech_response(omni_resp, request_config, run_level=self.run_level)
responses.append(omni_resp)
@@ -1113,19 +1180,21 @@ class OpenAIClientHandler:
"""
responses: list[DiffusionResponse] = []
def _create_from_config(cfg: dict[str, Any]):
def _create_from_config(cfg: dict[str, Any]) -> tuple[Any, float]:
stream = cfg.get("stream", False)
if stream:
raise NotImplementedError("Streaming is not currently implemented for diffusion model e2e test")
modalities = cfg.get("modalities", omit) # Most diffusion models don't require modalities param
eb = cfg.get("extra_body")
extra = copy.deepcopy(eb) if eb else None
return self.client.chat.completions.create(
wall_start = time.perf_counter()
chat_completion = self.client.chat.completions.create(
model=cfg.get("model"),
messages=cfg.get("messages"),
extra_body=extra,
modalities=modalities,
)
return chat_completion, wall_start
if isinstance(request_config, list):
if not request_config:
@@ -1133,12 +1202,20 @@ class OpenAIClientHandler:
if request_num != 1:
raise ValueError("request_num is not supported when request_config is a list")
with concurrent.futures.ThreadPoolExecutor(max_workers=len(request_config)) as executor:
futures = {executor.submit(_create_from_config, cfg): cfg for cfg in request_config}
futures = {
executor.submit(_create_from_config, cfg): (i + 1, cfg) for i, cfg in enumerate(request_config)
}
for future in concurrent.futures.as_completed(futures):
cfg = futures[future]
chat_completion = future.result()
response = self._process_diffusion_response(chat_completion)
request_idx, cfg = futures[future]
chat_completion, wall_start = future.result()
response = self._process_diffusion_response(chat_completion, wall_start=wall_start)
assert_diffusion_response(response, cfg, run_level=self.run_level)
if response.e2e_latency is not None:
self._print_client_stat(
f"[diffusion] request#{request_idx} success in {response.e2e_latency:.3f}s"
)
else:
self._print_client_stat(f"[diffusion] request#{request_idx} completed")
responses.append(response)
return responses
@@ -1152,27 +1229,40 @@ class OpenAIClientHandler:
with concurrent.futures.ThreadPoolExecutor(max_workers=len(size_splits)) as executor:
futures = [executor.submit(_create_from_config, sub) for sub in size_splits]
chat_completions = [f.result() for f in futures]
parts = [self._process_diffusion_response(cc) for cc in chat_completions]
parts = [self._process_diffusion_response(cc, wall_start=ws) for cc, ws in chat_completions]
merged = _merge_diffusion_responses(parts)
merged.e2e_latency = time.perf_counter() - t0
assert_diffusion_response(merged, request_config, run_level=self.run_level)
if merged.e2e_latency is not None:
self._print_client_stat(f"[diffusion] request#1 success in {merged.e2e_latency:.3f}s")
else:
self._print_client_stat("[diffusion] request#1 completed")
return [merged]
if request_num == 1:
# Send single request
chat_completion = _create_from_config(request_config)
response = self._process_diffusion_response(chat_completion)
chat_completion, wall_start = _create_from_config(request_config)
response = self._process_diffusion_response(chat_completion, wall_start=wall_start)
assert_diffusion_response(response, request_config, run_level=self.run_level)
if response.e2e_latency is not None:
self._print_client_stat(f"[diffusion] request#1 success in {response.e2e_latency:.3f}s")
else:
self._print_client_stat("[diffusion] request#1 completed")
responses.append(response)
return responses
# Send concurrent requests for the same request_config
with concurrent.futures.ThreadPoolExecutor(max_workers=request_num) as executor:
futures = [executor.submit(_create_from_config, request_config) for _ in range(request_num)]
futures = {executor.submit(_create_from_config, request_config): i + 1 for i in range(request_num)}
for future in concurrent.futures.as_completed(futures):
chat_completion = future.result()
response = self._process_diffusion_response(chat_completion)
request_idx = futures[future]
chat_completion, wall_start = future.result()
response = self._process_diffusion_response(chat_completion, wall_start=wall_start)
assert_diffusion_response(response, request_config, run_level=self.run_level)
if response.e2e_latency is not None:
self._print_client_stat(f"[diffusion] request#{request_idx} success in {response.e2e_latency:.3f}s")
else:
self._print_client_stat(f"[diffusion] request#{request_idx} completed")
responses.append(response)
return responses
@@ -1220,6 +1310,10 @@ class OpenAIClientHandler:
result.videos = [video_content]
result.e2e_latency = end_time - start_time
assert_diffusion_response(result, request_config, run_level=self.run_level)
if result.e2e_latency is not None:
self._print_client_stat(f"[diffusion] request#1 success in {result.e2e_latency:.3f}s")
else:
self._print_client_stat("[diffusion] request#1 completed")
return [result]
def _wait_until_video_completed(
@@ -1274,6 +1368,7 @@ class OmniRunner:
stage_configs_path: str | None = None,
**kwargs,
) -> None:
startup_t0 = time.perf_counter()
cleanup_dist_env_and_memory()
run_pre_test_cleanup()
run_post_test_cleanup()
@@ -1292,6 +1387,9 @@ class OmniRunner:
stage_configs_path=stage_configs_path,
**kwargs,
)
startup_s = time.perf_counter() - startup_t0
if log_stats:
print(f"OmniRunner startup took {startup_s:.3f}s (model={model_name})", flush=True)
def get_default_sampling_params_list(self) -> list[Any]:
if not hasattr(self.omni, "default_sampling_params_list"):
@@ -1797,6 +1895,9 @@ def iter_omni_server(
server_args = [*server_args, "--init-timeout", str(params.init_timeout)]
else:
server_args = [*server_args, "--init-timeout", "900"]
# ``omni_server`` / ``omni_server_function``: match ``serve`` (``--disable-log-stats`` wins).
if "--disable-log-stats" not in server_args and "--log-stats" not in server_args:
server_args = [*server_args, "--log-stats"]
if params.use_stage_cli:
if not params.use_omni:
raise ValueError("omni_server with use_stage_cli=True requires use_omni=True")