refactor(disagg): dedupe mooncake failure_exception into a mixin (#36031)
This commit is contained in:
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user