File size: 2,555 Bytes
85270f9 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 | """Tuning for the process-wide anyio worker thread pool.
Every blocking handler in this project is dispatched through
``fastapi.concurrency.run_in_threadpool`` (see ``services/log_service.py``), and
that helper draws its capacity from anyio's *default* thread limiter. anyio
hardcodes that limiter to 40 tokens and stores it in a ``RunVar``.
The scope of that ``RunVar`` is what makes this module work: it is keyed by the
running *event loop*, not by the current task (anyio 4.13 keeps a module-level
``WeakKeyDictionary`` indexed by ``get_running_loop()``). A single
``total_tokens`` assignment during startup therefore resizes the pool for every
request handler in the process, including ones created later.
Measured behaviour on this codebase: 60 concurrent requests to a handler that
blocks in ``run_in_threadpool`` only ever reach 40 simultaneous workers with the
default limiter, and reach all 60 once the limiter is raised. Raising the limit
is therefore a real lever -- but it only matters when the account pool
(``account_count x image_account_concurrency``) is larger than 40, and each extra
thread also means one more upstream request in flight.
"""
from __future__ import annotations
from anyio.to_thread import current_default_thread_limiter
from services.config import config
#: anyio's own default (``CapacityLimiter(40)`` in the asyncio backend).
DEFAULT_THREAD_POOL_TOKENS = 40
MIN_THREAD_POOL_TOKENS = 4
MAX_THREAD_POOL_TOKENS = 512
def normalize_thread_pool_tokens(value: object) -> int:
"""Clamp a configured worker-thread count into a usable range."""
try:
tokens = int(value) # type: ignore[arg-type]
except (TypeError, ValueError):
return DEFAULT_THREAD_POOL_TOKENS
return max(MIN_THREAD_POOL_TOKENS, min(MAX_THREAD_POOL_TOKENS, tokens))
def apply_thread_pool_tokens(tokens: object | None = None) -> int:
"""Resize the shared worker-thread limiter and return the applied value.
Must be called from inside the running event loop (anyio raises
``NoEventLoopError`` otherwise). The limiter object is mutated in place, so
callers that are already queued simply observe the new capacity.
"""
target = normalize_thread_pool_tokens(
config.thread_pool_tokens if tokens is None else tokens
)
limiter = current_default_thread_limiter()
limiter.total_tokens = target
return target
def current_thread_pool_tokens() -> int:
"""Return the limiter capacity currently in effect."""
return int(current_default_thread_limiter().total_tokens)
|