[Bugfix] Fix DP wave race condition re-arming engine while paused

[AjAnubolu]

Gate START_DP_WAVE on scheduler pause state to prevent race with collective_rpc.

Closes #36594

[gemini-code-assist]

Code Review

This pull request addresses a race condition in the data parallel engine by preventing the engine from starting a new wave if the scheduler is paused. The change in vllm/v1/engine/core.py adds a check for the scheduler's paused state before setting self.engines_running to True. This correctly gates the start of a new DP wave and prevents potential hangs during collective RPC calls when the system is in a paused state, for example during scaling operations. The fix is targeted and looks correct.

_{Note: Security Review did not run due to the size of the PR.}

[hao-aaron]

Hi @AjAnubolu, thanks for your contribution, appreciate the fix! Just wanted to check something, I am concerned about potential deadlock in the case that pause reqs to each engine are delayed:

Engine 0 (idle)          Coordinator            Engine 1 (idle)
     |                       |                       |
     |                       |                receives pause and return
     |                       |                       |
 recv req/wave boundary----->|---- START_DP_WAVE --> | engines running = False, never steps
     |                       |                       |
 Enter all reduce, deadlock  |                       |
     |                       |                       |
recv pause req, add          |                       |
to queue but never processed |                       |
     |                       |                       |

has this been tested?

EDIT: appears like this test fails

vllm/tests/v1/distributed/test_async_llm_dp.py

Line 348 in 6341d43

async def test_dp_pause_keep_race_staggered_engines():