the survivor sweep re-adopted one pod forever (v0.4.47)

v0.4.46's sweep loops "while _adopt_shared_pod() adopts something".
find_shared_pod knew nothing of the pool calling it, so it returned the
same pod every iteration; _add_pod ignored the duplicate, the call still
returned True, and the sweep never reached the next candidate. Live, on
the deploy that introduced it: one pod "adopted" 15-18 times in a second
and four pods left untracked behind it, each wasted pass costing up to
two 15-second probes inside the maintenance loop.

find_shared_pod now takes the caller's own pod ids and skips them, and
adoption reports True only for a pod the pool did not already hold.
Co-Authored-By: 's avatarClaude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01KnfkxBpcZ1v6ypRFFyhrVM
parent a930e93d
...@@ -16,7 +16,7 @@ ...@@ -16,7 +16,7 @@
# Canonical product version for CoderAI — single source of truth. Both the API # Canonical product version for CoderAI — single source of truth. Both the API
# metadata and the admin web UI read from here. # metadata and the admin web UI read from here.
__version__ = "0.4.46" __version__ = "0.4.47"
__tagline__ = "Complete orchestration, distribution and escalation of remotizable advanced inference." __tagline__ = "Complete orchestration, distribution and escalation of remotizable advanced inference."
# Configure the CUDA caching allocator BEFORE torch is imported anywhere. # Configure the CUDA caching allocator BEFORE torch is imported anywhere.
......
...@@ -3493,9 +3493,11 @@ class RunpodPodPool: ...@@ -3493,9 +3493,11 @@ class RunpodPodPool:
want_gpus = list(narrow.get("gpu_types") or []) want_gpus = list(narrow.get("gpu_types") or [])
except Exception: except Exception:
want_gpus = list(getattr(self.mcfg, "gpu_types", []) or []) want_gpus = list(getattr(self.mcfg, "gpu_types", []) or [])
with self._cv:
held = {str(p.pod_id) for p in self.pods}
shared_id, shared_url, shared_key, shared_rate = find_shared_pod( shared_id, shared_url, shared_key, shared_rate = find_shared_pod(
str(self.model_key), self.health_path, self.api_key, str(self.model_key), self.health_path, self.api_key,
image=want_image, gpus=want_gpus) image=want_image, gpus=want_gpus, exclude=held)
if not shared_url: if not shared_url:
return False return False
if shared_key and shared_key != self.api_key: if shared_key and shared_key != self.api_key:
...@@ -3556,10 +3558,13 @@ class RunpodPodPool: ...@@ -3556,10 +3558,13 @@ class RunpodPodPool:
# knows otherwise. # knows otherwise.
**_adopted_card(self.account, shared_id), **_adopted_card(self.account, shared_id),
healthy=True, last_used=time.time()) healthy=True, last_used=time.time())
self._add_pod(handle) added = self._add_pod(handle)
register_pod(shared_id, str(self.model_key), shared_url, self.api_key, register_pod(shared_id, str(self.model_key), shared_url, self.api_key,
hourly_usd=float(shared_rate or 0.0)) # image/gpu kept hourly_usd=float(shared_rate or 0.0)) # image/gpu kept
return True # True only for a pod this pool did not already hold. A duplicate is
# not progress, and a caller looping "while it adopts something" must
# be able to tell.
return bool(added)
def _warmup(self, pod_id: str, url: str) -> bool: def _warmup(self, pod_id: str, url: str) -> bool:
"""Make the pod load its weights before anyone calls it ready. """Make the pod load its weights before anyone calls it ready.
...@@ -6731,7 +6736,8 @@ def _gpu_name_matches(name: str, want: str) -> bool: ...@@ -6731,7 +6736,8 @@ def _gpu_name_matches(name: str, want: str) -> bool:
def find_shared_pod(pool_key: str, health_path: str = "/v1/models", def find_shared_pod(pool_key: str, health_path: str = "/v1/models",
api_key: str = "", image: str = "", api_key: str = "", image: str = "",
gpus: Optional[list] = None) -> tuple: gpus: Optional[list] = None,
exclude: Optional[set] = None) -> tuple:
"""A pod already running for this pool: (pod_id, url, key, hourly_usd). """A pod already running for this pool: (pod_id, url, key, hourly_usd).
``key`` is the token that pod was launched with, which is not necessarily ``key`` is the token that pod was launched with, which is not necessarily
...@@ -6774,6 +6780,13 @@ def find_shared_pod(pool_key: str, health_path: str = "/v1/models", ...@@ -6774,6 +6780,13 @@ def find_shared_pod(pool_key: str, health_path: str = "/v1/models",
continue continue
if _is_this_process(info): if _is_this_process(info):
continue # our own; the local pool handles it continue # our own; the local pool handles it
if exclude and str(pod_id) in exclude:
# Already in the caller's pool. Without this the survivor sweep
# found the SAME pod again on every iteration -- the pool ignored
# the duplicate, the sweep counted it as progress, and it never
# reached the next candidate: 15-18 adoptions of one pod in one
# second, and 4 pods left untracked behind it.
continue
url = str(info.get("url") or "") url = str(info.get("url") or "")
if not url: if not url:
# A pod that was still BOOTING when its owner died. register_pod # A pod that was still BOOTING when its owner died. register_pod
......
...@@ -131,7 +131,7 @@ def test_the_pool_passes_both_when_it_adopts(): ...@@ -131,7 +131,7 @@ def test_the_pool_passes_both_when_it_adopts():
"""Pinned against the source: the checks are only worth anything if the one """Pinned against the source: the checks are only worth anything if the one
caller actually uses them.""" caller actually uses them."""
src = (ROOT / "codai/api/runpod_worker.py").read_text() src = (ROOT / "codai/api/runpod_worker.py").read_text()
blk = src.split("def _adopt_shared_pod")[1][:2000] blk = src.split("def _adopt_shared_pod")[1][:3500]
assert "image=want_image, gpus=want_gpus" in blk assert "image=want_image, gpus=want_gpus" in blk
assert 'want_image = str(plan.get("image") or "")' in blk assert 'want_image = str(plan.get("image") or "")' in blk
assert "narrow_gpu_choice(self.mcfg" in blk assert "narrow_gpu_choice(self.mcfg" in blk
...@@ -93,3 +93,25 @@ def test_without_a_recorded_token_healthz_alone_is_not_enough(monkeypatch): ...@@ -93,3 +93,25 @@ def test_without_a_recorded_token_healthz_alone_is_not_enough(monkeypatch):
pid, url, _k, _r = rw.find_shared_pod("m", "/v1/models", "", pid, url, _k, _r = rw.find_shared_pod("m", "/v1/models", "",
image="img", gpus=["A40"]) image="img", gpus=["A40"])
assert pid is None assert pid is None
def test_a_pod_already_in_the_pool_is_not_offered_again(monkeypatch):
"""The sweep re-found one pod 15-18 times in a second and never reached
the four behind it."""
monkeypatch.setattr(rw, "_registry_update", lambda fn: {
"held": {"pool": "m", "url": "https://10.0.0.1:1/v1", "key": "k",
"image": "img", "gpu": "A40"},
"next": {"pool": "m", "url": "https://10.0.0.2:1/v1", "key": "k",
"image": "img", "gpu": "A40"}})
monkeypatch.setattr(rw, "_is_this_process", lambda info: False)
monkeypatch.setattr(rw, "_pod_health_ok", lambda *a, **k: True)
pid, *_ = rw.find_shared_pod("m", "/v1/models", "", image="img",
gpus=["A40"], exclude={"held"})
assert pid == "next"
def test_adopting_a_duplicate_is_not_reported_as_progress():
src = inspect.getsource(rw.RunpodPodPool._adopt_shared_pod)
assert "added = self._add_pod(handle)" in src
assert "return bool(added)" in src
assert "exclude=held" in src
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment