coffee/DeepInfer
1
0
Fork 0
mirror of https://github.com/deepseek-ai/DeepInfer.git synced 2026-10-29 23:37:28 +08:00
Code Issues Packages Projects Releases Wiki Activity

hicache L3: retry failed Store objects once before tripping the breaker

Browse Source 
Sporadic EFA transfer errors (~0.015% of puts with Mooncake 83f8f877) tripped the breaker on the first failure
and dropped every queued put for 30 s. Retry the failed objects once (puts and gets); only a failure that persists
counts and trips the breaker. New counters: l3_put_retried, l3_get_retried.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
whn09 2026-10-26 19:23:33 +08:00
parent 3657e00592
commit 625606db4e
1 changed files with 22 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
deepinfer/hicache/mooncake_store.py
View File

@ -21,7 +21,8 @@ learns about completions through drain_done(). Gets are synchronous (called from
Failure handling. A Store transfer that cannot complete blocks for Mooncake's fixed 60 s transfer timeout, so one bad
get stalls the engine thread that long. A circuit breaker therefore opens on the first transfer failure (any put or
get error other than "not found" / "already exists") or on a get slower than DEEPINFER_HICACHE_L3_SLOW_S (default 5 s).
get error other than "not found" / "already exists") that persists after one retry, or on a get slower than
DEEPINFER_HICACHE_L3_SLOW_S (default 5 s). The retry absorbs sporadic transfer errors (seen at ~0.015% of EFA puts).
While it is open, reads are skipped (host-only lookup) and queued puts are dropped. It closes after
DEEPINFER_HICACHE_L3_COOLDOWN_S (default 30 s, doubled on each consecutive trip up to 10 min).
@ -112,7 +113,7 @@ class MooncakeL3:
self._state_prefix = f"state/{fingerprint}/"
self._puts: queue.Queue = queue.Queue()
self._done: collections.deque = collections.deque() # tokens of finished puts (appended by the worker thread)
self.put_objects = self.put_failed = self.put_bytes = self.put_dropped = 0
self.put_objects = self.put_failed = self.put_bytes = self.put_dropped = self.put_retried = self.get_retried = 0
self.put_seconds = 0.0
self.slow_get_s = float(os.environ.get("DEEPINFER_HICACHE_L3_SLOW_S", "5"))
self.breaker = _Breaker(float(os.environ.get("DEEPINFER_HICACHE_L3_COOLDOWN_S", "30")))
@ -151,7 +152,16 @@ class MooncakeL3:
continue
t0 = time.perf_counter()
try:
rcs = self.store.batch_put_from_multi_buffers(keys, ptrs, sizes)
rcs = list(self.store.batch_put_from_multi_buffers(keys, ptrs, sizes))
# retry the failed objects once: sporadic transfer errors are not a sick Store
redo = [i for i, rc in enumerate(rcs) if rc not in (0, OBJECT_ALREADY_EXISTS)]
if redo:
self.put_retried += len(redo)
again = self.store.batch_put_from_multi_buffers(
[keys[i] for i in redo], [ptrs[i] for i in redo], [sizes[i] for i in redo]
)
for i, rc in zip(redo, again, strict=True):
rcs[i] = rc
# an object that already exists (another rank or instance stored it first) is not a failure
bad = [rc for rc in rcs if rc not in (0, OBJECT_ALREADY_EXISTS)]
self.put_objects += len(keys) - len(bad)
@ -182,7 +192,13 @@ class MooncakeL3:
if not keys:
return []
t0 = time.perf_counter()
rcs = self.store.batch_get_into_multi_buffers(keys, ptrs, sizes)
rcs = list(self.store.batch_get_into_multi_buffers(keys, ptrs, sizes))
redo = [i for i, (rc, s) in enumerate(zip(rcs, sizes, strict=True)) if rc != sum(s) and rc != OBJECT_NOT_FOUND]
if redo: # one retry, as for puts
self.get_retried += len(redo)
again = self.store.batch_get_into_multi_buffers([keys[i] for i in redo], [ptrs[i] for i in redo], [sizes[i] for i in redo])
for i, rc in zip(redo, again, strict=True):
rcs[i] = rc
dt = time.perf_counter() - t0
ok = [rc == sum(s) for rc, s in zip(rcs, sizes, strict=True)]
# "not found" just means evicted since exists(); anything else is a transfer problem
@ -201,6 +217,8 @@ class MooncakeL3:
l3_put_seconds=round(self.put_seconds, 3),
l3_put_queue=self._puts.qsize(),
l3_put_dropped=self.put_dropped,
l3_put_retried=self.put_retried,
l3_get_retried=self.get_retried,
l3_breaker_trips=self.breaker.trips,
l3_breaker_open=int(not self.breaker.allow()),
)