The floor freezes: blocking calls and mixing tools

Eleven o'clock. Table 3 orders a gratin. Gratin can only be baked in the old oven the chef bought secondhand. Once its door is shut, the handle locks until the baking is done, and someone has to stand there holding it.

Lou steps up to the oven. In that instant the floor freezes. Table 1 wants water, table 5 wants the check, and both just wait for Lou to come back.

Mr. Tak clicks his tongue. "Standing in front of an oven isn't working. It's just standing." The chef nods. "Lou can't wait. Hand the waiting to someone else."

11-1. When Lou freezes, the floor freezes (7 min)

The asyncio event loop runs on a single thread. The loop moves on to another task only when a coroutine yields at an await. But call a function that knows nothing about await, such as time.sleep or an old synchronous library, and the loop can't budge until that function returns. A call like that is called a blocking call.

import asyncio, time

start = time.perf_counter()

async def lu_rounds():                 # Lou does a lap of the floor every 0.1 s
    for _ in range(5):
        print(f"round {time.perf_counter() - start:.1f}s")
        await asyncio.sleep(0.1)

def old_oven():                        # an old function that knows nothing about await
    time.sleep(0.35)
    return "bread"

async def bake():
    await asyncio.sleep(0.15)
    print("oven", old_oven())          # the whole loop freezes here for 0.35 s

async def main():
    await asyncio.gather(lu_rounds(), bake())

asyncio.run(main())
# round 0.0s
# round 0.1s
# oven bread
# round 0.5s
# round 0.6s
# round 0.7s

• The laps at 0.2, 0.3 and 0.4 seconds vanished entirely. From 0.15 to 0.5 seconds, while Lou stood at the oven, no task could run.
• A blocking call doesn't just slow down its own coroutine. It freezes every task on the loop along with it. That's the biggest difference from having several threads.
• The usual suspects: time.sleep, synchronous network libraries like requests.get, reading big files, heavy computation.

In the timeline below, one slow table uses the old oven (time.sleep, 0.6 seconds). The other tables wait with asyncio.sleep, in 0.1-second slices.

• With the switch off, it takes about 0.9 seconds. T1 and T2 are 0.3-second dishes but stretch out to about 0.76 seconds, and T4 to T6 don't even start until the oven finishes at 0.6 seconds. Lou hadn't even reached those tables yet.
• Make table 1 the slow table and the other five all start only after 0.6 seconds.
• Turn the switch on and the oven moves to a thread: the other tables finish at about 0.3 seconds, and the total drops to the oven's time, about 0.6 seconds.

11-2. Finding the culprit with debug mode (6 min)

In a big program it's hard to tell which coroutine grabbed the loop. asyncio's debug mode logs callbacks and tasks that ran too long in one go.

import asyncio, logging, time

logging.basicConfig(level=logging.WARNING, format="%(name)s %(levelname)s: %(message)s")

def old_oven():
    time.sleep(0.3)

async def bake():
    old_oven()

async def main():
    await asyncio.create_task(bake(), name="T3-oven")

asyncio.run(main(), debug=True)
# e.g. asyncio WARNING: Executing <Task finished name='T3-oven' coro=<bake() done, defined at ...oven.py:8>
#     result=None created at ...\asyncio\tasks.py:395> took 0.301 seconds

• Turn it on with asyncio.run(main(), debug=True). You can also use the environment variable PYTHONASYNCIODEBUG=1 or python -X dev.
• One warning line gives you the task name (T3-oven), the file and line where the coroutine is defined, and how long it held the loop (took 0.301 seconds). Give your tasks a name= and the culprit is easier to spot.
• The message comes out as a WARNING from the asyncio logger. The paths differ on every machine.

The threshold is loop.slow_callback_duration, 0.1 seconds by default. Lower it to make the check more sensitive.

import asyncio, time

async def main():
    loop = asyncio.get_running_loop()
    print(loop.get_debug(), loop.slow_callback_duration)
    loop.slow_callback_duration = 0.05     # tell me about anything over 0.05 s
    time.sleep(0.07)

asyncio.run(main(), debug=True)
# True 0.1
# e.g. Executing <Task finished name='Task-1' coro=<main() done, defined at ...> result=None created at ...> took 0.070 seconds

Debug mode also tells you where a coroutine that was dropped without an await was created, and raises an error if another thread touches the loop the wrong way. Keep it on during development and off in production (it slows things down).

11-3. Leave the waiting to a thread: to_thread and run_in_executor (8 min)

If you can't fix the old oven, put someone else in front of it. asyncio.to_thread() (3.9) runs a blocking function in another thread and gives the loop something it can await.

import asyncio, time

start = time.perf_counter()

async def lu_rounds():
    for _ in range(5):
        print(f"round {time.perf_counter() - start:.1f}s")
        await asyncio.sleep(0.1)

def old_oven(item, *, minutes):
    time.sleep(0.35)
    return f"{item} {minutes}m"

async def bake():
    await asyncio.sleep(0.15)
    bread = await asyncio.to_thread(old_oven, "bread", minutes=40)
    print("oven", bread)

async def main():
    await asyncio.gather(lu_rounds(), bake())

asyncio.run(main())
# round 0.0s
# round 0.1s
# round 0.2s
# round 0.3s
# round 0.4s
# oven bread 40m

• Lou does her lap every 0.1 seconds without missing one. A thread minds the oven, and Lou just receives the result with await.
• to_thread(function, *args, **kwargs) passes the arguments straight through. It runs in Lesson 7's thread pool (the loop's default executor).
• It's a thread, so the GIL rules from Lesson 8 still apply. It works well for blocking code that waits, but it won't speed up heavy computation.

Heavy computation goes to a branch kitchen, meaning a process pool. Get the loop that's currently running with asyncio.get_running_loop(), then use loop.run_in_executor(executor, function, *args). Since it uses processes, save it to a file and run it.

import asyncio
from concurrent.futures import ProcessPoolExecutor

def knead(n):                          # hands-on work (CPU)
    return sum(i * i for i in range(n))

async def main():
    loop = asyncio.get_running_loop()
    with ProcessPoolExecutor(max_workers=2) as pool:
        doughs = await asyncio.gather(
            loop.run_in_executor(pool, knead, 2_000_000),
            loop.run_in_executor(pool, knead, 3_000_000),
        )
    print(doughs)

if __name__ == "__main__":
    asyncio.run(main())
# [2666664666667000000, 8999995500000500000]

• Pass None as the first argument of run_in_executor and it uses the default thread pool. to_thread is just a convenient wrapper for exactly that.
• run_in_executor doesn't accept keyword arguments. If you need them, wrap the call with functools.partial(knead, n=...).
• It's a process pool, so Lesson 8's rules still apply: you need if __name__ == "__main__":, and the function and values you pass must be picklable.

11-4. asyncio's synchronization tools (8 min)

The tools from Lessons 4 and 6 have asyncio versions of their own. The names are the same, but when they wait, instead of freezing the thread they yield with await.

Semaphores are used most often to limit the number of simultaneous connections. We need to make five calls to a supplier, but the supplier picks up only two calls at a time.

import asyncio, time

start = time.perf_counter()
supplier = asyncio.Semaphore(2)        # the supplier takes two calls at a time, max

async def call(n):
    async with supplier:
        print(f"call #{n} at {time.perf_counter() - start:.1f}s")
        await asyncio.sleep(0.2)

async def main():
    await asyncio.gather(*(call(n) for n in range(1, 6)))

asyncio.run(main())
# call #1 at 0.0s
# call #2 at 0.0s
# call #3 at 0.2s
# call #4 at 0.2s
# call #5 at 0.4s

asyncio.Queue has the same shape as Lesson 6's ticket rail. maxsize, task_done() and join() are all there too.

import asyncio

async def hall(q):
    for n in range(1, 6):
        await q.put(f"#{n}")           # waits here when the rail is full
    print("hall done")

async def cook(name, q, done):
    while True:
        ticket = await q.get()
        await asyncio.sleep(0.1)
        done.append((name, ticket))
        q.task_done()

async def main():
    q = asyncio.Queue(maxsize=2)
    done = []
    cooks = [asyncio.create_task(cook(c, q, done)) for c in ("A", "B")]
    await hall(q)
    await q.join()                     # until every ticket clipped on has had task_done
    for c in cooks:
        c.cancel()
    print(len(done), sorted(t for _, t in done))

asyncio.run(main())
# hall done
# 5 ['#1', '#2', '#3', '#4', '#5']

The loop is a single thread, so you might think race conditions can't happen. But if there's an await between the read and the write, you get Lesson 3's chalkboard accident all over again.

import asyncio

async def main():
    board = {"count": 0}

    async def cook():
        for _ in range(100):
            v = board["count"]
            await asyncio.sleep(0)     # control passes to another task here
            board["count"] = v + 1

    await asyncio.gather(*(cook() for _ in range(4)))
    print(board["count"])

asyncio.run(main())
# 100

The chalkboard should say 400, but it says 100. All four tasks read the same value, yield, and then write the same value. Wrap that stretch in an asyncio.Lock, and have the cooks wait for the "Open!" signal with an asyncio.Event, like this:

import asyncio

async def main():
    lock = asyncio.Lock()
    opened = asyncio.Event()
    board = {"count": 0}

    async def cook():
        await opened.wait()            # wait for "Open!"
        for _ in range(100):
            async with lock:           # an await between read and write means you need a lock
                v = board["count"]
                await asyncio.sleep(0)
                board["count"] = v + 1

    tasks = [asyncio.create_task(cook()) for _ in range(4)]
    await asyncio.sleep(0.1)
    opened.set()
    await asyncio.gather(*tasks)
    print(board["count"])

asyncio.run(main())
# 400

Don't mix tools from the two worlds. If a coroutine waits on a threading.Lock, the whole loop freezes, and the task holding the lock never gets a chance to come back and release it.

import asyncio, threading, time

lock = threading.Lock()                # what if you use a thread lock in a coroutine?
start = time.perf_counter()

async def cook(n):
    if not lock.acquire(timeout=0.2):  # the whole loop freezes while this waits
        print(f"#{n} gave up at {time.perf_counter() - start:.1f}s")
        return
    try:
        await asyncio.sleep(0.1)       # yields while still holding the lock
        print(f"#{n} done")
    finally:
        lock.release()

async def main():
    await asyncio.gather(cook(1), cook(2))

asyncio.run(main())
# #2 gave up at 0.2s
# #1 done

Without the timeout, #2 would wait forever and #1 would never wake up. The reverse holds too: asyncio's tools are not thread-safe. Don't put things into an asyncio.Queue from another thread. We'll see how to pass work from a thread to the loop in 11-6.

11-5. async with, async for, async generators (7 min)

with and for have waiting versions too.

• async with: a resource that needs an await on the way in and on the way out (__aenter__, __aexit__). Locks, semaphores, network connections.
• async for: a loop where getting the next value means waiting (__aiter__, __anext__). Tickets arriving one at a time, streaming responses.
• Async generators: put yield inside an async def and you get an iterator you loop over with async for.

import asyncio
from contextlib import asynccontextmanager

async def tickets(n):                  # async generator: async def + yield
    for i in range(1, n + 1):
        await asyncio.sleep(0.05)      # wait until the next ticket arrives
        yield f"#{i}"

@asynccontextmanager
async def stove(name):                 # a burner you use with async with
    await asyncio.sleep(0.01)
    print(f"{name} on")
    try:
        yield name
    finally:
        print(f"{name} off")

async def main():
    async with stove("S1"):
        async for t in tickets(3):
            print("cook", t)
    print([t async for t in tickets(2)])

asyncio.run(main())
# S1 on
# cook #1
# cook #2
# cook #3
# S1 off
# ['#1', '#2']

contextlib.asynccontextmanager turns a single yield into a resource for async with. You can use async for in list comprehensions too (only inside a coroutine).

Lesson 7's as_completed has an asyncio version as well. From 3.13 you can loop over it with async for, and it then hands back the very tasks you passed in, in the order they finish.

import asyncio

async def dish(n, sec):
    await asyncio.sleep(sec)
    return f"#{n}"

async def main():
    tasks = [asyncio.create_task(dish(n, s)) for n, s in [(1, 0.3), (2, 0.1), (3, 0.2)]]
    async for t in asyncio.as_completed(tasks):      # from 3.13
        print(t in tasks, await t)

asyncio.run(main())
# True #2
# True #3
# True #1

You can still write it the old way, for f in asyncio.as_completed(tasks): await f, but then each f is a new object, not the original task. If you want to know which order finished, async for is the easier choice.

11-6. Throwing work in from outside the loop, and seeing the running tasks (4 min)

Say the staff member who answers reservation calls works in another thread. To slot work into Lou's schedule from that thread, use asyncio.run_coroutine_threadsafe(coroutine, loop). What you get back is Lesson 7's concurrent.futures.Future, so the thread can wait on it with result(timeout=).

import asyncio

async def seat(table):
    await asyncio.sleep(0.05)
    return f"{table} seated"

def phone_line(loop):                  # a thread outside the loop (reservation calls)
    fut = asyncio.run_coroutine_threadsafe(seat("T7"), loop)
    print(fut.result(timeout=1))       # concurrent.futures.Future

async def main():
    loop = asyncio.get_running_loop()
    await asyncio.to_thread(phone_line, loop)

asyncio.run(main())
# T7 seated

Eager tasks (3.12): normally create_task only puts a task on the schedule and doesn't run it until its turn comes. With asyncio.eager_task_factory, a task runs up to its first await the moment it's created. A coroutine that finishes without ever hitting an await never even goes on the schedule; it finishes right there.

import asyncio

async def greet(n):
    print(f"greet T{n}")               # up to the first await
    await asyncio.sleep(0)
    return n

async def main(eager):
    if eager:
        asyncio.get_running_loop().set_task_factory(asyncio.eager_task_factory)
    t = asyncio.create_task(greet(1))
    print("created, done =", t.done())
    await t

asyncio.run(main(False))
asyncio.run(main(True))
# created, done = False
# greet T1
# greet T1
# created, done = False

Seeing the running tasks (3.14): if you know the process ID (PID) of a running asyncio program, you can run python -m asyncio ps <PID> in another terminal to list its tasks, or python -m asyncio pstree <PID> to see who is waiting on whom as a tree. For example, if a hall task is waiting on T1 and T2 through a TaskGroup, you'd see this (some columns and paths left out):

task name   coroutine stack                                   awaiter name
Task-1      main
hall        TaskGroup._aexit -> TaskGroup.__aexit__ -> hall   Task-1
T1          sleep -> table                                    hall
T2          sleep -> table                                    hall

This is the tool for the moment when "the floor has frozen and I can't tell where Lou is standing." Give your tasks a name= and the list becomes much easier to read.

Closing time (10 min)

Only table 3's gratin goes into the old oven, and a thread stands guard in front of it. One switch (fix) lets us compare.

import asyncio, time

def old_oven(sec):                     # an old library we can't touch
    time.sleep(sec)
    return "gratin"

async def table(n, start, fix):
    if n == 3:
        dish = await asyncio.to_thread(old_oven, 0.6) if fix else old_oven(0.6)
    else:
        await asyncio.sleep(0.3)
        dish = "noodle"
    return f"T{n} {dish} {time.perf_counter() - start:.1f}s"

async def service(fix):
    start = time.perf_counter()
    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(table(n, start, fix)) for n in range(1, 7)]
    print([t.result() for t in tasks])

asyncio.run(service(fix=False))
asyncio.run(service(fix=True))
# ['T1 noodle 0.6s', 'T2 noodle 0.6s', 'T3 gratin 0.6s', 'T4 noodle 0.9s', 'T5 noodle 0.9s', 'T6 noodle 0.9s']
# ['T1 noodle 0.3s', 'T2 noodle 0.3s', 'T3 gratin 0.6s', 'T4 noodle 0.3s', 'T5 noodle 0.3s', 'T6 noodle 0.3s']

• When Lou stands at the oven herself, not one of the 0.3-second noodle dishes goes out at 0.3 seconds. T1 and T2 had their pots on before the oven started but are served only at 0.6 seconds, and T4 to T6 don't get their pots on until the oven finishes, going out at 0.9 seconds. Exactly the shape you saw on the timeline.
• Hand the oven off with to_thread and the noodles go out at 0.3 seconds and the gratin at the oven's 0.6 seconds. That's tonight's goal.

After closing, the chef runs a finger along the recipe cards on the kitchen wall, one by one: threads, locks, queues, pools, processes, the loop, TaskGroup, to_thread. "That's a lot of tools piled up. But when a guest opens the door and walks in, how do you decide which one to reach for first?"

Practice

1. In the timeline, change the slow table from 1 through 6. How do the bars for tables numbered before the slow table differ from those after it, and why?
2. Run the first example from 11-1 with asyncio.run(main(), debug=True) and check which task the warning says held the loop, and for how many seconds.
3. If you change the semaphore example in 11-4 to asyncio.Semaphore(3), at what time will call #5 start? Work it out first, then run it.