Skip to content
Merged
Show file tree
Hide file tree
Changes from 2 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
79 changes: 79 additions & 0 deletions tests/entrypoints/test_omni_entrypoints.py
Original file line number Diff line number Diff line change
Expand Up @@ -1482,3 +1482,82 @@ def test_del_shutsdown_engine(monkeypatch: pytest.MonkeyPatch):
del app
gc.collect()
assert engine.shutdown_called


def _client_error_then_outputs(engine: FakeAsyncOmniEngine, msg: dict[str, Any]) -> None:
"""First request is rejected with a 4xx client error; later ones complete normally."""
if len(engine.submitted) == 1:
_enqueue_client_error_message(engine, msg)
else:
_enqueue_omni_final_only_outputs(engine, msg)


def test_omni_generate_client_error_keeps_engine_usable(monkeypatch: pytest.MonkeyPatch):
"""Issue #8449: a non-fatal client error must fail only that generate() call.

The rejected request is aborted and the error propagates with its status
metadata, but the engine stays up so the next generate() succeeds, matching
AsyncOmni. Before the fix list-mode generate() closed the engine and the
second call died with janus.ShutDown.
"""
sampling_params = [SamplingParams(max_tokens=8) for _ in range(3)]
engine = FakeAsyncOmniEngine(
stage_metadata=THREE_STAGE_META,
default_sampling_params_list=sampling_params,
on_add_request=_client_error_then_outputs,
)
_patch_engine(monkeypatch, engine)

app = Omni("dummy-model")
try:
with pytest.raises(OmniClientError) as exc_info:
app.generate(["blocked"], use_tqdm=False)

assert exc_info.value.status_code == 400
assert exc_info.value.error_type == "BadRequestError"
rejected_id = engine.submitted[0]["request_id"]
assert engine.aborted == [[rejected_id]]
assert rejected_id not in app.request_states
assert not engine.shutdown_called

outputs = app.generate(["p2"], use_tqdm=False)
finally:
app.shutdown()

assert [output.stage_id for output in outputs] == [0, 2]
assert len(engine.submitted) == 2


def test_omni_generate_server_error_keeps_engine_usable(monkeypatch: pytest.MonkeyPatch):
"""A non-fatal server-side error (RuntimeError, no 4xx status) also leaves the engine up."""
engine = FakeAsyncOmniEngine(
stage_metadata=THREE_STAGE_META,
on_add_request=_enqueue_error_message,
)
_patch_engine(monkeypatch, engine)

app = Omni("dummy-model")
try:
with pytest.raises(RuntimeError, match="engine boom") as exc_info:
app.generate(["boom"], use_tqdm=False)
assert not isinstance(exc_info.value, OmniClientError)
assert not engine.shutdown_called
assert engine.aborted == [[engine.submitted[0]["request_id"]]]
finally:
app.shutdown()


def test_omni_generate_fatal_error_still_closes_engine(monkeypatch: pytest.MonkeyPatch):
"""A fatal ErrorMessage means the engine is unusable: generate() keeps closing it."""
engine = FakeAsyncOmniEngine(
stage_metadata=THREE_STAGE_META,
on_add_request=_enqueue_fatal_error_message,
)
_patch_engine(monkeypatch, engine)

app = Omni("dummy-model")
with pytest.raises(EngineDeadError, match="engine dead") as exc_info:
app.generate(["dead"], use_tqdm=False)

assert isinstance(exc_info.value, OmniEngineDeadError)
assert engine.shutdown_called
11 changes: 10 additions & 1 deletion vllm_omni/entrypoints/omni.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@
from tqdm.auto import tqdm
from vllm.logger import init_logger
from vllm.sampling_params import RequestOutputKind
from vllm.v1.engine.exceptions import EngineDeadError

from vllm_omni.engine.async_omni_engine import AsyncOmniEngine
from vllm_omni.engine.messages import OutputMessage
Expand Down Expand Up @@ -95,9 +96,17 @@ def generate(
if py_generator:
return self._run_generation_with_generator(prompts, sampling_params_list, use_tqdm)
return list(self._run_generation(prompts, sampling_params_list, use_tqdm))
except EngineDeadError as e:
# Fatal: a stage or the orchestrator is gone, so the engine cannot
# serve anything else. Release the processes now.
logger.exception("[Omni] Engine died during generation: %s", e)
self.close()
raise
except Exception as e:

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P1] This PR removes self.close() from Omni.generate()'s except Exception (omni.py:1…

Evidence and suggested fix

This PR removes self.close() from Omni.generate()'s except Exception (omni.py:105-110) and claims in-flight requests were already aborted in _run_generation. That abort (omni.py:221-225 if "active_reqs" in locals() and active_reqs) only runs after omni.py:180 active_reqs = set(request_ids), which is after every add_request returns. request_states are registered at omni.py:160 before add_request at omni.py:169; a successful earlier add_request already did request_queue.sync_q.put(msg). A later prompt whose process_inputs fails (async_omni_engine.py:352-357, unchanged) therefore leaves earlier prompts in request_states and on the orchestrator queue of the now-kept-alive engine. Victim is generate([ok, bad]), not the single-prompt ErrorMessage path the new tests cover. Initialize active_reqs before the submit loop or abort leftover request_states in this except, and add that batch test.

Evidence: Trigger: Omni.generate([ok, bad]) where the second engine.add_request raises in process_inputs. omni.py:105-110 except Exception as e: / logger.exception("[Omni] Failed to run generation: %s", e) / raise — this PR dropped self.close() here and comments that in-flight requests were already aborted. That abort is omni.py:221-225 except Exception: / if "active_reqs" in locals() and active_reqs: / self.abort(list(active_reqs)). active_reqs is assigned only after the submit loop: omni.py:180 active_reqs = set(request_ids). Each prompt is registered first: omni.py:160 self.request_states[req_id] = req_state then omni.py:169 self.engine.add_request(. Unchanged by this diff, present in the PR-time tree: a successful add_request enqueues the request (async_omni_engine.py:576 self.request_queue.sync_q.put(msg)) before a later prompt's process_inputs re-raises (async_omni_engine.py:352-357 except Exception: ... raise). Adverse effect: active_reqs is never bound so abort is skipped; the first prompt remains in request_states and on the live orchestrator queue, occupying the engine this PR now keeps up. New tests only cover single-prompt ErrorMessage after the wait loop.

Suggestion: except Exception as e:
# Request-scoped failure. Abort leftovers from a submit-loop
# raise (active_reqs is assigned only after every add_request).
logger.exception("[Omni] Failed to run generation: %s", e)
leftover = list(self.request_states)
if leftover:
self.abort(leftover)
raise

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed in f6dddd4.

Thanks, confirmed. _run_generation() now initializes active_reqs before the try block and adds each request immediately when its ClientRequestState is registered, before sampling-parameter preparation or add_request(). A submission exception therefore aborts and removes all registered requests from that batch, including the failing request; prompts that were never registered are not aborted. Cleanup stays in _run_generation() so it also covers generator mode.

Added a regression test for ["ok", "bad", "not submitted"], parametrized over list and generator mode. It checks that the two registered requests are aborted, the third is never submitted, request_states is empty, the healthy engine is not shut down, and a subsequent generation succeeds. Both variants fail on the reviewed commit because no abort occurs.

Validation after the fix: 89 CPU tests passed across tests/entrypoints/test_omni_entrypoints.py and tests/engine/test_async_omni_engine_outputs.py (vLLM 0.31.0). All applicable pre-commit checks passed except the same 9 pre-existing mypy errors; no new type errors.

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P1] This diff removes self.close() from except Exception and only closes on `En…

Evidence and suggested fix

This diff removes self.close() from except Exception and only closes on EngineDeadError. List-mode Omni.generate() polls self.engine.try_get_output() (omni.py:187). On an empty queue with a dead orchestrator that API raises RuntimeError("Orchestrator died unexpectedly. See logs above.") (omni_engine_base.py:953-955), not EngineDeadError, so the new keep-alive path skips teardown. The next generate() then talks to a dead orchestrator, and stage processes are not released until GC/finalizer. Omni.errored is already not self.engine.is_alive() (omni_base.py:321). Close when self.errored in this except Exception branch, and add a CPU test that raises this RuntimeError from try_get_output and asserts shutdown_called (the new fatal test only covers ErrorMessage.fatal; FakeAsyncOmniEngine.try_get_output returns None on empty and never raises).

Evidence: Trigger: list-mode Omni.generate() consumes _run_generation, which polls try_get_output until requests finish. If the orchestrator thread dies with an empty output queue, that poll raises RuntimeError, not EngineDeadError; this PR's except Exception then re-raises without close(), so later generate() reuses the dead engine and stage GPU processes are not shut down. omni.py:187 (unchanged by this diff, present in the PR-time tree) msg = self.engine.try_get_output(). omni.py:99-110 (this diff) except EngineDeadError as e: … self.close() / raise then except Exception as e: … raise with no close(). Unchanged by this diff, present in the PR-time tree: omni_engine_base.py:953-955 except queue.Empty: / if not self.is_alive(): / raise RuntimeError("Orchestrator died unexpectedly. See logs above."). Unchanged: omni_base.py:321 return not self.engine.is_alive(). Unmet coverage: tests/entrypoints/test_omni_entrypoints.py:1550-1563 only drives ErrorMessage.fatal → EngineDeadError; FakeAsyncOmniEngine.try_get_output at tests/entrypoints/test_omni_entrypoints.py:162-166 except queue.Empty: / return None and never raises this RuntimeError. tests/engine/test_async_omni_engine_outputs.py:91-92 already pins the engine contract: with pytest.raises(RuntimeError, match="Orchestrator died unexpectedly"): engine.try_get_output().

Suggestion: except EngineDeadError as e:
# Fatal: a stage or the orchestrator is gone, so the engine cannot
# serve anything else. Release the processes now.
logger.exception("[Omni] Engine died during generation: %s", e)
self.close()
raise
except Exception as e:
# Request-scoped failure (e.g. OmniClientError -> 4xx). The in-flight
# requests were already aborted inside _run_generation; keep the
# engine alive so later generate() calls work, like AsyncOmni.
logger.exception("[Omni] Failed to run generation: %s", e)
if self.errored:
self.close()
raise

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed in f6dddd4.

Thanks, confirmed. The generic exception handler now checks self.errored and closes the engine when the orchestrator is dead, while preserving the original exception. Healthy engines still remain reusable after request-scoped errors, and the existing EngineDeadError shutdown branch is retained.

Added a CPU regression test whose try_get_output() marks the orchestrator dead and raises RuntimeError("Orchestrator died unexpectedly. See logs above."). It checks that the error propagates, shutdown is called, and the in-flight request is aborted and removed. This test fails on the reviewed commit because shutdown is not called.

Validation after the fix: 89 CPU tests passed across tests/entrypoints/test_omni_entrypoints.py and tests/engine/test_async_omni_engine_outputs.py (vLLM 0.31.0). All applicable pre-commit checks passed except the same 9 pre-existing mypy errors; no new type errors.

# Request-scoped failure (e.g. OmniClientError -> 4xx). The in-flight
# requests were already aborted inside _run_generation; keep the
# engine alive so later generate() calls work, like AsyncOmni.
logger.exception("[Omni] Failed to run generation: %s", e)
self.close()
raise

def _run_generation_with_generator(
Expand Down
Loading