[Core] Move pause and resume functions into engine

[hao-aaron]

Purpose

Relevant RFC: #32103

Follow up to #32351, enables pause and resume for api server count is > 1. This is necessary for data parallel deployments.

To keep a consistent pause state across multiple AsyncLLM instances, we push all pause logic to the EngineCore level. Pause and resume requests received by one AsyncLLM is broadcast to all EngineCores, so all engines instances share the same paused state and can be queried by any AsyncLLM to find the current paused state. New requests submitted while paused have the same behavior, e.g. new requests are frozen until resumed.

To ensure that pause only returns when the operation is completely finished (mode=wait finishes processing, mode=abort finishes processing all aborts), I introduce the FutureUtility mechanism, which allows a custom callback function to be specified which runs every step which decides when to return utility output and unblock the utility caller. An example is give in for mode=wait:

        elif mode == "wait":
            # wait: PAUSE_WAIT so adds are queued but step() still runs to drain.
            self._scheduler_pause_state = PauseState.PAUSE_WAIT

            def _step_wait() -> None:
                if self.scheduler.has_unfinished_requests():
                    return
                self._scheduler_pause_state = PauseState.PAUSE_KEEP
                if clear_cache:
                    self.reset_prefix_cache()
                    self.reset_mm_cache()
                    self.reset_encoder_cache()
                future.set_result(None)

            return UtilityFuture(future, _step_wait)

A return value of DeferredUtility lets the busy_loop know to stash the callback function and call it ever step, checking for some finish condition.

Support for DPEP is not included in this PR and will come in a separate one.

Test Plan

Unit tests

Test Result

passing

---

Essential Elements of an Effective PR Description Checklist

• The purpose of the PR, such as "Fix some issue (link existing issues this PR will resolve)".
• The test plan, such as providing test command.
• The test results, such as pasting the results comparison before and after, or e2e results
• (Optional) The necessary documentation update, such as updating supported_models.md and examples for a new model.
• (Optional) Release notes update. If your change is user facing, please update the release notes draft in the Google Doc.

[mergify]

Documentation preview: https://vllm--34125.org.readthedocs.build/en/34125/

[hao-aaron]

/gemini review

[gemini-code-assist]

Code Review

This pull request refactors the pause/resume functionality to support data-parallel deployments by centralizing the logic within EngineCore. Pause and resume commands are now broadcast to all engine instances, ensuring a consistent state. A new DeferredUtilityResult mechanism is introduced to handle asynchronous operations that require multiple steps to complete, making the pause operation more robust. The changes are well-tested with new unit tests and a new example for data-parallel pause/resume. My review identifies a couple of areas for improvement: one related to a potentially unreliable synchronization mechanism using queue.qsize(), and another concerning a possibly outdated NotImplementedError that may be overly restrictive with the new architecture.

[gemini-code-assist]

Code Review

This pull request introduces a significant refactoring to enable pause and resume functionality in data parallel deployments. The core logic for pausing and resuming is now centralized in EngineCore, and these actions are broadcast to all engine instances, ensuring a consistent state. A new DeferredUtilityResult mechanism is introduced to handle asynchronous operations that need to wait for specific conditions, which is a clean and effective pattern. The changes are well-implemented and supported by new and updated tests. My main concern is a potential IndexError in the new example test script, which could cause it to crash under certain conditions.

[kouroshHakha]

Did the first pass. Overall looks good. Needs some touch ups in the implementation of the deferred utility to bring down its complexity

[njhill]

Thanks @hao-aaron, overall I like the approach.

Have a couple of initial simplification comments from a first pass.

[hao-aaron]

/gemini review

[gemini-code-assist]

Code Review

This pull request is a significant and well-executed refactoring that moves the pause and resume logic into the EngineCore. This is a crucial step for enabling this functionality in data-parallel deployments. The introduction of the UtilityFuture mechanism for handling deferred operations is a solid design choice for managing asynchronous completion of pause modes. The code is cleaner, and the logic is now centralized, which greatly improves maintainability. I've found one critical issue related to aborting requests that are submitted while the engine is paused, which I've detailed in a specific comment. Otherwise, the changes look excellent.

[mergify]

This pull request has merge conflicts that must be resolved before it can be
merged. Please rebase the PR, @hao-aaron.

https://docs.github.com/en/pull-requests/collaborating-with-pull-requests/working-with-forks/syncing-a-fork

[njhill]

Thanks @hao-aaron!

[mergify]

This pull request has merge conflicts that must be resolved before it can be
merged. Please rebase the PR, @hao-aaron.

https://docs.github.com/en/pull-requests/collaborating-with-pull-requests/working-with-forks/syncing-a-fork

[mergify]

Hi @hao-aaron, the pre-commit checks have failed. Please run:

uv pip install pre-commit
pre-commit install
pre-commit run --all-files

Then, commit the changes and push to your branch.

For future commits, pre-commit will run automatically on changed files before each commit.

Tip

Is mypy or markdownlint failing?

mypy and markdownlint are run differently in CI. If the failure is related to either of these checks, please use the following commands to run them locally:

# For mypy (substitute "3.10" with the failing version if needed)
pre-commit run --hook-stage manual mypy-3.10
# For markdownlint
pre-commit run --hook-stage manual markdownlint

[mergify]

Hi @hao-aaron, the pre-commit checks have failed. Please run:

uv pip install pre-commit
pre-commit install
pre-commit run --all-files

Then, commit the changes and push to your branch.

For future commits, pre-commit will run automatically on changed files before each commit.

Tip

Is mypy or markdownlint failing?

mypy and markdownlint are run differently in CI. If the failure is related to either of these checks, please use the following commands to run them locally:

# For mypy (substitute "3.10" with the failing version if needed)
pre-commit run --hook-stage manual mypy-3.10
# For markdownlint
pre-commit run --hook-stage manual markdownlint