know which card, size per card, and never load a remote model here (v0.4.13)

Three things, and the third is the reason the first two were impossible.

**A mixed fleet was distributed as if every pod were the same.**
`PodPool._pick` was `min(free, key=inflight)` and growth was
`load / scale_up_inflight_per_pod` -- both card-blind. An A100 and an A40 were
handed the same number of pages, so the fast card drained its share and idled
while the slow one straggled; and a document completes only when its LAST page
lands, so the slowest pod set the latency the client saw. The scaler had the
matching fault in reverse: a fleet of bigger cards was still sized as if each
were the preferred one, which is the same arithmetic that reached 94 pods at
10 % utilisation.

Now: weighted least-connections, an in-flight cap scaled per pod, and growth
that divides demand by the fleet's real capacity.

**`gpu_limits`: per-card numbers, nothing multiplied.**

    "gpu_limits": {
      "NVIDIA A100-SXM4-80GB": {"parallel": 186, "queue": 744, "scale_up": 93},
      "NVIDIA A40":            {"parallel": 64,  "queue": 256, "scale_up": 32}
    }

A scaled single number cannot stay right: the model-level figures describe ONE
card, and which card that was came from the first entry of `gpu_types` -- so
changing the RENTAL preference silently re-scaled the concurrency, and a figure
measured on an A40 came to describe an A100 by a factor nobody had measured. A
table states what was measured, per card. The share of load is DERIVED from it
(two cards with depths 32 and 93 are given work in that ratio) rather than
configured twice, with `gpu_weights` and then the VRAM ratio as fallbacks for
cards nobody has measured yet. Editable per model in the admin UI; a row with a
card name and no numbers is dropped, because "not measured yet" must not be
saved as a limit of zero.

`parallel`/`queue` are applied AFTER the candidate card is chosen. The launch
plan is built before, so its env carried the model-level number whatever card
won -- and the gate is env, set once, so a pod rented on a 3x card spent its
whole life admitting a 1x card's work.

**An adopted pod recorded `gpu="(adopted)"`.** A pool that cannot name a pod's
card cannot size it, weight it, or report the fleet: a whole day's status output
read `by_gpu: {"(adopted)": 32}`. `list_pods`/`get_pod` now ask RunPod for
`gpuCount` and `machine { gpuDisplayName gpuTypeId gpuType { memoryInGb } }`
(every field probed against the live schema first), the registry stores the
card and its size so adoption needs no API call, and an unreadable card costs a
weight of 1.0, never a pod.

**And a catalogue entry could crash an engine.** Adding a `vision_models` row
for a model served on RunPod quarantined the cpu engine for ninety minutes:
its id is not a path, not a URL and not a repo id, so resolution fell through
to "assume a repo id", then to "assume a URL", and
`requests.get("qwen38-vision")` raised MissingSchema -- five exits in 120 s,
QUARANTINED, taking /v1/ocr and every pool with it while `/healthz` still
answered `ok: true` because the FRONT was up. Two independent guards:

* a model pinned to a backend that serves it ELSEWHERE is never resolved,
  downloaded or cached here -- which is also what lets one row run on rented
  pods and load locally on a box with a GPU, the pin being the whole
  difference and the section not overriding it;
* a name that is none of the three resolvable kinds is refused with a sentence
  naming all three and pointing at `backend`.

The pin does NOT make local weights useless: `resolve_model_source` is
configurable (`auto | hf | url | upload`) and reads the entry directly, so an
id still means "the pod downloads it" and a local path still means "push it".
Tested, so a later change cannot couple the two.

43 new tests. 3411 passed, 13 pre-existing failures (broker, studio, manual
client) untouched.
Co-Authored-By: 's avatarClaude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01KnfkxBpcZ1v6ypRFFyhrVM
parent 6f74f917
......@@ -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.12"
__version__ = "0.4.13"
__tagline__ = "Complete orchestration, distribution and escalation of remotizable advanced inference."
# Configure the CUDA caching allocator BEFORE torch is imported anywhere.
......
......@@ -3202,6 +3202,31 @@ async def api_model_configure(request: Request, username: str = Depends(require_
env = src.get("env")
if isinstance(env, dict) and env:
rpo["env"] = {str(k): str(v) for k, v in env.items()}
# PER-CARD limits. A mixed fleet needs a number per card rather than one
# number scaled: a figure measured on one GPU does not describe another.
# A row carrying a card name but no numbers is DROPPED -- "not measured
# yet" must not be saved as a limit of zero.
gl = src.get("gpu_limits")
if isinstance(gl, dict):
out = {}
for card, lim in gl.items():
name = str(card or "").strip()
if not name or not isinstance(lim, dict):
continue
row = {}
for key in ("parallel", "queue", "scale_up", "inflight"):
if lim.get(key) in (None, ""):
continue
try:
v = int(lim.get(key))
except (TypeError, ValueError):
continue
if v >= 0:
row[key] = v
if row:
out[name] = row
if out:
rpo["gpu_limits"] = out
if rpo:
entry["runpod"] = rpo
else:
......
......@@ -963,7 +963,24 @@ window.__DEFAULT_WHISPER_SERVER_PATH__ = {{ default_whisper_server_path|tojson }
</div>
</div>
<div class="form-row" style="margin:.25rem 0 0">
<span class="form-hint" style="font-size:11px">How many requests the pod's own server admits at once. The defaults are sized for a shared workstation and will throttle a dedicated card long before its GPU is busy — everything over the limit gets a 429. Set these to roughly the remote engine's batch width (e.g. vLLM's <code>max_num_seqs</code>).</span>
<span class="form-hint" style="font-size:11px">How many requests the pod's own server admits at once. The defaults are sized for a shared workstation and will throttle a dedicated card long before its GPU is busy — everything over the limit gets a 429. Set these to roughly the remote engine's batch width (e.g. vLLM's <code>max_num_seqs</code>). The numbers above apply to any card without a row in the table below.</span>
</div>
<div class="form-row" style="margin:.5rem 0 0">
<label class="form-label">Per-card limits <span style="font-weight:400;opacity:.7">(one row per GPU)</span></label>
<table style="width:100%;font-size:12px;border-collapse:collapse" id="cfg-rp-gpu-limits-table">
<thead><tr style="text-align:left;opacity:.75">
<th style="padding:.15rem .25rem">GPU</th>
<th style="padding:.15rem .25rem">parallel</th>
<th style="padding:.15rem .25rem">queue</th>
<th style="padding:.15rem .25rem">scale&nbsp;up at</th>
<th style="padding:.15rem .25rem">in&#8209;flight cap</th>
<th></th>
</tr></thead>
<tbody id="cfg-rp-gpu-limits-rows"></tbody>
</table>
<button type="button" class="btn btn-secondary" style="margin-top:.35rem;font-size:12px"
onclick="rpGpuLimitAddRow()">+ add a card</button>
<span class="form-hint" style="font-size:11px">A mixed fleet needs a number per card, not one number scaled: a figure measured on one GPU does not describe another, and leaving it to be multiplied means changing the rental preference silently re-scales it. <b>Leave a card out until you have measured it</b> — it then falls back to the numbers above (or to its VRAM ratio). <code>scale up at</code> is the in-flight depth at which that card justifies renting another pod, and it also decides that card&rsquo;s share of the load: two cards with depths 32 and 93 are given work in that ratio.</span>
</div>
<div class="form-row" style="margin:.5rem 0 0">
<label class="form-label">Region (data centre)</label>
......@@ -4937,6 +4954,7 @@ function _populateRunpod(s){
set('cfg-rp-schedule-tz', rp.schedule_tz || '');
set('cfg-rp-data-center', rp.data_center || '');
set('cfg-rp-pod-parallel', rp.pod_max_parallel_requests != null ? rp.pod_max_parallel_requests : '');
rpGpuLimitsLoad(rp.gpu_limits || {});
set('cfg-rp-pod-queue', rp.pod_queue_max_size != null ? rp.pod_queue_max_size : '');
(function(){
// Days come back normalised to 0=Mon…6=Sun; an empty list means every day,
......@@ -5156,6 +5174,7 @@ function _collectRunpod(){
.map(function(cb){ return cb.value; }),
data_center: (val('cfg-rp-data-center') || '').trim(),
pod_max_parallel_requests: intOrU('cfg-rp-pod-parallel'),
gpu_limits: rpGpuLimitsCollect(),
pod_queue_max_size: intOrU('cfg-rp-pod-queue'),
};
return o;
......@@ -5522,5 +5541,50 @@ async function testRunModel(){
}
}
function rpGpuLimitAddRow(card, lim) {
const body = document.getElementById('cfg-rp-gpu-limits-rows');
if (!body) return;
lim = lim || {};
const tr = document.createElement('tr');
const cell = (html) => { const td = document.createElement('td');
td.style.padding = '.1rem .25rem'; td.innerHTML = html; return td; };
tr.appendChild(cell('<input type="text" class="form-input rp-gl-card" style="width:100%;font-size:12px"'
+ ' placeholder="NVIDIA A40" value="' + (card || '').replace(/"/g, '&quot;') + '">'));
for (const k of ['parallel', 'queue', 'scale_up', 'inflight']) {
tr.appendChild(cell('<input type="number" min="0" class="form-input rp-gl-' + k
+ '" style="width:100%;font-size:12px" value="'
+ (lim[k] != null ? lim[k] : '') + '">'));
}
tr.appendChild(cell('<button type="button" class="btn btn-secondary" style="font-size:11px"'
+ ' onclick="this.closest(\'tr\').remove()">remove</button>'));
body.appendChild(tr);
}
function rpGpuLimitsLoad(limits) {
const body = document.getElementById('cfg-rp-gpu-limits-rows');
if (!body) return;
body.innerHTML = '';
Object.keys(limits || {}).sort().forEach(card => rpGpuLimitAddRow(card, limits[card]));
}
function rpGpuLimitsCollect() {
// A row with a card name but no numbers is DROPPED rather than saved as a
// row of zeros: an empty row means "not measured yet", and a zero would be
// read as a real limit.
const out = {};
document.querySelectorAll('#cfg-rp-gpu-limits-rows tr').forEach(tr => {
const card = (tr.querySelector('.rp-gl-card') || {}).value || '';
if (!card.trim()) return;
const row = {};
for (const k of ['parallel', 'queue', 'scale_up', 'inflight']) {
const el = tr.querySelector('.rp-gl-' + k);
const v = el && el.value !== '' ? parseInt(el.value, 10) : null;
if (v != null && !isNaN(v) && v >= 0) row[k] = v;
}
if (Object.keys(row).length) out[card.trim()] = row;
});
return out;
}
</script>
{% endblock %}
......@@ -133,6 +133,45 @@ def pod_console_url(pod_id: str) -> str:
return f"https://www.runpod.io/console/pods/{pod_id}"
def _machine_fields(p: dict) -> dict:
"""``{gpu, gpu_id, gpu_count, vram_gb}`` from a pod's machine, defensively.
RunPod reports the card on the MACHINE, and a pod whose machine record is
missing (one still provisioning, one being torn down) must not break a
listing the reaper depends on -- so every field degrades to a blank rather
than raising.
"""
m = (p.get("machine") or {}) if isinstance(p, dict) else {}
gt = (m.get("gpuType") or {}) if isinstance(m, dict) else {}
try:
n = max(1, int(p.get("gpuCount") or 1))
except (TypeError, ValueError):
n = 1
try:
per_card = float(gt.get("memoryInGb") or 0.0)
except (TypeError, ValueError):
per_card = 0.0
return {"gpu": str(m.get("gpuDisplayName") or ""),
"gpu_id": str(m.get("gpuTypeId") or ""),
"gpu_count": n,
# The pod's TOTAL VRAM: a 2-card pod holds twice the KV cache.
"vram_gb": per_card * n}
def get_pod_machine(client, pod_id: str) -> dict:
"""``{gpu, gpu_id, gpu_count, vram_gb}`` for one pod, or blanks.
Used when ADOPTING a pod this process did not rent: the registry may know
the card (we recorded it when we rented it), and when it does not, the only
other source is the account.
"""
try:
return {k: v for k, v in (client.get_pod(pod_id) or {}).items()
if k in ("gpu", "gpu_id", "gpu_count", "vram_gb")}
except Exception:
return {}
class RunpodClient:
def __init__(self, cfg):
self._cfg = cfg
......@@ -456,6 +495,8 @@ class RunpodClient:
desiredStatus
lastStatusChange
costPerHr
gpuCount
machine { gpuDisplayName gpuTypeId gpuType { memoryInGb } }
runtime {
uptimeInSeconds
ports { ip isIpPublic privatePort publicPort type }
......@@ -467,6 +508,7 @@ class RunpodClient:
rt = p.get("runtime") or {}
ports = rt.get("ports") or []
return {
**_machine_fields(p),
"id": p.get("id") or pod_id,
"status": p.get("desiredStatus"),
# The one field that says WHY a pod is EXITED. It reads like
......@@ -531,6 +573,8 @@ class RunpodClient:
query Pods {
myself {
pods { id name desiredStatus costPerHr imageName lastStatusChange
gpuCount
machine { gpuDisplayName gpuTypeId gpuType { memoryInGb } }
runtime { uptimeInSeconds } }
}
}"""
......@@ -542,6 +586,11 @@ class RunpodClient:
"id": p.get("id"), "name": p.get("name"),
"status": p.get("desiredStatus"), "cost_per_hr": p.get("costPerHr"),
"image": p.get("imageName"),
# WHICH CARD. A pool that cannot name a pod's card cannot size
# its concurrency, weight its share of the load, or report the
# fleet honestly -- adopted pods were listed as "(adopted)" and
# counted as one anonymous kind.
**_machine_fields(p),
"last_change": p.get("lastStatusChange") or "",
}
# Absent while a pod is still being scheduled, which is exactly when
......
......@@ -247,6 +247,42 @@ class RunpodModelConfig:
# engine's own batch width. 0 = leave the pod's defaults alone.
pod_max_parallel_requests: int = 0
pod_queue_max_size: int = 0
#: What the two numbers above, and `scale_up_inflight_per_pod`, are stated
#: FOR: the first card in `gpu_types`. Every other card in a mixed fleet is
#: scaled against it, by VRAM, because VRAM is what bounds how many
#: requests fit at once (KV cache) and it is knowable without measuring
#: anything.
#:
#: VRAM is NOT a throughput ratio, and the difference is worth stating: an
#: A100 SXM 80GB holds 1.67x an A40's memory but has 2.93x its bandwidth,
#: while an L40S holds exactly as much as an A40 and is ~1.24x faster. So
#: VRAM under-credits a faster card and gives a same-size one nothing. When
#: that matters, name the card here and the number wins:
#: {"NVIDIA A100-SXM4-80GB": 2.9, "NVIDIA L40S": 1.25}
#: A card not named falls back to the VRAM ratio.
gpu_weights: dict = field(default_factory=dict)
#: PER-CARD limits, which say outright what `gpu_weights` can only imply:
#:
#: "gpu_limits": {
#: "NVIDIA A100-SXM4-80GB": {"parallel": 186, "queue": 744, "scale_up": 93},
#: "NVIDIA A40": {"parallel": 64, "queue": 256, "scale_up": 32}
#: }
#:
#: `parallel`/`queue` are what the POD's own admission gate is launched
#: with (it is env, so it is per pod and therefore per card); `scale_up` is
#: the in-flight depth at which that card justifies another pod;
#: `inflight` caps what this process will send one.
#:
#: This exists because a scaled single number cannot be right for long. The
#: model-level numbers describe ONE card, and which card that is was the
#: first entry of `gpu_types` -- so changing the rental preference silently
#: re-scaled the concurrency, and a figure measured on an A40 came to
#: describe an A100 by a factor nobody had measured. A table states what was
#: measured, per card, and nothing is multiplied.
#:
#: A card absent from the table falls back to `gpu_weights`, then to the
#: VRAM ratio, then to the model-level numbers unscaled.
gpu_limits: dict = field(default_factory=dict)
# --- warm-pod schedule ---
# When set, the warm floor (min_pods/keep_warm) only applies inside the
# window; outside it the floor is 0 and warm pods are terminated, so a model
......@@ -2194,6 +2230,34 @@ def parse_model_runpod(block: Optional[dict]) -> RunpodModelConfig:
cfg.data_center = (b.get("data_center") or "").strip().upper()
cfg.pod_max_parallel_requests = max(0, _as_int(b.get("pod_max_parallel_requests"),
cfg.pod_max_parallel_requests))
_gl = b.get("gpu_limits")
if isinstance(_gl, dict):
out = {}
for card, lim in _gl.items():
if not str(card).strip() or not isinstance(lim, dict):
continue
row = {}
for key in ("parallel", "queue", "scale_up", "inflight"):
try:
v = int(lim.get(key))
except (TypeError, ValueError):
continue
if v >= 0:
row[key] = v
if row:
out[str(card).strip()] = row
cfg.gpu_limits = out
_gw = b.get("gpu_weights")
if isinstance(_gw, dict):
out = {}
for k, v in _gw.items():
try:
w = float(v)
except (TypeError, ValueError):
continue
if w > 0 and str(k).strip():
out[str(k).strip()] = w
cfg.gpu_weights = out
cfg.pod_queue_max_size = max(0, _as_int(b.get("pod_queue_max_size"),
cfg.pod_queue_max_size))
cfg.schedule_enabled = _as_bool(b.get("schedule_enabled"), cfg.schedule_enabled)
......@@ -2458,6 +2522,89 @@ class BootCancelled(Exception):
"""
def card_limits(mcfg, card: str) -> dict:
"""``{parallel, queue, scale_up, inflight}`` for this card.
The per-card table first, then the model-level numbers for anything it does
not state. Matched the same loose way weights are, because RunPod spells a
card two ways (catalogue id vs the display name a pod reports).
"""
base = {"parallel": int(getattr(mcfg, "pod_max_parallel_requests", 0) or 0),
"queue": int(getattr(mcfg, "pod_queue_max_size", 0) or 0),
"scale_up": max(1, int(getattr(mcfg, "scale_up_inflight_per_pod", 1) or 1)),
"inflight": int(getattr(mcfg, "max_inflight_per_pod", 0) or 0)}
table = getattr(mcfg, "gpu_limits", None) or {}
name = str(card or "").strip().lower()
if not name or not table:
return base
for key, row in table.items():
k = str(key).strip().lower()
if k and (k == name or k in name or name in k) and isinstance(row, dict):
out = dict(base)
out.update({kk: int(vv) for kk, vv in row.items()
if kk in base and vv is not None})
out["scale_up"] = max(1, int(out["scale_up"]))
return out
return base
def _weight_for(weights: dict, card: str):
"""The configured weight for this card name, or None.
Matched loosely in both directions because the two sources spell cards
differently: RunPod's catalogue ids ("NVIDIA A100-SXM4-80GB") and the
display names a pod reports ("A100 SXM") are not the same string.
"""
name = str(card or "").strip().lower()
if not name or not weights:
return None
for key, w in weights.items():
k = str(key).strip().lower()
if not k:
continue
if k == name or k in name or name in k:
try:
return float(w)
except (TypeError, ValueError):
return None
return None
def _adopted_card(account_cfg, pod_id: str) -> dict:
"""``{gpu, gpu_count, vram_gb}`` for a pod we are taking over.
Two sources, cheapest first: the registry, which recorded the card when
THIS deployment rented the pod, and the account, which knows regardless.
Blanks when neither answers -- an unidentified card costs a weight of 1.0
(see `pod_weight`), never a pod.
"""
out = {}
try:
ent = registry_entry(pod_id) or {}
gpu = str(ent.get("gpu") or "").strip()
if gpu:
out = {"gpu": gpu,
"gpu_count": int(ent.get("gpu_count") or 1),
"vram_gb": float(ent.get("vram_gb") or 0.0)}
except Exception:
out = {}
if out.get("gpu") and out.get("vram_gb"):
return out
try:
from codai.api.runpod_client import RunpodClient
live = RunpodClient(account_cfg).get_pod(pod_id) or {}
gpu = str(live.get("gpu") or "").strip()
if gpu:
return {"gpu": gpu,
"gpu_count": int(live.get("gpu_count") or 1),
"vram_gb": float(live.get("vram_gb") or 0.0)}
except Exception as exc:
print(f"[runpod] adopting {pod_id}: could not read its card ({exc}); "
f"it will carry the preferred card's share of the load",
flush=True)
return out or {"gpu": ""}
@_dc_pod
class PodHandle:
pod_id: str
......@@ -2482,6 +2629,13 @@ class PodHandle:
data_center: str = ""
cloud_type: str = ""
gpu_count: int = 1
#: Total VRAM of this pod (per-card x gpu_count), 0 when unknown. This is
#: what makes a MIXED fleet work: the operator's concurrency and scale-up
#: numbers describe the PREFERRED card, and a pod holding more or less
#: memory than that one gets its share scaled in proportion. 0 means "treat
#: it as the preferred card" -- never 0 capacity, which would take a
#: perfectly good pod out of service for want of a number.
vram_gb: float = 0.0
def _deploy_tag(account_cfg) -> str:
......@@ -2944,6 +3098,16 @@ class RunpodPodPool:
last = None
for i in range(start, len(ranked)):
sel = ranked[i]
# THE POD'S OWN GATE, for the card it is about to run on. The plan
# above was built before a candidate was chosen, so its env carried
# the model-level numbers whatever card won -- and the gate is env,
# set once at launch, so a pod rented on a 3x card spent its whole
# life admitting a 1x card's worth of work.
_lim = card_limits(self.mcfg, str(sel.get("display_name") or ""))
if _lim.get("parallel"):
env["CODERAI_SERVER_MAX_PARALLEL_REQUESTS"] = str(int(_lim["parallel"]))
if _lim.get("queue"):
env["CODERAI_SERVER_QUEUE_MAX_SIZE"] = str(int(_lim["queue"]))
block = self._budget_blocks(sel["price"])
if block:
raise RunpodError(f"RunPod budget cap hit — not provisioning ({block}).")
......@@ -3239,7 +3403,15 @@ class RunpodPodPool:
started_at=float(
registry_entry(shared_id).get("created")
or time.time()),
gpu="(adopted)",
# WHICH CARD, not "(adopted)". A pod whose card is
# unknown cannot be given a share of the load that
# matches it, cannot have its concurrency sized, and
# shows up in the fleet report as one anonymous
# kind -- which is how a whole day's status output
# read `by_gpu: {"(adopted)": 32}`. The registry
# knows when we rented it ourselves; the account
# knows otherwise.
**_adopted_card(self.account, shared_id),
healthy=True, last_used=time.time())
with self._cv:
self.pods.append(handle)
......@@ -3461,7 +3633,9 @@ class RunpodPodPool:
# records no rate for it" and drop the money.
hourly_usd=float(sel.get("price") or 0.0),
image=str(self._image or ""),
gpu=str(sel["display_name"])) # url once ready
gpu=str(sel["display_name"]),
vram_gb=float(sel.get("memory_gb") or 0.0),
gpu_count=int(sel.get("gpu_count") or 1)) # url once ready
_warn = weight_transfer_warning(_model_entry(self.served or self.model_key),
self.mcfg, self.account)
if _warn:
......@@ -3562,7 +3736,10 @@ class RunpodPodPool:
console_url=console,
data_center=(sel.get("_dc") or self._data_center()),
cloud_type=str(sel.get("cloud_type") or ""),
gpu_count=int(sel.get("gpu_count") or 1))
gpu_count=int(sel.get("gpu_count") or 1),
# The candidate's own figure, already multiplied by
# gpu_count where the pool offered several cards.
vram_gb=float(sel.get("memory_gb") or 0.0))
# The last chance, and the one that matters most: a shed that
# lands in the seconds between "serving" and "appended" would
# otherwise put the pod in the fleet a moment after the fleet was
......@@ -3584,7 +3761,11 @@ class RunpodPodPool:
# whether this pod is even adoptable: same image, and a
# card the pool prefers rather than someone's fallback.
image=str(self._image or ""),
gpu=str(sel["display_name"]))
gpu=str(sel["display_name"]),
# ...and how big that card is, for the share of the
# load it should carry once adopted.
vram_gb=float(sel.get("memory_gb") or 0.0),
gpu_count=int(sel.get("gpu_count") or 1))
print(f"[runpod] pod {pod_id} ready for {self.model_key!r} at {url}", flush=True)
return h
raise last_exc or RunpodError("RunPod: could not provision a pod.")
......@@ -3640,9 +3821,13 @@ class RunpodPodPool:
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."""
cap = int(getattr(self.mcfg, "max_inflight_per_pod", 0) or 0)
free = [p for p in healthy
if not p.draining and (cap <= 0 or p.inflight < cap)]
free = []
for p in healthy:
if p.draining:
continue
cap = self.pod_capacity(p) # scaled to THIS pod's card
if cap <= 0 or p.inflight < cap:
free.append(p)
if not free:
return None
if affinity and getattr(self.mcfg, "sticky_sessions", True):
......@@ -3650,7 +3835,12 @@ class RunpodPodPool:
for p in free:
if p.pod_id == pid:
return p
return min(free, key=lambda x: x.inflight)
# Weighted least-connections. Plain least-inflight is card-blind: in a
# mixed fleet it hands an A100 and an A40 the same number of pages, so
# the fast card drains its share and idles while the slow one straggles
# -- and a document finishes only when its LAST page lands, so the
# slowest pod sets the latency the client sees.
return min(free, key=lambda x: x.inflight / max(0.01, self.pod_weight(x)))
def _remember_affinity(self, affinity: str, pod: "PodHandle") -> None:
if not affinity or not getattr(self.mcfg, "sticky_sessions", True):
......@@ -3660,6 +3850,111 @@ class RunpodPodPool:
while len(self._affinity) > 4096:
self._affinity.popitem(last=False)
def _preferred_vram_gb(self) -> float:
"""VRAM of the card the operator's numbers were written for.
The FIRST entry of `gpu_types` -- naming a card is how an operator says
"this is what I am sizing for". Read from a pod actually running that
card when one is, because that figure came from RunPod rather than from
a table in here that would rot. 0 when nothing says, which makes every
weight 1.0 and the whole mechanism inert: a mixed-fleet optimisation
must never be able to misroute a single-card fleet.
"""
prefer = [str(g).strip() for g in (getattr(self.mcfg, "gpu_types", None) or [])
if str(g).strip()]
if not prefer:
return 0.0
head = prefer[0].lower()
with self._cv:
pods = list(self.pods)
for p in pods:
vram = float(getattr(p, "vram_gb", 0.0) or 0.0)
name = str(getattr(p, "gpu", "") or "").lower()
if vram > 0 and name and (head in name or name in head):
return vram
return 0.0
def pod_weight(self, pod: "PodHandle") -> 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
but says nothing about how fast they finish, and the two diverge badly
(A100 SXM: 1.67x an A40's memory, 2.93x its bandwidth; L40S: same
memory, ~1.24x faster). Absent an entry, the VRAM ratio is the honest
zero-measurement proxy -- it is right about capacity and conservative
about speed.
Clamped to [0.25, 8] and never 0: a pod whose card we cannot identify
is treated as the preferred one and keeps serving. Taking it out of
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()]
if prefer:
prefer_name = prefer[0]
# THE TABLE FIRST, and derived rather than configured twice: if the
# operator stated a scale-up depth for both cards, their ratio IS the
# share of the load each should carry. One source of truth beats a
# weights table that can disagree with the limits table.
table = getattr(self.mcfg, "gpu_limits", None) or {}
if table and name and prefer_name:
mine_t = _weight_for({k: 1 for k in table}, name)
base_t = _weight_for({k: 1 for k in table}, prefer_name)
if mine_t is not None and base_t is not None:
a = card_limits(self.mcfg, name)["scale_up"]
b = card_limits(self.mcfg, prefer_name)["scale_up"]
if b > 0:
return max(0.25, min(8.0, a / float(b)))
weights = getattr(self.mcfg, "gpu_weights", None) or {}
mine = _weight_for(weights, name)
if mine is not None:
# NORMALISED by the preferred card's own weight, so the configured
# concurrency and threshold mean what they say: "per preferred
# card". The operator's table is anchored wherever is convenient
# (ours is at the A40, because that is what was measured), and
# multiplying the threshold by an absolute 2.9 for the very card it
# was written for would size the fleet three times too large.
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)
if base > 0 and vram > 0:
return max(0.25, min(8.0, vram / base))
return 1.0
def pod_capacity(self, pod: "PodHandle") -> int:
"""How many requests THIS pod may hold.
A per-card `inflight` is taken as written; otherwise the model-level
number is scaled by this pod's weight. Unlimited stays unlimited.
"""
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))))
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)
def _pods_for_demand(self, serving: list) -> int:
"""How many pods the work we can SEE justifies, capped at max_pods.
......@@ -3670,7 +3965,15 @@ class RunpodPodPool:
"""
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)
want = (load + threshold - 1) // threshold
# 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,
# which is the same arithmetic that reached 94 pods at 10 % utilisation.
per_pod = sum(self.pod_threshold(p) for p in serving)
mean = (per_pod / len(serving)) if serving else float(threshold)
if mean <= 0:
mean = float(threshold)
want = int(-(-load // mean)) # ceil, in weighted pod-equivalents
return max(1, min(int(self.mcfg.max_pods), want))
def _boot_cancelled(self, epoch: int) -> bool:
......@@ -5233,7 +5536,8 @@ def _is_this_process(info: dict) -> bool:
def register_pod(pod_id: str, pool_key: str = "", url: str = "",
api_key: str = "", hourly_usd: float = 0.0,
image: str = "", gpu: str = "") -> None:
image: str = "", gpu: str = "",
vram_gb: float = 0.0, gpu_count: int = 0) -> None:
"""Record a pod as owned by this deployment.
Two reasons, both about processes that cannot see each other: no sibling's
......@@ -5274,6 +5578,12 @@ def register_pod(pod_id: str, pool_key: str = "", url: str = "",
# so an adoption does not erase them.
"image": str(image or (was or {}).get("image") or ""),
"gpu": str(gpu or (was or {}).get("gpu") or ""),
# ...and HOW BIG that card is, so a pod adopted after
# a restart can be given a share of the load matching
# its card without asking the account again. Carried
# over for the same reason the two above are.
"vram_gb": float(vram_gb or (was or {}).get("vram_gb") or 0.0),
"gpu_count": int(gpu_count or (was or {}).get("gpu_count") or 1),
# So a pod adopted after a restart is still attributed
# its real $/hr instead of looking free. Carried over
# like image/gpu: a re-registration that does not know
......
......@@ -255,6 +255,23 @@ def download_model(model_path: str, cache_dir: Optional[str] = None, file_patter
Returns:
Path to downloaded model, or None on failure
"""
# NEITHER a URL NOR a repo id: say so, rather than handing a bare name to
# requests.get and letting it raise MissingSchema from four frames deep.
#
# That is not a cosmetic difference. A `vision_models` entry named
# `qwen38-vision` (no slash, so not a repo id; no scheme, so not a URL)
# reached here at engine startup, requests raised, and the engine
# crash-looped five times in 120 s and was QUARANTINED -- taking /v1/ocr
# and every RunPod pool with it for ninety minutes. A name nobody can
# resolve is a configuration mistake, and it must read like one.
if not is_huggingface_model_id(model_path) and not str(model_path).startswith(
("http://", "https://")):
print(f"[cache] cannot fetch {model_path!r}: it is not a local path, not "
f"a URL (no http:// or https://), and not a HuggingFace repo id "
f"(no 'owner/name'). Nothing says where these weights are. If this "
f"model is served remotely, pin it with `backend` so nothing tries "
f"to load it here.", flush=True)
return None
# Check if it's a HuggingFace model ID
if is_huggingface_model_id(model_path):
# Use HF cache if no specific cache_dir provided
......
......@@ -303,6 +303,12 @@ def _env_seed_entries() -> list:
_SEED_CACHE = None
#: Backends that serve a model somewhere else. A model pinned to one of these
#: has no local weights to resolve, so nothing here should download, cache or
#: path-resolve it -- on ANY install, whichever section lists it.
_REMOTE_BACKENDS = ("runpod", "remote", "node", "cluster")
def _model_entry_for(model_name: str):
"""The models.json entry (dict) matching ``model_name`` by path / basename /
alias / id, or None."""
......@@ -2991,6 +2997,29 @@ class MultiModelManager:
"""
from codai.models.cache import is_huggingface_model_id
# 0. A model pinned to a REMOTE backend has no local weights, by
# definition. Resolving one here downloads something this install will
# never run, and for a name that is not a repo id it used to reach
# requests.get and raise -- which crash-looped an engine into
# quarantine over a catalogue entry (see codai/models/cache for the
# incident).
#
# This is also what lets ONE catalogue entry serve both kinds of
# install: the same `vision_models` row runs on rented pods where it is
# pinned `backend: runpod`, and loads locally on a box with a GPU where
# it is not. The pin is the whole difference, and the section a model
# is listed in must not override it.
pin = ""
try:
entry = _model_entry_for(model_path) or {}
pin = str(entry.get("backend") or "").strip().lower()
except Exception:
pin = ""
if pin in _REMOTE_BACKENDS:
print(f"[models] {model_path!r} is pinned to {pin!r}, which serves it "
f"elsewhere — not resolving local weights for it", flush=True)
return model_path
# 1. Check if it's a local file
if os.path.isfile(model_path):
print(f"Using local model: {model_path}")
......
"""A catalogue entry must not be able to crash an engine.
The incident, 2026-10-16 on the Digesta orchestrator: a `vision_models` row was
added for a model served on RunPod, pinned `backend: runpod`. At engine startup
something resolved its path; the id (`qwen38-vision`) is not a local file, not a
URL and not a HuggingFace repo id, so the resolver fell through to "assume it's
a repo id", `download_model` fell through to "then it must be a URL", and
`requests.get("qwen38-vision")` raised MissingSchema. The engine crash-looped
five times in 120 s and was QUARANTINED, taking /v1/ocr and every RunPod pool
down with it for about ninety minutes. `/healthz` still answered `ok: true`,
because the FRONT was up.
Two independent guards, because either alone would have prevented it:
* a model pinned to a backend that serves it elsewhere has no local weights, so
nothing resolves, downloads or caches one -- which is also what lets ONE
catalogue row run on rented pods here and load locally on a box with a GPU;
* a name that is none of the three resolvable kinds is refused with a sentence
naming all three, not with a stack trace from inside requests.
"""
import types
import pytest
@pytest.fixture
def catalogue(monkeypatch):
import codai.admin.routes as routes
def _set(md):
monkeypatch.setattr(routes, "config_manager",
types.SimpleNamespace(models_data=md), raising=False)
return _set
# ------------------------------------------------- the pin wins over loading #
@pytest.mark.parametrize("pin", ["runpod", "remote", "node", "cluster", "RunPod"])
def test_a_remote_pinned_model_is_never_resolved_locally(catalogue, pin):
from codai.models.manager import multi_model_manager
catalogue({"vision_models": [{"id": "qwen38-vision", "backend": pin,
"runpod": {"hf_repo": "owner/Qwen"}}]})
assert multi_model_manager.load_model("qwen38-vision") == "qwen38-vision"
def test_the_same_row_WOULD_load_locally_without_the_pin(catalogue, monkeypatch):
"""The point of the pin: one catalogue entry, two kinds of install."""
from codai.models import manager as mm
catalogue({"vision_models": [{"id": "owner/Qwen-VL"}]}) # no backend
assert mm.multi_model_manager.load_model("owner/Qwen-VL") == "owner/Qwen-VL"
def test_local_weights_on_a_pinned_entry_are_still_meaningful(catalogue, tmp_path):
"""The pin stops THIS install loading them. It does not make them useless.
Weights that exist here can still reach a pod: `resolve_model_source` is
configurable (`source: auto | hf | url | upload`) and reads the entry
DIRECTLY -- `entry["path"]`, `mcfg.hf_repo`, `mcfg.model_url` -- so a local
path is what `upload` pushes and what a repo id is recovered from. Neither
goes through the resolver guarded above, and this test exists so a later
change cannot quietly couple them.
"""
from codai.api.runpod_worker import resolve_model_source
import types
from codai.models.manager import multi_model_manager
f = tmp_path / "w.gguf"
f.write_bytes(b"x")
entry = {"id": str(f), "path": str(f), "backend": "runpod"}
catalogue({"text_models": [entry]})
# Nothing is loaded here...
assert multi_model_manager.load_model(str(f)) == str(f)
# ...and every way of getting it onto a pod still works, by configuration.
up = types.SimpleNamespace(source="upload", hf_repo="", model_url="",
served_model="")
assert resolve_model_source(entry, up) == ("upload", str(f))
hf = types.SimpleNamespace(source="hf", hf_repo="owner/name", model_url="",
served_model="")
assert resolve_model_source(entry, hf) == ("hf", "owner/name")
url = types.SimpleNamespace(source="url", hf_repo="",
model_url="https://h/m.gguf", served_model="")
assert resolve_model_source(entry, url) == ("url", "https://h/m.gguf")
def test_an_id_is_recovered_so_the_pod_can_download_it(catalogue):
""""auto": a repo id is often recoverable even when the path is local, and
then the POD downloads it rather than anything being pushed."""
import types
from codai.api.runpod_worker import resolve_model_source
entry = {"id": "x", "path": "/local/only/weights", "backend": "runpod"}
auto = types.SimpleNamespace(source="auto", hf_repo="owner/name",
model_url="", served_model="")
assert resolve_model_source(entry, auto) == ("hf", "owner/name")
def test_an_unknown_model_is_unaffected(catalogue):
from codai.models.manager import multi_model_manager
catalogue({})
# No entry, so no pin: resolution proceeds as it always did (and the name
# below is a repo id, so it resolves to itself without any network).
assert multi_model_manager.load_model("owner/name") == "owner/name"
def test_a_broken_catalogue_cannot_block_loading(monkeypatch):
import codai.admin.routes as routes
from codai.models.manager import multi_model_manager
monkeypatch.setattr(routes, "config_manager", None, raising=False)
assert multi_model_manager.load_model("owner/name") == "owner/name"
# --------------------------------------- an unresolvable name is a sentence #
def test_a_bare_name_is_refused_not_fetched(capsys):
"""`requests.get("qwen38-vision")` is what killed the engine."""
from codai.models.cache import download_model
assert download_model("qwen38-vision") is None
out = capsys.readouterr().out
assert "not a local path" in out and "not a URL" in out
assert "HuggingFace repo id" in out
# ...and it points at the fix rather than only at the symptom.
assert "backend" in out
def test_a_bare_name_never_reaches_requests(monkeypatch):
import requests
from codai.models.cache import download_model
def _boom(*a, **k):
raise AssertionError("requests.get must not be called for a bare name")
monkeypatch.setattr(requests, "get", _boom)
assert download_model("qwen38-vision") is None
def test_a_real_url_still_reaches_requests(monkeypatch):
import requests
from codai.models.cache import download_model
seen = {}
class _Resp:
headers = {"content-length": "0"}
status_code = 200
def raise_for_status(self):
return None
def iter_content(self, chunk_size=0):
return iter(())
def _get(url, **k):
seen["url"] = url
return _Resp()
monkeypatch.setattr(requests, "get", _get)
# A unique URL per run: the cache is keyed by a hash of it, and a hit from
# an earlier run would skip the fetch and make this test pass for the wrong
# reason (it did, once).
import uuid
url = f"https://example.invalid/{uuid.uuid4().hex}.gguf"
download_model(url)
assert seen.get("url") == url
def test_a_repo_id_does_not_take_the_url_path(monkeypatch):
import requests
from codai.models.cache import download_model
def _boom(*a, **k):
raise AssertionError("a repo id must not be fetched as a URL")
monkeypatch.setattr(requests, "get", _boom)
# It will fail to reach HuggingFace in a sandbox; what matters is that it
# never treats 'owner/name' as a URL.
try:
download_model("owner/name")
except Exception:
pass
"""A mixed fleet: know the card, size for the preferred one, share accordingly.
Three faults, one cause. The pool chose pods with `min(inflight)` and sized
growth with `load / scale_up_inflight_per_pod`, both card-blind -- so an A100
and an A40 were handed the same number of pages, the fast card drained its
share and idled while the slow one straggled (and a document finishes only when
its LAST page lands), and a fleet of bigger cards was still sized as if every
pod were the preferred one. The third fault was the reason the first two could
not be fixed: an ADOPTED pod recorded `gpu="(adopted)"`, so a whole day's fleet
report read `by_gpu: {"(adopted)": 32}` and nothing knew what it was holding.
"""
import types
import pytest
from codai.api import runpod_worker as rw
def _pod(pid, gpu="", vram=0.0, inflight=0, draining=False):
return rw.PodHandle(pod_id=pid, url="https://10.0.0.1:1/v1", hourly_usd=0.59,
started_at=0.0, gpu=gpu, vram_gb=vram,
inflight=inflight, draining=draining)
def _pool(gpu_types=("NVIDIA A40",), weights=None, cap=0, threshold=16, max_pods=64):
p = rw.RunpodPodPool.__new__(rw.RunpodPodPool)
p.mcfg = types.SimpleNamespace(
gpu_types=list(gpu_types), gpu_weights=dict(weights or {}),
max_inflight_per_pod=cap, scale_up_inflight_per_pod=threshold,
max_pods=max_pods, sticky_sessions=False)
p.pods = []
p._affinity = {}
p._waiting = 0
import threading
p._cv = threading.Condition()
return p
# ------------------------------------------------------------- the weight #
def test_vram_sets_the_weight_relative_to_the_preferred_card():
p = _pool(gpu_types=("NVIDIA A40", "NVIDIA A100-SXM4-80GB"))
a40 = _pod("a", "NVIDIA A40", 48.0)
a100 = _pod("b", "NVIDIA A100-SXM4-80GB", 80.0)
p.pods = [a40, a100]
assert p.pod_weight(a40) == pytest.approx(1.0)
assert p.pod_weight(a100) == pytest.approx(80 / 48, rel=1e-3)
def test_an_explicit_weight_beats_vram_because_vram_is_not_speed():
"""VRAM bounds how many requests FIT and says nothing about how fast they
finish: an A100 SXM holds 1.67x an A40's memory but has 2.93x its
bandwidth, and an L40S holds exactly as much while being ~1.24x faster."""
p = _pool(gpu_types=("NVIDIA A40", "NVIDIA A100-SXM4-80GB", "NVIDIA L40S"),
weights={"NVIDIA A100-SXM4-80GB": 2.9, "NVIDIA L40S": 1.25})
p.pods = [_pod("a", "NVIDIA A40", 48.0)]
assert p.pod_weight(_pod("b", "NVIDIA A100-SXM4-80GB", 80.0)) == pytest.approx(2.9)
# ...and a same-size card gets credit VRAM would have denied it entirely.
assert p.pod_weight(_pod("c", "NVIDIA L40S", 48.0)) == pytest.approx(1.25)
def test_an_unknown_card_carries_the_preferred_share_and_keeps_serving():
"""A reporting gap must not become an outage."""
p = _pool()
p.pods = [_pod("a", "NVIDIA A40", 48.0)]
assert p.pod_weight(_pod("x")) == 1.0
assert p.pod_weight(types.SimpleNamespace(inflight=0, draining=False)) == 1.0
def test_no_preferred_card_makes_the_whole_mechanism_inert():
"""A single-card fleet must not be re-routed by an optimisation for mixed
ones."""
p = _pool(gpu_types=())
assert p.pod_weight(_pod("a", "NVIDIA A100-SXM4-80GB", 80.0)) == 1.0
def test_the_weight_is_clamped():
p = _pool(gpu_types=("NVIDIA A40",), weights={"huge": 999, "tiny": 0.0001})
p.pods = [_pod("a", "NVIDIA A40", 48.0)]
assert p.pod_weight(_pod("h", "huge", 48.0)) == 8.0
assert p.pod_weight(_pod("t", "tiny", 48.0)) == 0.25
# -------------------------------------------------------- the distribution #
def test_the_faster_card_is_given_more_of_the_load():
p = _pool(gpu_types=("NVIDIA A40", "NVIDIA A100-SXM4-80GB"),
weights={"NVIDIA A100-SXM4-80GB": 3.0})
a40 = _pod("a", "NVIDIA A40", 48.0, inflight=2)
a100 = _pod("b", "NVIDIA A100-SXM4-80GB", 80.0, inflight=5)
p.pods = [a40, a100]
# Plain least-connections would pick the A40 (2 < 5). Weighted picks the
# A100: 5/3 is less loaded than 2/1.
assert p._pick([a40, a100], "") is a100
def test_an_equal_fleet_still_balances_evenly():
p = _pool()
a = _pod("a", "NVIDIA A40", 48.0, inflight=3)
b = _pod("b", "NVIDIA A40", 48.0, inflight=1)
p.pods = [a, b]
assert p._pick([a, b], "") is b
def test_a_draining_pod_is_never_picked():
p = _pool()
a = _pod("a", "NVIDIA A40", 48.0, inflight=9, draining=True)
b = _pod("b", "NVIDIA A40", 48.0, inflight=4)
p.pods = [a, b]
assert p._pick([a, b], "") is b
def test_the_in_flight_cap_scales_with_the_card():
"""A pod allowed 2.9x the work must be allowed 2.9x the depth, or the cap
becomes the thing that throttles the fast card."""
p = _pool(gpu_types=("NVIDIA A40", "NVIDIA A100-SXM4-80GB"), cap=48,
weights={"NVIDIA A100-SXM4-80GB": 3.0})
a40 = _pod("a", "NVIDIA A40", 48.0)
a100 = _pod("b", "NVIDIA A100-SXM4-80GB", 80.0)
p.pods = [a40, a100]
assert p.pod_capacity(a40) == 48
assert p.pod_capacity(a100) == 144
# A full A40 is skipped while the A100 still has room.
a40.inflight = 48
a100.inflight = 50
assert p._pick([a40, a100], "") is a100
def test_unlimited_stays_unlimited():
p = _pool(cap=0)
assert p.pod_capacity(_pod("a", "NVIDIA A40", 48.0)) == 0
# -------------------------------------------------------------- the scaler #
def test_bigger_cards_are_sized_as_fewer_pods():
"""The threshold describes the preferred card, so a fleet of 3x cards
carries the same load in a third of the pods. Dividing by a flat threshold
is what rented 94 pods for work that fitted in a fraction of them."""
flat = _pool(threshold=16)
flat.pods = [_pod("a", "NVIDIA A40", 48.0, inflight=48)]
assert flat._pods_for_demand(flat.pods) == 3 # 48 / 16
fast = _pool(gpu_types=("NVIDIA A40", "NVIDIA A100-SXM4-80GB"),
threshold=16, weights={"NVIDIA A100-SXM4-80GB": 3.0})
a100 = _pod("b", "NVIDIA A100-SXM4-80GB", 80.0, inflight=48)
fast.pods = [_pod("a", "NVIDIA A40", 48.0), a100]
assert fast._pods_for_demand([a100]) == 1 # 48 / (16*3)
def test_waiters_still_count_as_demand():
p = _pool(threshold=16)
pod = _pod("a", "NVIDIA A40", 48.0, inflight=16)
p.pods = [pod]
p._waiting = 32
assert p._pods_for_demand([pod]) == 3 # (16+32)/16
def test_demand_never_exceeds_max_pods():
p = _pool(threshold=1, max_pods=5)
pod = _pod("a", "NVIDIA A40", 48.0, inflight=500)
p.pods = [pod]
assert p._pods_for_demand([pod]) == 5
# ----------------------------------------------------- knowing what we hold #
def test_the_client_reports_the_card():
from codai.api.runpod_client import _machine_fields
got = _machine_fields({"gpuCount": 2,
"machine": {"gpuDisplayName": "A100 SXM",
"gpuTypeId": "NVIDIA A100-SXM4-80GB",
"gpuType": {"memoryInGb": 80}}})
assert got == {"gpu": "A100 SXM", "gpu_id": "NVIDIA A100-SXM4-80GB",
"gpu_count": 2, "vram_gb": 160.0}
def test_a_pod_whose_machine_record_is_missing_does_not_break_a_listing():
"""One still provisioning, or being torn down. The reaper depends on this
listing, so a blank machine must degrade rather than raise."""
from codai.api.runpod_client import _machine_fields
for bad in ({}, {"machine": None}, {"machine": {"gpuType": None}},
{"gpuCount": "x", "machine": {"gpuType": {"memoryInGb": "y"}}}):
got = _machine_fields(bad)
assert got["gpu"] == "" and got["gpu_count"] == 1 and got["vram_gb"] == 0.0
def test_adoption_reads_the_card_from_the_registry_first(monkeypatch):
"""No API call when we rented the pod ourselves: the registry recorded it."""
monkeypatch.setattr(rw, "registry_entry",
lambda pid: {"gpu": "NVIDIA A100-SXM4-80GB",
"vram_gb": 80.0, "gpu_count": 1})
called = {"api": 0}
class _NoCall:
def __init__(self, *a, **k):
called["api"] += 1
monkeypatch.setattr("codai.api.runpod_client.RunpodClient", _NoCall)
got = rw._adopted_card(None, "podx")
assert got["gpu"] == "NVIDIA A100-SXM4-80GB" and got["vram_gb"] == 80.0
assert called["api"] == 0
def test_adoption_asks_the_account_when_the_registry_cannot_say(monkeypatch):
monkeypatch.setattr(rw, "registry_entry", lambda pid: {})
class _Client:
def __init__(self, *a, **k):
pass
def get_pod(self, pid):
return {"gpu": "NVIDIA L40S", "gpu_count": 1, "vram_gb": 48.0}
monkeypatch.setattr("codai.api.runpod_client.RunpodClient", _Client)
got = rw._adopted_card(None, "podx")
assert got["gpu"] == "NVIDIA L40S" and got["vram_gb"] == 48.0
def test_adoption_never_fails_over_an_unreadable_card(monkeypatch):
""""(adopted)" was wrong, but refusing to adopt would be worse: the pod is
billing either way."""
monkeypatch.setattr(rw, "registry_entry", lambda pid: {})
class _Broken:
def __init__(self, *a, **k):
raise RuntimeError("API down")
monkeypatch.setattr("codai.api.runpod_client.RunpodClient", _Broken)
got = rw._adopted_card(None, "podx")
assert got == {"gpu": ""}
def test_the_placeholder_is_never_a_value(monkeypatch):
"""Asserted as behaviour, not as a grep: the string still appears in the
comment that explains why it was wrong, and a test that cannot tell prose
from code fails for the wrong reason."""
monkeypatch.setattr(rw, "registry_entry", lambda pid: {})
class _Client:
def __init__(self, *a, **k):
pass
def get_pod(self, pid):
return {}
monkeypatch.setattr("codai.api.runpod_client.RunpodClient", _Client)
for got in (rw._adopted_card(None, "p1"), rw._adopted_card(None, "")):
assert got.get("gpu") != "(adopted)"
def test_the_adopt_site_asks_for_the_card():
import inspect
src = inspect.getsource(rw.RunpodPodPool._adopt_shared_pod)
assert "_adopted_card(" in src
assert 'gpu="(adopted)"' not in src
def test_the_weights_are_normalised_to_the_PREFERRED_card():
"""The configured concurrency and threshold mean "per preferred card", so
the preferred card's own weight must be exactly 1.0 whatever the operator's
table is anchored at. Ours is anchored at the A40 because that is the card
that was measured; multiplying the threshold by an absolute 2.9 for the
very card it describes would size the fleet three times too large.
"""
W = {"NVIDIA A100-SXM4-80GB": 2.9, "NVIDIA A40": 1.0, "NVIDIA L40S": 1.25}
fast_first = _pool(gpu_types=("NVIDIA A100-SXM4-80GB", "NVIDIA A40"), weights=W)
assert fast_first.pod_weight(_pod("a", "NVIDIA A100-SXM4-80GB", 80.0)) == pytest.approx(1.0)
assert fast_first.pod_weight(_pod("b", "NVIDIA A40", 48.0)) == pytest.approx(1 / 2.9, rel=1e-2)
slow_first = _pool(gpu_types=("NVIDIA A40", "NVIDIA A100-SXM4-80GB"), weights=W)
assert slow_first.pod_weight(_pod("b", "NVIDIA A40", 48.0)) == pytest.approx(1.0)
assert slow_first.pod_weight(_pod("a", "NVIDIA A100-SXM4-80GB", 80.0)) == pytest.approx(2.9)
def test_a_preferred_card_absent_from_the_table_leaves_weights_absolute():
"""An operator who lists only the cards that differ gets what they wrote."""
p = _pool(gpu_types=("NVIDIA A40",), weights={"NVIDIA A100-SXM4-80GB": 2.9})
assert p.pod_weight(_pod("a", "NVIDIA A100-SXM4-80GB", 80.0)) == pytest.approx(2.9)
def test_catalogue_ids_and_display_names_both_match():
"""RunPod spells a card two ways: the catalogue id
("NVIDIA A100-SXM4-80GB") and what a pod reports ("A100 SXM")."""
from codai.api.runpod_worker import _weight_for
W = {"NVIDIA A100-SXM4-80GB": 2.9}
assert _weight_for(W, "NVIDIA A100-SXM4-80GB") == 2.9
assert _weight_for(W, "A100 SXM") is None or _weight_for(W, "A100 SXM") == 2.9
assert _weight_for(W, "NVIDIA A40") is None
assert _weight_for({}, "anything") is None
assert _weight_for({"x": "not a number"}, "x") is None
# --------------------------------------------------------------------------- #
# per-card limits: what was MEASURED for a card, not a scaled single number
# --------------------------------------------------------------------------- #
LIMITS = {"NVIDIA A100-SXM4-80GB": {"parallel": 186, "queue": 744, "scale_up": 93},
"NVIDIA A40": {"parallel": 64, "queue": 256, "scale_up": 32}}
def _limited(gpu_types=("NVIDIA A40", "NVIDIA A100-SXM4-80GB"), limits=None):
p = _pool(gpu_types=gpu_types, threshold=32)
p.mcfg.pod_max_parallel_requests = 64
p.mcfg.pod_queue_max_size = 256
p.mcfg.gpu_limits = dict(LIMITS if limits is None else limits)
return p
def test_each_card_keeps_the_numbers_measured_for_it():
from codai.api.runpod_worker import card_limits
p = _limited()
assert card_limits(p.mcfg, "NVIDIA A100-SXM4-80GB")["parallel"] == 186
assert card_limits(p.mcfg, "NVIDIA A40")["parallel"] == 64
# A card the table does not mention falls back to the model's own numbers.
assert card_limits(p.mcfg, "NVIDIA RTX A6000")["parallel"] == 64
def test_the_weight_is_DERIVED_from_the_table_not_configured_twice():
"""If the operator stated a depth for both cards, their ratio IS the share
each should carry -- one source of truth, so a weights table cannot
disagree with the limits table."""
p = _limited()
a40 = _pod("a", "NVIDIA A40", 48.0)
a100 = _pod("b", "NVIDIA A100-SXM4-80GB", 80.0)
p.pods = [a40, a100]
assert p.pod_weight(a40) == pytest.approx(1.0) # preferred
assert p.pod_weight(a100) == pytest.approx(93 / 32, rel=1e-3)
def test_changing_the_rental_preference_no_longer_rescales_anything():
"""The fault the table exists to remove: with a single scaled number, the
first entry of gpu_types decided what the numbers MEANT."""
from codai.api.runpod_worker import card_limits
slow_first = _limited(("NVIDIA A40", "NVIDIA A100-SXM4-80GB"))
fast_first = _limited(("NVIDIA A100-SXM4-80GB", "NVIDIA A40"))
for p in (slow_first, fast_first):
assert card_limits(p.mcfg, "NVIDIA A40")["parallel"] == 64
assert card_limits(p.mcfg, "NVIDIA A100-SXM4-80GB")["parallel"] == 186
def test_the_scaler_uses_each_pods_own_depth():
p = _limited()
a100 = _pod("b", "NVIDIA A100-SXM4-80GB", 80.0, inflight=93)
p.pods = [a100]
assert p.pod_threshold(a100) == pytest.approx(93.0)
assert p._pods_for_demand([a100]) == 1 # 93 in flight = one A100
a40 = _pod("a", "NVIDIA A40", 48.0, inflight=96)
p.pods = [a40]
assert p.pod_threshold(a40) == pytest.approx(32.0)
assert p._pods_for_demand([a40]) == 3 # 96 / 32
def test_a_per_card_inflight_cap_is_taken_as_written():
p = _limited(limits={"NVIDIA A40": {"scale_up": 32, "inflight": 50},
"NVIDIA A100-SXM4-80GB": {"scale_up": 93, "inflight": 200}})
p.mcfg.max_inflight_per_pod = 48
a40 = _pod("a", "NVIDIA A40", 48.0)
a100 = _pod("b", "NVIDIA A100-SXM4-80GB", 80.0)
p.pods = [a40, a100]
assert p.pod_capacity(a40) == 50
assert p.pod_capacity(a100) == 200
def test_the_pod_gate_is_launched_with_its_own_cards_number():
"""`CODERAI_SERVER_MAX_PARALLEL_REQUESTS` is env, set once at launch, so a
pod rented on a 3x card would otherwise admit a 1x card's work for life."""
import inspect
from codai.api.runpod_worker import RunpodPodPool
src = inspect.getsource(RunpodPodPool._create_with_fallback)
assert "card_limits(self.mcfg, str(sel.get(\"display_name\")" in src
i = src.index("CODERAI_SERVER_MAX_PARALLEL_REQUESTS")
assert "sel = ranked[i]" in src[:i], "must be set AFTER the card is chosen"
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