# Copyright 2025 Softwell S.r.l.
# Licensed under the Apache License, Version 2.0
"""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.
"""
from __future__ import annotations
import asyncio
import os
import pickle
from concurrent.futures import ProcessPoolExecutor
from functools import partial
from typing import Any, Callable
from .base import BaseExecutor, ExecutorError
__all__ = ["LocalExecutor"]
[docs]
class LocalExecutor(BaseExecutor):
"""
Executor using local ProcessPoolExecutor.
Runs functions in separate processes for true parallelism,
bypassing Python's GIL. Ideal for CPU-bound work.
Attributes:
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)
"""
__slots__ = ("max_pending", "_semaphore")
pool_mode = "process"
[docs]
def __init__(
self,
name: str = "default",
max_workers: int | None = None,
initializer: Callable[..., None] | None = None,
initargs: tuple[Any, ...] = (),
max_pending: int = 100,
bypass: bool = False,
) -> None:
"""
Initialize LocalExecutor.
Args:
name: Identifier for metrics and logging.
max_workers: Number of worker processes (default: CPU count).
initializer: Function called once per worker at startup.
initargs: Arguments passed to initializer.
max_pending: Maximum concurrent pending tasks.
bypass: If True, run synchronously without pool (for testing).
"""
super().__init__(name)
self.max_pending = max_pending
# Check environment for global bypass
env_bypass = os.environ.get("GENRO_EXECUTOR_BYPASS") == "1"
if bypass or env_bypass:
self.pool = None
self._semaphore: asyncio.Semaphore | None = None
else:
self.pool = ProcessPoolExecutor(
max_workers=max_workers,
initializer=initializer,
initargs=initargs,
)
self._semaphore = asyncio.Semaphore(max_pending)
async def _submit_pooled(self, func: Callable[..., Any], *args: Any, **kwargs: Any) -> Any:
"""Run func in the process pool behind the backpressure semaphore.
Functions and arguments must be pickle-serializable; metrics are
handled by BaseExecutor.submit().
"""
if self._semaphore is not None:
async with self._semaphore:
return await self._execute(func, *args, **kwargs)
return await self._execute(func, *args, **kwargs)
async def _execute(self, func: Callable[..., Any], *args: Any, **kwargs: Any) -> Any:
"""Execute function in the process pool."""
loop = asyncio.get_running_loop()
call = partial(func, *args, **kwargs)
try:
return await loop.run_in_executor(self.pool, call)
except pickle.PicklingError as e:
raise ExecutorError(
f"Cannot serialize arguments for {func.__name__}. "
f"Ensure all args are pickle-serializable. Original: {e}"
) from e
def __repr__(self) -> str:
"""Return string representation."""
mode = "bypass" if self.pool is None else "process"
return f"LocalExecutor(name={self.name!r}, mode={mode})"
if __name__ == "__main__":
import asyncio
async def main() -> None:
# Test bypass mode
executor = LocalExecutor(name="test", bypass=True)
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}")
asyncio.run(main())