a pool may not ask for more threads than the process is allowed (v0.4.18)

A 5,000-document run leased 48 warm pods and then read 278 pages ON THE
ORCHESTRATOR -- a box with no GPU -- while every one of those cards sat
idle. Each page logged "the pod-side pipeline could not run (can't start
new thread)". Zero documents were read on the pods.

The container's task budget was 2048 (podman's --pids-limit default) and
the pod pool had sized itself 2048 from ocr.max_concurrency: 4096. It
reached 1992 threads; the next one could not exist. Nothing had looked for
a limit, so nothing reported one.

Three separable mistakes, all three fixed:

  * `thread_pool.task_budget()` reads what the process is actually allowed
    -- cgroup v2 pids.max walked up to the root (a limit on any ancestor
    binds as hard as one on the leaf), minus pids.current because the other
    processes in the cgroup already spend it, and RLIMIT_NPROC -- and
    `pool_sizes()` fits inside it, keeping 256 tasks in reserve so a
    rasteriser worker, an isolated-venv engine or a `podman exec` to look at
    the box can still fork. The general pool is protected ahead of the pod
    pool: an install that cannot answer /healthz cannot be diagnosed. On the
    numbers above it now sizes 256 + 1506 instead of asking for 2304.

  * A lease took a POD-POOL thread per in-flight page, so the page width an
    install wants decided how many thread stacks it needs. A lease is a
    bounded wait (POD_LEASE_ATTEMPT_S), so a separate 256-thread lease pool
    serves thousands of pages by cycling. It also stops a cold start filling
    every thread with leases waiting for pods, leaving none to carry the
    pages those pods were for.

  * "Cannot create a thread" was treated as "the pods cannot run the
    pipeline", and the fallback for that is the local pipeline -- which needs
    MORE of this box than the path that just failed for want of it. Local
    exhaustion now answers 503 naming both ways out (the task budget, or
    ocr.max_concurrency) and leaves the pages for the cards.

AND THE TEARDOWN, from the same run: the client's own release sweep answered
500 five times in three seconds, because terminating a pod that was already
gone raised. By the time a fleet teardown reaches its last pod some are
always gone -- reaped for idle, shed by a scale-down, or released by a retry
of the same call -- so a teardown that was working looked like one that had
failed. Already gone is the goal, not a failure: it answers 200 with
`already_gone: true`, while a real failure still raises so a pod left
billing is never reported as stopped.

Also: a 400 on an upload path now logs why. A malformed or truncated
multipart body is rejected before any handler runs, so 19 of 137 batches
failed in that run with no record anywhere but the access line -- not the
filename, not the size, not even whether it was the body or a field.
Co-Authored-By: 's avatarClaude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01KnfkxBpcZ1v6ypRFFyhrVM
parent 174a3753
...@@ -16,7 +16,7 @@ ...@@ -16,7 +16,7 @@
# Canonical product version for CoderAI — single source of truth. Both the API # Canonical product version for CoderAI — single source of truth. Both the API
# metadata and the admin web UI read from here. # metadata and the admin web UI read from here.
__version__ = "0.4.17" __version__ = "0.4.18"
__tagline__ = "Complete orchestration, distribution and escalation of remotizable advanced inference." __tagline__ = "Complete orchestration, distribution and escalation of remotizable advanced inference."
# Configure the CUDA caching allocator BEFORE torch is imported anywhere. # Configure the CUDA caching allocator BEFORE torch is imported anywhere.
......
...@@ -804,6 +804,28 @@ app.include_router(admin_router, tags=["Admin"]) ...@@ -804,6 +804,28 @@ app.include_router(admin_router, tags=["Admin"])
# admin login page on a public address. A pod image is not an orchestrator. # admin login page on a public address. A pod image is not an orchestrator.
@app.exception_handler(400)
async def bad_request_says_why(request: Request, exc: HTTPException):
"""Log WHY a 400 happened on an upload path, then answer it unchanged.
A malformed or truncated multipart body is rejected by the ASGI framework
before any handler runs, so nothing of ours ever saw it and nothing was
logged: a live run had 19 of 137 batches answered 400 and the only record
anywhere was the access line. The client knew its documents had failed and
this side could not say what was wrong with them -- not the filename, not
the size, not even that it was the body rather than a field.
Behaviour is unchanged; this only makes the class diagnosable.
"""
detail = getattr(exc, "detail", "") or ""
print(f"[http] 400 on {request.method} {request.url.path}: {detail!r} "
f"(content-type {request.headers.get('content-type', '-')!r}, "
f"content-length {request.headers.get('content-length', '-')}, "
f"peer {getattr(getattr(request, 'client', None), 'host', '-')})",
flush=True)
return JSONResponse(status_code=400, content={"detail": detail or "bad request"})
@app.exception_handler(401) @app.exception_handler(401)
async def unauthorized_redirect(request: Request, exc: HTTPException): async def unauthorized_redirect(request: Request, exc: HTTPException):
"""Redirect browser clients to login page on 401; return JSON for API clients.""" """Redirect browser clients to login page on 401; return JSON for API clients."""
......
...@@ -5255,6 +5255,25 @@ def _scale_live_pool(pool, key: str, pods: int, teardown: bool = False, ...@@ -5255,6 +5255,25 @@ def _scale_live_pool(pool, key: str, pods: int, teardown: bool = False,
return out return out
def _already_gone(exc: BaseException) -> bool:
"""Does this error mean the pod does not exist any more?
RunPod says so several ways depending on which API answered and how long
ago the pod went, so the shapes are matched rather than one code. A false
positive here reports a live pod as gone -- which the reconcile sweep then
corrects from the account listing -- while a false negative answers 500 on
a teardown that has in fact succeeded, which is the failure this exists to
stop.
"""
text = f"{type(exc).__name__}: {exc}".lower()
if "404" in text or "not found" in text:
return True
return any(h in text for h in (
"does not exist", "no such pod", "already terminated",
"already been terminated", "pod not found", "unknown pod",
"invalid pod id", "has been removed"))
def release_pod(pod_id: str, reason: str = "released by client") -> dict: def release_pod(pod_id: str, reason: str = "released by client") -> dict:
"""Terminate ONE pod by id, wherever it is, and stop tracking it. """Terminate ONE pod by id, wherever it is, and stop tracking it.
...@@ -5304,16 +5323,34 @@ def release_pod(pod_id: str, reason: str = "released by client") -> dict: ...@@ -5304,16 +5323,34 @@ def release_pod(pod_id: str, reason: str = "released by client") -> dict:
RunpodClient(acct).terminate_pod(pid) RunpodClient(acct).terminate_pod(pid)
out["terminated"] = True out["terminated"] = True
except Exception as exc: except Exception as exc:
# Say which half happened. A pod dropped from the pool but still # ALREADY GONE IS THE GOAL, NOT A FAILURE. A client tearing down a
# billing is a worse state than either end, and the caller can only # fleet releases pods by id, and by the time it reaches the last one
# act on it if it is told. # some are already gone -- reaped for idle, shed by a scale-down,
out["error"] = str(exc) # released by a retry of this same call. Raising for those answered 500
if out["found_in_pool"]: # per pod, so a teardown that was working looked like a teardown that
out["warning"] = ( # had failed, and the operator was told the pods could not be stopped.
f"pod {pid} was removed from pool {out['found_in_pool']!r} but " # (Observed: five 500s in three seconds on a client's own release
f"RunPod did not confirm the termination — it may still be " # sweep.) Terminating something that does not exist has nothing left to
f"billing. Check the console.") # do, so it is success, said plainly.
raise RuntimeError(out.get("warning") or f"could not terminate {pid}: {exc}") if _already_gone(exc):
out["terminated"] = False
out["already_gone"] = True
out["detail"] = (f"pod {pid} no longer exists on the account; "
f"nothing to terminate")
print(f"[runpod] pod {pid} was already gone ({reason}) — release "
f"had nothing to do", flush=True)
else:
# Say which half happened. A pod dropped from the pool but still
# billing is a worse state than either end, and the caller can only
# act on it if it is told.
out["error"] = str(exc)
if out["found_in_pool"]:
out["warning"] = (
f"pod {pid} was removed from pool {out['found_in_pool']!r} but "
f"RunPod did not confirm the termination — it may still be "
f"billing. Check the console.")
raise RuntimeError(out.get("warning")
or f"could not terminate {pid}: {exc}")
finally: finally:
with _pools_lock: with _pools_lock:
_provisioning_ids.discard(pid) _provisioning_ids.discard(pid)
......
...@@ -69,8 +69,18 @@ _MAX_DEFAULT = 256 ...@@ -69,8 +69,18 @@ _MAX_DEFAULT = 256
#: the right answer, not more stacks. #: the right answer, not more stacks.
_MAX_POD = 2048 _MAX_POD = 2048
#: Width of the LEASE pool. Deliberately small and NOT derived from
#: `max_concurrency`: a lease is a bounded wait (`POD_LEASE_ATTEMPT_S`) that
#: either hands back a pod at once or comes back empty, so a few hundred
#: threads can serve thousands of pages by cycling. Sizing it per in-flight
#: page is what made the pod pool ask for 2048 threads and exhaust a 2048-task
#: cgroup; the page concurrency an install wants should not decide how many
#: thread stacks it needs.
_LEASE_WORKERS = 256
_lock = threading.Lock() _lock = threading.Lock()
_pod_pool: Optional[ThreadPoolExecutor] = None _pod_pool: Optional[ThreadPoolExecutor] = None
_lease_pool: Optional[ThreadPoolExecutor] = None
_installed = False _installed = False
...@@ -121,13 +131,104 @@ def sized_from_config() -> bool: ...@@ -121,13 +131,104 @@ def sized_from_config() -> bool:
return _configured_concurrency() > 0 return _configured_concurrency() > 0
def task_budget() -> int:
"""How many threads this PROCESS is actually allowed, or 0 if unbounded.
A pool sized from a config number is a pool sized from a wish. The ceiling
that decides whether those threads can exist is the process's task budget,
and in a container that is not the kernel's -- podman defaults
`--pids-limit` to 2048, and systemd units carry `TasksMax`. Both appear as
cgroup v2 `pids.max`, which is the number to read.
Measured on a live orchestrator: `ocr.max_concurrency: 4096` asked the pod
pool for its 2048 ceiling inside a cgroup whose TOTAL budget was 2048. The
pool reached 1992 threads, the 2049th raised "can't start new thread", and
the OCR manager read the pages on a GPU-less box instead -- the one outcome
the whole pod pipeline exists to avoid. Nothing reported a limit, because
nothing had looked for one.
Returns the tightest of cgroup v2 pids.max and RLIMIT_NPROC, minus what is
already running, or 0 when neither bounds us.
"""
import os
budget = 0
def _tighten(n):
nonlocal budget
if n and n > 0:
budget = n if budget <= 0 else min(budget, n)
# cgroup v2: this process's own cgroup, then its parents -- a limit on any
# ancestor binds us just as hard as one on our own leaf.
try:
with open("/proc/self/cgroup") as fh:
rel = fh.read().strip().split(":")[-1].lstrip("/")
node = os.path.join("/sys/fs/cgroup", rel)
current = 0
while True:
try:
with open(os.path.join(node, "pids.max")) as fh:
raw = fh.read().strip()
if raw and raw != "max":
_tighten(int(raw))
except Exception:
pass
try:
with open(os.path.join(node, "pids.current")) as fh:
current = max(current, int(fh.read().strip()))
except Exception:
pass
parent = os.path.dirname(node)
if parent == node or len(parent) <= len("/sys/fs/cgroup"):
break
node = parent
if budget > 0 and current > 0:
# What is left, not what the limit says: the other processes in this
# cgroup (the front, the system engine, every subprocess worker)
# are already spending it.
budget = max(0, budget - current)
except Exception:
pass
try:
import resource
soft, _hard = resource.getrlimit(resource.RLIMIT_NPROC)
if soft not in (resource.RLIM_INFINITY, -1):
_tighten(int(soft))
except Exception:
pass
return budget
#: Leave this many tasks unspent. Threads are not the only thing that needs to
#: fork: a rasteriser worker, an isolated-venv engine, a `podman exec` to look
#: at the box while it is busy. A pool that spends the last task takes the
#: operator's ability to diagnose it away with it.
_TASK_HEADROOM = 256
def pool_sizes() -> tuple: def pool_sizes() -> tuple:
"""(default_workers, pod_workers) for this install.""" """(default_workers, pod_workers) for this install, within its budget."""
want = _configured_concurrency() want = _configured_concurrency()
general = max(_MIN_DEFAULT, min(_MAX_DEFAULT, want)) general = max(_MIN_DEFAULT, min(_MAX_DEFAULT, want))
# Room for every concurrent page to be waiting on a pod at once, plus the # Room for every concurrent page to be waiting on a pod at once, plus the
# pool's own boots and probes. # pool's own boots and probes.
pod = max(_MIN_POD, min(_MAX_POD, want * 2)) pod = max(_MIN_POD, min(_MAX_POD, want * 2))
budget = task_budget()
if budget > 0:
spendable = budget - _TASK_HEADROOM
if spendable < _MIN_DEFAULT + _MIN_POD:
# Too tight to honour both floors: keep them, and let the operator
# see why from the boot line. Refusing to start would be worse --
# an install that serves slowly beats one that does not serve.
return _MIN_DEFAULT, _MIN_POD
if general + pod > spendable:
# The pod pool gives first: the general pool is what answers
# /healthz and the admin pages, and an install that cannot be
# looked at is an install that cannot be fixed.
general = min(general, max(_MIN_DEFAULT, spendable // 4))
pod = max(_MIN_POD, spendable - general)
return general, pod return general, pod
...@@ -196,6 +297,39 @@ def resize_pod_pool() -> int: ...@@ -196,6 +297,39 @@ def resize_pod_pool() -> int:
return max(cur, want) return max(cur, want)
def lease_executor() -> ThreadPoolExecutor:
"""The pool that pod LEASES wait in, created on first use.
Separate from the pod pool on purpose. Both block, but on different things
and for different lengths: a forward waits on a socket for as long as a
page takes to read, while a lease waits for a pod to be free or to boot.
Sharing one pool let a cold start fill every thread with leases waiting for
pods, leaving none to carry the pages those pods were for -- and sizing
that one pool per in-flight page is what exhausted the task budget.
"""
global _lease_pool
with _lock:
if _lease_pool is None:
budget = task_budget()
n = _LEASE_WORKERS
if budget > 0:
n = max(16, min(n, max(16, (budget - _TASK_HEADROOM) // 8)))
_lease_pool = ThreadPoolExecutor(
max_workers=n, thread_name_prefix="coderai-lease")
print(f"[threads] lease pool: {n} workers (a lease is a bounded "
f"wait, so this serves far more pages than it has threads)",
flush=True)
return _lease_pool
async def run_on_lease_pool(func, /, *args, **kwargs):
"""``run_on_pod_pool``, on the lease pool."""
loop = asyncio.get_running_loop()
ctx = contextvars.copy_context()
call = functools.partial(ctx.run, func, *args, **kwargs)
return await loop.run_in_executor(lease_executor(), call)
async def run_on_pod_pool(func, /, *args, **kwargs): async def run_on_pod_pool(func, /, *args, **kwargs):
"""``asyncio.to_thread``, but on the pod pool. """``asyncio.to_thread``, but on the pod pool.
......
...@@ -417,7 +417,7 @@ async def chat_completions(token: str, request: Request): ...@@ -417,7 +417,7 @@ async def chat_completions(token: str, request: Request):
if got is None: if got is None:
return _unknown_lease() return _unknown_lease()
from codai.api.thread_pool import run_on_pod_pool from codai.api.thread_pool import run_on_lease_pool
from codai.ocr.manager import POD_LEASE_ATTEMPT_S from codai.ocr.manager import POD_LEASE_ATTEMPT_S
from codai.models import placement from codai.models import placement
body = await request.body() body = await request.body()
...@@ -438,7 +438,7 @@ async def chat_completions(token: str, request: Request): ...@@ -438,7 +438,7 @@ async def chat_completions(token: str, request: Request):
# directly it blocked the whole process, not this request. Same # directly it blocked the whole process, not this request. Same
# defect as `_lease_a_pod`'s, on the chat path instead of the page # defect as `_lease_a_pod`'s, on the chat path instead of the page
# path, and with no retry loop above it to make the wait visible. # path, and with no retry loop above it to make the wait visible.
chosen = await run_on_pod_pool( chosen = await run_on_lease_pool(
placement.choose, placement.choose,
got["model_id"], got["model_id"],
spill_after=placement.spill_threshold(got["model_id"]), spill_after=placement.spill_threshold(got["model_id"]),
......
...@@ -33,7 +33,7 @@ import threading ...@@ -33,7 +33,7 @@ import threading
import time import time
from typing import Dict, List, Optional from typing import Dict, List, Optional
from codai.api.thread_pool import run_on_pod_pool from codai.api.thread_pool import run_on_lease_pool, run_on_pod_pool
from codai.ocr.base import (OcrEngine, OcrPage, OcrError, ink_coverage, from codai.ocr.base import (OcrEngine, OcrPage, OcrError, ink_coverage,
PAGE_OPTIONS, page_estimate, render_pages, PAGE_OPTIONS, page_estimate, render_pages,
iter_pdf_page_pdfs, pdf_page_count, split_pdf_page, iter_pdf_page_pdfs, pdf_page_count, split_pdf_page,
...@@ -752,6 +752,32 @@ def _pod_pipeline_policy(cfg) -> tuple: ...@@ -752,6 +752,32 @@ def _pod_pipeline_policy(cfg) -> tuple:
#: where the policy's ceiling is what decides how long to keep trying. #: where the policy's ceiling is what decides how long to keep trying.
POD_LEASE_ATTEMPT_S = 20.0 POD_LEASE_ATTEMPT_S = 20.0
def _is_local_exhaustion(exc: BaseException) -> bool:
"""Did this fail because THIS BOX has no resources left?
Told apart from every other failure because the answer is the opposite one.
An optimisation that cannot run should cost latency and not an error, so a
pod that cannot serve falls through to the local pipeline -- but the local
pipeline needs MORE of this machine than the pod path does: it rasterises
every page here and holds the bitmaps. Falling back when the reason is that
this process cannot even create a thread sends the work to the one place
guaranteed not to have room for it.
Seen live: a cgroup task budget of 2048 (podman's `--pids-limit` default)
against a pod pool sized 2048 from `ocr.max_concurrency`. 278 pages in ten
minutes were read on a GPU-less orchestrator while 28 rented cards sat
idle, each one logged as "the pod-side pipeline could not run".
"""
if isinstance(exc, (MemoryError,)):
return True
text = f"{type(exc).__name__}: {exc}".lower()
return any(h in text for h in (
"can't start new thread", "cant start new thread",
"can not start new thread", "unable to start thread",
"cannot allocate memory", "resource temporarily unavailable",
"too many open files", "thread limit"))
_POD_WAIT_COOLDOWN: dict = {} _POD_WAIT_COOLDOWN: dict = {}
POD_WAIT_COOLDOWN_S = 300.0 POD_WAIT_COOLDOWN_S = 300.0
...@@ -802,7 +828,7 @@ async def _lease_a_pod(model_id: str, policy: str, wait_s: float): ...@@ -802,7 +828,7 @@ async def _lease_a_pod(model_id: str, policy: str, wait_s: float):
# when it matters most. Measured on a 5,000-document run: 116 pods # when it matters most. Measured on a 5,000-document run: 116 pods
# rented, zero documents completed in ten minutes, and a front that # rented, zero documents completed in ten minutes, and a front that
# had stopped answering. # had stopped answering.
lease = await run_on_pod_pool( lease = await run_on_lease_pool(
placement.choose, model_id, placement.choose, model_id,
spill_after=placement.spill_threshold(model_id), spill_after=placement.spill_threshold(model_id),
acquire_timeout=POD_LEASE_ATTEMPT_S) acquire_timeout=POD_LEASE_ATTEMPT_S)
...@@ -1779,6 +1805,26 @@ class OcrManager: ...@@ -1779,6 +1805,26 @@ class OcrManager:
except OcrError: except OcrError:
raise raise
except Exception as exc: except Exception as exc:
if _is_local_exhaustion(exc):
# NOT a verdict on the pods, and not something the local
# pipeline can rescue: it needs more of this box than the
# path that just failed for want of it. Answer, so the
# caller retries and the pages still reach the cards this
# install is paying for.
_say_rate_limited(
("local-exhaustion", _remote_model),
f"[ocr] this process is out of local resources "
f"({exc}) — NOT reading on this box, which needs more "
f"of them than the pod path does. The pods are fine; "
f"this orchestrator cannot currently hand work to "
f"them. Raise the container's task budget "
f"(podman --pids-limit / systemd TasksMax, default "
f"2048) or lower ocr.max_concurrency.", every=30.0)
raise OcrError(
f"the orchestrator is out of local resources ({exc}) "
f"and will not rasterise this document itself; the "
f"pods for {_remote_model!r} are unaffected. Retry.",
status=503)
print(f"[ocr] the pod-side pipeline for {_remote_model!r} could " print(f"[ocr] the pod-side pipeline for {_remote_model!r} could "
f"not run ({exc}); reading locally instead", flush=True) f"not run ({exc}); reading locally instead", flush=True)
_out = None _out = None
......
"""Two failures from one live 5,000-document run, and the budget behind both.
The run rented 48 warm pods on a client lease, then read 278 pages ON THE
ORCHESTRATOR -- a box with no GPU -- while every one of those cards sat idle.
Each page logged "the pod-side pipeline could not run (can't start new
thread)". The cause was a container task budget of 2048 (podman's
`--pids-limit` default) against a pod pool sized 2048 from
`ocr.max_concurrency: 4096`: the pool reached 1992 threads and the next one
could not be created.
Three things were wrong, and they are separable:
1. The pools sized themselves from a config number without asking what the
process is allowed. `task_budget()` asks.
2. A lease took a pod-pool thread per in-flight page, so the width an
install wants decided how many thread stacks it needs. A lease is a
bounded wait, so a small pool serves many pages by cycling.
3. "Cannot create a thread" was treated as "the pods cannot run the
pipeline", and the fallback for that is the local pipeline -- which needs
MORE of this box than the path that just failed for want of it.
And separately, from the same run: the client's own teardown sweep answered
`500` five times in three seconds, because releasing a pod that was already
gone raised. A teardown that was working looked like one that had failed.
"""
import types
import pytest
from codai.api import runpod_worker, thread_pool
from codai.ocr import manager as om
from codai.ocr.base import OcrError
# ------------------------------------------------------------ task budget #
def test_the_budget_is_read_and_is_a_number():
n = thread_pool.task_budget()
assert isinstance(n, int) and n >= 0
def test_a_tight_budget_shrinks_the_pools(monkeypatch):
"""The live case: 2048 tasks total, 4096 asked for."""
monkeypatch.setattr(thread_pool, "_configured_concurrency", lambda: 4096)
monkeypatch.setattr(thread_pool, "task_budget", lambda: 2048 - 30)
general, pod = thread_pool.pool_sizes()
assert general + pod <= 2048 - 30 - thread_pool._TASK_HEADROOM
assert pod >= thread_pool._MIN_POD
assert general >= thread_pool._MIN_DEFAULT
def test_an_unbounded_budget_leaves_the_sizes_alone(monkeypatch):
monkeypatch.setattr(thread_pool, "_configured_concurrency", lambda: 4096)
monkeypatch.setattr(thread_pool, "task_budget", lambda: 0)
general, pod = thread_pool.pool_sizes()
assert pod == thread_pool._MAX_POD
assert general == thread_pool._MAX_DEFAULT
def test_the_general_pool_is_protected_so_the_box_stays_diagnosable(monkeypatch):
"""An install that cannot answer /healthz cannot be looked at."""
monkeypatch.setattr(thread_pool, "_configured_concurrency", lambda: 4096)
monkeypatch.setattr(thread_pool, "task_budget", lambda: 1200)
general, pod = thread_pool.pool_sizes()
assert general >= thread_pool._MIN_DEFAULT
assert general + pod <= 1200 - thread_pool._TASK_HEADROOM
def test_an_impossibly_tight_budget_still_starts(monkeypatch):
"""Serving slowly beats not serving."""
monkeypatch.setattr(thread_pool, "_configured_concurrency", lambda: 4096)
monkeypatch.setattr(thread_pool, "task_budget", lambda: 40)
assert thread_pool.pool_sizes() == (thread_pool._MIN_DEFAULT,
thread_pool._MIN_POD)
def test_the_lease_pool_is_not_sized_per_page(monkeypatch):
"""The whole point: page width must not decide thread count."""
monkeypatch.setattr(thread_pool, "_configured_concurrency", lambda: 100000)
monkeypatch.setattr(thread_pool, "_lease_pool", None)
monkeypatch.setattr(thread_pool, "task_budget", lambda: 0)
ex = thread_pool.lease_executor()
assert ex._max_workers <= thread_pool._LEASE_WORKERS
def test_the_lease_pool_also_respects_the_budget(monkeypatch):
monkeypatch.setattr(thread_pool, "_lease_pool", None)
monkeypatch.setattr(thread_pool, "task_budget", lambda: 400)
ex = thread_pool.lease_executor()
assert 16 <= ex._max_workers <= (400 - thread_pool._TASK_HEADROOM) // 8 + 1
# ------------------------------------------- never fall back to this box #
@pytest.mark.parametrize("exc", [
RuntimeError("can't start new thread"),
RuntimeError("can not start new thread"),
MemoryError(),
OSError("Cannot allocate memory"),
OSError("Resource temporarily unavailable"),
OSError("Too many open files"),
])
def test_local_exhaustion_is_recognised(exc):
assert om._is_local_exhaustion(exc) is True
@pytest.mark.parametrize("exc", [
RuntimeError("pod answered 503: no model"),
ValueError("bad PDF"),
OSError("connection refused"),
TimeoutError("read timed out"),
])
def test_an_ordinary_failure_still_falls_back(exc):
"""An optimisation that cannot run must cost latency, not an error."""
assert om._is_local_exhaustion(exc) is False
def test_the_message_names_the_limit_and_both_ways_out():
"""An operator reading it should not have to find this commit."""
src = open(om.__file__).read()
i = src.index("out of local resources")
window = src[i:i + 1400]
assert "pids-limit" in window and "TasksMax" in window
assert "max_concurrency" in window
assert "2048" in window # podman's default, which is the trap
# --------------------------------------------------- teardown idempotence #
@pytest.mark.parametrize("exc", [
RuntimeError("404 Not Found"),
RuntimeError("pod does not exist"),
RuntimeError("No such pod: abc"),
RuntimeError("pod has already been terminated"),
RuntimeError("pod not found"),
])
def test_an_already_gone_pod_is_recognised(exc):
assert runpod_worker._already_gone(exc) is True
@pytest.mark.parametrize("exc", [
RuntimeError("500 Internal Server Error"),
RuntimeError("connection refused"),
RuntimeError("unauthorized"),
RuntimeError("rate limited"),
])
def test_a_real_failure_is_still_a_failure(exc):
"""Reporting a live pod as gone leaves it billing."""
assert runpod_worker._already_gone(exc) is False
def _release_against(monkeypatch, terminate):
"""release_pod with no pools, a stub account and a stubbed terminate."""
monkeypatch.setattr(runpod_worker, "_pools", {}, raising=False)
class _Client:
def __init__(self, _acct):
pass
def terminate_pod(self, pid):
return terminate(pid)
import codai.api.runpod_client as rc
monkeypatch.setattr(rc, "RunpodClient", _Client)
import codai.models.manager as mm
monkeypatch.setattr(mm, "get_active_runpod_config",
lambda: types.SimpleNamespace(enabled=True))
return runpod_worker.release_pod("podxyz", "released by client")
def test_releasing_a_pod_that_is_already_gone_succeeds(monkeypatch, capsys):
"""THE TEARDOWN BUG: this answered 500, five times in three seconds."""
def _gone(_pid):
raise RuntimeError("404 Not Found: pod does not exist")
out = _release_against(monkeypatch, _gone)
assert out["already_gone"] is True
assert out["terminated"] is False
assert "no longer exists" in out["detail"]
assert "already gone" in capsys.readouterr().out
def test_releasing_a_live_pod_terminates_it(monkeypatch):
seen = []
out = _release_against(monkeypatch, lambda pid: seen.append(pid))
assert out["terminated"] is True
assert seen == ["podxyz"]
def test_a_real_termination_failure_still_raises(monkeypatch):
def _boom(_pid):
raise RuntimeError("500 upstream error")
with pytest.raises(RuntimeError):
_release_against(monkeypatch, _boom)
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