You must distinguish CPU-intensive functions (computations, rendering, etc.), IO-bound functions (file reading, API requests with requests or urllib) and asynchronous functions (the ones that use async and await such as AIOHTTP with AsyncIO) as this will dictate how you'll design your concurrent program.
First, you should not perform blocking calls (CPU-intensive work or blocking IO function such as requests.get) inside an asynchronous functions as it will block the event loop of the asynchronous runtime, resulting in concurrency issues. The following example illustrate this problem:
import asyncio
import requests
from math import sqrt
# Asynchronous function
async def fetch(i: int) -> int:
await asyncio.sleep(1.0)
return i
# CPU-intensive function
def render(i: int) -> int:
for _ in range(30_000_000):
_ = sqrt(i) ** 2
return i
# This function is flawed!
async def work(i: int) -> int:
value = await fetch(i)
return render(i) # CPU-intensive work inside `async`!
# This function is also flawed!
async def work(i: int) -> int:
value = await fetch(i)
return requests.get(f"https://url{i}"). # Blocking IO work inside `async`!
If you want to perform CPU-intensive work inside an async function, you should offload it to a process pool (for instance the ProcessPoolExecutor from concurrent.futures):
import asyncio
from concurrent.futures import ProcessPoolExecutor
async def work(i: int, executor: ProcessPoolExecutor) -> int:
i = await fetch(i)
loop = asyncio.get_running_loop()
return await loop.run_in_executor(executor, render, i)
If you want to perform an IO-bound operation inside an async function you can simply offload the work to a separate thread:
import asyncio
import requests
async def work(i: int) -> int:
i = await fetch(i)
return await asyncio.to_thread(requests.get, f"https://someurl{i}/")
Given that all your functions are now correct and depending on your use case you can use different frameworks for limiting the concurrency:
- AsyncIO queues for
async functions,
ProcessPoolExecutor with max_workers = 2 for example for CPU-intensive functions,- or dedicated API's such as
TCPConnector for AIOHTTP.
Here is how you would write a function that limits concurrent render tasks to two using AsyncIO and ProcessPoolExecutor:
import asyncio
from concurrent.futures import ProcessPoolExecutor
async def main():
with ProcessPoolExecutor(max_workers=2) as executor:
results = await asyncio.gather(
*(work(i, executor) for i in range(20))
)
print(results)
Not that this will not limit the concurrency of the fetch operations but only the CPU-intensive tasks running in the process pool. To limit the number of concurrent fetch operations you could use AIOHTTP's TCPConnector, an AsyncIO queue or a tool such as the throttler Python package.
