diff --git a/tests/v1/core/test_scheduler.py b/tests/v1/core/test_scheduler.py index c719d1975bba..3f82261a59a7 100644 --- a/tests/v1/core/test_scheduler.py +++ b/tests/v1/core/test_scheduler.py @@ -1307,13 +1307,18 @@ def create_requests_with_priority( mm_positions: Optional[list[PlaceholderRange]] = None, max_tokens: int = 16, stop_token_ids: Optional[list[int]] = None, - prompt_logprobs: Optional[int] = None): + prompt_logprobs: Optional[int] = None, + request_ids: Optional[list[str]] = None): """Create requests with specified priorities and arrival times.""" assert len(priorities) == num_requests if arrival_times is not None: assert len(arrival_times) == num_requests else: arrival_times = [float(i) for i in range(num_requests)] + if request_ids is not None: + assert len(request_ids) == num_requests + else: + request_ids = [f"{i}" for i in range(num_requests)] sampling_params = SamplingParams(ignore_eos=False, max_tokens=max_tokens, @@ -1328,7 +1333,7 @@ def create_requests_with_priority( mm_position = None mm_inputs = None request = Request( - request_id=f"{i}", + request_id=request_ids[i], prompt_token_ids=[i] * num_tokens, sampling_params=sampling_params, pooling_params=None, @@ -1829,3 +1834,91 @@ def test_schedule_skip_tokenizer_init_structured_output_request(): assert len(output.scheduled_new_reqs) == 0 assert len(scheduler.running) == 0 assert len(scheduler.waiting) == 1 + + +def test_priority_scheduling_preemption_victim_iterator_order(): + """Test that the scheduling order is maintained after + preempting lower-priority requests.""" + scheduler = create_scheduler_with_priority( + max_num_batched_tokens=200, + num_blocks=9, + ) + # Add three priority requests first. + priority_requests = create_requests_with_priority( + num_requests=3, + priorities=[3, 4, 5], + arrival_times=[1.0, 2.0, 3.0], + num_tokens=15, + request_ids=["1", "2", "3"], + ) + + for request in priority_requests: + scheduler.add_request(request) + # After scheduling, transfer from the waiting queue to the running queue. + # At this time, 3 blocks have been allocated, and 5 available blocks remain. + output = scheduler.schedule() + + model_output = ModelRunnerOutput( + req_ids=[req.request_id for req in priority_requests], + req_id_to_index={ + req.request_id: i + for i, req in enumerate(priority_requests) + }, + sampled_token_ids=[[15] for _ in priority_requests], + spec_token_ids=None, + logprobs=None, + prompt_logprobs_dict={}, + pooler_output=[], + ) + scheduler.update_from_output(output, model_output) + + # Add tow high priority requests. + high_priority_requests = create_requests_with_priority( + num_requests=2, + priorities=[1, 2], + arrival_times=[4.0, 5.0], + num_tokens=16, + request_ids=["4", "5"], + ) + for request in high_priority_requests: + scheduler.add_request(request) + + # After scheduling, transfer the two high-priority requests from + # the waiting queue to the running queue. + # the IDs of the requests in the running queue are: 1, 2, 3, 4, 5. + # At this time, 3+2 blocks have been allocated, + # and 3 available blocks remain. + output = scheduler.schedule() + + merge_requests = priority_requests + high_priority_requests + + model_output = ModelRunnerOutput( + req_ids=[req.request_id for req in merge_requests], + req_id_to_index={ + req.request_id: i + for i, req in enumerate(merge_requests) + }, + sampled_token_ids=[[1] for _ in merge_requests], + spec_token_ids=None, + logprobs=None, + prompt_logprobs_dict={}, + pooler_output=[], + ) + scheduler.update_from_output(output, model_output) + + # At this time, the request with the lowest priority + # (request.id = 2) will be preempted, freeing up 2 blocks, + # which exactly meets the resource allocation requirements + # for request.id = 4 and request.id = 5. + output = scheduler.schedule() + + # Should schedule the new request without preemption. + assert len(scheduler.running) == 4 # + assert len(scheduler.waiting) == 1 # + + running_priorities = [req.priority for req in scheduler.running] + running_req_ids = [req.request_id for req in scheduler.running] + + assert running_priorities == [3, 4, 1, 2] + assert running_req_ids == ["1", "2", "4", "5"] + assert scheduler.waiting.peek_request().priority == 5 diff --git a/vllm/v1/core/sched/scheduler.py b/vllm/v1/core/sched/scheduler.py index 49a744cfec69..413a853dfecb 100644 --- a/vllm/v1/core/sched/scheduler.py +++ b/vllm/v1/core/sched/scheduler.py @@ -257,7 +257,11 @@ class Scheduler(SchedulerInterface): self.running, key=lambda r: (r.priority, r.arrival_time), ) - self.running.remove(preempted_req) + preempted_index = self.running.index(preempted_req) + if preempted_index <= req_index: + req_index -= 1 + scheduled_running_reqs.remove(preempted_req) + self.running.pop(preempted_index) else: preempted_req = self.running.pop()