| """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 |
|
|
| |
| 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) |
| 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) |
|
|
