lesson 194 Async Python · advanced

Semaphores: limiting how much runs at once

Starting a thousand requests at once overloads servers and gets you rate-limited. A semaphore lets at most n tasks into a block at a time; the rest wait their turn.

You'll be able to

Use asyncio.Semaphore(n) with async with to cap concurrent work
Predict total time from the number of jobs, the limit and each job's duration
Give different resources their own limits

The idea

A semaphore holds a number of permits. async with sem: takes one (waiting if none are free) and gives it back when the block ends, even after an error. asyncio.Semaphore(2) therefore lets at most two tasks inside the block at once. Wrap only the part that needs limiting (the request itself), so waiting tasks don't hold permits.

Run each example to see what it prints. Running examples doesn't count toward progress.

example-1.py · Five pages, two at a time

import asyncio

class BrowserLoop(asyncio.BaseEventLoop):  # a real asyncio loop whose clock jumps instead of waiting
    def __init__(self):
        super().__init__()
        self._clock, self._selector = 0.0, self
    def time(self):
        return self._clock
    def select(self, timeout):
        self._clock += timeout
        return []
    def _process_events(self, events):
        pass
    def _write_to_self(self):
        pass

def run(main):  # on a computer: asyncio.run(main)
    outer = asyncio.events._get_running_loop()
    asyncio.events._set_running_loop(None)
    try:
        return asyncio.run(main, loop_factory=BrowserLoop)
    finally:
        asyncio.events._set_running_loop(outer)

async def main():
    loop = asyncio.get_running_loop()
    limit = asyncio.Semaphore(2)
    active = {"now": 0, "most": 0}

    async def fetch(page):
        async with limit:
            active["now"] += 1
            active["most"] = max(active["most"], active["now"])
            await asyncio.sleep(0.5)
            active["now"] -= 1
            return page

    pages = await asyncio.gather(*(fetch(p) for p in range(1, 6)))
    print(pages, active["most"], loop.time())

run(main())

The timing rule: with n jobs of equal length t and a limit of k, the total is about ceil(n / k) × t. Raising the limit helps until the server (or rate limit) becomes the bottleneck. Different resources often need different limits, such as 4 requests per API host but 2 database writes: give each its own semaphore.

example-2.py · One limit per host

import asyncio

class BrowserLoop(asyncio.BaseEventLoop):  # a real asyncio loop whose clock jumps instead of waiting
    def __init__(self):
        super().__init__()
        self._clock, self._selector = 0.0, self
    def time(self):
        return self._clock
    def select(self, timeout):
        self._clock += timeout
        return []
    def _process_events(self, events):
        pass
    def _write_to_self(self):
        pass

def run(main):  # on a computer: asyncio.run(main)
    outer = asyncio.events._get_running_loop()
    asyncio.events._set_running_loop(None)
    try:
        return asyncio.run(main, loop_factory=BrowserLoop)
    finally:
        asyncio.events._set_running_loop(outer)

async def main():
    loop = asyncio.get_running_loop()
    limits = {"api.runtime.market": asyncio.Semaphore(3), "images.runtime.market": asyncio.Semaphore(1)}

    async def get(host, path):
        async with limits[host]:
            await asyncio.sleep(0.2)
            return host.split(".")[0] + path

    jobs = [("api.runtime.market", f"/orders/{n}") for n in range(3)] + [("images.runtime.market", f"/p/{n}.jpg") for n in range(3)]
    results = await asyncio.gather(*(get(h, p) for h, p in jobs))
    print(len(results), results[0], results[-1], round(loop.time(), 2))

run(main())

When to use it: any time you fan out to an external service: APIs with rate limits, databases with connection limits, file systems. Mistakes to avoid: creating the semaphore inside each task (each gets its own, so nothing is limited); holding the permit while doing unrelated slow work; and confusing a limit on concurrency with a limit on rate (requests per second needs pacing too).

Review

asyncio.Semaphore(n) + async with lets at most n tasks into a block at once.
Total time ≈ ceil(jobs ÷ limit) × job time; the most constrained resource sets the pace.
Share one semaphore per resource; separate resources get separate semaphores.
Next: putting it together in an async request workflow.

Practise this

A fixed pool of async workers