From 8b46f1f4ecd22e08eb3e33e0665aecf5e43bd38b Mon Sep 17 00:00:00 2001 From: Shangming Cai Date: Wed, 18 Mar 2026 16:02:20 +0800 Subject: [PATCH] [PD] Add retry interval in ensure_prefill_info (#20832) Signed-off-by: Shangming Cai --- python/sglang/srt/disaggregation/decode.py | 18 +++++++++++++++++- 1 file changed, 17 insertions(+), 1 deletion(-) diff --git a/python/sglang/srt/disaggregation/decode.py b/python/sglang/srt/disaggregation/decode.py index 0ede52313..b64fa4ff9 100644 --- a/python/sglang/srt/disaggregation/decode.py +++ b/python/sglang/srt/disaggregation/decode.py @@ -21,6 +21,7 @@ Life cycle of a request in the decode server from __future__ import annotations import logging +import time from collections import deque from dataclasses import dataclass from http import HTTPStatus @@ -271,7 +272,9 @@ class DecodePreallocQueue: self.retracted_queue: List[Req] = [] self.pending_reqs: List[Req] = [] self._ensure_retry_count: Dict[str, int] = {} - self._max_ensure_retries: int = 30 # scheduling cycles + self._max_ensure_retries: int = 20 # scheduling cycles + self._ensure_last_attempt_time: Dict[str, float] = {} + self._ensure_retry_interval: float = 1.0 # seconds self.kv_manager = self._init_kv_manager() if self.scheduler.tp_worker.is_hybrid_swa: @@ -509,10 +512,22 @@ class DecodePreallocQueue: ready: Dict[str, List[Req]] = {} remaining: List[Req] = [] + now = time.monotonic() for bootstrap_addr, reqs in addr_to_reqs.items(): + last_attempt = self._ensure_last_attempt_time.get(bootstrap_addr) + if last_attempt is not None and ( + now - last_attempt < self._ensure_retry_interval + ): + remaining.extend(reqs) + continue + + self._ensure_last_attempt_time[bootstrap_addr] = now + if self.kv_manager.try_ensure_parallel_info(bootstrap_addr): if bootstrap_addr in self._ensure_retry_count: del self._ensure_retry_count[bootstrap_addr] + if bootstrap_addr in self._ensure_last_attempt_time: + del self._ensure_last_attempt_time[bootstrap_addr] ready[bootstrap_addr] = reqs continue @@ -530,6 +545,7 @@ class DecodePreallocQueue: self.scheduler.metrics_collector.increment_bootstrap_failed_reqs() self.scheduler.stream_output([req], req.return_logprob) del self._ensure_retry_count[bootstrap_addr] + del self._ensure_last_attempt_time[bootstrap_addr] else: remaining.extend(reqs)