diff --git a/tests/e2e/online_serving/test_cosyvoice3_tts_expansion.py b/tests/e2e/online_serving/test_cosyvoice3_tts_expansion.py index bf347db6f..e770baf56 100644 --- a/tests/e2e/online_serving/test_cosyvoice3_tts_expansion.py +++ b/tests/e2e/online_serving/test_cosyvoice3_tts_expansion.py @@ -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", ) diff --git a/tests/e2e/online_serving/test_qwen3_tts_base.py b/tests/e2e/online_serving/test_qwen3_tts_base.py index 87c25f5b6..5bacd7ced 100644 --- a/tests/e2e/online_serving/test_qwen3_tts_base.py +++ b/tests/e2e/online_serving/test_qwen3_tts_base.py @@ -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", ) diff --git a/tests/e2e/online_serving/test_qwen3_tts_base_expansion.py b/tests/e2e/online_serving/test_qwen3_tts_base_expansion.py index a4267c096..866d7483c 100644 --- a/tests/e2e/online_serving/test_qwen3_tts_base_expansion.py +++ b/tests/e2e/online_serving/test_qwen3_tts_base_expansion.py @@ -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", ), diff --git a/tests/e2e/online_serving/test_qwen3_tts_customvoice.py b/tests/e2e/online_serving/test_qwen3_tts_customvoice.py index 06bb7b44b..a03e7ad8b 100644 --- a/tests/e2e/online_serving/test_qwen3_tts_customvoice.py +++ b/tests/e2e/online_serving/test_qwen3_tts_customvoice.py @@ -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", ) diff --git a/tests/e2e/online_serving/test_qwen3_tts_customvoice_expansion.py b/tests/e2e/online_serving/test_qwen3_tts_customvoice_expansion.py index 9d9da2847..ce3bec928 100644 --- a/tests/e2e/online_serving/test_qwen3_tts_customvoice_expansion.py +++ b/tests/e2e/online_serving/test_qwen3_tts_customvoice_expansion.py @@ -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", ), diff --git a/tests/helpers/assertions.py b/tests/helpers/assertions.py index ca9bcb4f2..65a7c1404 100644 --- a/tests/helpers/assertions.py +++ b/tests/helpers/assertions.py @@ -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: diff --git a/tests/helpers/fixtures/runtime.py b/tests/helpers/fixtures/runtime.py index b20d1e471..3bffa475d 100644 --- a/tests/helpers/fixtures/runtime.py +++ b/tests/helpers/fixtures/runtime.py @@ -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") diff --git a/tests/helpers/runtime.py b/tests/helpers/runtime.py index a50390d6c..1483aef3e 100644 --- a/tests/helpers/runtime.py +++ b/tests/helpers/runtime.py @@ -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")