From 2c92b63ccea7ce6471bbd4e8ae48ac6f974e523d Mon Sep 17 00:00:00 2001 From: Sihao Li <111170255+GG-li@users.noreply.github.com> Date: Mon, 12 Jan 2026 15:38:47 +0800 Subject: [PATCH] fix:#issue 432 (#517) Signed-off-by: GG-li <3226868735@qq.com> Signed-off-by: Sihao Li <111170255+GG-li@users.noreply.github.com> Co-authored-by: Hongsheng Liu --- vllm_omni/entrypoints/omni_stage.py | 30 ++++++++++++++++++++++++++--- 1 file changed, 27 insertions(+), 3 deletions(-) diff --git a/vllm_omni/entrypoints/omni_stage.py b/vllm_omni/entrypoints/omni_stage.py index 4fe6b0083..af6f60f04 100644 --- a/vllm_omni/entrypoints/omni_stage.py +++ b/vllm_omni/entrypoints/omni_stage.py @@ -585,9 +585,18 @@ def _stage_worker( lock_files = acquired_lock_fds except Exception as e: - logger.debug("[Stage-%s] Failed to set up sequential initialization lock: %s", stage_id, e) + logger.debug( + "[Stage-%s] Failed to set up sequential initialization lock: %s", + stage_id, + e, + ) # Init engine based on stage_type - logger.debug("[Stage-%s] Initializing %s engine with args keys=%s", stage_id, stage_type, list(engine_args.keys())) + logger.debug( + "[Stage-%s] Initializing %s engine with args keys=%s", + stage_id, + stage_type, + list(engine_args.keys()), + ) try: if stage_type == "diffusion": engine_args.pop("model_stage") @@ -1111,6 +1120,20 @@ async def _stage_worker_async( except (OSError, ValueError): pass omni_stage.set_async_engine(stage_engine) + if hasattr(omni_stage.async_engine, "log_stats") and omni_stage.async_engine.log_stats: + + async def _force_log(): + try: + while True: + await asyncio.sleep(10.0) + await omni_stage.async_engine.do_log_stats() + except asyncio.CancelledError: + pass + + log_stats_task = asyncio.create_task(_force_log()) + else: + log_stats_task = None + # Don't keep the dummy data in memory (only for LLM engines) if stage_type != "diffusion": await stage_engine.reset_mm_cache() @@ -1335,7 +1358,8 @@ async def _stage_worker_async( } ) logger.debug("Enqueued result for request %s to downstream", rid) - + if log_stats_task is not None: + log_stats_task.cancel() logger.info("Stage worker exiting")