[Perf][Qwen3-TTS] Enable event-driven orchestration by default

Signed-off-by: Sy03 <1370724210@qq.com>
This commit is contained in:
Sy03 committed 2026-09-16 11:23:36 +08:00
1 parent 78934753ff
commit 521ef843c6
5 files changed
+71 -7

No files matched your search

+1 -1
View File
@@ -83,7 +83,7 @@ depends on the installed kernels and model path.
| `SPEAKER_SAMPLES_DIR` | Filesystem path; default `~/.cache/vllm-omni/speakers` | Speech server; read when speaker storage initializes | Environment-only setting. The directory is created; filesystem errors propagate. | Stable |
| `SPEAKER_MAX_UPLOADED` | Integer; default `1000` | Speech server; read when speaker storage initializes | Environment-only setting. A non-integer logs a warning and uses `1000`; range is not otherwise validated. | Stable |
| `VLLM_OMNI_ASYNC_OUTPUT_TIMEOUT` | Float seconds; default `600` | Diffusion engine async-output wait in `step_streaming`; resolved per call on the request path, not at import | Environment-only setting. A non-float or `<=0` value warns once and uses the default. | Experimental |
| `VLLM_OMNI_EVENT_DRIVEN_ORCH` | `1`, `true`, `yes` or `on` enables; default `0` (off) | Orchestration loop and the serving-side final-output drain; read once when the `Orchestrator` is constructed | Environment-only setting. Values are stripped and case-normalized; any unrecognized value leaves the legacy poll loop selected. | Experimental |
| `VLLM_OMNI_EVENT_DRIVEN_ORCH` | `1`, `true`, `yes` or `on` enables; default `0` (off), except Qwen3-TTS defaults to on | Orchestration loop and the serving-side final-output drain; read once when the `Orchestrator` is constructed | Environment-only setting. An explicit value always wins. Values are stripped and case-normalized; any unrecognized value leaves the legacy poll loop selected. | Experimental |
| `VLLM_OMNI_INPUT_WAIT_TIMEOUT_S` | Float seconds; default `600`; `<=0` disables | Full-payload input coordinator, not async-chunk transfer; read when the scheduler module imports in each worker | Environment-only setting. A non-float logs a warning and uses `600`. | Stable operational control |
| `VLLM_OMNI_ORCH_MONITOR_PATH` | Filesystem path; default `<current-working-directory>/vllm_omni_orch_monitor_<timestamp>.json` | Orchestrator monitor enabled by `--enable-orch-monitor`; read when the monitor is created | Environment-only path override. Parent directories are created; write errors are logged. | Diagnostic |
| `VLLM_OMNI_VIDEO_SYNC_TIMEOUT` | Float seconds; default `600` | Synchronous Videos API; read when the API server module imports | Environment-only setting. A non-float raises `ValueError` during import. | Experimental |
+1 -1
View File
@@ -72,7 +72,7 @@ the boundary affected by the in-flight stage client/process refactor in
responsibilities proposed only by that PR are not current contracts.
The orchestration loop also has an opt-in event-driven mode
(`VLLM_OMNI_EVENT_DRIVEN_ORCH=1`, default off) proposed in
(`VLLM_OMNI_EVENT_DRIVEN_ORCH=1`, default off except for Qwen3-TTS) proposed in
[#5221](https://github.com/vllm-project/vllm-omni/pull/5221). It changes poll
cadence only: the routing, ordering, and terminal-state contracts below hold
identically on both loops.
+17 -1
View File
@@ -19,7 +19,10 @@ import pytest
from vllm_omni.engine.async_omni_engine import AsyncOmniEngine
from vllm_omni.engine.messages import ShutdownRequestMessage
from vllm_omni.engine.orchestrator import _event_driven_orch_enabled
from vllm_omni.engine.orchestrator import (
_event_driven_orch_default_for_pipeline,
_event_driven_orch_enabled,
)
from . import test_orchestrator as legacy
from . import test_orchestrator_error_handling as legacy_errors
@@ -125,6 +128,19 @@ def test_flag_parsing(monkeypatch) -> None:
assert _event_driven_orch_enabled() is False
def test_only_qwen3_tts_has_a_pipeline_default() -> None:
assert _event_driven_orch_default_for_pipeline("qwen3_tts") is True
for model_type in (None, "qwen3_omni_moe", "minicpmo_4_5", "moss_tts_delay"):
assert _event_driven_orch_default_for_pipeline(model_type) is False
def test_explicit_event_driven_value_overrides_pipeline_default(monkeypatch) -> None:
monkeypatch.delenv("VLLM_OMNI_EVENT_DRIVEN_ORCH", raising=False)
assert _event_driven_orch_enabled(default=True) is True
monkeypatch.setenv("VLLM_OMNI_EVENT_DRIVEN_ORCH", "0")
assert _event_driven_orch_enabled(default=True) is False
def test_default_is_legacy_loop(monkeypatch) -> None:
"""Without the env flag, the harness runs the legacy poll loop."""
monkeypatch.delenv("VLLM_OMNI_EVENT_DRIVEN_ORCH", raising=False)
+41 -1
View File
@@ -13,6 +13,8 @@ from __future__ import annotations
import asyncio
import concurrent.futures
import copy
import json
import os
import queue
import shutil
import threading
@@ -69,7 +71,11 @@ from vllm_omni.engine.messages import (
OutputMessage,
StageSubmissionMessage,
)
from vllm_omni.engine.orchestrator import Orchestrator
from vllm_omni.engine.orchestrator import (
_EVENT_DRIVEN_ORCH_ENV,
Orchestrator,
_event_driven_orch_default_for_pipeline,
)
from vllm_omni.engine.rpc_result_router import CorrelatedRpcClient
from vllm_omni.engine.serialization import deserialize_additional_information
from vllm_omni.engine.stage_client import StageClient
@@ -212,6 +218,40 @@ class AsyncOmniEngine:
# retaining the richer resolver result for pipeline-wide settings.
# Overrides of that private seam fall back to the factory below.
deploy_config_path = kwargs.get("deploy_config")
# ``trust_remote_code`` is tri-state (bool | None): ``None`` means "not
# specified" so stage-config resolution can defer to the deploy yaml's
# per-stage value (see ``with_trust_remote_code_override``). The
# restriction path below loads the top-level HF config via vLLM's
# ``get_config``, which needs a real bool, so collapse ``None`` to the
# default ``False`` here (#5495).
pipeline_config = StageConfigFactory.get_pipeline_config(
model=model,
trust_remote_code=bool(trust_remote_code),
deploy_config_path=deploy_config_path,
)
self.endpoint_restrictions = pipeline_config.endpoint_restrictions if pipeline_config is not None else ()
self._duplex_runtime_extension_path = (
pipeline_config.duplex_runtime_extension if pipeline_config is not None else None
)
self.duplex_serving_adapter_path = (
pipeline_config.duplex_serving_adapter if pipeline_config is not None else None
)
self._duplex_control_enabled = bool(pipeline_config and pipeline_config.duplex_control_enabled)
if (
_EVENT_DRIVEN_ORCH_ENV not in os.environ
and pipeline_config is not None
and _event_driven_orch_default_for_pipeline(pipeline_config.model_type)
):
os.environ[_EVENT_DRIVEN_ORCH_ENV] = "1"
self.duplex_session_config = DuplexSessionRuntimeConfig()
if deploy_config_path is not None:
self.duplex_session_config = load_deploy_config(deploy_config_path).duplex_session
# Tri-state: None means "not specified" — the deploy yaml's per-stage
# trust_remote_code stays in effect. An explicit True/False here is a
# global override (precedence: caller > deploy yaml > default False);
# the merge rule lives in with_trust_remote_code_override.
kwargs = with_trust_remote_code_override(kwargs, trust_remote_code)
self._config_resolution: OmniConfigResolution | None = None
self.config_path, self.stage_configs = self._resolve_stage_configs(
model,
+11 -3
View File
@@ -96,7 +96,7 @@ def cleanup_request_artifact_dirs(artifact_dirs: set[str] | list[str]) -> None:
# 1 ms poll cadence to event-driven wakeups: one reader task per live LLM stage
# replica awaits `client.get_output_async()` directly — the same pattern vLLM's
# own AsyncLLM output handler uses — and feeds a single serial dispatch queue.
# Default off; the legacy poll loop remains the fallback.
# Default is off except for pipelines with an explicit validated default.
_EVENT_DRIVEN_ORCH_ENV = "VLLM_OMNI_EVENT_DRIVEN_ORCH"
# How often the event-driven loop reconciles its reader-task set against
@@ -104,8 +104,16 @@ _EVENT_DRIVEN_ORCH_ENV = "VLLM_OMNI_EVENT_DRIVEN_ORCH"
_ORCH_READER_RECONCILE_INTERVAL_S = 0.5
def _event_driven_orch_enabled() -> bool:
return os.environ.get(_EVENT_DRIVEN_ORCH_ENV, "0").strip().lower() in ("1", "true", "yes", "on")
def _event_driven_orch_enabled(*, default: bool = False) -> bool:
value = os.environ.get(_EVENT_DRIVEN_ORCH_ENV)
if value is None:
return default
return value.strip().lower() in ("1", "true", "yes", "on")
def _event_driven_orch_default_for_pipeline(pipeline_model_type: str | None) -> bool:
"""Return whether a pipeline has a validated event-driven default."""
return pipeline_model_type == "qwen3_tts"
def _build_terminal_empty_output(