Compare commits

..
Author SHA1 Message Date
Hongsheng Liu fd071ae081 Merge branch 'main' into codex-diffusion-shutdown-escalation 2026-09-10 13:27:51 +08:00
Zhou TaichangandClaude Opus 5 aff7d64948 [Rebase] Rebase to vLLM 0.29.0 (#7230)
Signed-off-by: tzhouam <tzhouam@connect.ust.hk>
Signed-off-by: Zhou Taichang <tzhouam@connect.ust.hk>
Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
2026-09-10 12:08:28 +08:00
Sy03 333f14e97e [Bugfix][MiniCPM-o] Align structured chat content with native omni rendering (#7344)
Signed-off-by: Sy03 <1370724210@qq.com>
2026-09-10 11:38:11 +08:00
Hongsheng Liu c79c841e00 [Bugfix] Escalate diffusion worker shutdown and retain survivors
Signed-off-by: Hongsheng Liu <liuhongsheng4@huawei.com>
2026-09-06 15:56:17 +08:00
9 changed files with 389 additions and 30 deletions

No files matched your search

+1
View File
@@ -155,6 +155,7 @@ steps:
source_file_dependencies:
- tests/e2e/offline_inference/test_minicpmo_4_5.py
- tests/e2e/online_serving/test_minicpmo_4_5.py
- vllm_omni/transformers_utils/chat_templates/minicpmo45_native.jinja
- vllm_omni/model_executor/models/minicpmo_4_5/
- vllm_omni/model_executor/models/step_audio2/
- vllm_omni/model_executor/models/cosyvoice3/code2wav_core/
+1
View File
@@ -188,6 +188,7 @@ steps:
- label: "Omni · MiniCPM-o 4.5 Online Test"
source_file_dependencies:
- tests/e2e/online_serving/test_minicpmo_4_5.py
- vllm_omni/transformers_utils/chat_templates/minicpmo45_native.jinja
- vllm_omni/model_executor/models/minicpmo_4_5/
- vllm_omni/model_executor/models/step_audio2/
- vllm_omni/model_executor/models/cosyvoice3/code2wav_core/
+1 -1
View File
@@ -161,7 +161,7 @@ where = ["."]
include = ["vllm_omni*"]
[tool.setuptools.package-data]
"vllm_omni" = ["_version.py", "py.typed"]
"vllm_omni" = ["_version.py", "py.typed", "transformers_utils/chat_templates/*.jinja"]
[tool.setuptools_scm]
# no extra settings needed, presence enables setuptools-scm
+17 -4
View File
@@ -28,13 +28,13 @@ also provided.
- Default deploy configs (auto-loaded by HF `model_type=minicpmo` +
`hf_config.version="4.5"`):
- Default single-GPU compatibility layout (auto-loaded):
- Default single-GPU compatibility layout (auto-loaded):
[`vllm_omni/deploy/minicpmo_4_5.yaml`](../../vllm_omni/deploy/minicpmo_4_5.yaml)
- Recommended 2-GPU continuous-batching layout:
- Recommended 2-GPU continuous-batching layout:
[`vllm_omni/deploy/minicpmo_4_5_2gpu.yaml`](../../vllm_omni/deploy/minicpmo_4_5_2gpu.yaml),
- 3-GPU layout:
- 3-GPU layout:
[`vllm_omni/deploy/minicpmo_4_5_3gpu.yaml`](../../vllm_omni/deploy/minicpmo_4_5_3gpu.yaml)
- 8x RTX 4090 layout:
- 8x RTX 4090 layout:
[`vllm_omni/deploy/minicpmo_4_5_8x4090.yaml`](../../vllm_omni/deploy/minicpmo_4_5_8x4090.yaml)
- Online example + Gradio demo:
[`examples/online_serving/minicpmo/`](../../examples/online_serving/minicpmo/)
@@ -126,9 +126,20 @@ video. Use the two-GPU profile for production throughput.
```bash
vllm serve openbmb/MiniCPM-o-4_5 --omni \
--trust-remote-code \
--chat-template vllm_omni/transformers_utils/chat_templates/minicpmo45_native.jinja \
--chat-template-content-format openai \
--host 0.0.0.0 --port 8099
```
Run these commands from the repository root. The bundled chat template
matches the Hugging Face `chat(omni_mode=True)` content layout: media and text
parts are concatenated without adding separators, while whitespace explicitly
included in a text part is preserved. This matters for speech generation because
an inserted newline changes the Thinker condition passed to Talker. For image
questions, place the image before the question, as in the native image-chat
layout. Continue to request `use_tts_template: true` and
`enable_thinking: false` for spoken answers.
The deploy config is auto-loaded by the model registry — no
`--deploy-config` flag needed for this default single-GPU layout.
@@ -271,6 +282,8 @@ uses GPU 5. GPUs 6–7 are left free.
vllm serve openbmb/MiniCPM-o-4_5 --omni \
--deploy-config vllm_omni/deploy/minicpmo_4_5_8x4090.yaml \
--trust-remote-code \
--chat-template vllm_omni/transformers_utils/chat_templates/minicpmo45_native.jinja \
--chat-template-content-format openai \
--host 0.0.0.0 --port 8099
```
@@ -4,8 +4,10 @@
import asyncio
import multiprocessing as mp
import queue
import signal
import threading
import time
import weakref
from types import SimpleNamespace
from unittest.mock import MagicMock, Mock
@@ -1330,6 +1332,7 @@ class TestExecutorShutdownCleaner:
class FakeProcess:
def __init__(self, name):
self.name = name
self.pid = 123
self.alive = True
self.terminated = False
self.join_timeouts = []
@@ -1358,6 +1361,132 @@ class TestExecutorShutdownCleaner:
assert first.terminated and second.terminated
assert not first.is_alive() and not second.is_alive()
@pytest.mark.parametrize("cooperative", [False, True])
def test_real_workers_exit_and_are_reaped(self, monkeypatch, cooperative):
from vllm_omni.diffusion.executor import multiproc_executor as executor_module
ctx = mp.get_context("fork")
ready, stop = ctx.Event(), ctx.Event()
def run_worker():
signal.signal(signal.SIGTERM, signal.SIG_IGN)
ready.set()
stop.wait()
dead = ctx.Process(target=lambda: None)
worker = ctx.Process(target=run_worker)
monkeypatch.setattr(executor_module, "_WORKER_SHUTDOWN_GRACE_S", 0.2)
monkeypatch.setattr(executor_module, "_WORKER_TERMINATE_GRACE_S", 0.1)
monkeypatch.setattr(executor_module, "_WORKER_KILL_GRACE_S", 5.0, raising=False)
dead.start()
worker.start()
try:
dead.join(5)
assert not dead.is_alive()
assert ready.wait(5), "worker did not install its signal handler"
mq = Mock()
if cooperative:
mq.enqueue.side_effect = lambda *args, **kwargs: stop.set()
else:
mq.enqueue.side_effect = OSError("shutdown queue unavailable")
cleaner = executor_module._ExecutorShutdownCleaner(mq, 2, [dead, worker])
cleaner()
assert not worker.is_alive()
assert worker.exitcode == (0 if cooperative else -signal.SIGKILL)
assert cleaner.processes == []
assert cleaner.broadcast_mq is None
cleaner()
finally:
for proc in (dead, worker):
if proc.is_alive():
proc.kill()
proc.join(5)
proc.close()
@pytest.mark.parametrize("failed_action", ["terminate", "kill"])
def test_kill_phase_shares_deadline_and_continues_after_os_errors(self, monkeypatch, failed_action):
from vllm_omni.diffusion.executor import multiproc_executor as executor_module
first, second = Mock(pid=101), Mock(pid=102)
first.name, second.name = "first", "second"
for proc in (first, second):
proc.is_alive.side_effect = [True, True, True, False]
getattr(first, failed_action).side_effect = OSError("signal failed")
first.join.side_effect = [OSError("join failed"), None, None]
monotonic = Mock(side_effect=[100, 100, 110, 120, 120, 124, 130, 130, 134])
monkeypatch.setattr(executor_module, "time", SimpleNamespace(monotonic=monotonic))
cleaner = executor_module._ExecutorShutdownCleaner(processes=[first, second])
cleaner()
first.kill.assert_called_once()
second.kill.assert_called_once()
assert [call.args[0] for call in first.join.call_args_list] == [15, 5, 5]
assert [call.args[0] for call in second.join.call_args_list] == [5, 1, 1]
assert cleaner.processes == []
def test_executor_retains_survivor_for_explicit_shutdown_retry(self, monkeypatch):
from vllm_omni.diffusion.executor import multiproc_executor as executor_module
survivor = Mock(pid=123)
survivor.name = "surviving-worker"
survivor.is_alive.return_value = True
cleaner = executor_module._ExecutorShutdownCleaner(processes=[survivor])
executor, _, _ = _make_executor()
executor._shutdown_cleaner = cleaner
executor._processes = [survivor]
executor._finalizer = weakref.finalize(executor, cleaner)
executor._pump_stop = threading.Event()
executor._futures_lock = threading.RLock()
executor._rpc_futures = {}
executor._output_futures = {}
executor._batch_split_map = {}
log_error = Mock()
monkeypatch.setattr(executor_module.logger, "error", log_error)
executor.shutdown()
assert not executor._finalizer.alive
assert executor._shutdown_cleaner is cleaner
assert executor._processes == [survivor]
log_error.assert_called_once()
assert log_error.call_args.args[1] == [("surviving-worker", 123)]
survivor.is_alive.return_value = False
executor.shutdown()
executor.shutdown()
assert executor._shutdown_cleaner is None
assert executor._processes == []
def test_concurrent_cleanup_does_not_repeat_process_operations(self):
from vllm_omni.diffusion.executor import multiproc_executor as executor_module
joining, release = threading.Event(), threading.Event()
proc = Mock(pid=123)
proc.is_alive.return_value = True
def join(timeout):
joining.set()
assert release.wait(5)
proc.is_alive.return_value = False
proc.join.side_effect = join
cleaner = executor_module._ExecutorShutdownCleaner(processes=[proc])
thread = threading.Thread(target=cleaner)
thread.start()
try:
assert joining.wait(5)
cleaner()
proc.join.assert_called_once()
finally:
release.set()
thread.join(5)
assert not thread.is_alive()
assert cleaner.processes == []
# ───────── monitor thread & death sentinel integration tests ─────────
@@ -1406,7 +1535,7 @@ class TestMultiprocExecutorWorkerMonitor:
executor._result_mq = None
executor._shutdown_cleaner = None
# Use a no-op so shutdown() doesn't crash on None resources.
executor._finalizer = lambda: None
executor._finalizer = weakref.finalize(executor, lambda: None)
# ------------------------------------------------------------------
# Attributes added by remove_bubble_v2 (async D2H); shutdown() iterates
# over them, so they need to exist even when constructed via __new__.
@@ -1441,7 +1570,7 @@ class TestMultiprocExecutorWorkerMonitor:
executor._broadcast_mq = None
executor._result_mq = None
executor._shutdown_cleaner = None
executor._finalizer = lambda: None
executor._finalizer = weakref.finalize(executor, lambda: None)
proc = _make_short_lived_process()
executor._processes = [proc]
+23 -6
View File
@@ -24,6 +24,9 @@ from tests.helpers.stage_config import get_deploy_config_path, modify_stage_conf
os.environ["VLLM_WORKER_MULTIPROC_METHOD"] = "spawn"
_MODEL = "openbmb/MiniCPM-o-4_5"
_NATIVE_CHAT_TEMPLATE = (
Path(__file__).resolve().parents[3] / "vllm_omni/transformers_utils/chat_templates/minicpmo45_native.jinja"
)
_CI_DEPLOY = modify_stage_config(
get_deploy_config_path("minicpmo_4_5.yaml"),
updates={
@@ -63,7 +66,14 @@ test_params = [
model=_MODEL,
stage_config_path=_CI_DEPLOY,
use_stage_cli=False,
server_args=["--trust-remote-code", "--async-chunk"],
server_args=[
"--trust-remote-code",
"--async-chunk",
"--chat-template",
str(_NATIVE_CHAT_TEMPLATE),
"--chat-template-content-format",
"openai",
],
env_dict={
"VLLM_CONFIGURE_LOGGING": "1",
"VLLM_LOGGING_CONFIG_PATH": str(_PROMPT_LOGGING_CONFIG_PATH),
@@ -265,11 +275,18 @@ def test_image_to_text_audio_001(omni_server, openai_client) -> None:
Input Setting: stream=True
"""
image_data_url = f"data:image/jpeg;base64,{generate_synthetic_image(24, 24)['base64']}"
messages = dummy_messages_from_mix_data(
system_prompt=get_system_prompt(),
image_data_url=image_data_url,
content_text=get_prompt("text_image"),
)
# Match the native image-chat layout explicitly; do not rely on the
# renderer moving media placeholders ahead of text.
messages = [
get_system_prompt(),
{
"role": "user",
"content": [
{"type": "image_url", "image_url": {"url": image_data_url}},
{"type": "text", "text": get_prompt("text_image")},
],
},
]
request_config = {
"model": omni_server.model,
@@ -0,0 +1,45 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
"""MiniCPM structured-content rendering must preserve native part boundaries."""
from pathlib import Path
import pytest
from jinja2 import TemplateError
from transformers.utils.chat_template_utils import _compile_jinja_template
pytestmark = [pytest.mark.core_model, pytest.mark.cpu]
TEMPLATE = Path(__file__).resolve().parents[4] / "vllm_omni/transformers_utils/chat_templates/minicpmo45_native.jinja"
@pytest.mark.parametrize(
"parts,expected",
[
([{"type": "image"}, {"type": "text", "text": "Question?"}], "(<image>./</image>)Question?"),
([{"type": "text", "text": "Question?"}, {"type": "image"}], "Question?(<image>./</image>)"),
([{"type": "image"}, {"type": "text", "text": "\n\nQuestion?"}], "(<image>./</image>)\n\nQuestion?"),
([{"type": "text", "text": " A "}, {"type": "text", "text": " B "}], " A B "),
([{"type": "audio"}, {"type": "text", "text": "Transcribe."}], "(<audio>./</audio>)Transcribe."),
([{"type": "video"}, {"type": "image"}], "(<video>./</video>)(<image>./</image>)"),
("\nKeep this whitespace.\n", "\nKeep this whitespace.\n"),
],
)
def test_native_content_concatenation(parts, expected):
rendered = _compile_jinja_template(TEMPLATE.read_text()).render(
messages=[{"role": "user", "content": parts}],
add_generation_prompt=True,
enable_thinking=False,
use_tts_template=True,
)
assert rendered == (
f"<|im_start|>user\n{expected}<|im_end|>\n<|im_start|>assistant\n<think>\n\n</think>\n\n<|tts_bos|>"
)
def test_unsupported_part_is_not_silently_discarded():
with pytest.raises(TemplateError, match="Unsupported MiniCPM-o content part"):
_compile_jinja_template(TEMPLATE.read_text()).render(
messages=[{"role": "user", "content": [{"type": "unknown"}]}],
add_generation_prompt=True,
)
@@ -1,3 +1,6 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
from __future__ import annotations
import concurrent.futures
@@ -10,7 +13,7 @@ import threading
import time
import weakref
from collections.abc import Callable
from dataclasses import dataclass
from dataclasses import dataclass, field
from multiprocessing.synchronize import Event
from typing import TYPE_CHECKING, Any, cast
@@ -42,6 +45,7 @@ _DEQUEUE_TIMEOUT_S = 5.0
_DLO_DP_WAVE_TIMEOUT_S = float(os.environ.get("VLLM_OMNI_DLO_DP_WAVE_TIMEOUT", 600.0))
_WORKER_SHUTDOWN_GRACE_S = 15.0
_WORKER_TERMINATE_GRACE_S = 5.0
_WORKER_KILL_GRACE_S = 5.0
_RESULT_PUMP_JOIN_TIMEOUT_S = 2.0
@@ -80,33 +84,61 @@ class _ExecutorShutdownCleaner:
broadcast_mq: MessageQueue | None = None
num_workers: int = 0
processes: list[mp.Process] | None = None
_lock: threading.Lock = field(default_factory=threading.Lock, init=False, repr=False)
def __call__(self) -> None:
"""Clean up background resources."""
# The worker monitor and an explicit shutdown may race. A retry must
# not signal/join the same Process objects while cleanup is in flight.
if not self._lock.acquire(blocking=False):
return
try:
self._cleanup()
finally:
self._lock.release()
def _cleanup(self) -> None:
if self.broadcast_mq is not None:
try:
for _ in range(self.num_workers):
self.broadcast_mq.enqueue(SHUTDOWN_MESSAGE, timeout=1.0)
self.broadcast_mq = None
except Exception as exc:
logger.warning("Failed to send shutdown signal: %s", exc)
finally:
self.broadcast_mq = None
if self.processes:
join_deadline = time.monotonic() + _WORKER_SHUTDOWN_GRACE_S
for proc in self.processes:
if not proc.is_alive():
continue
proc.join(max(0.0, join_deadline - time.monotonic()))
alive = [proc for proc in self.processes if proc.is_alive()]
for proc in alive:
logger.warning("Terminating diffusion worker %s after timeout", proc.name)
proc.terminate()
for action, grace in (
(None, _WORKER_SHUTDOWN_GRACE_S),
("terminate", _WORKER_TERMINATE_GRACE_S),
("kill", _WORKER_KILL_GRACE_S),
):
if not alive:
break
if action is not None:
for proc in alive:
try:
logger.warning("Calling %s on diffusion worker %s (pid=%s)", action, proc.name, proc.pid)
getattr(proc, action)()
except OSError:
logger.exception("Failed to %s diffusion worker %s (pid=%s)", action, proc.name, proc.pid)
terminate_deadline = time.monotonic() + _WORKER_TERMINATE_GRACE_S
for proc in alive:
proc.join(max(0.0, terminate_deadline - time.monotonic()))
deadline = time.monotonic() + grace
for proc in alive:
try:
proc.join(max(0.0, deadline - time.monotonic()))
except OSError:
logger.exception("Failed to join diffusion worker %s (pid=%s)", proc.name, proc.pid)
alive = [proc for proc in alive if proc.is_alive()]
self.processes = alive
if alive:
logger.error(
"Diffusion worker cleanup incomplete after kill: %s; retaining processes for shutdown retry",
[(proc.name, proc.pid) for proc in alive],
)
class MultiprocDiffusionExecutor(DiffusionExecutor):
@@ -1006,8 +1038,12 @@ class MultiprocDiffusionExecutor(DiffusionExecutor):
def shutdown(self) -> None:
self._closed = True
self._pump_stop.set()
cleaner = self._shutdown_cleaner
try:
self._finalizer()
if self._finalizer.alive:
self._finalizer()
elif cleaner is not None:
cleaner()
finally:
pump_threads = getattr(self, "_result_pump_threads", [])
for thread in pump_threads:
@@ -1031,5 +1067,6 @@ class MultiprocDiffusionExecutor(DiffusionExecutor):
self._rpc_futures.clear()
self._output_futures.clear()
self._batch_split_map.clear()
self._shutdown_cleaner = None
self._processes = []
self._processes = (cleaner.processes or []) if cleaner is not None else []
if not self._processes:
self._shutdown_cleaner = None
@@ -0,0 +1,116 @@
{# SPDX-License-Identifier: Apache-2.0
Adapted from OpenBMB/MiniCPM-o-4_5 tokenizer_config.json.
Match chat(omni_mode=True): concatenate structured content parts without
injecting separators, preserving whitespace explicitly supplied as text.
#}
{%- set normalized = namespace(messages=[]) -%}
{%- for message in messages -%}
{%- if message.content is string or message.content is none -%}
{%- set normalized.messages = normalized.messages + [message] -%}
{%- else -%}
{%- set content = namespace(text='') -%}
{%- for part in message.content -%}
{%- if part.type == 'text' -%}
{%- set content.text = content.text + part.text -%}
{%- elif part.type == 'image' -%}
{%- set content.text = content.text + '(<image>./</image>)' -%}
{%- elif part.type == 'audio' -%}
{%- set content.text = content.text + '(<audio>./</audio>)' -%}
{%- elif part.type == 'video' -%}
{%- set content.text = content.text + '(<video>./</video>)' -%}
{%- else -%}
{{- raise_exception('Unsupported MiniCPM-o content part: ' + part.type) -}}
{%- endif -%}
{%- endfor -%}
{%- set normalized.messages = normalized.messages + [dict(message, content=content.text)] -%}
{%- endif -%}
{%- endfor -%}
{%- set messages = normalized.messages -%}
{%- if tools %}
{{- '<|im_start|>system\n' }}
{%- if messages[0].role == 'system' %}
{{- messages[0].content + '\n\n' }}
{%- endif %}
{{- "# Tools\n\nYou may call one or more functions to assist with the user query.\n\nYou are provided with function signatures within <tools></tools> XML tags:\n<tools>" }}
{%- for tool in tools %}
{{- "\n" }}
{{- tool | tojson }}
{%- endfor %}
{{- "\n</tools>\n\nFor each function call, return a json object with function name and arguments within <tool_call></tool_call> XML tags:\n<tool_call>\n{\"name\": <function-name>, \"arguments\": <args-json-object>}\n</tool_call><|im_end|>\n" }}
{%- else %}
{%- if messages[0].role == 'system' %}
{{- '<|im_start|>system\n' + messages[0].content + '<|im_end|>\n' }}
{%- endif %}
{%- endif %}
{%- set ns = namespace(multi_step_tool=true, last_query_index=messages|length - 1) %}
{%- for message in messages[::-1] %}
{%- set index = (messages|length - 1) - loop.index0 %}
{%- if ns.multi_step_tool and message.role == "user" and not(message.content.startswith('<tool_response>') and message.content.endswith('</tool_response>')) %}
{%- set ns.multi_step_tool = false %}
{%- set ns.last_query_index = index %}
{%- endif %}
{%- endfor %}
{%- for message in messages %}
{%- if (message.role == "user") or (message.role == "system" and not loop.first) %}
{{- '<|im_start|>' + message.role + '\n' + message.content + '<|im_end|>' + '\n' }}
{%- elif message.role == "assistant" %}
{%- set content = message.content %}
{%- set reasoning_content = '' %}
{%- if message.reasoning_content is defined and message.reasoning_content is not none %}
{%- set reasoning_content = message.reasoning_content %}
{%- else %}
{%- if '</think>' in message.content %}
{%- set content = message.content.split('</think>')[-1].lstrip('\n') %}
{%- set reasoning_content = message.content.split('</think>')[0].rstrip('\n').split('<think>')[-1].lstrip('\n') %}
{%- endif %}
{%- endif %}
{%- if loop.index0 > ns.last_query_index %}
{%- if loop.last or (not loop.last and reasoning_content) %}
{{- '<|im_start|>' + message.role + '\n<think>\n' + reasoning_content.strip('\n') + '\n</think>\n\n' + content.lstrip('\n') }}
{%- else %}
{{- '<|im_start|>' + message.role + '\n' + content }}
{%- endif %}
{%- else %}
{{- '<|im_start|>' + message.role + '\n' + content }}
{%- endif %}
{%- if message.tool_calls %}
{%- for tool_call in message.tool_calls %}
{%- if (loop.first and content) or (not loop.first) %}
{{- '\n' }}
{%- endif %}
{%- if tool_call.function %}
{%- set tool_call = tool_call.function %}
{%- endif %}
{{- '<tool_call>\n{"name": "' }}
{{- tool_call.name }}
{{- '", "arguments": ' }}
{%- if tool_call.arguments is string %}
{{- tool_call.arguments }}
{%- else %}
{{- tool_call.arguments | tojson }}
{%- endif %}
{{- '}\n</tool_call>' }}
{%- endfor %}
{%- endif %}
{{- '<|im_end|>\n' }}
{%- elif message.role == "tool" %}
{%- if loop.first or (messages[loop.index0 - 1].role != "tool") %}
{{- '<|im_start|>user' }}
{%- endif %}
{{- '\n<tool_response>\n' }}
{{- message.content }}
{{- '\n</tool_response>' }}
{%- if loop.last or (messages[loop.index0 + 1].role != "tool") %}
{{- '<|im_end|>\n' }}
{%- endif %}
{%- endif %}
{%- endfor %}
{%- if add_generation_prompt %}
{{- '<|im_start|>assistant\n' }}
{%- if enable_thinking is defined and enable_thinking is false %}
{{- '<think>\n\n</think>\n\n' }}
{%- endif %}
{%- if use_tts_template is defined and use_tts_template is true %}
{{- '<|tts_bos|>' }}
{%- endif %}
{%- endif -%}