Bounding Concurrency

“Do them all at once” is the first thing people reach for with asyncio, and it is almost never what you want. With ten sources it’s fine. With ten thousand it opens ten thousand connections, and something between your process and the other end falls over — file descriptors, the remote’s rate limit, your own memory.

A library that does I/O on behalf of callers has to decide how much at once, and let them change it.

A semaphore is a counter of permits

asyncio.Semaphore(3) allows three holders at a time. A fourth waits inside async with until someone leaves.

async def read_one(name, limit):
    async with limit:                       # wait for a free slot
        async with Source(name) as source:
            return await source.read()

The nesting order matters: the semaphore is acquired before the source is opened, so at most three sources exist at a time. Open the source first and you’ve limited nothing that costs anything.

import asyncio

in_flight = 0
peak = 0


async def read(name, limit):
    global in_flight, peak
    async with limit:
        in_flight += 1
        peak = max(peak, in_flight)
        try:
            await asyncio.sleep(0.02)
            return f"{name}: ok"
        finally:
            in_flight -= 1


async def main(limit_size):
    global peak
    peak = 0
    limit = asyncio.Semaphore(limit_size)
    async with asyncio.TaskGroup() as group:
        for index in range(12):
            group.create_task(read(f"s{index}", limit))
    print(f"limit {limit_size}: peak in flight {peak}")


asyncio.run(main(3))
asyncio.run(main(6))
asyncio.run(main(100))

With a limit of 100 and twelve sources, the peak is twelve — the limit only binds when there’s more work than permits.

Test that the limit is reached, not just respected

This is the test almost everyone writes:

    def test_never_more_than_the_limit_at_once(self) -> None:
        asyncio.run(read_all([f"s{index}" for index in range(8)], limit=3))
        self.assertLessEqual(Source.peak_in_flight, 3)

It passes if the limit works. It also passes if the code accidentally became sequential — peak 1 is comfortably ≤ 3. Half the assertion is missing:

    def test_the_limit_is_actually_reached(self) -> None:
        """A limit nothing reaches would make the test above pass for free."""
        asyncio.run(read_all([f"s{index}" for index in range(8)], limit=3))
        self.assertEqual(Source.peak_in_flight, 3)

Any upper bound deserves this pair. One says you never exceed it, the other says you actually use it — and the second one is what fails the day somebody replaces group.create_task(worker(name)) with await worker(name).

Limiting tasks is not the same as limiting work

A semaphore caps how many coroutines are inside a block. That is not the same as capping how many tasks exist, or how much memory the results take.

import asyncio


async def read(name, limit):
    async with limit:
        await asyncio.sleep(0)
        return f"{name}: 42"


async def main():
    limit = asyncio.Semaphore(3)
    names = [f"s{index}" for index in range(10_000)]

    async with asyncio.TaskGroup() as group:
        tasks = [group.create_task(read(name, limit)) for name in names]
        print("tasks created at once:", len(tasks))
        print("still pending:", sum(1 for task in tasks if not task.done()))

    results = [task.result() for task in tasks]
    print("results held in memory:", len(results))


asyncio.run(main())

Ten thousand Task objects, all created before any of them finished, each holding a coroutine frame; plus ten thousand strings at the end. The semaphore did its job — only three were ever reading — and memory still scales with the input.

Two ways out, depending on what you’re protecting:

• Protecting the other end (connections, rate limits): a semaphore is exactly right.
• Protecting yourself (memory, backlog): you need to stop creating work until earlier work is finished. That’s backpressure, and it’s what the queue in the next lesson is for.

Timeouts belong here too

An unbounded wait is the other way concurrency goes wrong: one source that never answers holds a permit forever, and everything queued behind it stops.

        async with asyncio.timeout(2):
            return await source.read()

asyncio.timeout cancels the block when the deadline passes and raises TimeoutError. It’s a cancellation, which means whatever was awaiting gets CancelledError and must clean up on the way out — the subject of the next lesson, and the source of today’s bug hunt.

import asyncio
import unittest
from collections import defaultdict

in_flight = defaultdict(int)
peak = defaultdict(int)


async def fetch(host, path):
    in_flight[host] += 1
    peak[host] = max(peak[host], in_flight[host])
    try:
        await asyncio.sleep(0.01)
        return f"{host}{path}"
    finally:
        in_flight[host] -= 1


class HostLimiter:
    """TODO: hand out at most `per_host` permits for each host, independently.

    Used as:   async with limiter(host): ...
    """

    def __init__(self, per_host=2):
        self.per_host = per_host

    def __call__(self, host):
        raise NotImplementedError("HostLimiter.__call__")


async def crawl(urls, per_host=2):
    limiter = HostLimiter(per_host)
    results = []

    async def one(host, path):
        async with limiter(host):
            results.append(await fetch(host, path))

    async with asyncio.TaskGroup() as group:
        for host, path in urls:
            group.create_task(one(host, path))
    return results


class PerHost(unittest.TestCase):
    def setUp(self):
        in_flight.clear()
        peak.clear()

    def test_everything_is_fetched(self):
        urls = [("a.example", f"/{i}") for i in range(4)]
        self.assertEqual(len(asyncio.run(crawl(urls))), 4)

    def test_one_host_is_limited(self):
        urls = [("a.example", f"/{i}") for i in range(6)]
        asyncio.run(crawl(urls, per_host=2))
        self.assertEqual(peak["a.example"], 2)

    def test_hosts_do_not_block_each_other(self):
        urls = [("a.example", f"/{i}") for i in range(4)]
        urls += [("b.example", f"/{i}") for i in range(4)]
        asyncio.run(crawl(urls, per_host=2))
        self.assertEqual(peak["a.example"], 2)
        self.assertEqual(peak["b.example"], 2)

    def test_a_host_seen_twice_shares_its_limit(self):
        urls = [("a.example", "/1"), ("a.example", "/2"), ("a.example", "/3")]
        asyncio.run(crawl(urls, per_host=1))
        self.assertEqual(peak["a.example"], 1)


unittest.TextTestRunner(verbosity=2).run(
    unittest.defaultTestLoader.loadTestsFromTestCase(PerHost)
)

class HostLimiter:
    def __init__(self, per_host=2):
        self.per_host = per_host
        self.limits = {}

    def __call__(self, host):
        if host not in self.limits:
            self.limits[host] = asyncio.Semaphore(self.per_host)
        return self.limits[host]

What you learned

• A semaphore caps concurrent holders; acquire it before opening the expensive thing.
• Bound tests come in pairs: never more than the limit, and the limit is actually reached.
• Count events instead of timing them, or your suite will fail on a busy machine.
• Capping concurrency doesn’t cap memory; that needs backpressure.
• A timeout is a cancellation, which is why cleanup is the next lesson.