[Bugfix] Fix illegal memory access from a forward during a partial wake_up (#44483)

Signed-off-by: Meihan-chen <zr010426ztt@outlook.com>
Signed-off-by: aoshen02 <aoshen@inferact.ai>
Co-authored-by: aoshen02 <aoshen@inferact.ai>
Co-authored-by: Nick Hill <nickhill123@gmail.com>
(cherry picked from commit 93ec645878)

vllm/v1/engine/core.py

@@ -809,8 +809,10 @@ class EngineCore:
         if tags is None or tags:
             self.model_executor.wake_up(tags)
 
-        # Resume scheduling (applies to all levels)
-        self.resume_scheduler()
+        # Partial wakes intentionally keep the remaining allocations asleep.
+        # Resume scheduling only once all executor memory is resident again.
+        if not self.model_executor.is_sleeping:
+            self.resume_scheduler()
 
     def is_sleeping(self) -> bool:
         """Check if engine is sleeping at any level."""
@@ -1947,10 +1949,16 @@ class DPEngineCoreProc(EngineCoreProc):
                     # All engines are idle.
                     continue
 
+                # Execute a dummy pass when no ready requests ran, unless the
+                # engine is sleeping. self.is_sleeping() also covers the KV-offload
+                # window before model_executor.is_sleeping flips.
+                elif not self.is_sleeping():
+                    with self.log_iteration_details(None):
                 # We are in a running state and so must execute a dummy pass
                 # if the model didn't execute any ready requests.
-                with self.log_iteration_details(None):
-                    self.execute_dummy_batch()
+                if not self.model_executor.is_sleeping:
+                    with self.log_iteration_details(None):
+                        self.execute_dummy_batch()
 
             # 3) All-reduce operation to determine global unfinished reqs.
             self.engines_running = self._has_global_unfinished_reqs(