aiomultiprocess

Warnings

• breaking By default, aiomultiprocess uses 'spawned' processes on all platforms, which is different from the standard `multiprocessing` module's default 'forked' on Linux/macOS (for Python versions prior to 3.8). This means any objects or coroutines passed to child processes *must* be pickleable (importable from a fresh child process). Unpickleable objects will cause errors.
  Fix: Ensure all functions, classes, and global objects passed to worker processes are defined at the top-level of a module and are importable. If 'forked' behavior is strictly required (e.g., for sharing non-pickleable resources that are copied by fork), call `aiomultiprocess.set_start_method('fork')` before creating any workers or pools.
• gotcha Global variables and shared state behave differently with spawned processes. Each child process gets its own independent memory space; changes to global variables in one process are not reflected in others. This can lead to unexpected behavior if not accounted for.
  Fix: Avoid relying on shared mutable global state. Use queues (`multiprocessing.Manager().Queue()`) for explicit inter-process communication or pass data explicitly as arguments and return values. For resource initialization per process, use the `initializer` and `initargs` parameters of the `Pool` constructor.
• gotcha Exceptions raised within worker processes are automatically caught and re-raised in the main process as `ProxyException` objects. While convenient, this might obscure the original traceback or prevent in-worker exception handling (e.g., for logging or reporting).
  Fix: To handle exceptions within the worker process itself, provide an `exception_handler` callable to the `Pool` (or `Process`/`Worker`) constructor. This handler will be called with the exception object before it's propagated back to the main process.
• gotcha Long-running tasks or high concurrency with `aiomultiprocess.Pool` can sometimes lead to 'OSError: [Errno 24] Too many open files' or memory leaks if worker processes are not periodically refreshed.
  Fix: Utilize the `maxtasksperchild` parameter when creating a `Pool`. Setting it to a positive integer will cause worker processes to exit and be respawned after completing the specified number of tasks, helping to release resources and prevent file handle exhaustion.
• gotcha If you are using an alternative `asyncio` event loop implementation (e.g., `uvloop`), `aiomultiprocess` will not automatically use it in child processes. The default `asyncio` loop will be used instead, potentially leading to suboptimal performance.
  Fix: Explicitly specify the event loop initializer using the `loop_initializer` parameter in the `Pool` (or `Process`/`Worker`) constructor. For `uvloop`, this would be `loop_initializer=uvloop.new_event_loop`.

Install

• pip install aiomultiprocess Install with pip

Imports

• Pool

  from aiomultiprocess import Pool

  The primary class for managing a pool of worker processes.
• Process

  from aiomultiprocess import Process

  For running a single coroutine in a dedicated subprocess, similar to `multiprocessing.Process`.
• Worker

  from aiomultiprocess import Worker

  Similar to `Process` but specifically designed to return results from a coroutine task in a subprocess.

Quickstart

This quickstart demonstrates using `aiomultiprocess.Pool` to fetch multiple URLs concurrently across several processes. It defines an asynchronous function `fetch_url_content` that uses `aiohttp` to make an HTTP GET request. The `main` function then creates a `Pool` and uses its `map` method to distribute the `fetch_url_content` coroutine calls across the worker processes, iterating over results as they complete.

import asyncio
from aiohttp import ClientSession
from aiomultiprocess import Pool

async def fetch_url_content(url: str) -> str:
    """An example async coroutine to be run by the pool."""
    async with ClientSession() as session:
        async with session.get(url) as response:
            response.raise_for_status() # Raise an exception for bad status codes
            return await response.text()

async def main():
    urls = [
        "https://www.google.com",
        "https://www.python.org",
        "https://docs.python.org/3/library/asyncio.html",
        "https://www.wikipedia.org"
    ]

    print(f"Fetching {len(urls)} URLs using aiomultiprocess Pool...")
    async with Pool(processes=2) as pool: # Use 2 processes for demonstration
        # Use pool.map to apply the coroutine to each URL
        async for result in pool.map(fetch_url_content, urls):
            if result: # Check if result is not None (e.g., if an exception was caught by handler)
                print(f"Fetched content size: {len(result)} bytes for one URL.")
            else:
                print("Failed to fetch content for a URL.")

if __name__ == '__main__':
    # Ensure aiohttp is installed for this example: pip install aiohttp
    try:
        asyncio.run(main())
    except ImportError:
        print("Please install aiohttp for this example: pip install aiohttp")
    except Exception as e:
        print(f"An error occurred: {e}")