Supercharge Your Python Asyncio With Aiomultiprocess: A Complete Information | by Peng Qian | Jul, 2023

Lastly, primarily based on my expertise, let me share some extra sensible finest practices.

Use pool solely

Though aiomultiprocess additionally offers the Course of and Employee lessons for us to select from, we should always at all times use the Pool class to make sure most effectivity because of the important useful resource consumption of making processes.

Learn how to use queues

In a earlier article, I defined tips on how to use asyncio.Queue to implement the producer-consumer sample to steadiness sources and efficiency.
In aiomultiprocess, we will additionally use queues. Nevertheless, since we’re in a course of pool, we can not use asyncio.Queue. On the identical time, we can not instantly use multiprocessing.Queue within the course of pool.
On this case, you must use multiprocessing.Supervisor().Queue() to create a queue, with the code as follows:

import random
import asyncio
from multiprocessing import Supervisor
from multiprocessing.queues import Queue
from aiomultiprocess import Pool
async def employee(identify: str, queue: Queue):
whereas True:
merchandise = queue.get()
if not merchandise:
print(f"employee: {identify} obtained the top sign, and can cease operating.")
queue.put(merchandise)
break
await asyncio.sleep(random.uniform(0.2, 0.7))
print(f"employee: {identify} start to course of worth {merchandise}", flush=True)
async def producer(queue: Queue):
for i in vary(20):
await asyncio.sleep(random.uniform(0.2, 0.7))
queue.put(random.randint(1, 3))
queue.put(None)
async def most important():
queue: Queue = Supervisor().Queue()
producer_task = asyncio.create_task(producer(queue))
async with Pool() as pool:
c_tasks = [pool.apply(worker, args=(f"worker-{i}", queue)) 
for i in range(5)]
await asyncio.collect(*c_tasks)
await producer_task
if __name__ == "__main__":
asyncio.run(most important())

Utilizing initializer to initialize sources

Suppose you must use an aiohttp session or a database connection pool in a coroutine technique, however we can not move arguments when creating duties in the primary course of as a result of these objects can’t be pickled.

An alternate is to outline a world object and an initialization technique. On this initialization technique, entry the worldwide object and carry out initialization.

Similar to multiprocessing.Pool, aiomultiprocess.Pool can settle for an initialization technique and corresponding initialization parameters when initialized. This technique might be known as to finish the initialization when every course of begins:

import asyncio
from aiomultiprocess import Pool
import aiohttp
from aiohttp import ClientSession, ClientTimeout
session: ClientSession | None = None
def init_session(timeout: ClientTimeout = None):
international session
session = aiohttp.ClientSession(timeout=timeout)
async def get_status(url: str) -> int:
international session
async with session.get(url) as response:
status_code = response.standing
return status_code
async def most important():
url = "https://httpbin.org/get"
timeout = ClientTimeout(2)
async with Pool(initializer=init_session, initargs=(timeout,)) as pool:
duties = [asyncio.create_task(pool.apply(get_status, (url,))) 
for i in range(3)]
standing = await asyncio.collect(*duties)
print(standing)
if __name__ == "__main__":
asyncio.run(most important())

Exception dealing with and retries

Though aiomultiprocess.Pool offers the exception_handler parameter to assist with exception dealing with, for those who want extra flexibility, you must mix it with asyncio.wait. For the utilization of asyncio.wait, you possibly can seek advice from my earlier article.

With asyncio.wait, you will get duties that encounter exceptions. After extracting the duty, you may make some changes after which re-execute the duty, as proven within the code under:

import asyncio
import random
from aiomultiprocess import Pool
async def employee():
await asyncio.sleep(0.2)
outcome = random.random()
if outcome > 0.5:
print("will increase an exception")
increase Exception("one thing error")
return outcome
async def most important():
pending, outcomes = set(), []
async with Pool() as pool:
for i in vary(7):
pending.add(asyncio.create_task(pool.apply(employee)))
whereas len(pending) > 0:
finished, pending = await asyncio.wait(pending, return_when=asyncio.FIRST_EXCEPTION)
print(f"now the rely of finished, pending is {len(finished)}, {len(pending)}")
for end in finished:
if outcome.exception():
pending.add(asyncio.create_task(pool.apply(employee)))
else:
outcomes.append(await outcome)
print(outcomes)
if __name__ == "__main__":
asyncio.run(most important())

Utilizing Tenacity for retries

In fact, we’ve got extra versatile and highly effective choices for exception dealing with and retries, similar to utilizing the Tenacity library, which I defined on this article.

With Tenacity, the code above might be considerably simplified. You simply want so as to add a decorator to the coroutine technique, and the strategy will robotically retry when an exception is thrown.

import asyncio
from random import random
from aiomultiprocess import Pool
from tenacity import *
@retry()
async def employee(identify: str):
await asyncio.sleep(0.3)
outcome = random()
if outcome > 0.6:
print(f"{identify} will increase an exception")
increase Exception("one thing mistaken")
return outcome
async def most important():
async with Pool() as pool:
duties = pool.map(employee, [f"worker-{i}" for i in range(5)])
outcomes = await duties
print(outcomes)
if __name__ == "__main__":
asyncio.run(most important())

Utilizing tqdm to point progress

I like tqdm as a result of it could possibly at all times inform me how far the code has run after I’m ready in entrance of the display. This text additionally explains tips on how to use it.

Since aiomultiprocess makes use of asyncio’s API to attend for duties to finish, additionally it is suitable with tqdm:

import asyncio
from random import uniform
from aiomultiprocess import Pool
from tqdm.asyncio import tqdm_asyncio
async def employee():
delay = uniform(0.5, 5)
await asyncio.sleep(delay)
return delay * 10
async def most important():
async with Pool() as pool:
duties = [asyncio.create_task(pool.apply(worker)) for _ in range(1000)]
outcomes = await tqdm_asyncio.collect(*duties)
print(outcomes[:10])
if __name__ == "__main__":
asyncio.run(most important())