Executors

Execution contexts for handlers.

Base Executor

Base executor interface and exceptions.

Purpose

Defines the abstract base class for all executor implementations and common exceptions. BaseExecutor owns the shared machinery — bypass mode, metrics accounting around submit, pool shutdown, decorator interface — while subclasses provide the pooled execution core.

Definition:

class BaseExecutor(ABC):
    name: str
    pool: Any            # None in bypass mode
    pool_mode: str       # "process", "thread", ... (metrics label)

    async def submit(self, func: Callable, *args, **kwargs) -> Any
        '''Template: bypass check + metrics around _submit_pooled().'''

    @abstractmethod
    async def _submit_pooled(self, func: Callable, *args, **kwargs) -> Any
        '''Pooled execution core (backpressure + dispatch).'''

    def shutdown(self, wait: bool = True) -> None
        '''Shutdown self.pool if present.'''

    @property
    def metrics(self) -> dict[str, Any]
        '''base_metrics(); subclasses may extend with extra gauges.'''

    def __call__(self, func: Callable) -> Callable
        '''Decorator interface - wraps func to use submit().'''

Exceptions:

ExecutorError
    Base exception for executor operations.

ExecutorOverloadError(ExecutorError)
    Raised when executor has too many pending tasks (backpressure).

Design Notes

• BaseExecutor provides __call__ for decorator pattern (uses submit internally)

• Subclasses implement _submit_pooled() and set pool/pool_mode in __init__

• Metrics dict includes: name, mode, pending, submitted, completed, failed, avg_duration_ms; subclasses may add gauges on top of base_metrics()

class genro_asgi.executors.base.BaseExecutor(name='default')

Bases: ABC

Abstract base class for all executor implementations.

Owns bypass handling, metrics accounting and pool shutdown; implements the decorator pattern via __call__ which uses submit().

Subclasses must: - call BaseExecutor.__init__ and set pool (None means bypass mode) - set the pool_mode class attribute (metrics label) - implement _submit_pooled(): backpressure + pooled dispatch

name

Identifier for this executor instance.

pool

The underlying pool, or None in bypass mode.

Example

>>> class MyExecutor(BaseExecutor):
...     pool_mode = "inline"
...     async def _submit_pooled(self, func, *args, **kwargs):
...         return func(*args, **kwargs)
>>>
>>> executor = MyExecutor("test")
>>>
>>> @executor
... def my_func(x):
...     return x * 2
>>>
>>> result = await my_func(5)  # returns 10

__init__(name='default')

Initialize shared executor state.

Parameters:

name (str) – Identifier for metrics and logging.

base_metrics()

Common metrics core shared by all executors.

Return type:

dict[str, Any]

Returns:

Dict with name, mode, pending, submitted, completed, failed, avg_duration_ms.

property metrics: dict[str, Any]

Return executor metrics (base_metrics; subclasses may extend).

Returns:

• name: Executor name

• mode: pool_mode, or “bypass” when no pool

• pending: Number of pending tasks

• submitted: Total submitted tasks

• completed: Total completed tasks

• failed: Total failed tasks

• avg_duration_ms: Mean completed-task duration

Return type:

Dict containing at minimum

name

pool: Any

pool_mode: str = 'pool'

shutdown(wait=True)

Shutdown the executor and release resources.

Parameters:

wait (bool) – If True, wait for pending tasks to complete.

Return type:

None

async submit(func, *args, **kwargs)

Submit a function for execution.

Template method: in bypass mode (no pool) runs func synchronously; otherwise wraps _submit_pooled() with metrics accounting.

Parameters:

• func (Callable[..., Any]) – The function to execute.

• *args (Any) – Positional arguments for func.

• **kwargs (Any) – Keyword arguments for func.

Return type:

Any

Returns:

The result of func(*args, **kwargs).

Raises:

• ExecutorError – If execution fails.

• ExecutorOverloadError – If too many tasks are pending.

exception genro_asgi.executors.base.ExecutorError

Bases: Exception

Base exception for executor operations.

exception genro_asgi.executors.base.ExecutorOverloadError

Bases: ExecutorError

Raised when executor has too many pending tasks.

Local Executor

Local executor for CPU-bound work using ProcessPoolExecutor.

Classes:

LocalExecutor – wraps ProcessPoolExecutor for async submission

with backpressure (semaphore) and metrics.

Supports bypass mode (no processes) for testing. Environment variable GENRO_EXECUTOR_BYPASS=1 forces bypass globally. Functions and arguments must be pickle-serializable.

class genro_asgi.executors.local.LocalExecutor(name='default', max_workers=None, initializer=None, initargs=(), max_pending=100, bypass=False)

Bases: BaseExecutor

Executor using local ProcessPoolExecutor.

Runs functions in separate processes for true parallelism, bypassing Python’s GIL. Ideal for CPU-bound work.

name

Identifier for this executor (used in metrics/logging).

pool

The ProcessPoolExecutor, or None in bypass mode.

max_pending

Maximum pending tasks before backpressure.

Example

>>> executor = LocalExecutor(name="compute", max_workers=4)
>>>
>>> @executor
... def heavy_work(data):
...     return process(data)
>>>
>>> result = await heavy_work(my_data)

__init__(name='default', max_workers=None, initializer=None, initargs=(), max_pending=100, bypass=False)

Initialize LocalExecutor.

Parameters:

• name (str) – Identifier for metrics and logging.

• max_workers (int | None) – Number of worker processes (default: CPU count).

• initializer (Optional[Callable[..., None]]) – Function called once per worker at startup.

• initargs (tuple[Any, ...]) – Arguments passed to initializer.

• max_pending (int) – Maximum concurrent pending tasks.

• bypass (bool) – If True, run synchronously without pool (for testing).

max_pending

pool_mode: str = 'process'

Thread Executor

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.

class genro_asgi.executors.thread.ThreadExecutor(name='worker', max_workers=None, thread_name_prefix='gnr-worker', max_pending=100, bypass=False)

Bases: 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).

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)

__init__(name='worker', max_workers=None, thread_name_prefix='gnr-worker', max_pending=100, bypass=False)

Initialize ThreadExecutor.

Parameters:

• name (str) – Identifier for metrics and logging.

• max_workers (int | None) – Number of worker threads (default: ThreadPoolExecutor’s, i.e. min(32, cpu + 4)).

• thread_name_prefix (str) – Prefix for the pool’s thread names.

• max_pending (int) – Maximum concurrent pending tasks before backpressure.

• bypass (bool) – If True, run synchronously without a pool (for testing).

max_pending

max_workers

property metrics: dict[str, Any]

Return current executor metrics.

Returns:

base_metrics() plus the pressure gauges total/busy/queue_depth/ occupancy (raw, instantaneous).

pool_mode: str = 'thread'

shutdown(wait=True)

Shutdown the thread pool.

Parameters:

wait (bool) – If True, wait for pending tasks to complete.

Return type:

None

Executor Registry

Executor registry for managing multiple named executor instances.

Classes:

ExecutorRegistry – lazy creation, caching, and coordinated shutdown

of named executors.

Supports multiple executor types via factory pattern (default: LocalExecutor). Provides centralized metrics collection and bulk shutdown.

class genro_asgi.executors.registry.ExecutorRegistry

Bases: object

Registry for managing named executor instances.

Provides centralized management of executors with lazy creation, caching, and coordinated shutdown.

executors

Dict of name -> executor instance.

Example

>>> registry = ExecutorRegistry()
>>> executor = registry.get_or_create("compute", max_workers=4)
>>> registry.shutdown_all()

__init__()

Initialize ExecutorRegistry with default factories.

all_metrics()

Collect metrics from all executors.

Return type:

list[dict[str, Any]]

Returns:

List of metric dicts, one per executor.

property executors: dict[str, BaseExecutor]

Return dict of all registered executors.

get(name)

Get executor by name without creating.

Parameters:

name (str) – Executor identifier.

Return type:

BaseExecutor | None

Returns:

Executor instance or None if not found.

get_or_create(name, executor_type='local', **kwargs)

Get existing executor or create new one.

Parameters:

• name (str) – Unique identifier for the executor.

• executor_type (str) – Type of executor (“local” by default).

• **kwargs (Any) – Arguments passed to executor constructor.

Return type:

BaseExecutor

Returns:

Executor instance (cached if already exists).

Raises:

ValueError – If executor_type is not registered.

Example

>>> executor = registry.get_or_create("pdf", max_workers=2)
>>> same_executor = registry.get_or_create("pdf")  # returns cached

register_factory(executor_type, factory)

Register a factory for a custom executor type.

Parameters:

• executor_type (str) – Type identifier (e.g., “remote”, “hybrid”).

• factory (Callable[..., BaseExecutor]) – Callable that creates executor instances. Signature: factory(name: str, **kwargs) -> BaseExecutor

Return type:

None

Example

>>> registry.register_factory("remote", lambda name, **kw: RemoteExecutor(name, **kw))

shutdown_all(wait=True)

Shutdown all registered executors.

Parameters:

wait (bool) – If True, wait for pending tasks to complete.

Return type:

None