# Copyright 2025 Softwell S.r.l.
# Licensed under the Apache License, Version 2.0
"""Thread executor for blocking I/O-bound work using ThreadPoolExecutor.
Classes:
ThreadExecutor -- wraps ThreadPoolExecutor for async submission, preserves
the contextvars context across the thread boundary, and
exposes occupancy/queue gauges on top of the base metrics.
This is the dispatch for synchronous code that must run off the event loop while
staying observable: unlike ``asyncio.to_thread`` (which uses the loop's anonymous
default executor), the pool is owned, so its pressure can be measured. The worker
reads ``metrics`` to report how busy it is; it makes no decisions itself.
Supports bypass mode (no pool) for testing. Environment variable
``GENRO_EXECUTOR_BYPASS=1`` forces bypass globally.
"""
from __future__ import annotations
import asyncio
import contextvars
import os
from concurrent.futures import ThreadPoolExecutor
from functools import partial
from typing import Any, Callable
from .base import BaseExecutor
__all__ = ["ThreadExecutor"]
[docs]
class ThreadExecutor(BaseExecutor):
"""
Executor using an owned ThreadPoolExecutor.
Runs blocking functions on threads (GIL-bound, no pickling) while keeping the
event loop free. The pool is owned so occupancy and queue depth are
observable: ``metrics`` reports ``busy``/``total``/``queue_depth``/
``occupancy`` as raw instantaneous gauges (no trend, no thresholds — those
live in the scaler).
Attributes:
name: Identifier for this executor (used in metrics/logging).
pool: The ThreadPoolExecutor, or None in bypass mode.
max_workers: Total slots (pool size); stored explicitly.
max_pending: Maximum pending tasks before backpressure blocks.
Example:
>>> executor = ThreadExecutor(name="worker", max_workers=8)
>>>
>>> @executor
... def blocking_io(data):
... return slow_call(data)
>>>
>>> result = await blocking_io(my_data)
"""
__slots__ = (
"max_workers",
"max_pending",
"_gate",
"_pending_sem",
"_busy",
)
pool_mode = "thread"
[docs]
def __init__(
self,
name: str = "worker",
max_workers: int | None = None,
thread_name_prefix: str = "gnr-worker",
max_pending: int = 100,
bypass: bool = False,
) -> None:
"""
Initialize ThreadExecutor.
Args:
name: Identifier for metrics and logging.
max_workers: Number of worker threads (default: ThreadPoolExecutor's,
i.e. ``min(32, cpu + 4)``).
thread_name_prefix: Prefix for the pool's thread names.
max_pending: Maximum concurrent pending tasks before backpressure.
bypass: If True, run synchronously without a pool (for testing).
"""
super().__init__(name)
self.max_pending = max_pending
env_bypass = os.environ.get("GENRO_EXECUTOR_BYPASS") == "1"
if bypass or env_bypass:
self.pool = None
self.max_workers = 0
self._gate: asyncio.Semaphore | None = None
self._pending_sem: asyncio.Semaphore | None = None
else:
self.pool = ThreadPoolExecutor(
max_workers=max_workers,
thread_name_prefix=thread_name_prefix,
)
# The pool resolves max_workers to a concrete number; mirror it from
# our own argument resolution rather than reading a private attribute.
self.max_workers = (
max_workers if max_workers is not None else min(32, (os.cpu_count() or 1) + 4)
)
self._gate = asyncio.Semaphore(self.max_workers)
self._pending_sem = asyncio.Semaphore(max_pending)
self._busy = 0
@property
def metrics(self) -> dict[str, Any]:
"""
Return current executor metrics.
Returns:
base_metrics() plus the pressure gauges total/busy/queue_depth/
occupancy (raw, instantaneous).
"""
result = self.base_metrics()
total = self.max_workers
busy = self._busy
result.update(
{
"total": total,
"busy": busy,
"queue_depth": max(0, result["pending"] - busy),
"occupancy": busy / total if total > 0 else 0.0,
}
)
return result
async def _submit_pooled(self, func: Callable[..., Any], *args: Any, **kwargs: Any) -> Any:
"""Run func on a pool thread behind backpressure and the slot gate.
Backpressure first, then a real slot. busy is the number of slots
held; everything past total that cleared backpressure waits on the
gate and shows up as queue_depth. All counters are touched on the
event loop (single-threaded), never inside the worker thread.
Metrics are handled by BaseExecutor.submit().
"""
if self._gate is None or self._pending_sem is None:
return await self._execute(func, *args, **kwargs)
async with self._pending_sem:
async with self._gate:
self._busy += 1
try:
return await self._execute(func, *args, **kwargs)
finally:
self._busy -= 1
async def _execute(self, func: Callable[..., Any], *args: Any, **kwargs: Any) -> Any:
"""Execute func on a pool thread, preserving the contextvars context.
Mirrors ``asyncio.to_thread`` semantics (which runs
``copy_context().run(func, ...)`` on the loop's default executor) but on
the owned pool, so any ContextVar set before the call is visible inside
the thread.
"""
loop = asyncio.get_running_loop()
ctx = contextvars.copy_context()
return await loop.run_in_executor(self.pool, partial(ctx.run, func, *args, **kwargs))
[docs]
def shutdown(self, wait: bool = True) -> None:
"""
Shutdown the thread pool.
Args:
wait: If True, wait for pending tasks to complete.
"""
if self.pool is not None:
self.pool.shutdown(wait=wait)
def __repr__(self) -> str:
"""Return string representation."""
mode = "bypass" if self.pool is None else "thread"
return f"ThreadExecutor(name={self.name!r}, mode={mode}, max_workers={self.max_workers})"
if __name__ == "__main__":
async def main() -> None:
executor = ThreadExecutor(name="test", max_workers=4)
print(f"Executor: {executor}")
@executor
def square(x: int) -> int:
return x * x
result = await square(5) # type: ignore[misc]
print(f"square(5) = {result}")
print(f"Metrics: {executor.metrics}")
executor.shutdown()
asyncio.run(main())