Executors

Version: 2.0.0 Status: SOURCE OF TRUTH Last Updated: 2026-06-28

Overview

Executors offload blocking and CPU-bound work off the async event loop, keeping it free to serve other requests.

The package (genro_asgi.executors) provides:

BaseExecutor — abstract interface; implements the decorator pattern via __call__, which delegates to submit().
ThreadExecutor — owned ThreadPoolExecutor for blocking I/O-bound work (GIL-bound, no pickling), with occupancy/queue pressure gauges and contextvars propagation.
LocalExecutor — ProcessPoolExecutor for CPU-bound work (true parallelism, bypasses the GIL); functions and arguments must be pickle-serializable.
ExecutorRegistry — lazy creation, caching, and coordinated shutdown of named executors, with a factory mechanism for custom types.

Cross-cutting features common to both concrete executors:

Decorator usage — @executor wraps a function so calling it awaits a submit().
Backpressure — a max_pending semaphore bounds in-flight tasks.
Metrics — each executor exposes a metrics property for observability.
Bypass mode — run synchronously without a pool for testing (bypass=True, or GENRO_EXECUTOR_BYPASS=1 globally).

Public exports: genro_asgi.executors.__all__ exposes BaseExecutor, ExecutorError, ExecutorOverloadError, ExecutorRegistry, LocalExecutor, and ThreadExecutor. The top-level genro_asgi package re-exports all of these except ThreadExecutor; import it from genro_asgi.executors.

Architecture

Async event loop
│
├── ExecutorRegistry
│     ├── "worker" → ThreadExecutor (ThreadPoolExecutor, I/O-bound)
│     ├── "pdf"    → LocalExecutor   (ProcessPoolExecutor, 2 procs)
│     └── "ml"     → LocalExecutor   (ProcessPoolExecutor, 4 procs)
│
└── async handlers `await` executor results

Each named executor owns its own pool, so a slow workload on one pool does not block another.

BaseExecutor

The abstract base (base.py) defines the contract every executor implements and the shared decorator behaviour. Subclasses implement only submit(), shutdown(), and the metrics property; BaseExecutor.__call__ provides the decorator that wraps a function into an async callable backed by submit().

from genro_asgi.executors import BaseExecutor

class MyExecutor(BaseExecutor):
    name = "test"

    async def submit(self, func, *args, **kwargs):
        return func(*args, **kwargs)

    def shutdown(self, wait=True):
        ...

    @property
    def metrics(self):
        return {"name": self.name}

executor = MyExecutor()

@executor
async def double(x):
    return x * 2

result = await double(5)  # 10

Exceptions (also in base.py):

ExecutorError — base exception for executor operations.
ExecutorOverloadError(ExecutorError) — reserved for backpressure overload.

The metrics dict produced by concrete executors includes at minimum: name, pending, submitted, completed, failed.

ThreadExecutor

For blocking I/O-bound work that must run off the event loop while staying observable. Unlike asyncio.to_thread (which uses the loop’s anonymous default executor), ThreadExecutor owns its ThreadPoolExecutor, so its pressure can be measured. It also preserves the contextvars context across the thread boundary (copy_context().run(...)), mirroring asyncio.to_thread semantics.

from genro_asgi.executors import ThreadExecutor

executor = ThreadExecutor(name="worker", max_workers=8)

@executor
def blocking_io(data):
    return slow_call(data)  # runs on a pool thread

result = await blocking_io(my_data)

Constructor: ThreadExecutor(name="worker", max_workers=None, thread_name_prefix="gnr-worker", max_pending=100, bypass=False). When max_workers is None, the pool defaults to min(32, cpu + 4).

Backpressure is two-stage: a max_pending semaphore admits tasks, then a max_workers “gate” semaphore grants a real slot. Tasks that cleared backpressure but are waiting on the gate count as queue_depth.

Pressure gauges

Beyond the base metrics, ThreadExecutor.metrics adds raw, instantaneous gauges (no trend, no thresholds — those belong to a scaler, not the executor):

{
    "name": "worker",
    "mode": "thread",        # "bypass" when no pool
    "pending": ...,          # submitted - completed - failed
    "submitted": ...,
    "completed": ...,
    "failed": ...,
    "avg_duration_ms": ...,
    "total": ...,            # max_workers (pool size)
    "busy": ...,             # slots currently held
    "queue_depth": ...,      # max(0, pending - busy)
    "occupancy": ...,        # busy / total
}

LocalExecutor

For CPU-bound work. Runs functions in separate processes via ProcessPoolExecutor, achieving true parallelism past the GIL. Functions, arguments, and return values must be pickle-serializable.

from genro_asgi.executors import LocalExecutor

executor = LocalExecutor(name="compute", max_workers=4)

@executor
def heavy_work(data):
    return process(data)  # runs in a worker process

result = await heavy_work(my_data)

Constructor: LocalExecutor(name="default", max_workers=None, initializer=None, initargs=(), max_pending=100, bypass=False). When max_workers is None, the pool uses the CPU count.

If a non-pickle-serializable argument reaches the pool, _execute raises ExecutorError wrapping the original pickle.PicklingError.

Worker initialization (preloaded data)

LocalExecutor forwards initializer / initargs to the underlying ProcessPoolExecutor, so each worker process can load heavy data once at startup and reuse it across tasks:

# Worker-side module globals
_model = None

def init_ml_worker(model_path):
    global _model
    _model = load_heavy_model(model_path)

def predict(data):
    return _model.predict(data)  # model already loaded

executor_ml = LocalExecutor(
    name="ml",
    max_workers=4,
    initializer=init_ml_worker,
    initargs=("/models/v1.pkl",),
)

@executor_ml
def predict(data):
    return _model.predict(data)

Typical preload targets: ML models, large lookup tables, compiled regexes, per-worker connection pools, static configuration.

Metrics

{
    "name": "compute",
    "mode": "process",       # "bypass" when no pool
    "pending": ...,
    "submitted": ...,
    "completed": ...,
    "failed": ...,
    "avg_duration_ms": ...,
}

ExecutorRegistry

The registry (registry.py) is the public way to obtain a named, cached executor. It ships two built-in factories, "local" (→ LocalExecutor) and "thread" (→ ThreadExecutor); "local" is the default executor_type.

from genro_asgi.executors import ExecutorRegistry

registry = ExecutorRegistry()

# Process pool for CPU-bound work (default type "local")
pdf = registry.get_or_create("pdf", max_workers=2)

# Thread pool for blocking I/O
io = registry.get_or_create("io", executor_type="thread", max_workers=8)

# Same name returns the cached instance (kwargs ignored on a cache hit)
same_pdf = registry.get_or_create("pdf")
assert same_pdf is pdf

@pdf
def generate_pdf(data):
    return create_pdf(data)

result = await generate_pdf(report_data)

get_or_create(name, executor_type="local", **kwargs):

Returns the cached executor if name already exists (extra kwargs ignored).
Otherwise looks up executor_type in the factory table; an unknown type raises ValueError listing the available types.
Calls the factory as factory(name=name, **kwargs) and caches the result.

Other registry API:

register_factory(executor_type, factory) — register a custom type. The factory signature is factory(name: str, **kwargs) -> BaseExecutor.
get(name) — return the executor or None without creating it.
shutdown_all(wait=True) — shut down every executor and clear the cache.
all_metrics() — list of each executor’s metrics dict.
executors (property) — copy of the name -> executor map.
len(registry), name in registry — size and membership.

Custom factory

def make_remote(name, **kw):
    return RemoteExecutor(name, **kw)

registry.register_factory("remote", make_remote)
heavy = registry.get_or_create("heavy", executor_type="remote", url="...")

GenroAsgiWorker owns a ThreadExecutor

GenroAsgiWorker (server/worker.py) — the minimal single-app ASGI server a commander spawns to distribute load — owns a ThreadExecutor as its dispatch for blocking work:

self.executor = ThreadExecutor(name="worker", max_workers=max_workers)

The worker constructor accepts max_workers and passes it straight to the ThreadExecutor. The mounted app reaches the executor through the dual parent-child relationship as app.server.executor. The executor’s metrics are how the worker reports its own pressure (busy / total / queue_depth / occupancy); the executor makes no scaling decisions itself. On shutdown (run() finally-block) the worker calls self.executor.shutdown().

This is the only place in genro-asgi that wires an executor into the request path today.

Bypass mode for testing

Both concrete executors accept bypass=True (or honour the global GENRO_EXECUTOR_BYPASS=1 environment variable). In bypass mode no pool is created and submit() runs the function synchronously, so decorated functions still await correctly:

from genro_asgi.executors import LocalExecutor

executor = LocalExecutor(name="test", bypass=True)

@executor
def square(x):
    return x * x

async def test_square():
    assert await square(5) == 25
    assert executor.metrics["mode"] == "bypass"

Backpressure

Both executors bound in-flight work with a max_pending semaphore (default 100). LocalExecutor acquires that one semaphore around each task. ThreadExecutor adds a second max_workers “gate” semaphore: a task first passes max_pending admission, then waits for a free worker slot, and the wait between the two shows up as queue_depth.

ExecutorOverloadError is defined in base.py for backpressure overload scenarios; the current LocalExecutor / ThreadExecutor implementations block on the semaphore rather than raising it.

Constraints

LocalExecutor: decorated functions must be top-level (not lambdas or bound methods), and all arguments / return values must be pickle-serializable.
LocalExecutor worker-preloaded data is effectively read-only across tasks (separate process memory).
ThreadExecutor runs on the GIL (no pickling, no extra processes), so it suits I/O-bound — not CPU-bound — work; it propagates the contextvars context into the thread.
Pool workers are persistent: created at pool construction, reused for all tasks.

No server-level executor entry point yet

AsgiServer has no executor() method and wires no ExecutorRegistry in server/server.py; there is no ExecutorDecorator class. (The former ServerBinder, which delegated to a nonexistent server.executor(...), was dead code and has been removed.)

To obtain a pool today, use ExecutorRegistry.get_or_create (or instantiate ThreadExecutor / LocalExecutor directly), exactly as GenroAsgiWorker does. A server-level executor() convenience, backed by an ExecutorRegistry owned by the server, is a reasonable future addition but is not implemented.

Remote executors (roadmap — not implemented)

A remote/distributed executor is a planned extension, not present in the code. The intended shape is a custom factory registered on the registry:

# Roadmap — RemoteExecutor does not exist yet.
registry.register_factory("remote", lambda name, **kw: RemoteExecutor(name, **kw))
heavy = registry.get_or_create("heavy", executor_type="remote", url="...")

@heavy
def heavy_task(data):
    return process(data)  # would run on a remote worker

Goals: horizontal scaling across machines, load balancing, and fault tolerance, behind the same BaseExecutor decorator API as the local executors.