TopolearnShared state and rate limits still need coordination, even on a single thread.
Coroutines run on one thread, so a single statement like x += 1 can't be interrupted. But any sequence that awaits in the middle can: read a value, await something, write it back, and another coroutine may have changed it in between. asyncio provides async versions of the classic primitives (not thread-safe; for threads use threading):
asyncio.Lock:async with lock:ensures only one coroutine is inside the block at a time.asyncio.Semaphore(n): at most n holders at once, the standard way to cap concurrent requests to a service.BoundedSemaphoreraises if released too often.asyncio.Event: one coroutine callsevent.set(), any number blocked onawait event.wait()wake up. Good for "ready" or "shutdown" signals.asyncio.Conditionandasyncio.Barrier(3.11) cover rarer cases.
These primitives no longer take a loop argument and bind to the running loop when first used, so create them inside async code or at least use them in only one loop.
A Semaphore around a fan-out is the simplest rate limiter: gather all tasks, and each acquires the semaphore before its request.
Example
import asyncio
sem, active, peak = asyncio.Semaphore(2), [], 0
async def call_api(i):
global peak
async with sem: # at most 2 inside at once
active.append(i); peak = max(peak, len(active))
await asyncio.sleep(0.05)
active.remove(i)
return i
async def main():
ready = asyncio.Event()
async def waiter():
await ready.wait()
print("waiter: go!")
w = asyncio.create_task(waiter())
print(await asyncio.gather(*(call_api(i) for i in range(6))))
ready.set(); await w
print("peak concurrency:", peak)
asyncio.run(main())
# [0, 1, 2, 3, 4, 5] / waiter: go! / peak concurrency: 2Remember
- Races can happen across await points, even on one thread
- Semaphore(n) caps concurrent operations
- Event broadcasts a one-time signal to many waiters
Watch out for
- Using threading.Lock in async code: acquiring it blocks the whole event loop.