Python: Blocking Calls, Locks and Queues in asyncio

Description

asyncio is cooperative: every coroutine shares one thread and has to give it back at each await. When all the code cooperates, one process serves thousands of connections smoothly. When one piece of code doesn't, everything stalls - and the symptoms show up far away from the cause. This lesson covers the mistakes that cause most asyncio incidents in real services, and the standard fixes: keeping blocking calls off the loop, races that happen across an await, and limiting how much work runs at once.

Here's the kind of bug we mean. An async web service runs happily until someone adds a "quick" call using a normal HTTP library:

async def get_rates(request):
    rates = requests.get(RATES_URL).json()      # blocks for 300 ms
    return json_response(rates)

Every request that reaches this line freezes the whole service for 300 ms. Health checks time out, WebSocket heartbeats stop, and endpoints that have nothing to do with rates get slow. By the end of the first section you'll know why, and three ways to fix it.

We'll go from simple to more technical:

1. Blocking calls: what they are, why they freeze everything, and the fixes.
2. Finding the code that blocks.
3. Races across await, and asyncio.Lock.
4. Limiting concurrency with semaphores and queues.
5. Talking to the event loop from other threads.

Blocking calls

The event loop runs on one thread and can only switch coroutines at an await. A blocking call is any function that makes the thread wait without an await: the loop is stuck inside it and can't run anything else until it returns. Three coroutines that each called time.sleep(0.2) inside gather took 0.6 s in total, not 0.2 - they couldn't overlap at all.

Common blocking calls in async code:

• time.sleep;
• synchronous HTTP clients: requests, urllib;
• synchronous database drivers (psycopg2, the standard sqlite3);
• subprocess.run, and reading big files;
• heavy computation: parsing a large JSON document, hashing passwords, resizing images.

The fixes, best first:

1. Use the async version. await asyncio.sleep(...), httpx.AsyncClient or aiohttp for HTTP, asyncpg or an async SQLAlchemy engine for databases, asyncio.create_subprocess_exec for subprocesses.
2. Run blocking I/O in a thread with await asyncio.to_thread(func, *args) (Python 3.9+). The loop keeps serving while a worker thread waits: three time.sleep(0.2) calls through to_thread finished together in about 0.2 s.
3. Run heavy computation in a process. A thread won't help CPU-bound Python because of the GIL; use a process pool: await loop.run_in_executor(process_pool, func, *args).

Finding the code that blocks

Blocking calls hide inside libraries, so look for them deliberately. Run with debug mode - asyncio.run(main(), debug=True) or the environment variable PYTHONASYNCIODEBUG=1 - and the loop logs every step that held it for more than 100 ms. A handler containing a 0.2 s time.sleep was reported as Executing <Task finished name='Task-1' coro=<handler() done, ...> ...> took 0.210 seconds - the task's name and coroutine tell you where to look. Run your test suite in debug mode now and then; it catches blocking calls before production does.

Races across await

Because coroutines only switch at await, code between two awaits never gets interrupted - that removes most of the races threads have. But it doesn't remove races across an await:

async def book(user):
    global seats
    if seats > 0:                 # both coroutines see 1...
        await charge(user)        # ...and switch here...
        seats -= 1                # ...so both book the last seat

Two bookings for the last seat both succeeded and left seats at -1. The fix is an asyncio.Lock held across the whole check-await-update sequence:

seat_lock = asyncio.Lock()

async def book(user):
    global seats
    async with seat_lock:
        if seats > 0:
            await charge(user)
            seats -= 1

Now only one booking gets the seat. Two warnings about locks:

• asyncio.Lock is for coroutines on one event loop. It isn't thread-safe, and using it from a second event loop raises RuntimeError: ... is bound to a different event loop.
• Never hold a threading.Lock across an await. If another coroutine on the same loop then tries to take it, it blocks the loop's thread - and the coroutine holding the lock can never run to release it.

Limiting concurrency: semaphores and queues

Starting 10,000 HTTP calls at once with gather will overwhelm the other service, or get you rate-limited. Two tools cap how much runs at a time:

• asyncio.Semaphore(n) lets at most n coroutines into a section at once:

limit = asyncio.Semaphore(10)

async def fetch_one(url):
    async with limit:                         # at most 10 calls in flight
        return await client.get(url)

pages = await asyncio.gather(*(fetch_one(u) for u in urls))

30 calls of 0.1 s each with Semaphore(10) took about 0.3 s - three waves of ten.

• asyncio.Queue(maxsize=n) connects producers and a fixed number of worker coroutines. await queue.put(item) waits when the queue is full - that's backpressure, slowing the producer down to the speed of the workers - and await queue.get() waits when it's empty. queue.task_done() plus await queue.join() waits until every item has been processed. To stop the workers, Python 3.13 added queue.shutdown(): workers waiting in get() get a QueueShutDown exception and can exit cleanly. In a test, a worker processed five items and then exited on shutdown.

Talking to the loop from other threads

asyncio objects - tasks, locks, queues, the loop itself - are not thread-safe. Code running in another thread (a callback from a C library, a to_thread worker) must hand work over to the loop:

• loop.call_soon_threadsafe(callback, *args) schedules a plain function on the loop's thread.
• asyncio.run_coroutine_threadsafe(coro, loop) runs a coroutine on the loop and returns a concurrent.futures.Future the thread can wait on with .result().

Which tool for which job?

Common mistakes

Check yourself

Try to answer before opening each one. Recalling beats re-reading.

1. Why does a requests.get inside one handler slow down every other request?

2. Coroutines only switch at await. How can two bookings still take the last seat?

3. How do you make 10,000 fetches run at most 20 at a time?

Key takeaways

• Never block the loop: async libraries first, to_thread for blocking I/O, processes for heavy computation.
• Debug mode logs every step that holds the loop for over 100 ms.
• An await is where other coroutines run: lock check-await-update sequences with asyncio.Lock.
• Semaphores cap fan-out; queues give backpressure; Queue.shutdown stops workers.
• asyncio objects aren't thread-safe: go through run_coroutine_threadsafe or call_soon_threadsafe.