Skip to content

C
coderai
Commit 72bbed69 authored by Stefy Lanza (nextime / spora )'s avatar Stefy Lanza (nextime / spora )
Browse files
Options

a lease may not wait in a thread (v0.4.19) 

Third run, third failure, and this one was mine. A 5,000-document run
dispatched ZERO pages to 50 healthy pods in 41 minutes: not one
`POST /v1/ocr` in any pod's log, every GPU at `KV 0.0%`, 811 pages waiting,
and the scaler reading that unservable queue as demand and renting to 73
pods at $43/hr. `/healthz` answered in 1.6 ms the whole time, so the loop
fixes from 0.4.16/0.4.17 were holding and nothing looked wrong from outside.

0.4.18 bounded a lease attempt at 20 s and ran it on a 256-thread pool. That
made the LEASE STAGE the ceiling on everything: 256 threads at one attempt
per 20 s passes ~13 pages a second however many pods are up, and a thread
that drew the growth branch sat through an entire ~9-minute boot. Against
811 pages each retrying every couple of seconds, nothing got through.

So `acquire(grow_inline=False)`: answer now, or start growth in the
background and raise `NoPodRightNow`. The caller waits on its event loop,
where waiting is free, and `POD_LEASE_ATTEMPT_S` drops 20 s -> 2 s for the
pools whose acquire predates the flag. Threaded through
`placement.choose(grow_inline=)` by signature inspection, so a candidate
that cannot take it keeps working.

Demand had to survive that. `_waiting` only ever counted THREADS parked in
acquire, and the non-blocking path has none, so the pool would have stopped
growing. It now counts DENIALS over a window equal to the caller's retry
interval, which makes the count distinct waiting PAGES rather than the
attempts they have made between them -- a page retrying every 2 s appears
once in any 2-second window. A longer window multiplies demand by the retry
rate, which is how the same run read its own retries as new work.

Two more things found on the way:

  * `pod_weight`/`pod_capacity`/`pod_threshold` depend only on a card's NAME,
    but were recomputed per POD per acquire -- `_pick` once per free pod,
    `_pods_for_demand` once per serving pod, `_preferred_vram_gb` walking
    every pod inside each -- all under the pool lock. Memoised per card:
    measured 286us -> 69us per acquire at 70 pods, 528us -> 102us at 128. The
    memo self-initialises via getattr, because a pool built without __init__
    must not have an optimisation turn into an AttributeError.

  * `_pick` returned a bare None for four different causes -- no pods,
    none healthy, all draining, all at their in-flight cap -- which need
    opposite responses, and the caller could only report the queue it could
    not drain. It now says which, once per 15 s, with every count in the line.

I also mis-read `futex_do_wait` on 324 of 352 threads as lock contention. A
condition-variable wait IS a futex wait, so that is what pages legitimately
waiting for a pod look like. The pod logs settled it instead: zero POSTs
means the dispatch never happened, not that it was slow.
Co-Authored-By: 's avatarClaude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01KnfkxBpcZ1v6ypRFFyhrVM
parent 4478272e
Hide whitespace changes
Inline Side-by-side
Showing with 722 additions and 47 deletions
codai/__init__.py
View file @ 72bbed69
......@@ -16,7 +16,7 @@
# Canonical product version for CoderAI — single source of truth. Both the API
# metadata and the admin web UI read from here.
__version__ = "0.4.18"
__version__ = "0.4.19"
__tagline__ = "Complete orchestration, distribution and escalation of remotizable advanced inference."
# Configure the CUDA caching allocator BEFORE torch is imported anywhere.
......
codai/api/runpod_worker.py
View file @ 72bbed69
......@@ -2915,6 +2915,37 @@ class RunpodPodPool:
#: Set from the pod plan at provision time; the default suits an LLM pod.
self.health_path = mcfg.health_path or "/v1/models"
self._cv = threading.Condition(threading.RLock())
#: Per-CARD arithmetic, memoised. `pod_weight`, `pod_capacity` and
#: `pod_threshold` depend only on the card's NAME and the model config,
#: but they were recomputed per pod on every acquire(): `_pick` once per
#: free pod, `_pods_for_demand` once per serving pod, and
#: `_preferred_vram_gb` iterating every pod again inside each of those
#: -- all while holding `_cv`.
#:
#: At 70 pods that is ~140 recomputations per acquire, each scanning the
#: gpu_limits/gpu_weights tables with substring matches. Measured on a
#: live run: 811 pages waiting, 324 of 352 engine threads parked in
#: futex_do_wait, /healthz answering in 1.6 ms while every lease
#: serialised behind the pool lock and the scaler kept renting more pods
#: -- each new pod making the per-acquire cost higher. The fleet grew
#: because it could not serve, and could not serve because it had grown.
#:
#: Keyed by card name and a generation that bumps when the pod set or
#: the config changes, because `_preferred_vram_gb` reads the pods.
self._card_cache: dict = {}
self._card_gen = 0
#: When `acquire(grow_inline=False)` last turned pages away. This is the
#: demand signal now that a waiting page no longer parks a thread in
#: `acquire` -- `_waiting` only ever counted THREADS, and the
#: non-blocking path has none to count.
#:
#: Counted over a window equal to the caller's RETRY interval, which
#: makes the count the number of DISTINCT pages waiting rather than the
#: attempts they have made between them: a page retrying every 2 s
#: appears exactly once in any 2-second window. A longer window inflates
#: demand by the retry rate, which is how an earlier build read its own
#: retries as new work and rented 73 pods for a queue it never served.
self._denied: list = []
#: A warm floor a client leased through /v1/runpod/scale, and when the
#: lease runs out. 0/0 = nobody asked.
self._client_floor = 0
......@@ -3423,6 +3454,7 @@ class RunpodPodPool:
healthy=True, last_used=time.time())
with self._cv:
self.pods.append(handle)
self._bump_cards()
self._cv.notify_all()
register_pod(shared_id, str(self.model_key), shared_url, self.api_key,
hourly_usd=float(shared_rate or 0.0)) # image/gpu kept
......@@ -3757,6 +3789,7 @@ class RunpodPodPool:
raise BootCancelled(pod_id)
with self._cv:
self.pods.append(h)
self._bump_cards()
self._cv.notify_all()
with _pools_lock:
_provisioning_ids.discard(pod_id)
......@@ -3826,17 +3859,54 @@ class RunpodPodPool:
self._booting = max(0, self._booting - want)
self._cv.notify_all()
def _say_no_pod(self, total: int, healthy: int, draining: int,
at_cap: int) -> None:
"""One rate-limited line naming WHICH of the four reasons it was."""
now = time.time()
if now < getattr(self, "_no_pod_said", 0.0):
return
self._no_pod_said = now + 15.0
if total == 0:
why = "the pool holds no pods at all"
elif healthy == 0:
why = (f"none of its {total} pod(s) is healthy — they are booting, "
f"or failing their health probe")
elif draining >= healthy:
why = f"all {draining} healthy pod(s) are draining"
elif at_cap:
why = (f"all {at_cap} serving pod(s) are at their in-flight cap "
f"(max_inflight_per_pod, scaled per card)")
else:
why = "no pod was eligible, for a reason this code cannot name"
print(f"[runpod] {self.model_key!r}: NO POD to hand out — {why} "
f"[{total} in pool, {healthy} healthy, {draining} draining, "
f"{at_cap} at cap, {self._booting} booting, "
f"{self._queued()} waiting]", flush=True)
def _pick(self, healthy: list, affinity: str):
"""Which pod serves this request: the one that already holds the
conversation's cache, else the least loaded."""
free = []
free, draining, at_cap = [], 0, 0
for p in healthy:
if p.draining:
draining += 1
continue
cap = self.pod_capacity(p) # scaled to THIS pod's card
if cap <= 0 or p.inflight < cap:
free.append(p)
else:
at_cap += 1
if not free:
# SAY WHY, or the next incident is the last one again. "811 pages
# waiting with 50 pods up" has four different causes -- every pod
# unhealthy, every pod draining, every pod at its in-flight cap, or
# no pod in the list at all -- and they need opposite responses.
# A live run spent 41 minutes and $43/hr indistinguishable between
# them, because this returned a bare None and the caller could only
# report the queue it could not drain.
with self._cv:
total = len(self.pods)
self._say_no_pod(total, len(healthy), draining, at_cap)
return None
if affinity and getattr(self.mcfg, "sticky_sessions", True):
pid = self._affinity.get(affinity)
......@@ -3882,7 +3952,90 @@ class RunpodPodPool:
return vram
return 0.0
#: Must match the caller's retry interval -- see `_denied`.
DENY_WINDOW_S = 2.0
def _note_denied(self) -> None:
"""Record one page turned away. Trimmed as it goes."""
now = time.time()
with self._cv:
self._denied.append(now)
cut = now - (self.DENY_WINDOW_S * 4)
if self._denied[0] < cut:
self._denied = [t for t in self._denied if t >= cut]
if len(self._denied) > _DENY_CAP:
# A hard cap as well as the age filter. Under a burst every
# entry is recent, so filtering by age drops nothing and the
# list grows with the burst -- and keeping more than the cap
# buys nothing, because only the newest window is ever counted.
del self._denied[:-_DENY_CAP]
def _denied_recent(self) -> int:
"""Distinct pages waiting, as far as the pool can tell."""
cut = time.time() - self.DENY_WINDOW_S
with self._cv:
return sum(1 for t in getattr(self, "_denied", ()) if t >= cut)
def _queued(self) -> int:
"""Work that wants a pod but has none: threads parked, or pages denied.
Either signal, never their sum -- a thread parked in acquire and a page
turned away that will retry are the same page counted two ways.
"""
return max(max(0, self._waiting), self._denied_recent())
def _bump_cards(self) -> None:
"""Invalidate the per-card memo. Call with `_cv` held.
Cheap on purpose: a counter, not a clear, so a caller mid-read of the
old generation finishes against a consistent value rather than a dict
that emptied underneath it.
"""
self._card_gen = getattr(self, "_card_gen", 0) + 1
cache = getattr(self, "_card_cache", None)
if cache is not None and len(cache) > 256:
cache.clear()
def _card_facts(self, name: str) -> dict:
"""`{weight, capacity, threshold}` for this card, computed once.
The three were separate methods doing overlapping work -- both weight
and threshold read `gpu_limits`, and capacity called weight -- so one
memo covers all of them and the tables are scanned once per card per
generation instead of once per pod per call.
"""
# getattr-initialised: a pool may be built by a caller that did not run
# __init__ (a test double, a duck-typed stand-in), and a memo that
# raises on such a pool would turn an optimisation into an outage.
cache = getattr(self, "_card_cache", None)
if cache is None:
cache = self._card_cache = {}
key = (name or "", getattr(self, "_card_gen", 0))
got = cache.get(key)
if got is not None:
return got
facts = {"weight": self._compute_weight(name),
"capacity": 0, "threshold": 1.0}
lim = card_limits(self.mcfg, name)
base_inflight = int(getattr(self.mcfg, "max_inflight_per_pod", 0) or 0)
stated_inflight = int(lim.get("inflight") or 0)
if stated_inflight != base_inflight and stated_inflight > 0:
facts["capacity"] = stated_inflight
elif base_inflight > 0:
facts["capacity"] = max(1, int(round(base_inflight * facts["weight"])))
base_up = max(1, int(getattr(self.mcfg, "scale_up_inflight_per_pod", 1) or 1))
stated_up = int(lim.get("scale_up") or 0)
facts["threshold"] = (float(stated_up) if stated_up and stated_up != base_up
else base_up * facts["weight"])
cache[key] = facts
return facts
def pod_weight(self, pod: "PodHandle") -> float:
"""This pod's share of the load — memoised per card. See `_card_facts`."""
return self._card_facts(
str(getattr(pod, "gpu", "") or "").strip())["weight"]
def _compute_weight(self, name: str) -> float:
"""This pod's share of the load, relative to the preferred card.
An explicit `gpu_weights` entry wins: VRAM bounds how many requests FIT
......@@ -3897,11 +4050,6 @@ class RunpodPodPool:
rotation for want of a number would turn a reporting gap into an
outage.
"""
# getattr throughout: this is a duck-typed pod, and a caller holding a
# lighter object must get the neutral weight rather than an exception.
# A missing attribute is the same situation as an unknown card, and the
# docstring above promises that keeps the pod serving.
name = str(getattr(pod, "gpu", "") or "").strip()
prefer_name = ""
prefer = [str(g).strip() for g
in (getattr(self.mcfg, "gpu_types", None) or []) if str(g).strip()]
......@@ -3932,36 +4080,41 @@ class RunpodPodPool:
base = _weight_for(weights, prefer_name) if prefer_name else None
return max(0.25, min(8.0, mine / (base or 1.0)))
base = self._preferred_vram_gb()
vram = float(getattr(pod, "vram_gb", 0.0) or 0.0)
vram = self._vram_of_card(name)
if base > 0 and vram > 0:
return max(0.25, min(8.0, vram / base))
return 1.0
def _vram_of_card(self, name: str) -> float:
"""VRAM reported by any pod running this card, or 0.
By card rather than per pod because the weight is a property of the
card: two A40s do not deserve two different weights, and asking per pod
is what made this an O(pods) walk inside an O(pods) loop.
"""
want = str(name or "").strip().lower()
if not want:
return 0.0
with self._cv:
pods = list(self.pods)
for p in pods:
if str(getattr(p, "gpu", "") or "").strip().lower() == want:
return float(getattr(p, "vram_gb", 0.0) or 0.0)
return 0.0
def pod_capacity(self, pod: "PodHandle") -> int:
"""How many requests THIS pod may hold.
"""How many requests THIS pod may hold. 0 = unlimited.
A per-card `inflight` is taken as written; otherwise the model-level
number is scaled by this pod's weight. Unlimited stays unlimited.
number is scaled by the card's weight. Memoised — see `_card_facts`.
"""
name = str(getattr(pod, "gpu", "") or "").strip()
lim = card_limits(self.mcfg, name)
stated = int(lim.get("inflight") or 0)
base = int(getattr(self.mcfg, "max_inflight_per_pod", 0) or 0)
if stated != base and stated > 0:
return stated # the table said so, for this card
if base <= 0:
return 0
return max(1, int(round(base * self.pod_weight(pod))))
return self._card_facts(
str(getattr(pod, "gpu", "") or "").strip())["capacity"]
def pod_threshold(self, pod: "PodHandle") -> float:
"""The in-flight depth at which THIS pod justifies another one."""
name = str(getattr(pod, "gpu", "") or "").strip()
lim = card_limits(self.mcfg, name)
stated = int(lim.get("scale_up") or 0)
base = max(1, int(getattr(self.mcfg, "scale_up_inflight_per_pod", 1) or 1))
if stated and stated != base:
return float(stated) # written for this card
return base * self.pod_weight(pod)
return self._card_facts(
str(getattr(pod, "gpu", "") or "").strip())["threshold"]
def _pods_for_demand(self, serving: list) -> int:
"""How many pods the work we can SEE justifies, capped at max_pods.
......@@ -3972,7 +4125,7 @@ class RunpodPodPool:
much, so the only question it could answer was "one more?".
"""
threshold = max(1, int(self.mcfg.scale_up_inflight_per_pod or 1))
load = sum(p.inflight for p in serving) + max(0, self._waiting)
load = sum(p.inflight for p in serving) + self._queued()
# Per POD, weighted: the threshold describes the preferred card, so a
# fleet of bigger cards carries the same load in fewer pods. Dividing
# by a flat threshold would rent ~3x more A100s than the work needs,
......@@ -4049,7 +4202,7 @@ class RunpodPodPool:
want = self._pods_for_demand(serving)
if in_hand < want:
threshold = max(1, int(self.mcfg.scale_up_inflight_per_pod or 1))
load = sum(p.inflight for p in serving) + max(0, self._waiting)
load = sum(p.inflight for p in serving) + self._queued()
return (f"{load} page(s) in flight or waiting at {threshold} per pod "
f"needs {want} pod(s), and {in_hand} are in hand")
# And the original, anticipatory trigger: every pod is at its in-flight
......@@ -4107,13 +4260,29 @@ class RunpodPodPool:
threading.Thread(target=_run, daemon=True,
name=f"runpod-grow-{self.model_key}").start()
def acquire(self, timeout: float = 1200.0, affinity: str = ""):
def acquire(self, timeout: float = 1200.0, affinity: str = "",
grow_inline: bool = True):
"""Return (PodHandle, url) for a ready pod, provisioning on demand.
Bumps in-flight; the caller MUST pair it with release(pod). Picks the pod
holding this conversation's cache when ``affinity`` is given, else the
least-loaded one, and grows the pool in the background once the least
loaded pod is at ``scale_up_inflight_per_pod``.
``grow_inline=False`` means "answer now": trigger growth in the
background and raise rather than renting a pod in the caller's thread
or waiting on the condition variable. For a caller that can wait
somewhere cheaper than a thread -- a coroutine with its own retry loop
-- this is the only safe mode, and the reason is arithmetic:
A page's lease ran with a 20 s bounded attempt in a 256-thread pool, so
the lease STAGE could pass at most ~13 pages a second however many pods
were up, and a thread that drew the growth branch sat through an entire
~9-minute boot. Against 811 pages, each retrying every couple of
seconds, nothing ever got through: a live run dispatched ZERO pages to
50 healthy pods in 41 minutes, every pod's log showing not one
`POST /v1/ocr`, while the queue the scaler could see made it rent more.
Waiting is free on an event loop and costs a whole thread here.
"""
from codai.api.runpod_client import RunpodError
deadline = time.time() + timeout
......@@ -4142,6 +4311,19 @@ class RunpodPodPool:
self._grow_in_background() # absorbs the NEXT request
return chosen, url
if grow_now and not grow_inline:
# Growth belongs in the background; the caller waits on its own
# loop. Released before raising, or the booting count leaks and
# `_grow_reason` stops firing for everyone.
with self._cv:
self._booting = max(0, self._booting - 1)
self._cv.notify_all()
self._note_denied()
self._grow_in_background()
raise NoPodRightNow(
f"no pod free for {self.model_key!r} right now; growth "
f"started in the background")
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
......@@ -4181,6 +4363,16 @@ class RunpodPodPool:
self._cv.notify_all()
continue
if not grow_inline:
# Nothing to hand out and growth is someone else's job: say so
# at once instead of holding this thread on the condvar. The
# count still rises, because the scaler reads it as demand --
# it is just no longer a thread each.
self._note_denied()
raise NoPodRightNow(
f"no pod slot for {self.model_key!r} right now "
f"(max_pods={self.mcfg.max_pods})")
# At max_pods with every pod at its in-flight ceiling, or someone else
# is provisioning: wait for a slot rather than piling on. Counted
# while waiting, so the next pass through _should_grow can see how
......@@ -4220,6 +4412,7 @@ class RunpodPodPool:
return
pod.healthy = False
self.pods.remove(pod)
self._bump_cards()
print(f"[runpod] pod {pod.pod_id} dropped from the pool: {reason}", flush=True)
threading.Thread(target=self._terminate, args=(pod, reason), daemon=True).start()
......@@ -4241,6 +4434,7 @@ class RunpodPodPool:
return False
pod.healthy = False
self.pods.remove(pod)
self._bump_cards()
if getattr(pod, "gpu", "") == "(shared)":
# Another process owns the bill; it writes the ledger entry.
print(f"[runpod] borrowed pod {pod.pod_id} is gone ({reason}) — "
......@@ -4272,6 +4466,7 @@ class RunpodPodPool:
with self._cv:
if pod in self.pods:
self.pods.remove(pod)
self._bump_cards()
return
from codai.api.runpod_client import RunpodClient
from codai.api import runpod_ledger
......@@ -4502,6 +4697,7 @@ class RunpodPodPool:
if victim.inflight == 0:
victim.healthy = False
self.pods.remove(victim)
self._bump_cards()
terminated.append(victim.pod_id)
threading.Thread(target=self._terminate,
args=(victim, "released by client"),
......@@ -4569,6 +4765,7 @@ class RunpodPodPool:
if p.inflight == 0:
p.healthy = False
self.pods.remove(p)
self._bump_cards()
threading.Thread(target=self._terminate, args=(p, "scaled down"),
daemon=True).start()
......@@ -4594,6 +4791,7 @@ class RunpodPodPool:
if victim.inflight == 0:
victim.healthy = False
self.pods.remove(victim)
self._bump_cards()
threading.Thread(target=self._terminate, args=(victim, "scaled down"),
daemon=True).start()
why = "terminated"
......@@ -4648,6 +4846,7 @@ class RunpodPodPool:
if p.inflight == 0 and idle_to > 0 and (now - p.last_used) > idle_to:
p.healthy = False
self.pods.remove(p)
self._bump_cards()
to_kill = p
# terminate outside the lock
threading.Thread(target=self._terminate, args=(to_kill, "idle"),
......@@ -4760,6 +4959,20 @@ class RunpodPodPool:
# module-level pool registry + scaler ------------------------------------- #
#: Most denial timestamps a pool keeps. Only the newest `DENY_WINDOW_S` of
#: them is ever counted, so this bounds memory under a burst and nothing else.
_DENY_CAP = 4096
class NoPodRightNow(RuntimeError):
"""No pod is free THIS INSTANT and the caller asked not to wait here.
Distinct from a pool that cannot serve at all: growth has been started and
retrying shortly is the right response. Raised only for
``acquire(grow_inline=False)``.
"""
_pools_lock = threading.RLock()
_pools: dict = {} # model_key -> RunpodPodPool
_scaler_started = False
......
codai/models/placement.py
View file @ 72bbed69
......@@ -399,9 +399,15 @@ def _spill_pod_candidate(model_id: str) -> Optional[Candidate]:
if pool is None:
return None
def _acquire(timeout: float = 0.0):
# The pool's own default (1200 s) unless the caller bounds the attempt.
pod, url = pool.acquire(timeout=timeout) if timeout > 0 else pool.acquire()
def _acquire(timeout: float = 0.0, grow_inline: bool = True):
# `grow_inline=False` is "answer now": no renting in this thread, no
# waiting on the pool's condvar. See PodPool.acquire.
if not grow_inline:
pod, url = pool.acquire(timeout=0.0, grow_inline=False)
elif timeout > 0:
pod, url = pool.acquire(timeout=timeout)
else:
pod, url = pool.acquire()
key = getattr(pool, "api_key", "") or ""
return (url.rstrip("/") + "/v1", key, lambda: pool.release(pod),
lambda: pool.discard(pod, "connection refused"))
......@@ -475,7 +481,7 @@ def _load_of(cand: Candidate) -> int:
def choose(model_id: str, spill_after: int = 0,
acquire_timeout: float = 0.0) -> Lease:
acquire_timeout: float = 0.0, grow_inline: bool = True) -> Lease:
"""A lease on the least-loaded server for the model.
``spill_after`` is the in-flight count at which a fixed server (the local one,
......@@ -511,7 +517,7 @@ def choose(model_id: str, spill_after: int = 0,
# pages over the pods it has and grows when they are all at their
# threshold. This is "on runpod only", scaling.
for cand in [c for c in cands if c.primary and c.acquire is not None]:
lease = _take(cand, acquire_timeout)
lease = _take(cand, acquire_timeout, grow_inline)
if lease is not None:
return lease
last_exc = "pool unavailable"
......@@ -533,7 +539,7 @@ def choose(model_id: str, spill_after: int = 0,
# 3. Everything we own is busy: rent. This is "local + spillover" — the
# overflow goes to a pool instead of queueing behind a saturated GPU.
for cand in pools:
lease = _take(cand, acquire_timeout)
lease = _take(cand, acquire_timeout, grow_inline)
if lease is not None:
return lease
last_exc = "pool unavailable"
......@@ -578,7 +584,16 @@ def _say_once(key: str, message: str, every: float = 30.0) -> None:
print(message, flush=True)
def _take(cand: Candidate, acquire_timeout: float = 0.0) -> Optional[Lease]:
def _accepts_grow_inline(fn) -> bool:
try:
import inspect
return "grow_inline" in inspect.signature(fn).parameters
except Exception:
return False
def _take(cand: Candidate, acquire_timeout: float = 0.0,
grow_inline: bool = True) -> Optional[Lease]:
"""Acquire a pool candidate, or None when it cannot provide a server.
``acquire_timeout`` bounds ONE attempt. A pool's acquire waits for a pod and
......@@ -588,9 +603,12 @@ def _take(cand: Candidate, acquire_timeout: float = 0.0) -> Optional[Lease]:
thread per page for the length of a boot. 0 keeps the pool's own default.
"""
try:
got = (cand.acquire(timeout=acquire_timeout)
if acquire_timeout > 0 and _accepts_timeout(cand.acquire)
else cand.acquire())
if not grow_inline and _accepts_grow_inline(cand.acquire):
got = cand.acquire(grow_inline=False)
elif acquire_timeout > 0 and _accepts_timeout(cand.acquire):
got = cand.acquire(timeout=acquire_timeout)
else:
got = cand.acquire()
except Exception as exc:
_say_once(f"take:{cand.key}",
f"[ocr] {cand.key} cannot serve this page: {exc}")
......
codai/ocr/manager.py
View file @ 72bbed69
......@@ -744,13 +744,21 @@ def _pod_pipeline_policy(cfg) -> tuple:
#: budget, or at its ceiling, stays that way for minutes: without this every
#: page of every document would pay the full wait before going local, turning a
#: 60s ceiling into a 60s tax per page.
#: How long ONE lease attempt may hold a pod-pool thread. A pool's acquire
#: waits for a free pod and may rent one inline, so an unbounded attempt parks a
#: thread for a whole boot -- several minutes on a cold start, times every page
#: of every document in flight. Bounded, the attempt comes back and the waiting
#: happens in `_lease_a_pod`'s own loop, on the event loop, where it is free and
#: where the policy's ceiling is what decides how long to keep trying.
POD_LEASE_ATTEMPT_S = 20.0
#: How long ONE lease attempt may hold a lease-pool thread.
#:
#: Near zero, and it has to be: the attempt is made with `grow_inline=False`,
#: so the pool either has a pod this instant or says so and starts growing in
#: the background. Nothing in that path blocks, and `_lease_a_pod` does the
#: waiting on the event loop where it is free.
#:
#: This was 20 s in 0.4.18, which made the lease STAGE the ceiling on
#: everything: 256 threads x one attempt per 20 s is ~13 pages a second however
#: many pods are up, and a thread that drew the growth branch sat through a
#: ~9-minute boot. Measured consequence -- zero pages dispatched to 50 healthy
#: pods in 41 minutes, not one `POST /v1/ocr` in any pod's log, while the
#: unservable queue made the scaler rent 73 pods at $43/hr. The value is kept
#: only as the ceiling for a pool whose acquire predates `grow_inline`.
POD_LEASE_ATTEMPT_S = 2.0
def _is_local_exhaustion(exc: BaseException) -> bool:
......@@ -831,7 +839,8 @@ async def _lease_a_pod(model_id: str, policy: str, wait_s: float):
lease = await run_on_lease_pool(
placement.choose, model_id,
spill_after=placement.spill_threshold(model_id),
acquire_timeout=POD_LEASE_ATTEMPT_S)
acquire_timeout=POD_LEASE_ATTEMPT_S,
grow_inline=False)
except OcrError:
raise
except Exception as exc:
......
tests/test_runpod_card_memo.py 0 → 100644
View file @ 72bbed69
"""The fleet grew because it could not serve, and could not serve because it had grown.
`pod_weight`, `pod_capacity` and `pod_threshold` depend only on a card's NAME
and the model config. They were recomputed per POD on every `acquire()`:
`_pick` once per free pod, `_pods_for_demand` once per serving pod, and
`_preferred_vram_gb` walking every pod again inside each of those -- all while
holding the pool lock.
At 70 pods that is ~140 recomputations per acquire, each doing substring
matches across the gpu_limits and gpu_weights tables. Measured on a live
5,000-document run: 811 pages waiting, 324 of the engine's 352 threads parked
in `futex_do_wait`, `/healthz` still answering in 1.6 ms -- so not a blocked
event loop, but every lease serialised behind the pool lock. The scaler read
the waiting pages as demand and rented another pod every 15 s, and each new pod
made the per-acquire cost higher: 69 -> 73 pods, $43/hr, zero documents read.
So the arithmetic is memoised per card. These tests pin both halves: that the
memo returns what the computation returned (a cache that is subtly wrong is
worse than a slow one), and that the per-acquire cost stops growing with the
fleet.
"""
import time
import types
import pytest
from codai.api import runpod_worker as rw
def _mcfg(**kw):
base = dict(max_pods=128, min_pods=0, max_inflight_per_pod=0,
scale_up_inflight_per_pod=32, pod_max_parallel_requests=64,
pod_queue_max_size=256, scale_up_max_parallel_boots=32,
gpu_types=["NVIDIA A100-SXM4-80GB", "NVIDIA A40"],
gpu_weights={"NVIDIA A100-SXM4-80GB": 2.9, "NVIDIA L40S": 1.25,
"NVIDIA A40": 1.0, "A100 PCIe": 2.8},
gpu_limits=None, health_path="/v1/models", env={})
base.update(kw)
return types.SimpleNamespace(**base)
def _pool(mcfg=None, cards=()):
"""A pool with no account work done: only the bits the memo touches."""
pool = rw.RunpodPodPool.__new__(rw.RunpodPodPool)
import threading
pool.mcfg = mcfg or _mcfg()
pool._cv = threading.Condition(threading.RLock())
pool._card_cache = {}
pool._card_gen = 0
pool._booting = 0
pool._waiting = 0
pool._affinity = {}
pool.model_key = "m"
pool.pods = []
for i, (card, vram) in enumerate(cards):
pool.pods.append(types.SimpleNamespace(
pod_id=f"p{i}", gpu=card, vram_gb=vram, inflight=0, healthy=True,
draining=False, last_used=0.0, is_spot=False, hourly_usd=1.0,
started_at=time.time(), url="https://10.0.0.1:1/v1"))
return pool
def _pod(card, vram=48.0, inflight=0):
return types.SimpleNamespace(gpu=card, vram_gb=vram, inflight=inflight,
healthy=True, draining=False, pod_id="x")
# ------------------------------------------------------------- correctness #
def test_the_preferred_card_weighs_one():
"""Everything is relative to gpu_types[0] — an absolute 2.9 would size the
fleet three times too large for the very card it was measured on."""
p = _pool()
assert p.pod_weight(_pod("NVIDIA A100-SXM4-80GB")) == pytest.approx(1.0)
def test_a_slower_card_weighs_less_than_one():
p = _pool()
assert p.pod_weight(_pod("NVIDIA A40")) == pytest.approx(1.0 / 2.9, rel=1e-6)
def test_an_unknown_card_is_neutral_and_keeps_serving():
"""A reporting gap must not become an outage."""
p = _pool()
assert p.pod_weight(_pod("Some Card Nobody Listed", vram=0.0)) == 1.0
def test_the_weight_is_clamped():
p = _pool(_mcfg(gpu_types=["NVIDIA A40"],
gpu_weights={"NVIDIA A40": 1.0, "Monster": 500.0}))
assert p.pod_weight(_pod("Monster")) == 8.0
p2 = _pool(_mcfg(gpu_types=["Monster"],
gpu_weights={"NVIDIA A40": 1.0, "Monster": 500.0}))
assert p2.pod_weight(_pod("NVIDIA A40")) == 0.25
def test_a_per_card_limits_row_wins_over_the_weights_table():
"""One source of truth: if the operator stated a depth per card, their
ratio IS the share each should carry."""
p = _pool(_mcfg(gpu_limits={"NVIDIA A40": {"scale_up": 16},
"NVIDIA A100-SXM4-80GB": {"scale_up": 64}}))
assert p.pod_weight(_pod("NVIDIA A40")) == pytest.approx(16 / 64)
def test_unlimited_inflight_stays_unlimited():
p = _pool()
assert p.pod_capacity(_pod("NVIDIA A40")) == 0
def test_capacity_scales_with_the_weight():
p = _pool(_mcfg(max_inflight_per_pod=100))
assert p.pod_capacity(_pod("NVIDIA A100-SXM4-80GB")) == 100
assert p.pod_capacity(_pod("NVIDIA A40")) == max(1, round(100 / 2.9))
def test_a_stated_per_card_inflight_is_taken_as_written():
p = _pool(_mcfg(max_inflight_per_pod=100,
gpu_limits={"NVIDIA A40": {"inflight": 7}}))
assert p.pod_capacity(_pod("NVIDIA A40")) == 7
def test_threshold_scales_with_the_weight():
p = _pool()
assert p.pod_threshold(_pod("NVIDIA A100-SXM4-80GB")) == pytest.approx(32.0)
assert p.pod_threshold(_pod("NVIDIA A40")) == pytest.approx(32 / 2.9, rel=1e-6)
def test_a_stated_per_card_scale_up_is_taken_as_written():
p = _pool(_mcfg(gpu_limits={"NVIDIA A40": {"scale_up": 9}}))
assert p.pod_threshold(_pod("NVIDIA A40")) == pytest.approx(9.0)
# ------------------------------------------------------------------ memo #
def test_the_tables_are_scanned_once_per_card_not_once_per_pod(monkeypatch):
calls = []
real = rw.card_limits
monkeypatch.setattr(rw, "card_limits",
lambda m, c: calls.append(c) or real(m, c))
p = _pool(cards=[("NVIDIA A40", 48.0)] * 40 + [("NVIDIA L40S", 48.0)] * 30)
for pod in p.pods:
p.pod_weight(pod)
p.pod_capacity(pod)
p.pod_threshold(pod)
# Two distinct cards. Without the memo this was ~210 scans for 70 pods.
assert len(set(calls)) <= 4
assert len(calls) <= 12, f"{len(calls)} table scans for 2 cards"
def test_a_new_pod_invalidates_the_memo():
"""`_preferred_vram_gb` reads the pods, so the generation must move."""
p = _pool(cards=[("NVIDIA A40", 48.0)])
before = p._card_gen
with p._cv:
p.pods.append(_pod("NVIDIA L40S"))
p._bump_cards()
assert p._card_gen > before
def test_the_memo_does_not_grow_without_bound():
p = _pool()
for i in range(300):
p._card_cache[(f"card{i}", 0)] = {"weight": 1.0}
with p._cv:
p._bump_cards()
assert len(p._card_cache) == 0
def test_the_memo_answers_the_same_as_a_cold_pool():
"""A cache that is subtly wrong is worse than a slow one."""
cards = ["NVIDIA A40", "NVIDIA L40S", "A100 PCIe",
"NVIDIA A100-SXM4-80GB", "Unlisted Card"]
warm = _pool(_mcfg(max_inflight_per_pod=100))
for c in cards:
for _ in range(3):
warm.pod_weight(_pod(c))
for c in cards:
cold = _pool(_mcfg(max_inflight_per_pod=100))
assert warm.pod_weight(_pod(c)) == cold.pod_weight(_pod(c))
assert warm.pod_capacity(_pod(c)) == cold.pod_capacity(_pod(c))
assert warm.pod_threshold(_pod(c)) == cold.pod_threshold(_pod(c))
# --------------------------------------------------------------- scaling #
def _acquire_cost(n_pods: int) -> float:
"""Seconds for the per-acquire arithmetic over a fleet of n pods."""
cards = [("NVIDIA A40", 48.0), ("NVIDIA L40S", 48.0),
("A100 PCIe", 80.0)]
p = _pool(cards=[cards[i % 3] for i in range(n_pods)])
healthy = list(p.pods)
t0 = time.monotonic()
for _ in range(40):
p._pick(healthy, "")
p._pods_for_demand(healthy)
return time.monotonic() - t0
def test_the_per_acquire_cost_does_not_explode_with_the_fleet():
"""The runaway: each pod rented made every later acquire more expensive."""
small = _acquire_cost(8)
large = _acquire_cost(80)
# 10x the pods is 10x the pods to iterate, and that part is unavoidable --
# but it must not be the ~100x of recomputing per pod per call.
assert large < small * 30, (
f"8 pods: {small * 1000:.1f}ms, 80 pods: {large * 1000:.1f}ms — "
f"{large / max(small, 1e-9):.0f}x for 10x the fleet")
# ------------------------------------------------------- saying WHY not #
# "811 pages waiting with 50 pods up" has four causes needing opposite
# responses, and the pool used to return a bare None for all of them.
def _no_pod_line(pool, healthy, capsys):
pool._no_pod_said = 0.0
assert pool._pick(healthy, "") is None
return capsys.readouterr().out
def test_an_empty_pool_says_so(capsys):
p = _pool()
assert "holds no pods at all" in _no_pod_line(p, [], capsys)
def test_all_pods_unhealthy_says_booting_or_failing(capsys):
p = _pool(cards=[("NVIDIA A40", 48.0)] * 50)
for pod in p.pods:
pod.healthy = False
out = _no_pod_line(p, [], capsys)
assert "none of its 50 pod(s) is healthy" in out
assert "booting" in out
def test_all_pods_draining_says_draining(capsys):
p = _pool(cards=[("NVIDIA A40", 48.0)] * 4)
for pod in p.pods:
pod.draining = True
assert "are draining" in _no_pod_line(p, list(p.pods), capsys)
def test_all_pods_at_cap_says_the_cap(capsys):
p = _pool(_mcfg(max_inflight_per_pod=2),
cards=[("NVIDIA A100-SXM4-80GB", 80.0)] * 3)
for pod in p.pods:
pod.inflight = 2
out = _no_pod_line(p, list(p.pods), capsys)
assert "in-flight cap" in out
assert "max_inflight_per_pod" in out
def test_the_counts_are_all_there_for_an_operator(capsys):
p = _pool(cards=[("NVIDIA A40", 48.0)] * 7)
for pod in p.pods:
pod.healthy = False
p._booting, p._waiting = 22, 811
out = _no_pod_line(p, [], capsys)
for frag in ("7 in pool", "0 healthy", "22 booting", "811 waiting"):
assert frag in out, f"missing {frag!r} in: {out}"
def test_it_is_rate_limited_not_once_per_page(capsys):
"""811 waiting pages must not write 811 lines."""
p = _pool()
p._no_pod_said = 0.0
for _ in range(200):
p._pick([], "")
assert capsys.readouterr().out.count("NO POD") == 1
def test_a_pool_with_a_free_pod_says_nothing(capsys):
p = _pool(cards=[("NVIDIA A40", 48.0)])
assert p._pick(list(p.pods), "") is not None
assert "NO POD" not in capsys.readouterr().out
tests/test_runpod_nonblocking_lease.py 0 → 100644
View file @ 72bbed69
"""A lease must not wait in a thread.
0.4.18 bounded a lease attempt at 20 s and ran it on a 256-thread pool, which
made the LEASE STAGE the ceiling on everything downstream: 256 threads at one
attempt per 20 s passes ~13 pages a second however many pods are up, and a
thread that drew the growth branch sat through an entire ~9-minute boot.
Measured consequence, on a live 5,000-document run: **zero** pages dispatched
to 50 healthy pods in 41 minutes. Not one `POST /v1/ocr` in any pod's log, GPUs
at `KV 0.0%`, 811 pages waiting, and the scaler reading that unservable queue
as demand and renting to 73 pods at $43/hr. `/healthz` answered in 1.6 ms
throughout, so nothing looked broken from outside.
So `acquire(grow_inline=False)` answers now or raises `NoPodRightNow`, and the
caller waits on its event loop, where waiting is free. The pool keeps a demand
signal by counting DENIALS over a window equal to the caller's retry interval,
so the count is distinct waiting pages rather than the attempts they have made.
"""
import threading
import time
import types
import pytest
from codai.api import runpod_worker as rw
def _mcfg(**kw):
base = dict(max_pods=128, min_pods=0, max_inflight_per_pod=0,
scale_up_inflight_per_pod=32, scale_up_max_parallel_boots=32,
gpu_types=["NVIDIA A40"], gpu_weights={"NVIDIA A40": 1.0},
gpu_limits=None, sticky_sessions=True, health_path="/v1/models",
env={})
base.update(kw)
return types.SimpleNamespace(**base)
def _pool(mcfg=None, pods=()):
p = rw.RunpodPodPool.__new__(rw.RunpodPodPool)
p.mcfg = mcfg or _mcfg()
p._cv = threading.Condition(threading.RLock())
p._card_cache, p._card_gen, p._denied = {}, 0, []
p._booting = p._waiting = 0
p._affinity = {}
p.model_key = "m"
p.pods = list(pods)
p.grew = 0
def _grow():
p.grew += 1
p._grow_in_background = _grow
return p
def _pod(inflight=0, healthy=True, draining=False):
return types.SimpleNamespace(gpu="NVIDIA A40", vram_gb=48.0,
inflight=inflight, healthy=healthy,
draining=draining, pod_id="p",
url="https://10.0.0.1:1", last_used=0.0)
# ------------------------------------------------------------ answers now #
def test_an_empty_pool_raises_at_once_instead_of_renting_here():
p = _pool()
t0 = time.monotonic()
with pytest.raises(rw.NoPodRightNow):
p.acquire(timeout=600.0, grow_inline=False)
assert time.monotonic() - t0 < 0.5, "it waited"
assert p.grew == 1, "growth must still be triggered, in the background"
def test_it_does_not_hold_the_thread_for_the_timeout():
"""THE REGRESSION: 20 s per attempt x 256 threads = ~13 pages/s."""
p = _pool()
t0 = time.monotonic()
for _ in range(50):
with pytest.raises(rw.NoPodRightNow):
p.acquire(timeout=1200.0, grow_inline=False)
took = time.monotonic() - t0
assert took < 1.0, f"50 denials took {took:.1f}s — still waiting somewhere"
def test_a_free_pod_is_still_handed_out():
p = _pool(pods=[_pod()])
pod, url = p.acquire(timeout=600.0, grow_inline=False)
assert pod is not None and url
assert pod.inflight == 1
def test_the_booting_count_is_not_leaked_when_it_raises():
"""A leaked count stops `_grow_reason` firing for everyone."""
p = _pool()
for _ in range(5):
with pytest.raises(rw.NoPodRightNow):
p.acquire(timeout=600.0, grow_inline=False)
assert p._booting == 0
def test_blocking_mode_is_untouched_for_callers_that_want_it():
"""A caller with a thread to spare and no retry loop still waits."""
p = _pool(_mcfg(max_pods=0), pods=[_pod()])
pod, _url = p.acquire(timeout=5.0)
assert pod is not None
# ------------------------------------------------- demand without threads #
def test_a_denial_is_counted_as_demand():
p = _pool()
with pytest.raises(rw.NoPodRightNow):
p.acquire(grow_inline=False)
assert p._queued() >= 1
def test_retries_of_one_page_are_not_counted_as_many_pages():
"""The runaway: reading its own retries as new work rented 73 pods."""
p = _pool()
p.DENY_WINDOW_S = 0.2
for _ in range(20): # one page, retrying fast
with pytest.raises(rw.NoPodRightNow):
p.acquire(grow_inline=False)
time.sleep(0.03)
# Within one window only the attempts made in that window count.
assert p._queued() <= 10, p._queued()
def test_the_demand_decays_when_the_pages_stop_asking():
p = _pool()
p.DENY_WINDOW_S = 0.1
with pytest.raises(rw.NoPodRightNow):
p.acquire(grow_inline=False)
assert p._queued() >= 1
time.sleep(0.25)
assert p._queued() == 0
def test_threads_and_denials_are_not_added_together():
"""The same page counted two ways would double the fleet."""
p = _pool()
p._waiting = 7
for _ in range(3):
with pytest.raises(rw.NoPodRightNow):
p.acquire(grow_inline=False)
assert p._queued() == max(7, p._denied_recent())
def test_the_record_does_not_grow_without_bound():
p = _pool()
for _ in range(5000):
p._note_denied()
assert len(p._denied) <= 4096
def test_demand_still_drives_the_fleet_size():
"""With no threads parked, denials must still rent pods."""
p = _pool(pods=[_pod(inflight=1)])
for _ in range(40):
p._note_denied()
assert p._pods_for_demand([p.pods[0]]) > 1
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