@tduncan / Nvidia-Optimization-Stack-VLLM / commits / 42dbe4578c

[Core] Rework scheduler `skipped_waiting` queue (#58947)

Signed-off-by: Nick Hill <nickhill123@gmail.com> Co-authored-by: Robert Shaw <114415538+robertgshaw2-redhat@users.noreply.github.com>

Nick Hill committed Sep 28, 2026 at 13:03 UTC 42dbe4578c7bbeceadc28bd8c6a3826949163f5c

13 files changed +220 -149

rust/src/engine-core-client/src/protocol/stats.rs

+2 -2

@@ -278,9 +278,9 @@ pub use kv_connector::*;

278 pub struct SchedulerStats {

279 /// Number of requests in model execution batches.

280 pub num_running_reqs: u64,

281 - /// Length of the "waiting" request queue.

281 + /// Number of waiting requests not deferred.

282 pub num_waiting_reqs: u64,

283 - /// Length of the "skipped waiting" queue.

283 + /// Number of waiting requests deferred by transient constraints or a blocked status.

284 #[serde(default)]

285 pub num_skipped_waiting_reqs: u64,

286 /// Internal DP load-balancing step counter.

tests/v1/core/test_scheduler.py

+28 -32

@@ -2273,7 +2273,8 @@ def test_spec_decode_padding_skipped_with_prefill_in_batch():

2273

2274

2275 def test_scheduler_stats_waiting_queues():

2276 - """Test that scheduler stats correctly report waiting and skipped_waiting queues."""

2276 + """Test that scheduler stats correctly report capacity-bound and blocked

2277 + waiting requests."""

2278 # Create scheduler with limited capacity so we can have waiting requests

2279 scheduler = create_scheduler(max_num_batched_tokens=100)

2280

@@ -2286,18 +2287,18 @@ def test_scheduler_stats_waiting_queues():

2287 for request in all_requests[:3]:

2288 scheduler.add_request(request)

2289

2289 - # Manually add 2 more to skipped_waiting to simulate constraint-blocked

2290 + # Add 2 more that are blocked on grammar compilation

2291 for request in all_requests[3:]:

2291 - request.status = RequestStatus.WAITING_FOR_REMOTE_KVS

2292 - scheduler.skipped_waiting.add_request(request)

2292 + request.status = RequestStatus.WAITING_FOR_STRUCTURED_OUTPUT_GRAMMAR

2293 + scheduler.add_request(request)

2294

2295 # Schedule - this will schedule 2 requests, leaving 1 in waiting

2296 output = scheduler.schedule()

2297

2298 # Verify: 2 scheduled, 1 still waiting on capacity, 2 blocked by constraints

2299 assert len(output.scheduled_new_reqs) == 2

2299 - assert len(scheduler.waiting) == 1

2300 - assert len(scheduler.skipped_waiting) == 2

2300 + assert len(scheduler.waiting) == 3

2301 + assert len(scheduler.deferred_waiting) == 2

2302

2303 # Call update_from_output() to get frontend-facing stat

2304 scheduled_req_ids = list(output.num_scheduled_tokens.keys())

@@ -2395,7 +2396,7 @@ def _step_until_done(

2396

2397

2398 def _num_waiting_requests(scheduler: Scheduler) -> int:

2398 - return len(scheduler.waiting) + len(scheduler.skipped_waiting)

2399 + return len(scheduler.waiting) + len(scheduler.kv_holding_waiting)

2400

2401

2402 def _step_until_kv_transfer_finished(scheduler: Scheduler, req_ids: list[str]):

@@ -2944,7 +2945,7 @@ def test_kv_connector_handles_preemption(

2945 if is_async:

2946 waiting_req_ids = [

2947 req.request_id

2947 - for req in scheduler.skipped_waiting

2948 + for req in scheduler.kv_holding_waiting

2949 if req.status == RequestStatus.WAITING_FOR_REMOTE_KVS

2950 ]

2951 assert len(waiting_req_ids) == 1

@@ -3807,8 +3808,8 @@ def test_schedule_skip_tokenizer_init_structured_output_request():

3808 output = scheduler.schedule()

3809 assert len(output.scheduled_new_reqs) == 0

3810 assert len(scheduler.running) == 0

3810 - assert len(scheduler.waiting) == 0

3811 - assert len(scheduler.skipped_waiting) == 1

3811 + assert len(scheduler.waiting) == 1

3812 + assert len(scheduler.deferred_waiting) == 1

3813

3814

3815 @pytest.mark.parametrize("async_grammar", [True, False])

@@ -5227,7 +5228,7 @@ def test_prepend_skipped_requests_order():

5228 req.status = RequestStatus.WAITING_FOR_REMOTE_KVS

5229 scheduler.waiting.remove_requests(expected_waiting_reqs[:2])

5230 for req in expected_waiting_reqs[:2]:

5230 - scheduler.skipped_waiting.add_request(req)

5231 + scheduler.kv_holding_waiting.add_request(req)

5232

5233 # schedule step

5234 # expect the first 2 waiting to be skipped, the third running,

@@ -5237,21 +5238,18 @@ def test_prepend_skipped_requests_order():

5238 # pop the third request which is expected to be running

5239 expected_waiting_reqs.pop(2)

5240

5240 - # verify waiting order is preserved

5241 - waiting_reqs = list(scheduler.skipped_waiting) + list(scheduler.waiting)

5241 + # verify waiting order is preserved, with the KV-holding requests first

5242 + waiting_reqs = list(scheduler.kv_holding_waiting) + list(scheduler.waiting)

5243 assert waiting_reqs == expected_waiting_reqs

5244

5245

5245 -def test_remote_kv_promotion_keeps_fcfs_with_grammar_prefix():

5246 +def test_remote_kv_promotion_drains_before_grammar_prefix():

5247 scheduler = create_scheduler(max_num_seqs=1)

5248 scheduler.connector = Mock()

5249 scheduler.connector.get_num_new_matched_tokens.return_value = (0, False)

5250

5251 requests = create_requests(num_requests=4)

5251 - for request in requests:

5252 - scheduler.add_request(request)

5253 -

5254 - req_grammar_1, req_grammar_2, req_remote, req_tail = list(scheduler.waiting)

5252 + req_grammar_1, req_grammar_2, req_remote, req_tail = requests

5253

5254 # simulate two structured-output grammar requests at the waiting head

5255 # that become ready now.

@@ -5260,31 +5258,28 @@ def test_remote_kv_promotion_keeps_fcfs_with_grammar_prefix():

5258 req_grammar_2.status = RequestStatus.WAITING_FOR_STRUCTURED_OUTPUT_GRAMMAR

5259 req_grammar_2.structured_output_request = Mock(grammar=object())

5260

5263 - # simulate a remote-KV request that is ready to be promoted now.

5261 + # simulate a remote-KV request that is ready to be promoted now. Parked

5262 + # loads hold KV blocks, so they drain ahead of the grammar requests.

5263 req_remote.status = RequestStatus.WAITING_FOR_REMOTE_KVS

5265 - scheduler.waiting.remove_requests([req_grammar_1, req_grammar_2, req_remote])

5266 - scheduler.skipped_waiting.add_request(req_grammar_1)

5267 - scheduler.skipped_waiting.add_request(req_grammar_2)

5268 - scheduler.skipped_waiting.add_request(req_remote)

5264 + for request in requests:

5265 + scheduler.add_request(request)

5266 + assert list(scheduler.kv_holding_waiting) == [req_remote]

5267 scheduler.finished_recving_kv_req_ids.add(req_remote.request_id)

5268 scheduler._update_waiting_for_remote_kv = Mock()

5269

5270 output = scheduler.schedule()

5271

5272 assert output.scheduled_new_reqs

5275 - assert output.scheduled_new_reqs[0].req_id == req_grammar_1.request_id

5276 - waiting_req_ids = [

5277 - req.request_id

5278 - for req in list(scheduler.skipped_waiting) + list(scheduler.waiting)

5279 - ]

5273 + assert output.scheduled_new_reqs[0].req_id == req_remote.request_id

5274 + waiting_req_ids = [req.request_id for req in scheduler.waiting]

5275 assert waiting_req_ids == [

5276 + req_grammar_1.request_id,

5277 req_grammar_2.request_id,

5282 - req_remote.request_id,

5278 req_tail.request_id,

5279 ]

5280

5281

5287 -def test_fcfs_mixed_skipped_waiting_types_keep_order():

5282 +def test_fcfs_mixed_blocked_types_remote_load_drains_first():

5283 scheduler = create_scheduler(max_num_batched_tokens=20)

5284 scheduler._update_waiting_for_remote_kv = Mock()

5285

@@ -5305,7 +5300,8 @@ def test_fcfs_mixed_skipped_waiting_types_keep_order():

5300 for req in (req_grammar, req_remote, req_stream, req_regular, req_tail):

5301 scheduler.add_request(req)

5302 scheduler.schedule()

5308 - assert list(scheduler.skipped_waiting) == [req_grammar, req_remote, req_stream]

5303 + assert list(scheduler.waiting) == [req_grammar, req_stream, req_tail]

5304 + assert list(scheduler.kv_holding_waiting) == [req_remote]

5305

5306 scheduler.finish_requests(req_regular.request_id, RequestStatus.FINISHED_ABORTED)

5307 assert not scheduler.running

@@ -5316,8 +5312,8 @@ def test_fcfs_mixed_skipped_waiting_types_keep_order():

5312

5313 second_output = scheduler.schedule()

5314 expected_order = [

5319 - req_grammar.request_id,

5315 req_remote.request_id,

5316 + req_grammar.request_id,

5317 req_stream.request_id,

5318 req_tail.request_id,

5319 ]

tests/v1/core/test_worker_slot_overflow.py

+2 -2

@@ -105,8 +105,8 @@ def test_streaming_pause_does_not_over_admit_worker_slot():

105 # still holds its slot.

106 scheduler.update_from_output(out, _model_runner_output(["session"], [[STOP_TOKEN]]))

107 assert session.status == RequestStatus.WAITING_FOR_STREAMING_REQ

108 - # A blocked waiting status lands in skipped_waiting, not the main queue.

109 - assert session in scheduler.skipped_waiting

108 + # The paused session keeps its KV blocks, so it lands in kv_holding_waiting.

109 + assert session in scheduler.kv_holding_waiting

110 assert scheduler.num_waiting_for_streaming_input == 1

111

112 # A different request arrives while the session is paused.

tests/v1/kv_connector/unit/test_error_propagation.py

+3 -2

@@ -119,7 +119,7 @@ def test_error_propagation_async_load(fail_scheduler: Scheduler):

119

120 scheduler_output = fail_scheduler.schedule()

121

122 - assert len(fail_scheduler.skipped_waiting) == 1

122 + assert len(fail_scheduler.kv_holding_waiting) == 1

123 assert request.status == RequestStatus.WAITING_FOR_REMOTE_KVS

124 assert request.num_computed_tokens == num_external_computed_tokens

125

@@ -145,4 +145,5 @@ def test_error_propagation_async_load(fail_scheduler: Scheduler):

145 assert output.finish_reason == FinishReason.ERROR

146

147 assert len(fail_scheduler.waiting) == 0

148 - assert len(fail_scheduler.skipped_waiting) == 0

148 + assert len(fail_scheduler.kv_holding_waiting) == 0

149 + assert not fail_scheduler.deferred_waiting

tests/v1/kv_connector/unit/test_invalid_blocks_correctness.py

+1 -1

@@ -333,7 +333,7 @@ def test_async_recompute_blocks_not_cached_when_invalid(

333 scheduler_output = recompute_scheduler.schedule()

334

335 # request should be waiting for remote KVs

336 - assert len(recompute_scheduler.skipped_waiting) == 1

336 + assert len(recompute_scheduler.kv_holding_waiting) == 1

337 assert request.status == RequestStatus.WAITING_FOR_REMOTE_KVS

338 assert request.num_computed_tokens == num_external_computed_tokens

339

tests/v1/kv_connector/unit/test_kv_load_failure_recovery.py

+10 -8

@@ -111,8 +111,8 @@ def test_async_load_failure(

111 scheduler_output = scheduler.schedule()

112

113 assert len(scheduler.waiting) == 0

114 - assert len(scheduler.skipped_waiting) == 3

115 - for request in scheduler.skipped_waiting:

114 + assert len(scheduler.kv_holding_waiting) == 3

115 + for request in scheduler.kv_holding_waiting:

116 assert request.num_computed_tokens == num_external_computed_tokens

117 assert request.status == RequestStatus.WAITING_FOR_REMOTE_KVS

118 assert scheduler.connector.get_num_new_matched_tokens.call_count == 3

@@ -132,8 +132,8 @@ def test_async_load_failure(

132 min_invalid_block_idx = min(invalid_block_idxs)

133

134 assert len(scheduler.waiting) == 0

135 - assert len(scheduler.skipped_waiting) == 3

136 - for request in scheduler.skipped_waiting:

135 + assert len(scheduler.kv_holding_waiting) == 3

136 + for request in scheduler.kv_holding_waiting:

137 if request.request_id == request2.request_id:

138 assert request.num_computed_tokens == (

139 min_invalid_block_idx * scheduler.block_size

@@ -340,8 +340,9 @@ def test_async_progressive_load_failure(

340 scheduler_output = scheduler.schedule()

341

342 assert len(scheduler.waiting) == 0

343 - assert len(scheduler.skipped_waiting) == 1

344 - assert scheduler.skipped_waiting.peek_request().request_id == request.request_id

343 + assert len(scheduler.kv_holding_waiting) == 1

344 + peeked = scheduler.kv_holding_waiting.peek_request()

345 + assert peeked.request_id == request.request_id

346 assert request.num_computed_tokens == num_external_computed_tokens

347 assert request.status == RequestStatus.WAITING_FOR_REMOTE_KVS

348 assert scheduler.connector.get_num_new_matched_tokens.call_count == 1

@@ -363,8 +364,9 @@ def test_async_progressive_load_failure(

364 min_invalid_block_idx = min(min_invalid_block_idx, invalid_block_idx)

365

366 assert len(scheduler.waiting) == 0

366 - assert len(scheduler.skipped_waiting) == 1

367 - assert scheduler.skipped_waiting.peek_request().request_id == request.request_id

367 + assert len(scheduler.kv_holding_waiting) == 1

368 + peeked = scheduler.kv_holding_waiting.peek_request()

369 + assert peeked.request_id == request.request_id

370 assert request.num_computed_tokens == (

371 min_invalid_block_idx * scheduler.block_size

372 )

tests/v1/kv_connector/unit/test_nixl_connector.py

+1 -1

@@ -3026,7 +3026,7 @@ def test_empty_recv_is_reported_only_when_awaited(

3026 request is named in neither _recving_transfers nor _failed_recv_reqs and

3027 never reaches finished_recving. The scheduler has no other way to release a

3028 WAITING_FOR_REMOTE_KVS request, and there is no timeout, so it sits in

3029 - skipped_waiting holding its blocks for the life of the process.

3029 + kv_holding_waiting holding its blocks for the life of the process.

3030

3031 Notify-only (awaiting_kvs=False): request_finished seeding an empty recv to

3032 free P's blocks for a request aborted before it was scheduled, or a readback

tests/v1/kv_connector/unit/test_remote_prefill_lifecycle.py

+70 -2

@@ -25,7 +25,7 @@ pytestmark = pytest.mark.cpu_test

27 def _num_waiting_requests(scheduler) -> int:

28 - return len(scheduler.waiting) + len(scheduler.skipped_waiting)

28 + return len(scheduler.waiting) + len(scheduler.kv_holding_waiting)

31 def test_basic_lifecycle():

@@ -64,7 +64,7 @@ def test_basic_lifecycle():

65 # Req waiting for KVs with no computed/scheduled toks ...

66 assert _num_waiting_requests(scheduler) == 1

67 - assert request in scheduler.skipped_waiting

67 + assert request in scheduler.kv_holding_waiting

68 assert request.status == RequestStatus.WAITING_FOR_REMOTE_KVS

69 assert request.num_computed_tokens == NUM_TOKENS

@@ -791,3 +791,71 @@ def test_async_load_reserves_blocks_for_promotion_margin():

791 scheduler_output = scheduler.schedule()

792 assert req_a.status == RequestStatus.RUNNING

793 assert scheduler_output.num_scheduled_tokens[req_a.request_id] > 0

794 +

795 +

796 +def test_deferred_lookup_does_not_block_parked_load():

797 + """A request with a deferred connector lookup must not strand an async

798 + load queued behind it.

799 +

800 + req_a's lookup is deferred (connector returns None), so it is parked in

801 + waiting holding no blocks; req_b is then admitted as an async

802 + load holding most of the pool. Once req_a's lookup resolves it cannot

803 + allocate, and the waiting scan stops at it every step. The parked load

804 + must still be promotable: it lives in the KV-holding queue

805 + that is drained first, so it runs, finishes, and frees its blocks for

806 + req_a.

807 + """

808 + vllm_config = create_vllm_config()

809 + BLOCK_SIZE = vllm_config.cache_config.block_size

810 + scheduler = create_scheduler(vllm_config, num_blocks=8) # usable = 7

811 +

812 + req_a = create_request(

813 + request_id=1, block_size=BLOCK_SIZE, num_tokens=BLOCK_SIZE * 4, max_tokens=1

814 + )

815 + req_b = create_request(

816 + request_id=2,

817 + block_size=BLOCK_SIZE,

818 + num_tokens=BLOCK_SIZE * 5,

819 + do_remote_prefill=True,

820 + max_tokens=1,

821 + )

822 + scheduler.add_request(req_a)

823 + scheduler.add_request(req_b)

824 +

825 + deferrals = [(None, False)]

826 +

827 + def lookup(request, num_computed_tokens):

828 + if request is req_b:

829 + return BLOCK_SIZE * 5, True

830 + return deferrals.pop() if deferrals else (0, False)

831 +

832 + with patch.object(

833 + scheduler.connector, "get_num_new_matched_tokens", side_effect=lookup

834 + ):

835 + # Step 1: req_a's lookup is deferred; req_b is admitted as an async

836 + # load holding 5 of 7 usable blocks.

837 + scheduler.schedule()

838 + assert req_a.status == RequestStatus.WAITING

839 + assert req_b.status == RequestStatus.WAITING_FOR_REMOTE_KVS

840 +

841 + # req_a's lookup resolves (no hit) while req_b's transfer is in

842 + # flight; req_a needs 4 blocks but only 2 are free, so the waiting

843 + # scan stops at it. req_b's transfer then finishes.

844 + scheduler.update_from_output(

845 + scheduler.schedule(),

846 + create_model_runner_output([], finished_recving={req_b.request_id}),

847 + )

848 +

849 + # req_b must still be reached, promoted, and scheduled despite the

850 + # scan stopping at req_a.

851 + scheduler_output = scheduler.schedule()

852 + assert req_b.status == RequestStatus.RUNNING

853 + assert scheduler_output.num_scheduled_tokens[req_b.request_id] > 0

854 +

855 + # req_b finishes and frees its blocks; now req_a fits and runs.

856 + scheduler.update_from_output(

857 + scheduler_output, create_model_runner_output([req_b])

858 + )

859 + scheduler_output = scheduler.schedule()

860 + assert req_a.status == RequestStatus.RUNNING

861 + assert scheduler_output.num_scheduled_tokens[req_a.request_id] > 0

tests/v1/kv_connector/unit/utils.py

@@ -54,6 +54,8 @@ def assert_scheduler_empty(scheduler: Scheduler):

54 # Scheduler Metadata.

55 assert len(scheduler.requests) == 0

56 assert len(scheduler.waiting) == 0

57 + assert len(scheduler.kv_holding_waiting) == 0

58 + assert not scheduler.deferred_waiting

59 assert len(scheduler.running) == 0

60 assert len(scheduler.finished_req_ids) == 0

61 assert len(scheduler.finished_recving_kv_req_ids) == 0

tests/v1/streaming_input/test_scheduler_streaming.py

+3 -3

@@ -554,9 +554,9 @@ class TestStreamingScheduler(unittest.TestCase):

554 eco_cycle2 = eco_dict_cycle2[session.client_index].outputs[0]

555 assert eco_cycle2.finish_reason == FinishReason.STOP

556 assert session.status == RequestStatus.WAITING_FOR_STREAMING_REQ

557 - # Sessions paused for streaming input are blocked-waiting, so they

558 - # live in the skipped_waiting queue rather than the main waiting queue.

559 - assert session in scheduler.skipped_waiting

557 + # Sessions paused for streaming input keep their KV blocks, so they

558 + # live in the kv_holding_waiting queue.

559 + assert session in scheduler.kv_holding_waiting

560 assert session._all_token_ids == [1, 2, 3, 10, STOP_TOKEN]

561

562 # CRITICAL ASSERTION: Cached prompt_token_ids STILL must not have changed

vllm/v1/core/sched/request_queue.py

+3 -3

@@ -41,7 +41,7 @@ class RequestQueue(ABC):

41 pass

43 @abstractmethod

44 - def prepend_requests(self, requests: "RequestQueue") -> None:

44 + def prepend_requests(self, requests: Iterable[Request]) -> None:

45 """Prepend all requests from another queue to the front of this

46 queue."""

47 pass

@@ -93,7 +93,7 @@ class FCFSRequestQueue(deque[Request], RequestQueue):

93 """Prepend a request to the front of the queue."""

94 self.appendleft(request)

96 - def prepend_requests(self, requests: RequestQueue) -> None:

96 + def prepend_requests(self, requests: Iterable[Request]) -> None:

97 """Prepend all requests from another queue to the front of this

98 queue.

@@ -163,7 +163,7 @@ class PriorityRequestQueue(RequestQueue):

163 front. Requests are ordered by (priority, arrival_time)."""

164 self.add_request(request)

165

166 - def prepend_requests(self, requests: RequestQueue) -> None:

166 + def prepend_requests(self, requests: Iterable[Request]) -> None:

167 """Add all requests from another queue according to priority policy.

168

169 Note: In a priority queue, there is no concept of prepending to the

vllm/v1/core/sched/scheduler.py

+92 -91

@@ -210,8 +210,13 @@ class Scheduler(SchedulerInterface):

210 ) from e

211 # Priority queues for requests.

212 self.waiting = create_request_queue(self.policy)

213 - # requests skipped in waiting flow due async deps or constraints.

214 - self.skipped_waiting = create_request_queue(self.policy)

213 + # Waiting requests that hold KV blocks are always drained before

214 + # self.waiting: a block-holder must never sit behind a request whose

215 + # failed allocation would stop the scheduling scan.

216 + self.kv_holding_waiting = create_request_queue(self.policy)

217 + # Waiting requests that are deferred, i.e. were skipped by the scheduling

218 + # scan or enqueued in a blocked status.

219 + self.deferred_waiting: set[Request] = set()

220 self.running: list[Request] = []

221

222 # The request IDs that are finished in between the previous and the

@@ -606,9 +611,8 @@ class Scheduler(SchedulerInterface):

611 # `long_prefill_token_threshold` exists to stop a long prefill from

612 # starving other requests of the token budget. When it is the only

613 # request there is nobody to starve, so let it use the whole budget.

609 - num_eligible_reqs = (

610 - len(self.running) + len(self.waiting) + len(self.skipped_waiting)

611 - )

614 + num_running, num_waiting = self.get_request_counts()

615 + num_eligible_reqs = num_running + num_waiting

616 long_prefill_token_threshold = (

617 self.scheduler_config.long_prefill_token_threshold

618 if num_eligible_reqs > 1

@@ -867,10 +871,21 @@ class Scheduler(SchedulerInterface):

871

872 # Next, schedule the WAITING requests.

873 if not preempted_reqs and self._pause_state == PauseState.UNPAUSED:

870 - step_skipped_waiting = create_request_queue(self.policy)

874 + step_skipped_waiting: deque[Request] = deque()

875 + step_skipped_kv_holding: deque[Request] = deque()

876 +

877 + def skip_request(from_queue: RequestQueue) -> None:

878 + request_to_skip = from_queue.pop_request()

879 + self.deferred_waiting.add(request_to_skip)

880 + if self._holds_kv_blocks(request_to_skip):

881 + step_skipped_kv_holding.appendleft(request_to_skip)

882 + else:

883 + step_skipped_waiting.appendleft(request_to_skip)

884

872 - while (self.waiting or self.skipped_waiting) and token_budget > 0:

873 - if input_budget <= draft_slots:

885 + while token_budget > 0:

886 + # Requests holding KV blocks are always drained first.

887 + request_queue = self.kv_holding_waiting or self.waiting

888 + if not request_queue or input_budget <= draft_slots:

889 break

890 # Paused streaming sessions (WAITING_FOR_STREAMING_REQ) are not

891 # in `running` but still hold a model-runner request slot.

@@ -878,38 +893,22 @@ class Scheduler(SchedulerInterface):

893 if num_running >= self.max_num_active_reqs:

894 break

895

881 - request_queue = self._select_waiting_queue_for_scheduling()

882 - assert request_queue is not None

883 -

896 request = request_queue.peek_request()

897 request_id = request.request_id

898

887 - # try to promote blocked statuses while traversing skipped queue.

888 - if self._is_blocked_waiting_status(

889 - request.status

890 - ) and not self._try_promote_blocked_waiting_request(request):

891 - if request.status == RequestStatus.WAITING_FOR_REMOTE_KVS:

892 - logger.debug(

893 - "%s is still in WAITING_FOR_REMOTE_KVS state.",

894 - request_id,

895 - )

896 - request_queue.pop_request()

897 - step_skipped_waiting.prepend_request(request)

899 + ready_to_schedule = self._handle_blocked_waiting_request(request)

900 + if not ready_to_schedule:

901 + skip_request(request_queue)

902 continue

903

900 - if (

901 - request.num_stale_output_tokens > 0

902 - and not request.drop_stale_output

903 - ):

904 + if request.num_stale_output_tokens and not request.drop_stale_output:

905 # Deliverable stale output still in flight: resuming now

906 # could resample a position that output later delivers.

907 # It drains within the pipeline depth.

907 - request_queue.pop_request()

908 - step_skipped_waiting.prepend_request(request)

908 + skip_request(request_queue)

909 continue

910

911 - # Check that adding the request still respects the max_loras

912 - # constraint.

911 + # Check that adding the request still respects the max_loras constraint.

912 if (

913 self.lora_config

914 and request.lora_request

@@ -919,8 +918,7 @@ class Scheduler(SchedulerInterface):

918 )

919 ):

920 # Scheduling would exceed max_loras, skip.

922 - request_queue.pop_request()

923 - step_skipped_waiting.prepend_request(request)

921 + skip_request(request_queue)

922 continue

923

924 num_external_computed_tokens = 0

@@ -959,8 +957,7 @@ class Scheduler(SchedulerInterface):

957 # The request cannot be scheduled because

958 # the KVConnector couldn't determine

959 # the number of matched tokens.

962 - request_queue.pop_request()

963 - step_skipped_waiting.prepend_request(request)

960 + skip_request(request_queue)

961 continue

962

963 if self.prefix_replay_tokens:

@@ -1030,8 +1027,7 @@ class Scheduler(SchedulerInterface):

1027

1028 # Skip request with pending mm encoding prefetches

1029 if self._ec_transfer_pending(request, num_computed_tokens):

1033 - request_queue.pop_request()

1034 - step_skipped_waiting.prepend_request(request)

1030 + skip_request(request_queue)

1031 continue

1032

1033 # Track first scheduled prefill, not post-preemption repeat prefills

@@ -1052,8 +1048,7 @@ class Scheduler(SchedulerInterface):

1048 num_computed_tokens = request.num_computed_tokens

1049

1050 if self._ec_transfer_pending(request, num_computed_tokens):

1055 - request_queue.pop_request()

1056 - step_skipped_waiting.prepend_request(request)

1051 + skip_request(request_queue)

1052 continue

1053

1054 encoder_inputs_to_schedule = None

@@ -1260,12 +1255,10 @@ class Scheduler(SchedulerInterface):

1255 request, num_new_local_computed_tokens

1256 )

1257

1263 - request = request_queue.pop_request()

1258 if load_kv_async:

1259 # If loading async, allocate memory and put request

1260 # into the WAITING_FOR_REMOTE_KV state.

1261 request.status = RequestStatus.WAITING_FOR_REMOTE_KVS

1268 - step_skipped_waiting.prepend_request(request)

1262 # Set num_computed_tokens even though KVs are not yet loaded.

1263 # request.num_computed_tokens will not be used anywhere until

1264 # the request finished the KV transfer.

@@ -1286,13 +1279,16 @@ class Scheduler(SchedulerInterface):

1279 # overwrite; the zeroing could race the write.

1280 self._skip_zero_block_ids.update(

1281 self.kv_cache_manager.get_zeroing_block_ids_in_range(

1289 - request.request_id,

1282 + request_id,

1283 num_new_local_computed_tokens,

1284 num_computed_tokens,

1285 )

1286 )

1287 + skip_request(request_queue)

1288 continue

1289

1290 + request = request_queue.pop_request()

1291 + self.deferred_waiting.discard(request)

1292 self.running.append(request)

1293 if num_external_computed_tokens > 0:

1294 # load_kv_async is False here

@@ -1343,13 +1339,16 @@ class Scheduler(SchedulerInterface):

1339 self.ec_connector.update_state_after_alloc(request, i)

1340

1341 # re-queue requests skipped in this pass ahead of older skipped items.

1342 + if step_skipped_kv_holding:

1343 + self.kv_holding_waiting.prepend_requests(step_skipped_kv_holding)

1344 if step_skipped_waiting:

1347 - self.skipped_waiting.prepend_requests(step_skipped_waiting)

1345 + self.waiting.prepend_requests(step_skipped_waiting)

1346

1347 # DP prefill balancing: on a step that admitted prefills (release),

1348 # record whether it was capacity-bound.

1349 if not defer_prefills:

1352 - self.prefill_capacity_bound = bool(self.waiting)

1350 + _, num_waiting = self.get_request_counts()

1351 + self.prefill_capacity_bound = num_waiting > len(self.deferred_waiting)

1352

1353 # Check if the scheduling constraints are satisfied.

1354 total_num_scheduled_tokens = sum(num_scheduled_tokens.values())

@@ -2237,7 +2236,8 @@ class Scheduler(SchedulerInterface):

2236 if stopped_preempted_reqs:

2237 # This is a rare case and unlikely to impact performance.

2238 self.waiting.remove_requests(stopped_preempted_reqs)

2240 - self.skipped_waiting.remove_requests(stopped_preempted_reqs)

2239 + self.kv_holding_waiting.remove_requests(stopped_preempted_reqs)

2240 + self.deferred_waiting.difference_update(stopped_preempted_reqs)

2241

2242 error_req_ids = set(self.grammar_compile_error_reqs)

2243 self.grammar_compile_error_reqs.clear()

@@ -2381,24 +2381,27 @@ class Scheduler(SchedulerInterface):

2381 RequestStatus.WAITING_FOR_STREAMING_REQ,

2382 )

2383

2384 + @staticmethod

2385 + def _holds_kv_blocks(request: Request) -> bool:

2386 + # Whether a request currently has kv blocks allocated.

2387 + # While always the case when status is WAITING_FOR_REMOTE_KVS

2388 + # or WAITING_FOR_STREAMING_REQ, there are such cases where it's

2389 + # WAITING or PREEMPTED.

2390 + return (

2391 + request.num_computed_tokens > 0

2392 + # num_computed_tokens is reset to 0 if kv transfer fails for requests in

2393 + # WAITING_FOR_REMOTE_KVS state, before their allocated blocks are freed.

2394 + or request.status == RequestStatus.WAITING_FOR_REMOTE_KVS

2395 + )

2396 +

2397 def _enqueue_waiting_request(self, request: Request) -> None:

2398 if self._is_blocked_waiting_status(request.status):

2386 - self.skipped_waiting.add_request(request)

2399 + self.deferred_waiting.add(request)

2400 + if self._holds_kv_blocks(request):

2401 + self.kv_holding_waiting.add_request(request)

2402 else:

2403 self.waiting.add_request(request)

2404

2390 - def _select_waiting_queue_for_scheduling(self) -> RequestQueue | None:

2391 - if self.policy == SchedulingPolicy.FCFS:

2392 - return self.skipped_waiting or self.waiting or None

2393 -

2394 - # PRIORITY mode: compare queue heads when both queues are non-empty.

2395 - if self.waiting and self.skipped_waiting:

2396 - waiting_req = self.waiting.peek_request()

2397 - skipped_req = self.skipped_waiting.peek_request()

2398 - return self.waiting if waiting_req < skipped_req else self.skipped_waiting

2399 -

2400 - return self.waiting or self.skipped_waiting or None

2401 -

2405 def _handle_stopped_request(self, request: Request) -> bool:

2406 """Return True if finished (can be False for resumable requests)."""

2407 if not request.resumable:

@@ -2544,7 +2547,7 @@ class Scheduler(SchedulerInterface):

2547

2548 def get_request_counts(self) -> tuple[int, int]:

2549 """Returns (num_running_reqs, num_waiting_reqs)."""

2547 - return len(self.running), len(self.waiting) + len(self.skipped_waiting)

2550 + return len(self.running), len(self.waiting) + len(self.kv_holding_waiting)

2551

2552 def get_kv_cache_usage(self) -> float:

2553 """Returns the fraction of the KV cache currently in use (0.0-1.0)."""

@@ -2625,7 +2628,8 @@ class Scheduler(SchedulerInterface):

2628 self.running = remove_all(self.running, running_requests_to_remove)

2629 if waiting_requests_to_remove:

2630 self.waiting.remove_requests(waiting_requests_to_remove)

2628 - self.skipped_waiting.remove_requests(waiting_requests_to_remove)

2631 + self.kv_holding_waiting.remove_requests(waiting_requests_to_remove)

2632 + self.deferred_waiting.difference_update(waiting_requests_to_remove)

2633

2634 # Second pass: set status and free requests

2635 for request in valid_requests:

@@ -2733,14 +2737,11 @@ class Scheduler(SchedulerInterface):

2737 def get_num_unfinished_requests(self) -> int:

2738 if self._pause_state == PauseState.PAUSED_ALL:

2739 return 0

2740 + num_running, num_waiting = self.get_request_counts()

2741 if self._pause_state == PauseState.PAUSED_NEW:

2737 - return len(self.running)

2738 - num_waiting = (

2739 - len(self.waiting)

2740 - + len(self.skipped_waiting)

2741 - - self.num_waiting_for_streaming_input

2742 - )

2743 - return num_waiting + len(self.running)

2742 + return num_running

2743 + num_waiting -= self.num_waiting_for_streaming_input

2744 + return num_waiting + num_running

2745

2746 def has_finished_requests(self) -> bool:

2747 if self.finished_req_ids:

@@ -2749,10 +2750,8 @@ class Scheduler(SchedulerInterface):

2750 return False

2751 # Finished requests waiting on delayed connector cleanup remain in

2752 # self.requests after they have been removed from scheduling queues.

2752 - num_in_queues = (

2753 - len(self.waiting) + len(self.skipped_waiting) + len(self.running)

2754 - )

2755 - return len(self.requests) > num_in_queues

2753 + num_running, num_waiting = self.get_request_counts()

2754 + return len(self.requests) > (num_running + num_waiting)

2755

2756 def has_requests(self) -> bool:

2757 # Override the interface default to also keep the engine alive while a

@@ -2884,10 +2883,12 @@ class Scheduler(SchedulerInterface):

2883 ec_connector_stats_payload = (

2884 ec_connector_stats.data if ec_connector_stats else None

2885 )

2886 + num_running, num_waiting = self.get_request_counts()

2887 + num_deferred = len(self.deferred_waiting)

2888 return SchedulerStats(

2888 - num_running_reqs=len(self.running),

2889 - num_waiting_reqs=len(self.waiting),

2890 - num_skipped_waiting_reqs=len(self.skipped_waiting),

2889 + num_running_reqs=num_running,

2890 + num_waiting_reqs=num_waiting - num_deferred,

2891 + num_skipped_waiting_reqs=num_deferred,

2892 kv_cache_usage=self.kv_cache_manager.usage,

2893 prefix_cache_stats=prefix_cache_stats,

2894 connector_prefix_cache_stats=connector_prefix_cache_stats,

@@ -3093,39 +3094,39 @@ class Scheduler(SchedulerInterface):

3094 request.num_computed_tokens = request.num_tokens - 1

3095 self.finished_recving_kv_req_ids.remove(request.request_id)

3096

3096 - def _try_promote_blocked_waiting_request(self, request: Request) -> bool:

3097 - """Try to promote a blocked waiting request back to schedulable states."""

3097 + def _handle_blocked_waiting_request(self, request: Request) -> bool:

3098 + """Returns True if scheduling of this request should proceed, False if

3099 + it should be skipped for this step."""

3100 if request.status == RequestStatus.WAITING_FOR_REMOTE_KVS:

3099 - # finished_recving_kv_req_ids is populated during

3100 - # update_from_output(), based on worker-side connector signals

3101 - # in KVConnectorOutput.finished_recving

3102 - if request.request_id not in self.finished_recving_kv_req_ids:

3101 + # Check whether async loading requests are finished.

3102 + request_id = request.request_id

3103 + if request_id not in self.finished_recving_kv_req_ids:

3104 + logger.debug("%s is still in WAITING_FOR_REMOTE_KVS state.", request_id)

3105 return False

3106 + # Promote request to waiting or preempted.

3107 self._update_waiting_for_remote_kv(request)

3108 if request.num_preemptions:

3109 request.status = RequestStatus.PREEMPTED

3110 else:

3111 request.status = RequestStatus.WAITING

3109 - return True

3112

3111 - if request.status == RequestStatus.WAITING_FOR_STRUCTURED_OUTPUT_GRAMMAR:

3112 - structured_output_req = request.structured_output_request

3113 - if not structured_output_req or structured_output_req.grammar is None:

3113 + elif request.status == RequestStatus.WAITING_FOR_STRUCTURED_OUTPUT_GRAMMAR:

3114 + # Check whether grammar compilation has finished.

3115 + struct_output_req = request.structured_output_request

3116 + if not struct_output_req or struct_output_req.grammar is None:

3117 return False

3115 - if isinstance(structured_output_req.grammar, Exception):

3118 + if isinstance(struct_output_req.grammar, Exception):

3119 self.grammar_compile_error_reqs.add(request.request_id)

3120 return False

3121 + # Grammar is ready, promote to waiting state.

3122 request.status = RequestStatus.WAITING

3119 - return True

3123

3121 - if request.status == RequestStatus.WAITING_FOR_STREAMING_REQ:

3124 + elif request.status == RequestStatus.WAITING_FOR_STREAMING_REQ:

3125 + # Streaming input request waiting for more input.

3126 assert not request.streaming_queue

3127 return False

3128

3125 - raise AssertionError(

3126 - "Unexpected blocked waiting status in promotion: "

3127 - f"{request.status.name} for request {request.request_id}"

3128 - )

3129 + return True

3130

3131 def _update_from_kv_xfer_finished(self, kv_connector_output: KVConnectorOutput):

3132 """KV Connector: update the scheduler state based on the output.

@@ -3298,7 +3299,7 @@ class Scheduler(SchedulerInterface):

3299 # handle async KV loads (not cached yet, evict_blocks=False)

3300 async_load_reqs = (

3301 req

3301 - for req in self.skipped_waiting

3302 + for req in self.kv_holding_waiting

3303 if req.status == RequestStatus.WAITING_FOR_REMOTE_KVS

3304 )

3305 async_failed_req_ids, num_failed_tokens, _ = (

vllm/v1/metrics/stats.py

+3 -2

@@ -188,8 +188,9 @@ class SchedulerStats:

188

189 num_running_reqs: int = 0

190

191 - num_waiting_reqs: int = 0 # length of the "waiting" request queue

192 - num_skipped_waiting_reqs: int = 0 # length of the "skipped waiting" queue

191 + num_waiting_reqs: int = 0 # waiting requests not deferred

192 + # waiting requests deferred by transient constraints or a blocked status

193 + num_skipped_waiting_reqs: int = 0

194

195 # These are used for internal DP load-balancing.

196 step_counter: int = 0