aiomultiprocess

The core concept of aiomultiprocess is to combine the benefits of asyncio and multiprocessing.

• AsyncIO alone is limited by the Global Interpreter Lock (GIL).
• Multiprocessing alone typically works on one task at a time per process.

aiomultiprocess runs a full AsyncIO event loop on each child process. This allows each child process to execute multiple coroutines at once, enabling high levels of concurrency that scale with the number of available CPU cores.

Install aiomultiprocess using pip3. This package requires Python 3.6 or newer.

$ pip3 install aiomultiprocess

The map and starmap methods return a PoolResult object. This object is both an Awaitable and an AsyncIterable, providing two ways to consume results:

1. Await it: Returns the entire sequence of results once all tasks are complete.
2. Async iterate: Yields results one-by-one as they become available (though note that results_generator currently iterates through the task_ids in order).

If any task raises an exception, awaiting the PoolResult or iterating through it will raise a ProxyException containing the traceback.

You can optimize Pool performance using the following configuration options:

Performance Formulas:

• total_concurrency = processes * childconcurrency
• throughput = total_concurrency / job_time
• contention = throughput / queuecount

The map() and starmap() methods queue one job for each element in the provided iterable. You can await these methods to get all results as a list in the original input order, or iterate over them with async for to process results as they complete (while still maintaining order).

import math
from aiomultiprocess import Pool

async with Pool() as pool:
    data = [1, 4, 9, 16, 25]
    
    # Get all results as a list (awaited)
    results = await pool.map(math.sqrt, data)
    # [1, 2, 3, 4, 5]

    # Process results as they complete (iterated)
    async for value in pool.map(math.sqrt, data):
        ...

Use pool.apply() to queue individual jobs into the process pool. This is useful when you want to use asyncio.gather to manage multiple specific tasks being distributed across the pool.

from asyncio import gather
from aiomultiprocess import Pool

async def get(url):
    async with request("GET", url) as response:
        return await response.text("utf-8")

async with Pool() as pool:
    a, b, c = gather(
        pool.apply(get, "https://github.com"),
        pool.apply(get, "https://noswap.com"),
        pool.apply(get, "https://omnilib.dev"),
    )

The aiomultiprocess.Pool class is the primary way to run multiple coroutines concurrently across a pool of worker processes. You can use pool.map() to iterate over results as they complete, which maintains the original input order.

Note: Pool is best used as an asynchronous context manager to ensure proper cleanup.

from aiohttp import request
from aiomultiprocess import Pool

async def get(url):
    async with request("GET", url) as response:
        return await response.text("utf-8")

async def main():
    urls = ["https://noswap.com", ...]
    async with Pool() as pool:
        async for result in pool.map(get, urls):
            ...  # process result

Use the initializer and initargs parameters in Pool to run arbitrary code in each child process after its async event loop has been created. This is useful for setting up logging, database connections, or other per-process resources.

import logging
from aiomultiprocess import Pool

def setup_logging(level=logging.WARNING):
    logging.basicConfig(level=level)

async with Pool(
    initializer=setup_logging, initargs=(logging.DEBUG,)
) as pool:
    ...

Exceptions raised in workers are wrapped in aiomultiprocess.types.ProxyException when they reach the main process. To handle exceptions within the worker process itself (e.g., for external monitoring), use the exception_handler hook.

import sentry_sdk
from aiomultiprocess import Pool

async with Pool(
    exception_handler=sentry_sdk.capture_exception
) as pool:
    ...