runpod: one pod per pool across engines, not one per engine

The second live test showed nvidia and radeon each renting their OWN pod
for 'capability:embeddings' — two GPUs, two bills, while max_pods said 1.
Same root cause as the reaper bug: one engine per GPU, each with its own
pools, none able to see the others.

Before renting, a pool now looks in the shared registry for a healthy pod
another process already has for the SAME pool key, and borrows it. The
registry entry carries the URL, recorded once the pod is serving.

A borrowed pod is released, never terminated: its owner holds the
lifecycle and the bill, and killing it would take it away from the engine
still using it. It is also billed at 0 here, so the cost is not
double-counted against the caps.
Co-Authored-By: 's avatarClaude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0182DhR1eFNPQyGmDrpHbedY
parent c35fbb17
...@@ -1176,7 +1176,7 @@ class RunpodPodPool: ...@@ -1176,7 +1176,7 @@ class RunpodPodPool:
"console_url": console, "started_at": time.time()} "console_url": console, "started_at": time.time()}
# …and from every SIBLING engine's reaper, which cannot see the set # …and from every SIBLING engine's reaper, which cannot see the set
# above: each engine has its own pools and its own reaper. # above: each engine has its own pools and its own reaper.
register_pod(pod_id, str(self.model_key)) register_pod(pod_id, str(self.model_key)) # url recorded once ready
print(f"[runpod] pod {pod_id} created for {self.model_key!r} " print(f"[runpod] pod {pod_id} created for {self.model_key!r} "
f"({sel['display_name']} {sel['cloud_type']}) — booting; logs: {console}", f"({sel['display_name']} {sel['cloud_type']}) — booting; logs: {console}",
flush=True) flush=True)
...@@ -1223,6 +1223,9 @@ class RunpodPodPool: ...@@ -1223,6 +1223,9 @@ class RunpodPodPool:
with _pools_lock: with _pools_lock:
_provisioning_ids.discard(pod_id) _provisioning_ids.discard(pod_id)
_provisioning_info.pop(pod_id, None) _provisioning_info.pop(pod_id, None)
# Publish the URL so a sibling engine needing the same pool reuses
# this pod instead of renting a second GPU for the same work.
register_pod(pod_id, str(self.model_key), url)
print(f"[runpod] pod {pod_id} ready for {self.model_key!r} at {url}", flush=True) print(f"[runpod] pod {pod_id} ready for {self.model_key!r} at {url}", flush=True)
return h return h
raise last_exc or RunpodError("RunPod: could not provision a pod.") raise last_exc or RunpodError("RunPod: could not provision a pod.")
...@@ -1325,6 +1328,23 @@ class RunpodPodPool: ...@@ -1325,6 +1328,23 @@ class RunpodPodPool:
return chosen, url return chosen, url
if grow_now: if grow_now:
# Before renting: is a sibling process already running a pod for
# this exact pool? One engine per GPU means the same capability
# was being rented a card each.
shared_id, shared_url = find_shared_pod(
str(self.model_key), self.health_path, self.api_key)
if shared_url:
print(f"[runpod] reusing pod {shared_id} from another engine "
f"for {self.model_key!r}", flush=True)
handle = PodHandle(pod_id=shared_id, url=shared_url,
hourly_usd=0.0, # billed by its owner
started_at=time.time(), gpu="(shared)",
healthy=True, last_used=time.time())
with self._cv:
self.pods.append(handle)
self._provisioning = False
self._cv.notify_all()
continue
# Nothing to serve this request yet — provision inline and retry. # Nothing to serve this request yet — provision inline and retry.
try: try:
self._provision_one() self._provision_one()
...@@ -1353,6 +1373,15 @@ class RunpodPodPool: ...@@ -1353,6 +1373,15 @@ class RunpodPodPool:
self._cv.notify_all() self._cv.notify_all()
def _terminate(self, pod: PodHandle, reason: str): def _terminate(self, pod: PodHandle, reason: str):
if getattr(pod, "gpu", "") == "(shared)":
# Borrowed from another engine: it owns the lifecycle and the bill.
# Terminating it here would kill a pod that process is still using.
print(f"[runpod] releasing borrowed pod {pod.pod_id} ({reason}); its "
"owner reaps it", flush=True)
with self._cv:
if pod in self.pods:
self.pods.remove(pod)
return
from codai.api.runpod_client import RunpodClient from codai.api.runpod_client import RunpodClient
from codai.api import runpod_ledger from codai.api import runpod_ledger
cost = (time.time() - pod.started_at) / 3600.0 * pod.hourly_usd cost = (time.time() - pod.started_at) / 3600.0 * pod.hourly_usd
...@@ -1697,15 +1726,47 @@ def _registry_update(fn): ...@@ -1697,15 +1726,47 @@ def _registry_update(fn):
return {} return {}
def register_pod(pod_id: str, pool_key: str = "") -> None: def register_pod(pod_id: str, pool_key: str = "", url: str = "") -> None:
"""Record a pod as owned by this deployment, so no sibling process reaps it.""" """Record a pod as owned by this deployment.
Two reasons, both about processes that cannot see each other: no sibling's
reaper may treat it as an orphan, and a sibling that needs the SAME pool can
reuse this pod instead of renting a second GPU for the same work.
"""
if not pod_id: if not pod_id:
return return
_registry_update(lambda d: d.__setitem__( _registry_update(lambda d: d.__setitem__(
str(pod_id), {"pool": str(pool_key), "pid": os.getpid(), "at": time.time()}) str(pod_id), {"pool": str(pool_key), "pid": os.getpid(), "at": time.time(),
"url": str(url or "")})
or True) or True)
def find_shared_pod(pool_key: str, health_path: str = "/v1/models",
api_key: str = "") -> tuple:
"""A pod another process already has for this pool: (pod_id, url) or (None, '').
coderai runs one engine per GPU, and each has its own pools — so the same
capability was rented a pod PER ENGINE, paying twice for one job while
max_pods said 1. Checked against the shared registry and health-probed, so a
dead entry is never handed out.
"""
data = _registry_update(lambda d: False)
if not isinstance(data, dict):
return None, ""
mine = os.getpid()
for pod_id, info in data.items():
if not isinstance(info, dict) or info.get("pool") != str(pool_key):
continue
if info.get("pid") == mine:
continue # our own; the local pool handles it
url = str(info.get("url") or "")
if not url:
continue
if _pod_health_ok(url, path=health_path, api_key=api_key):
return pod_id, url
return None, ""
def unregister_pod(pod_id: str) -> None: def unregister_pod(pod_id: str) -> None:
"""Forget a pod we terminated, so the registry does not grow without bound.""" """Forget a pod we terminated, so the registry does not grow without bound."""
if not pod_id: if not pod_id:
......
...@@ -332,3 +332,51 @@ def test_a_young_pod_is_never_reaped(monkeypatch, tmp_path): ...@@ -332,3 +332,51 @@ def test_a_young_pod_is_never_reaped(monkeypatch, tmp_path):
monkeypatch.setattr("codai.api.runpod_client.RunpodClient", _Client) monkeypatch.setattr("codai.api.runpod_client.RunpodClient", _Client)
assert rw.reap_orphans() == 1 assert rw.reap_orphans() == 1
assert killed == ["ancient"] # the 30-second-old pod survives assert killed == ["ancient"] # the 30-second-old pod survives
def test_a_sibling_engines_pod_is_reused_not_duplicated(tmp_path, monkeypatch):
"""One engine per GPU means each had its own pool: the same capability was
rented a card PER ENGINE while max_pods said 1. Observed live — nvidia took
pod y6tg…, radeon took yymi… for the same 'capability:embeddings'."""
import codai.api.runpod_worker as rw
monkeypatch.setattr(rw, "_pod_registry_path", lambda: str(tmp_path / "pods.json"))
monkeypatch.setattr(rw, "_pod_health_ok", lambda url, **kw: True)
# Another process recorded a healthy pod for this pool.
rw._registry_update(lambda d: d.__setitem__(
"pod-from-nvidia", {"pool": "capability:embeddings", "pid": 999999,
"at": 0, "url": "http://pod-a:8000"}) or True)
pod_id, url = rw.find_shared_pod("capability:embeddings")
assert (pod_id, url) == ("pod-from-nvidia", "http://pod-a:8000")
# A different pool must not borrow it.
assert rw.find_shared_pod("capability:images") == (None, "")
# Our own pods are left to the local pool, not "borrowed" from ourselves.
import os
rw._registry_update(lambda d: d.__setitem__(
"pod-mine", {"pool": "capability:images", "pid": os.getpid(),
"at": 0, "url": "http://pod-b:8000"}) or True)
assert rw.find_shared_pod("capability:images") == (None, "")
def test_a_borrowed_pod_is_released_not_terminated(monkeypatch):
"""Terminating a pod we merely borrowed would kill it under the engine that
owns it — and that engine is the one paying for it."""
import codai.api.runpod_worker as rw
killed = []
class _Client:
def __init__(self, *a, **k): pass
def terminate_pod(self, pid): killed.append(pid)
monkeypatch.setattr("codai.api.runpod_client.RunpodClient", _Client)
pool = _pool(max_pods=1)
borrowed = rw.PodHandle(pod_id="pod-x", url="http://x", hourly_usd=0.0,
started_at=0.0, gpu="(shared)")
pool.pods = [borrowed]
pool._terminate(borrowed, "idle")
assert killed == [] and pool.pods == []
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