diff --git a/vllm/v1/core/sched/scheduler.py b/vllm/v1/core/sched/scheduler.py index f62cc24f22..5a71e7333f 100644 --- a/vllm/v1/core/sched/scheduler.py +++ b/vllm/v1/core/sched/scheduler.py @@ -56,10 +56,12 @@ from vllm.v1.core.sched.request_queue import ( from vllm.v1.core.sched.utils import check_stop, remove_all from vllm.v1.engine import EngineCoreEventType, EngineCoreOutput, EngineCoreOutputs from vllm.v1.kv_cache_interface import ( + CircularBufferSpec, KVCacheConfig, MambaSpec, get_mamba_prefill_checkpoint_position, is_mamba_prefill_checkpoint_valid, + iter_layer_specs, ) from vllm.v1.metrics.perf import ModelMetrics, PerfStats from vllm.v1.metrics.stats import ( @@ -192,6 +194,26 @@ class Scheduler(SchedulerInterface): self.requires_kv_delivery = self.connector.requires_kv_delivery + # Ring-skip telemetry: the QSA ring block (CircularBufferSpec) is rewritten + # by every step that touches the request, so with overlapping batches + # (async scheduling or PP) an in-flight step may still be committing a + # freed/preempted request's suffix rows when the scheduler hands that + # block to another request. A KV connector is not the only writer that + # can race the stale in-flight write; the next request's own first + # step is. Defer the return of freed blocks to the pool until the + # newest scheduled step's output has been processed whenever a ring + # group exists. Same mechanism and fence as the connector gate above. + if ( + not self.defer_block_free + and self.vllm_config.max_concurrent_batches > 1 + and any( + isinstance(spec, CircularBufferSpec) + for group in kv_cache_config.kv_cache_groups + for spec in iter_layer_specs(group.kv_cache_spec) + ) + ): + self.defer_block_free = True + self.kv_event_publisher = EventPublisherFactory.create( self.kv_events_config, self.parallel_config.data_parallel_index, @@ -816,6 +838,7 @@ class Scheduler(SchedulerInterface): else: preempted_req = self.running.pop() + num_deferred_before = len(self.deferred_frees) self._preempt_request( preempted_req, scheduled_timestamp, @@ -825,6 +848,16 @@ class Scheduler(SchedulerInterface): if preempted_req == request: # No more request to preempt. Cannot schedule this request. break + if len(self.deferred_frees) > num_deferred_before: + # The ring-skip fix: the victim's blocks are fenced + # behind its in-flight step and return to the pool only + # when that step's output is processed, so retrying the + # allocation now cannot succeed on their account and would + # preempt the next request for nothing -- with every running + # request in flight (PP / async) the whole batch. Stop at one + # deferred victim; this request stays running, unscheduled, + # and retries next step after the drain. + break if new_blocks is None: # Cannot schedule this request.