diff --git a/deepinfer/hicache/mooncake_store.py b/deepinfer/hicache/mooncake_store.py index 5fb899d..6271ad7 100644 --- a/deepinfer/hicache/mooncake_store.py +++ b/deepinfer/hicache/mooncake_store.py @@ -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()), )