Async I/O in Anger
Async context managers and iterators, connection pools, backpressure when producers outrun consumers, and bridging sync and async code without deadlocking.
Async context managers and iterators, connection pools, backpressure when producers outrun consumers, and bridging sync and async code without deadlocking.
The scraper works perfectly against ten URLs. Pointed at eighty thousand, it opens eighty thousand sockets in the first second, exhausts the file descriptor limit, gets rate-limited by everything it touches, and accumulates eighty thousand pending results in memory because the database writer is slower than the fetcher.
None of that is an asyncio bug. It is what happens when work is started without limit and results are produced faster than they are consumed. By the end of this lesson you will bound concurrency, apply backpressure, share connection pools correctly, and cross the sync boundary without deadlocking.
Bad — one task per item.
async def scrape_all(urls):
async with asyncio.TaskGroup() as group:
for url in urls: # 80,000 tasks
group.create_task(fetch(url))Good — a semaphore capping what is in flight.
async def scrape_all(urls, limit=20):
semaphore = asyncio.Semaphore(limit)
async def bounded(url):
async with semaphore:
return await fetch(url)
async with asyncio.TaskGroup() as group:
tasks = [group.create_task(bounded(u)) for u in urls]
return [task.result() for task in tasks]Creating eighty thousand tasks is cheap — a few hundred megabytes — but each one immediately tries to open a socket. You hit the process file-descriptor limit, the remote service starts returning 429s, and the failures look like network problems rather than a decision you made.
The semaphore lets twenty run and makes the rest wait. Note it
wraps the work, not the task creation: all eighty thousand
tasks still exist, and they queue at the async with.
For very large inputs, avoid creating the tasks at all by processing in batches or with a worker pattern:
async def worker(queue, results):
while True:
url = await queue.get()
try:
results.append(await fetch(url))
finally:
queue.task_done()queue = asyncio.Queue(maxsize=1000)
async with asyncio.TaskGroup() as group:
workers = [group.create_task(worker(queue, results))
for _ in range(20)]
for url in urls:
await queue.put(url) # blocks when the queue is full
await queue.join()
for task in workers:
task.cancel()Twenty coroutines, one bounded queue, and constant memory regardless of input size.
That maxsize=1000 is the important part, and it is the
principle the opening scenario violated.
Backpressure is a slow consumer being able to slow the
producer down. await queue.put(url) on a full queue suspends
until there is room — so the fetcher cannot outrun the writer,
and memory stays flat.
asyncio.Queue() # unbounded: the producer never waits
asyncio.Queue(maxsize=1000) # bounded: the producer waitsA memory leak with a scheduler attached.
The producer never waits, so eight hundred items a second accumulate. Nothing errors and no log line appears; the process simply grows until it is killed.
The same shape as the unbounded cache from the memory lesson.
The producer runs at the consumer's speed.
await queue.put(...) on a full queue suspends until there
is room, so memory stays flat and the slowness is visible as
slowness rather than as an out-of-memory kill.
The number itself matters less than that there is one. A bounded queue converts "we run out of memory in twenty minutes" into "the producer runs at the consumer's speed", which is almost always the behaviour you wanted.
That creates a connection pool per request, does a TLS handshake per request, and throws the pool away — the same mistake as the HTTP lesson, with the extra cost that the handshakes now compete for one event loop.
Create one client for the program and pass it in:
limits is a second bound, at a different layer: the semaphore
controls how many of your tasks are working, and the pool
controls how many sockets exist. Setting both is not
redundant — the pool limit protects the remote service and your
descriptor table even if a bug lets the semaphore grow.
The same applies to database pools:
Acquire late and release early. Holding a connection across
an unrelated await — an HTTP call, a sleep — occupies a pool
slot while doing nothing with it, and twenty tasks doing that
deadlock a pool of twenty.
Async code calling blocking code has one answer, from the asyncio lesson:
The other direction is harder, and where you are calling from decides everything.
A plain synchronous program
asyncio.run, once, at the top. Never inside a coroutine —
nesting raises.
Inside a running loop
You cannot block on a coroutine at all. asyncio.run raises
and loop.run_until_complete deadlocks — you are asking the
loop to run something while occupying its only thread.
Another thread
asyncio.run_coroutine_threadsafe, which schedules the work
on the loop's thread and hands your thread something ordinary
to wait on.
That third one, in full:
That schedules the coroutine on the loop's thread and gives you
a concurrent.futures.Future to wait on from yours. It is the
correct bridge, and it needs the loop object, captured earlier
with asyncio.get_running_loop().
The general shape that avoids all of this: async at the edges, sync in the middle. Keep I/O and orchestration async; keep the decision-making in ordinary functions that take values and return values. Those are also the functions that are easy to test, which the testing lesson argued for on entirely separate grounds.
Streaming a large response keeps memory flat and lets work start before the download finishes:
stream is a context manager because the connection stays open
for the duration — reading the body is the point, and the pool
slot is held until you finish. Consume it promptly.
Combining several async iterators is where people reach for something that does not exist in the standard library:
The aiostream library provides this and more. It is shown here
because the pattern — one pending task per source, replaced as
each completes — is worth recognising.
pytest-asyncio or anyio's plugin runs the coroutine. The
comprehension over an async iterator is async for in
expression form, and it collects a stream for assertion.
Two practical notes. Give every async test a timeout, because a
deadlocked test hangs the suite rather than failing it. And
prefer a mock transport to patching the client — the mocking
lesson's point about patching at the boundary applies, and
httpx.MockTransport exercises your real client configuration.
You can now run async work at scale without exhausting sockets or memory, share pools correctly, and move between async and sync code deliberately. The idea underneath all of it is one sentence: anything unbounded will find its bound in production, and it will be a resource limit rather than a number you chose.
Next is Performance Engineering, which returns to the profiling discipline from the practice course and goes deeper — what to do once you have a profile, where the boundary with native code sits, and how to know when the answer is a different data structure rather than faster code.
Before you move on, take an async script you have and add a semaphore around its concurrent work. Then run it against a deliberately large input with the limit set to 5, then 50, then 500, and watch where throughput stops improving and errors start. That curve is the number you should have been setting all along.
BOUND EVERYTHING
Semaphore(20) around the WORK, not the task creation
a bounded Queue(maxsize=N) + N workers, for very large inputs
client limits: max_connections, max_keepalive_connections
the semaphore bounds your tasks; the pool bounds sockets -
set both
BACKPRESSURE
await queue.put(x) on a FULL queue suspends the producer
an unbounded queue between fast producer and slow consumer
is a memory leak with a scheduler attached
the size matters less than having one
POOLS AND CLIENTS
one client for the program; pass it in
a client per request = a TLS handshake per request
acquire a connection LATE, release EARLY
holding one across an unrelated await deadlocks the pool
create pools inside the async entry point, never at import
a pool is bound to the loop that made it
CROSSING THE BOUNDARY
sync from async await asyncio.to_thread(fn, arg)
async from sync asyncio.run(main()) once, at the top
from a running loop you CANNOT block on a coroutine
from another thread asyncio.run_coroutine_threadsafe(coro, loop)
-> future.result(timeout=...)
design: async at the edges, sync in the middle
STREAMING
async with client.stream(...) as r:
async for chunk in r.aiter_bytes():
the pool slot is held until you finish - consume promptly
TESTING
pytest-asyncio / anyio; a timeout on every async test
[x async for x in stream()] to collect
a mock transport, not a patched client
debug=True in tests: slow callbacks, never-awaited coroutinesasync def fetch(url):
async with httpx.AsyncClient() as client: # a new pool, per call
return await client.get(url)async def main():
async with httpx.AsyncClient(
timeout=10.0,
limits=httpx.Limits(max_connections=100,
max_keepalive_connections=20),
) as client:
await scrape_all(client, urls)pool = await asyncpg.create_pool(dsn, min_size=5, max_size=20)
async with pool.acquire() as connection:
rows = await connection.fetch(query)text = await asyncio.to_thread(path.read_text)future = asyncio.run_coroutine_threadsafe(handle(event), loop)
result = future.result(timeout=10)async with client.stream("GET", url) as response:
response.raise_for_status()
async for chunk in response.aiter_bytes():
await handle.write(chunk)async def merge(*iterators):
"""Yield from several async iterators as items arrive."""
tasks = {
asyncio.ensure_future(anext(it)): it for it in iterators
}
while tasks:
done, _ = await asyncio.wait(
tasks, return_when=asyncio.FIRST_COMPLETED
)
for task in done:
iterator = tasks.pop(task)
try:
yield task.result()
except StopAsyncIteration:
continue
tasks[asyncio.ensure_future(anext(iterator))] = iteratorimport pytest
@pytest.mark.anyio
async def test_fetches_every_page():
async with httpx.AsyncClient(transport=mock_transport) as client:
pages = [p async for p in read_pages(client, "/photos")]
assert len(pages) == 3