From 849ce719767a432297c6eb5aa6479584c82138ab Mon Sep 17 00:00:00 2001 From: Shangming Cai Date: Sun, 23 Aug 2026 14:17:44 +0800 Subject: [PATCH] refactor(disagg): dedupe mooncake failure_exception into a mixin (#36031) --- .../srt/disaggregation/mooncake/conn.py | 62 +++++++++---------- 1 file changed, 29 insertions(+), 33 deletions(-) diff --git a/python/sglang/srt/disaggregation/mooncake/conn.py b/python/sglang/srt/disaggregation/mooncake/conn.py index 6f4a2425f..503818092 100644 --- a/python/sglang/srt/disaggregation/mooncake/conn.py +++ b/python/sglang/srt/disaggregation/mooncake/conn.py @@ -2244,7 +2244,34 @@ class MooncakeKVManager(StagingManagerMixin, CommonKVManager): self._run_one_probe_pass() -class MooncakeKVSender(CommonKVSender): +class MooncakeFailureExceptionMixin: + """Shared `failure_exception` for the Mooncake sender and receiver. + + Both sides conclude a failed room identically: latch Failed, clear local + state, then raise with the recorded reason -- or, when no reason was + recorded locally, report it as propagated from another rank. Expects the + concrete class to provide ``conclude_state``, ``clear()``, + ``bootstrap_room`` and ``kv_mgr``. + """ + + def failure_exception(self): + # A room with no locally recorded reason failed on another rank. + if self.conclude_state is None: + self.conclude_state = KVPoll.Failed + + self.clear() + + 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 = "Failed due to an unknown reason from another rank" + raise KVTransferError( + self.bootstrap_room, failure_reason, is_from_another_rank=is_propagated + ) + + +class MooncakeKVSender(MooncakeFailureExceptionMixin, CommonKVSender): def __init__( self, @@ -2324,22 +2351,6 @@ class MooncakeKVSender(CommonKVSender): else: return self.conclude_state - def failure_exception(self): - # Explicitly set the status to failure since this request has failed in another rank - if self.conclude_state is None: - self.conclude_state = KVPoll.Failed - - self.clear() - - 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 = "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: self.trace_ctx = TraceReqContext( @@ -2361,7 +2372,7 @@ class MooncakeKVSender(CommonKVSender): self.trace_ctx.trace_req_finish() -class MooncakeKVReceiver(CommonKVReceiver): +class MooncakeKVReceiver(MooncakeFailureExceptionMixin, CommonKVReceiver): def __init__( self, mgr: MooncakeKVManager, @@ -2532,21 +2543,6 @@ class MooncakeKVReceiver(CommonKVReceiver): return status - def failure_exception(self): - if self.conclude_state is None: - self.conclude_state = KVPoll.Failed - - self.clear() - - 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 = "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): pass