diff --git a/python/sglang/srt/managers/prefill_delayer.py b/python/sglang/srt/managers/prefill_delayer.py index d775a35c9..eacae42a7 100644 --- a/python/sglang/srt/managers/prefill_delayer.py +++ b/python/sglang/srt/managers/prefill_delayer.py @@ -244,13 +244,25 @@ class PrefillDelayer: self.skip_first_delayer = False pass else: - next_state = prev_state or _State() - next_state = next_state.bump_delayed_count() + # Bound the wait like the "mixed" branch: on a saturated + # engine slot_condition may never turn false, so cap the + # delay by max_delay_passes. + prev_delayed_count = prev_state.delayed_count if prev_state else 0 + if prev_delayed_count < self._max_delay_passes - 1: + next_state = prev_state or _State() + next_state = next_state.bump_delayed_count() + return _NegotiateOutput( + next_state=next_state, + output_allow=False, + output_reason="delay", + **debug_info, + ) return _NegotiateOutput( - next_state=next_state, - output_allow=False, - output_reason="delay", + next_state=None, + output_allow=True, + output_reason="wait_timeout", **debug_info, + **wait_info, ) exist_previous_wait = prev_state is not None return _NegotiateOutput( diff --git a/python/sglang/srt/managers/scheduler.py b/python/sglang/srt/managers/scheduler.py index 573627ecd..6a4a0880a 100644 --- a/python/sglang/srt/managers/scheduler.py +++ b/python/sglang/srt/managers/scheduler.py @@ -1141,7 +1141,7 @@ class Scheduler( self.schedule_low_priority_values_first, ) self.prefill_delayer: Optional[PrefillDelayer] = None - self.max_prefill_bs: int = 0 + self.max_prefill_bs: float = 0.0 if get_schedule().enable_prefill_delayer: if get_disagg().disaggregation_mode == "decode": logger.info( @@ -3010,6 +3010,11 @@ class Scheduler( def get_new_batch_prefill(self, running_batch: ScheduleBatch) -> NextBatchPlan: prefill_delayer_single_pass = None if self.prefill_delayer: + # Decay the max-prefill-bs high-watermark once per pass so one + # unusually large admission burst does not permanently raise the + # slot_condition bar in the delayer (0.998/pass ~= half-life of + # ~350 forward passes). + self.max_prefill_bs *= 0.998 # Get max usage across all pools for prefill delay decision max_pool_usage = ( self.pool_stats_observer.get_pool_stats().get_max_pool_usage() @@ -3104,7 +3109,7 @@ class Scheduler( chunked_prefill_size, running_bs if self.is_mixed_chunk else 0, self.priority_scheduling_preemption_threshold, - max_prefill_bs=self.max_prefill_bs, + max_prefill_bs=int(self.max_prefill_bs), max_running_requests=self.max_running_requests, prefill_max_requests=get_schedule().prefill_max_requests, prefill_delayer_single_pass=prefill_delayer_single_pass, diff --git a/test/registered/scheduler/test_prefill_delayer.py b/test/registered/scheduler/test_prefill_delayer.py index 58074fbce..64ce40841 100644 --- a/test/registered/scheduler/test_prefill_delayer.py +++ b/test/registered/scheduler/test_prefill_delayer.py @@ -371,6 +371,53 @@ _NEGOTIATE_TEST_CASES = [ # One queue-trigger delay was recorded before the wall-clock release. expected_wait_forward_passes=1, ), + # slot_condition (all-branch) must not delay forever: with 128-100=28 + # free slots < max_prefill_bs=80 the delay holds, but it must release + # with wait_timeout after max_delay_passes, like the mixed branch. + NegotiateTestCase( + name="slot_condition_pass_cap_timeout", + max_delay_passes=3, + token_usage_low_watermark=0.8, + calls=[ + # skip_first_delayer consumes the first would-be delay. + NegotiateCall( + prefillable=[True, True, True, True], + token_usage=[0.9, 0.9, 0.9, 0.9], + running_batch=[100, 100, 100, 100], + max_prefill_bs=[80, 80, 80, 80], + waiting_queue_len=[10, 10, 10, 10], + max_running_requests=128, + ), + NegotiateCall( + prefillable=[True, True, True, True], + token_usage=[0.9, 0.9, 0.9, 0.9], + running_batch=[100, 100, 100, 100], + max_prefill_bs=[80, 80, 80, 80], + waiting_queue_len=[10, 10, 10, 10], + max_running_requests=128, + ), + NegotiateCall( + prefillable=[True, True, True, True], + token_usage=[0.9, 0.9, 0.9, 0.9], + running_batch=[100, 100, 100, 100], + max_prefill_bs=[80, 80, 80, 80], + waiting_queue_len=[10, 10, 10, 10], + max_running_requests=128, + ), + NegotiateCall( + prefillable=[True, True, True, True], + token_usage=[0.9, 0.9, 0.9, 0.9], + running_batch=[100, 100, 100, 100], + max_prefill_bs=[80, 80, 80, 80], + waiting_queue_len=[10, 10, 10, 10], + max_running_requests=128, + ), + ], + expected_allow=True, + expected_reason="wait_timeout", + # Two slot-condition delays accumulated after the skip-first pass. + expected_wait_forward_passes=2, + ), ]