From 8ff0c9fef90ef28676c1e4b2e1662992601a32d4 Mon Sep 17 00:00:00 2001 From: Shangming Cai Date: Mon, 8 Jun 2026 16:56:46 +0800 Subject: [PATCH] [PD] Downgrade propagated rank failure logs from error to debug (#27534) Signed-off-by: Shangming Cai --- .../sglang/srt/disaggregation/common/conn.py | 16 +++++++++ python/sglang/srt/disaggregation/decode.py | 16 +++++++-- .../srt/disaggregation/mooncake/conn.py | 33 +++++++++---------- python/sglang/srt/disaggregation/mori/conn.py | 23 ++++++++----- python/sglang/srt/disaggregation/nixl/conn.py | 16 +++++++-- python/sglang/srt/disaggregation/prefill.py | 16 +++++++-- 6 files changed, 87 insertions(+), 33 deletions(-) diff --git a/python/sglang/srt/disaggregation/common/conn.py b/python/sglang/srt/disaggregation/common/conn.py index b4ed4f0db..f025bbd7a 100644 --- a/python/sglang/srt/disaggregation/common/conn.py +++ b/python/sglang/srt/disaggregation/common/conn.py @@ -49,6 +49,22 @@ from sglang.srt.utils.network import ( logger = logging.getLogger(__name__) +class KVTransferError(Exception): + def __init__( + self, + bootstrap_room: int, + failure_reason: str, + is_from_another_rank: bool = False, + ): + super().__init__(failure_reason) + self.bootstrap_room = bootstrap_room + self.failure_reason = failure_reason + self.is_from_another_rank = is_from_another_rank + + def __str__(self): + return f"KVTransferError(bootstrap_room={self.bootstrap_room}): {self.failure_reason}" + + @dataclasses.dataclass class PrefillServerInfo: # Topology fields (fetched from bootstrap server) diff --git a/python/sglang/srt/disaggregation/decode.py b/python/sglang/srt/disaggregation/decode.py index b53302cb6..e9efdcdd9 100644 --- a/python/sglang/srt/disaggregation/decode.py +++ b/python/sglang/srt/disaggregation/decode.py @@ -643,11 +643,17 @@ class DecodePreallocQueue(DecodeHiCachePreallocMixin): decode_req.req.time_stats.set_bootstrap_done_time() elif poll == KVPoll.Failed: error_message = f"Decode handshake failed for request rank={self.tp_rank} {decode_req.req.rid=} {decode_req.req.bootstrap_room=}" + is_propagated = False try: decode_req.kv_receiver.failure_exception() except Exception as e: error_message += f" with exception {e}" - logger.error(error_message) + is_propagated = getattr(e, "is_from_another_rank", False) + # Mute error message for propagated exceptions to avoid duplicate logging + if is_propagated: + logger.debug(error_message) + else: + logger.error(error_message) prepare_abort( decode_req.req, error_message, @@ -1609,13 +1615,19 @@ class DecodeTransferQueue(DecodeHiCacheTransferMixin): f"Decode transfer failed for request rank={self.tp_rank} " f"{decode_req.req.rid=} {decode_req.req.bootstrap_room=}" ) + is_propagated = False if poll == KVPoll.Failed: try: decode_req.kv_receiver.failure_exception() except Exception as e: error_message += f" with exception {e}" + is_propagated = getattr(e, "is_from_another_rank", False) self._clean_hicache_prefetch_resources(decode_req) - logger.error(error_message) + # Mute error message for propagated exceptions to avoid duplicate logging + if is_propagated: + logger.debug(error_message) + else: + logger.error(error_message) prepare_abort( decode_req.req, error_message, diff --git a/python/sglang/srt/disaggregation/mooncake/conn.py b/python/sglang/srt/disaggregation/mooncake/conn.py index ccae5cc41..b21aee9f7 100644 --- a/python/sglang/srt/disaggregation/mooncake/conn.py +++ b/python/sglang/srt/disaggregation/mooncake/conn.py @@ -20,6 +20,7 @@ from sglang.srt.disaggregation.common.conn import ( CommonKVManager, CommonKVReceiver, CommonKVSender, + KVTransferError, ) from sglang.srt.disaggregation.common.staging_handler import ( DecodeStagingContext, @@ -62,16 +63,6 @@ FAILED_SESSION_RECOVERIES = Counter( ) -class KVTransferError(Exception): - def __init__(self, bootstrap_room: int, failure_reason: str): - super().__init__(failure_reason) - self.bootstrap_room = bootstrap_room - self.failure_reason = failure_reason - - def __str__(self): - return f"KVTransferError(bootstrap_room={self.bootstrap_room}): {self.failure_reason}" - - # decode @dataclasses.dataclass class TransferInfo: @@ -1747,10 +1738,13 @@ class MooncakeKVSender(CommonKVSender): self.clear() with self.kv_mgr.failure_lock: - failure_reason = self.kv_mgr.failure_records.pop( - self.bootstrap_room, "Failed due to an unknown reason from another rank" - ) - raise KVTransferError(self.bootstrap_room, failure_reason) + failure_reason = self.kv_mgr.failure_records.pop(self.bootstrap_room, None) + is_propagated = failure_reason is None + if is_propagated: + failure_reason = "Failed due to an unknown reason from another rank" + raise KVTransferError( + self.bootstrap_room, failure_reason, is_from_another_rank=is_propagated + ) def _init_trace_ctx(self): if self.kv_mgr.enable_trace: @@ -1908,10 +1902,13 @@ class MooncakeKVReceiver(CommonKVReceiver): self.clear() with self.kv_mgr.failure_lock: - failure_reason = self.kv_mgr.failure_records.pop( - self.bootstrap_room, "Failed due to an unknown reason from another rank" - ) - raise KVTransferError(self.bootstrap_room, failure_reason) + failure_reason = self.kv_mgr.failure_records.pop(self.bootstrap_room, None) + is_propagated = failure_reason is None + if is_propagated: + failure_reason = "Failed due to an unknown reason from another rank" + raise KVTransferError( + self.bootstrap_room, failure_reason, is_from_another_rank=is_propagated + ) class MooncakeKVBootstrapServer(CommonKVBootstrapServer): diff --git a/python/sglang/srt/disaggregation/mori/conn.py b/python/sglang/srt/disaggregation/mori/conn.py index c1e6ae22f..81bb7e960 100644 --- a/python/sglang/srt/disaggregation/mori/conn.py +++ b/python/sglang/srt/disaggregation/mori/conn.py @@ -32,6 +32,7 @@ from sglang.srt.disaggregation.common.conn import ( CommonKVManager, CommonKVReceiver, CommonKVSender, + KVTransferError, ) from sglang.srt.disaggregation.common.utils import ( AuxDataCodec, @@ -1539,10 +1540,13 @@ class MoriKVSender(CommonKVSender): self._finalize_failure() self.clear() with self.kv_mgr.failure_lock: - failure_reason = self.kv_mgr.failure_records.pop( - self.bootstrap_room, "KV transfer failed" - ) - raise RuntimeError(failure_reason) + failure_reason = self.kv_mgr.failure_records.pop(self.bootstrap_room, None) + is_propagated = failure_reason is None + if is_propagated: + failure_reason = "KV transfer failed" + raise KVTransferError( + self.bootstrap_room, failure_reason, is_from_another_rank=is_propagated + ) def abort(self): self._finalize_failure("Aborted by AbortReq.") @@ -1675,10 +1679,13 @@ class MoriKVReceiver(CommonKVReceiver): self.clear() with self.kv_mgr.failure_lock: - failure_reason = self.kv_mgr.failure_records.pop( - self.bootstrap_room, "KV transfer failed" - ) - raise RuntimeError(failure_reason) + failure_reason = self.kv_mgr.failure_records.pop(self.bootstrap_room, None) + is_propagated = failure_reason is None + if is_propagated: + failure_reason = "KV transfer failed" + raise KVTransferError( + self.bootstrap_room, failure_reason, is_from_another_rank=is_propagated + ) def abort(self): if self.bootstrap_room is None: diff --git a/python/sglang/srt/disaggregation/nixl/conn.py b/python/sglang/srt/disaggregation/nixl/conn.py index 2554439e9..d180f2a35 100644 --- a/python/sglang/srt/disaggregation/nixl/conn.py +++ b/python/sglang/srt/disaggregation/nixl/conn.py @@ -22,6 +22,7 @@ from sglang.srt.disaggregation.common.conn import ( CommonKVManager, CommonKVReceiver, CommonKVSender, + KVTransferError, ) from sglang.srt.disaggregation.common.staging_handler import StagingRegisterInfo from sglang.srt.disaggregation.common.utils import ( @@ -1973,8 +1974,10 @@ class NixlKVSender(CommonKVSender): if exc is not None: raise exc if failure_reason is not None: - raise RuntimeError(failure_reason) - raise RuntimeError("NIXL KVSender Exception") + raise KVTransferError(self.bootstrap_room, failure_reason) + raise KVTransferError( + self.bootstrap_room, "NIXL KVSender Exception", is_from_another_rank=True + ) class NixlKVReceiver(CommonKVReceiver): @@ -2138,7 +2141,14 @@ class NixlKVReceiver(CommonKVReceiver): ) def failure_exception(self): - raise RuntimeError("NIXL KVReceiver Exception") + with self.kv_mgr.failure_lock: + failure_reason = self.kv_mgr.failure_records.pop(self.bootstrap_room, None) + is_propagated = failure_reason is None + if is_propagated: + failure_reason = "NIXL KVReceiver Exception" + raise KVTransferError( + self.bootstrap_room, failure_reason, is_from_another_rank=is_propagated + ) class NixlKVBootstrapServer(CommonKVBootstrapServer): diff --git a/python/sglang/srt/disaggregation/prefill.py b/python/sglang/srt/disaggregation/prefill.py index 21695efee..ce1afdac3 100644 --- a/python/sglang/srt/disaggregation/prefill.py +++ b/python/sglang/srt/disaggregation/prefill.py @@ -721,11 +721,17 @@ class SchedulerDisaggregationPrefillMixin: req.time_stats.set_prefill_kv_transfer_finish_time() elif poll == KVPoll.Failed: error_message = f"Prefill transfer failed for request rank={self.ps.tp_rank} {req.rid=} {req.bootstrap_room=}" + is_propagated = False try: req.disagg_kv_sender.failure_exception() except Exception as e: error_message += f" with exception {e}" - logger.warning(error_message) + is_propagated = getattr(e, "is_from_another_rank", False) + # Mute error message for propagated exceptions to avoid duplicate logging + if is_propagated: + logger.debug(error_message) + else: + logger.warning(error_message) req.time_stats.trace_ctx.abort(abort_info={"reason": error_message}) release_kv_cache(req, self.tree_cache) # unlock the tree prepare_abort( @@ -802,11 +808,17 @@ class SchedulerDisaggregationPrefillMixin: f"Prefill bootstrap failed for request rank={self.ps.tp_rank} " f"{req.rid=} {req.bootstrap_room=}" ) + is_propagated = False try: req.disagg_kv_sender.failure_exception() except Exception as e: error_message += f" with exception {e}" - logger.warning(error_message) + is_propagated = getattr(e, "is_from_another_rank", False) + # Mute error message for propagated exceptions to avoid duplicate logging + if is_propagated: + logger.debug(error_message) + else: + logger.warning(error_message) req.time_stats.trace_ctx.abort(abort_info={"reason": error_message}) if req.req_pool_idx is not None or self.tree_cache.supports_mamba(): release_kv_cache(req, self.tree_cache)