151 lines
6.1 KiB
Python
151 lines
6.1 KiB
Python
# #region BlockingExecutorsModule [C:4] [TYPE Module] [SEMANTICS async, blocking, executors, run_blocking, threadpool]
|
|
# @BRIEF Named bounded executors for sync DB/file/git work. All production blocking operations use
|
|
# loop.run_in_executor via the run_blocking helper instead of default asyncio.to_thread.
|
|
# @LAYER Infrastructure
|
|
# @RELATION DEPENDS_ON -> [EXT:concurrent.futures:ThreadPoolExecutor]
|
|
# @RELATION DEPENDS_ON -> [EXT:asyncio:AbstractEventLoop]
|
|
# @RATIONALE asyncio.to_thread uses default executor without backpressure. Named bounded executors
|
|
# prevent thread pool exhaustion under load. Semaphore per kind limits queue depth.
|
|
# @REJECTED Default asyncio.to_thread — no backpressure, no bounded queue, can exhaust thread pool.
|
|
|
|
import asyncio
|
|
from collections.abc import Callable
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
from functools import partial
|
|
from typing import Any
|
|
|
|
from ..logger import logger
|
|
|
|
# Module-level executors — one per kind, created once.
|
|
_db_executor: ThreadPoolExecutor | None = None
|
|
_file_executor: ThreadPoolExecutor | None = None
|
|
_git_executor: ThreadPoolExecutor | None = None
|
|
|
|
# Per-kind semaphores for queue backpressure.
|
|
_db_semaphore: asyncio.Semaphore | None = None
|
|
_file_semaphore: asyncio.Semaphore | None = None
|
|
_git_semaphore: asyncio.Semaphore | None = None
|
|
|
|
# Default config — overridden by init_executors().
|
|
_DEFAULT_MAX_WORKERS = 10
|
|
_DEFAULT_QUEUE_TIMEOUT = 30.0
|
|
|
|
|
|
# #region init_executors [C:2] [TYPE Function]
|
|
# @BRIEF Initialize named executors with given config. Called once at application startup.
|
|
# @PRE event loop is running.
|
|
# @POST Executors and semaphores are ready for use.
|
|
# @SIDE_EFFECT Creates ThreadPoolExecutor instances.
|
|
def init_executors(
|
|
db_workers: int = 10,
|
|
file_workers: int = 10,
|
|
git_workers: int = 5,
|
|
queue_timeout: float = 30.0,
|
|
) -> None:
|
|
global _db_executor, _file_executor, _git_executor
|
|
global _db_semaphore, _file_semaphore, _git_semaphore
|
|
|
|
_db_executor = ThreadPoolExecutor(max_workers=db_workers, thread_name_prefix="db")
|
|
_file_executor = ThreadPoolExecutor(max_workers=file_workers, thread_name_prefix="file")
|
|
_git_executor = ThreadPoolExecutor(max_workers=git_workers, thread_name_prefix="git")
|
|
|
|
_db_semaphore = asyncio.Semaphore(db_workers * 2)
|
|
_file_semaphore = asyncio.Semaphore(file_workers * 2)
|
|
_git_semaphore = asyncio.Semaphore(git_workers * 2)
|
|
|
|
logger.reason("init_executors",
|
|
extra={"payload": {"db": db_workers, "file": file_workers, "git": git_workers, "queue_timeout": queue_timeout}})
|
|
|
|
|
|
# #endregion init_executors
|
|
|
|
|
|
# #region shutdown_executors [C:2] [TYPE Function]
|
|
# @BRIEF Shut down all executors. Called at application shutdown.
|
|
# @POST All executors are shut down, pending futures cancelled.
|
|
# @SIDE_EFFECT Waits for running tasks up to timeout.
|
|
def shutdown_executors(wait: bool = True, cancel_futures: bool = True) -> None:
|
|
global _db_executor, _file_executor, _git_executor
|
|
for executor in (_db_executor, _file_executor, _git_executor):
|
|
if executor is not None:
|
|
executor.shutdown(wait=wait, cancel_futures=cancel_futures)
|
|
_db_executor = _file_executor = _git_executor = None
|
|
logger.reason("shutdown_executors", extra={"payload": {}})
|
|
|
|
|
|
# #endregion shutdown_executors
|
|
|
|
|
|
# #region _get_executor [C:1] [TYPE Function]
|
|
# @BRIEF Return named executor and semaphore for given kind.
|
|
# @PRE init_executors has been called.
|
|
# @POST Returns (executor, semaphore) tuple.
|
|
def _get_executor(kind: str) -> tuple[ThreadPoolExecutor, asyncio.Semaphore | None]:
|
|
if kind == "db":
|
|
return _db_executor, _db_semaphore
|
|
if kind == "file":
|
|
return _file_executor, _file_semaphore
|
|
if kind == "git":
|
|
return _git_executor, _git_semaphore
|
|
raise ValueError(f"Unknown executor kind: {kind}. Use 'db', 'file', or 'git'.")
|
|
|
|
|
|
# #endregion _get_executor
|
|
|
|
|
|
# #region run_blocking [C:4] [TYPE Function]
|
|
# @BRIEF Execute a blocking function in a named bounded executor.
|
|
# @PRE init_executors has been called. kind is one of 'db', 'file', 'git'.
|
|
# @POST fn(*args) executed in named executor. Returns result or raises timeout/exception.
|
|
# @SIDE_EFFECT Acquires/releases per-kind semaphore. Runs function in thread pool.
|
|
# @RATIONALE asyncio.to_thread does not accept an executor parameter and uses default
|
|
# unbounded pool. run_blocking uses loop.run_in_executor with named executors and
|
|
# semaphore-based backpressure.
|
|
# @REJECTED asyncio.to_thread — default executor without queue backpressure.
|
|
async def run_blocking(
|
|
kind: str,
|
|
fn: Callable[..., Any],
|
|
*args: Any,
|
|
timeout: float | None = None,
|
|
**kwargs: Any,
|
|
) -> Any:
|
|
executor, semaphore = _get_executor(kind)
|
|
|
|
if semaphore is not None:
|
|
try:
|
|
await asyncio.wait_for(semaphore.acquire(), timeout=timeout or _DEFAULT_QUEUE_TIMEOUT)
|
|
except TimeoutError:
|
|
logger.explore(f"run_blocking.{kind}",
|
|
extra={"error": f"Queue timeout ({timeout or _DEFAULT_QUEUE_TIMEOUT}s)"})
|
|
raise TimeoutError(f"Blocking executor '{kind}' queue timeout after {timeout or _DEFAULT_QUEUE_TIMEOUT}s")
|
|
|
|
try:
|
|
loop = asyncio.get_running_loop()
|
|
bound_fn = partial(fn, *args, **kwargs)
|
|
result = await loop.run_in_executor(executor, bound_fn)
|
|
return result
|
|
except asyncio.CancelledError:
|
|
logger.explore(f"run_blocking.{kind}",
|
|
extra={"error": "Cancelled — function continues in thread pool"})
|
|
raise
|
|
finally:
|
|
if semaphore is not None:
|
|
semaphore.release()
|
|
|
|
|
|
# #endregion run_blocking
|
|
|
|
|
|
# #region run_cpu_blocking [C:2] [TYPE Function]
|
|
# @BRIEF Execute CPU-bound work in default executor (no bounded pool needed).
|
|
# @POST fn(*args) executed in default thread pool.
|
|
# @SIDE_EFFECT Runs function in thread pool. Does not acquire semaphore.
|
|
async def run_cpu_blocking(fn: Callable[..., Any], *args: Any, **kwargs: Any) -> Any:
|
|
loop = asyncio.get_running_loop()
|
|
bound_fn = partial(fn, *args, **kwargs)
|
|
return await loop.run_in_executor(None, bound_fn)
|
|
|
|
|
|
# #endregion run_cpu_blocking
|
|
# #endregion BlockingExecutorsModule
|