Repository: vllm-project/vllm
File: vllm/v1/engine/core.py (Python)

Code before:

        # Reset the scheduler's encoder cache manager (logical state)
        self.scheduler.reset_encoder_cache()
        # Reset the GPU model runner's encoder cache (physical storage)
        self.model_executor.reset_encoder_cache()

    def pause_scheduler(
        self, mode: PauseMode = "abort", clear_cache: bool = True
    ) -> Future | None:
        """Pause generation; behavior depends on mode.

        All pause states queue new adds. PAUSE_ABORT and PAUSE_KEEP skip step();
        PAUSE_WAIT allows step() so in-flight requests can drain.

        - ``abort``: Set PAUSE_ABORT, abort all requests, wait for abort
          outputs to be sent (when running with output_queue), clear caches,
          then complete the returned Future.
        - ``wait``: Set PAUSE_WAIT (queue adds, keep stepping); when drained,
          set PAUSE_KEEP, clear caches, complete the returned Future.
        - ``keep``: Set PAUSE_KEEP; return a Future that completes when the
          output queue is empty.
        """
        if mode not in ("keep", "abort", "wait"):
            raise ValueError(f"Invalid pause mode: {mode}")

        output_queue = getattr(self, "output_queue", None)
        if output_queue is None and mode == "wait":
            raise ValueError("Cannot use 'wait' pause mode with inline-engine mode")

        def wait_until_idle(engine: "EngineCore", future: Future[Any] | None) -> bool:
            if engine.scheduler.has_unfinished_requests() or engine.batch_queue:
                return False
            if output_queue is not None and not output_queue.empty():
                return False
            if clear_cache:
                engine.reset_prefix_cache(reset_running_requests=True)
                engine.reset_mm_cache()
                engine.reset_encoder_cache()
            if future is not None:
                future.set_result(None)
            return True

        if mode == "abort":
            aborted_reqs = self.scheduler.finish_requests(
                None, RequestStatus.FINISHED_ABORTED
            )
            if output_queue is not None:
                self._send_abort_outputs(output_queue, aborted_reqs)

        pause_state = PauseState.PAUSED_ALL if mode == "keep" else PauseState.PAUSED_NEW
        self.scheduler.set_pause_state(pause_state)

Review comment (refactor):
In theory, this suggests we should delegate some of this to `EngineCoreProc`

Here's Claude's suggestion for `EngineCore` hooks with `EngineCoreProc` overrides:

```python
def _validate_pause_mode(self, mode: PauseMode) -> None:
    """Validate pause mode compatibility. Override in subclasses.

    Base implementation: 'wait' mode not supported in inline mode.
    """
    if mode == "wait":
        raise ValueError("Cannot use 'wait' pause mode with inline-engine mode")

def _is_async_idle(self) -> bool:
    """Check if async operations are idle (e.g., output queue empty).

    Base implementation: Always True (no async ops in inline mode).
    Override in EngineCoreProc to check output_queue.
    """
    return True

def _send_pause_abort_outputs(self, aborted_reqs: list[tuple[str, int]]) -> None:
    """Send abort outputs for paused requests.

    Base implementation: No-op (inline mode handles outputs synchronously).
    Override in EngineCoreProc to send via output_queue.
    """
    pass

def _can_wait_async(self) -> bool:
    """Check if engine supports async waiting via per_step_hooks.

    Base implementation: False (inline mode runs synchronously).
    Override in EngineCoreProc to return True.
    """
    return False
```

with `EngineCoreProc` overrides:

```python
def _validate_pause_mode(self, mode: PauseMode) -> None:
    """Validate pause mode - all modes supported in process mode."""
    pass  # All modes work with output_queue

def _is_async_idle(self) -> bool:
    """Check if output queue is empty."""
    return self.output_queue.empty()

def _send_pause_abort_outputs(self, aborted_reqs: list[tuple[str, int]]) -> None:
    """Send abort outputs via output_queue."""
    self._send_abort_outputs(self.output_queue, aborted_reqs)

def _can_wait_async(self) -> bool:
    """Process mode can wait asynchronously via run_busy_loop."""
    return True
```

The result is certainly meh, but I think I hate it less than the abstraction violation


Code after:

        # Reset the scheduler's encoder cache manager (logical state)
        self.scheduler.reset_encoder_cache()
        # Reset the GPU model runner's encoder cache (physical storage)
        self.model_executor.reset_encoder_cache()

    def _reset_caches(self, reset_running_requests=True) -> None:
        self.reset_prefix_cache(reset_running_requests=reset_running_requests)
        self.reset_mm_cache()
        self.reset_encoder_cache()

    def pause_scheduler(
        self, mode: PauseMode = "abort", clear_cache: bool = True
    ) -> Future | None:
        """Pause generation; behavior depends on mode.

        All pause modes queue new adds -- "abort" and "keep" skip step();
        "wait" allows step() so in-flight requests can drain.

        - ``abort``: Set PAUSED_NEW, abort all requests, wait for abort
          outputs to be sent (when running with output_queue), optionally
          clear caches, then complete the returned Future.
        - ``wait``: Set PAUSED_NEW (queue adds, keep stepping); when drained,
          optionally clear caches, then complete the returned Future.
        - ``keep``: Set PAUSED_ALL; return a Future that completes when the
          output queue is empty.
        """
        if mode not in ("keep", "abort", "wait"):
            raise ValueError(f"Invalid pause mode: {mode}")
        if mode == "wait":
            raise ValueError("'wait' mode can't be used in inproc-engine mode")

        if mode == "abort":
            self.scheduler.finish_requests(None, RequestStatus.FINISHED_ABORTED)

        pause_state = PauseState.PAUSED_ALL if mode == "keep" else PauseState.PAUSED_NEW
        self.scheduler.set_pause_state(pause_state)
        if clear_cache:
            self._reset_caches()

        return None

    def resume_scheduler(self) -> None:
        """Resume the scheduler and flush any requests queued while paused."""
        self.scheduler.set_pause_state(PauseState.UNPAUSED)

    def is_scheduler_paused(self) -> bool:
        """Return whether the scheduler is in any pause state."""
        return self.scheduler.pause_state != PauseState.UNPAUSED

    def sleep(self, level: int = 1, mode: PauseMode = "abort") -> None | Future: