Tasks and the Event Loop

gather starts a batch of coroutines and waits for all of them. That’s often exactly what you want. But sometimes you want to start something now and check on it later, or deal with each result the moment it arrives. For that, you need tasks.

Tasks: started now, awaited later

asyncio.create_task(coroutine) hands a coroutine to the event loop, which starts running it as soon as the current coroutine next pauses. You get back a Task, which you can await later for the result:

import asyncio
import time


async def fetch_log(server, delay):
    print(f"  {time.perf_counter() - START:.1f}s  {server}: started")
    await asyncio.sleep(delay)
    print(f"  {time.perf_counter() - START:.1f}s  {server}: done")
    return f"{server} log"


async def main():
    auth = asyncio.create_task(fetch_log("auth", 0.3))       # both start running...
    search = asyncio.create_task(fetch_log("search", 0.1))
    print(f"  {time.perf_counter() - START:.1f}s  main: tasks created, doing other things")
    await asyncio.sleep(0.2)
    print(f"  {time.perf_counter() - START:.1f}s  main: now I need the results")
    print(await auth, "and", await search)                   # ...and are awaited later


START = time.perf_counter()
asyncio.run(main())

Both fetches began when main reached its first await, and they kept running while main did other things. When main finally awaited search, it had already finished, so the result came back immediately.

gather is built on this. It wraps each coroutine in a task, then waits for them all.

Results as they arrive

gather hands back everything at the end, in the order you asked. When you’d rather act on each result as soon as it’s ready, for example to show progress, use asyncio.as_completed:

import asyncio

DELAYS = {"auth": 0.3, "payments": 0.2, "search": 0.1}


async def fetch_log(server):
    await asyncio.sleep(DELAYS[server])
    return server, [f"{server}: ok"] * 3


async def main():
    for next_done in asyncio.as_completed([fetch_log(s) for s in DELAYS]):
        server, lines = await next_done
        print(f"{server} arrived with {len(lines)} lines")


asyncio.run(main())

The loop runs once per coroutine, in the order they finish. The coroutines don’t come back out of as_completed in order, so fetch_log returns the server’s name alongside its lines. Otherwise you couldn’t tell which result is whose.

Structured concurrency: TaskGroup

A task you create and then forget keeps running in the background, and if it fails, nobody hears about it. Since Python 3.11, asyncio.TaskGroup makes the lifetime of tasks explicit:

async with asyncio.TaskGroup() as group:
    auth = group.create_task(fetch_log("auth"))
    search = group.create_task(fetch_log("search"))
# Here, every task in the group has finished.
print(auth.result(), search.result())

The async with block doesn’t end until every task in it has finished, so nothing can outlive the code that started it. It’s the async version of Day 4’s lesson: resources, tasks included, should have a clear end. Lesson 3 covers what a TaskGroup does when one of its tasks fails.

Bug hunt: the timeout that never fired

A service fetches data and then does some heavy work on it. The caller allows it 0.3 seconds. A heartbeat task ticks every 0.1 seconds, the way a progress display would. The heavy work is even written as an async def:

import asyncio
import time


async def crunch(n):
    """Stand-in for slow, ordinary work: parsing a huge log, compressing a file..."""
    return sum(i * i for i in range(n))    # an async def, without a single await


async def fetch_and_crunch(n):
    await asyncio.sleep(0.05)          # fetching: waiting on the network
    return await crunch(n)             # crunching: pure work, no waiting


async def heartbeat(beats):
    """Tick every 0.1s, the way a progress display would."""
    while True:
        beats.append(round(time.perf_counter() - START, 1))
        await asyncio.sleep(0.1)


async def main():
    beats = []
    ticker = asyncio.create_task(heartbeat(beats))
    try:
        async with asyncio.timeout(0.3):
            result = await fetch_and_crunch(15_000_000)
        print("finished:", result)
    except TimeoutError:
        print("gave up at the 0.3s timeout")
    ticker.cancel()
    print(f"took {time.perf_counter() - START:.1f}s")
    print("heartbeats at", beats)


START = time.perf_counter()
asyncio.run(main())

This playground is an experiment, and its timings are observations. The same amount of work takes different times on different computers. If yours finishes in under 0.3 seconds, add a zero to 15_000_000 and run it again.

Reproduce. It took well over 0.3 seconds, and it finished rather than timing out. The limit was simply ignored.

Inspect. The heartbeat is the state to look at. It should have ticked every 0.1 seconds throughout. It ticked once, at the very start, and then not again until the work was over.

Before you open this, explain why neither the heartbeat nor the timeout could run, even though crunch is an async def. Then try a fix. Read the reveal either way, even if your fix works: it introduces await asyncio.sleep(0), and a way to test that a loop stays responsive without timing anything.

def test_other_tasks_run_while_crunching(self):
    turns = 0

    async def ticker():
        nonlocal turns
        while True:
            turns += 1
            await asyncio.sleep(0)

    async def attempt():
        ticking = asyncio.create_task(ticker())
        await asyncio.sleep(0)                    # let the ticker start
        before = turns
        await crunch(1_000_000)
        ticking.cancel()
        return turns - before

    self.assertGreater(asyncio.run(attempt()), 0)

async def crunch(n, chunk=20_000):
    """Slow, ordinary work, done in chunks so other tasks get a turn between them."""
    total = 0
    for start in range(0, n, chunk):
        total += sum(i * i for i in range(start, min(start + chunk, n)))
        await asyncio.sleep(0)         # a chance for timers and other tasks to run
    return total

Your turn

import asyncio
import sys
import unittest

DELAYS = {"auth": 0.3, "payments": 0.2, "search": 0.1}


async def fetch_log(server):
    await asyncio.sleep(DELAYS[server])
    return server, [f"{server}: line {i}" for i in range(3)]


async def fetch_all(servers, on_arrival):
    raise NotImplementedError("your code here")


class TestFetchAll(unittest.TestCase):
    def test_progress_is_reported_as_logs_arrive(self):
        arrived = []
        asyncio.run(fetch_all(list(DELAYS), arrived.append))
        self.assertEqual(arrived, ["search", "payments", "auth"])

    def test_results_come_back_in_the_order_asked(self):
        logs = asyncio.run(fetch_all(["auth", "search"], lambda server: None))
        self.assertEqual(list(logs), ["auth", "search"])
        self.assertEqual(logs["search"][0], "search: line 0")


suite = unittest.defaultTestLoader.loadTestsFromTestCase(TestFetchAll)
unittest.TextTestRunner(stream=sys.stdout, verbosity=2).run(suite)

async def fetch_all(servers, on_arrival):
    logs = {}
    for next_done in asyncio.as_completed([fetch_log(server) for server in servers]):
        server, lines = await next_done
        on_arrival(server)
        logs[server] = lines
    return {server: logs[server] for server in servers}

What you can now do

• Start work with create_task and collect it later.
• Handle results in the order they finish with as_completed.
• Keep tasks’ lifetimes explicit with TaskGroup.
• Recognise a coroutine that blocks the loop, and give control back with await, or move blocking calls off the loop on your own machine.

Next: timeouts on purpose, and what to do when one source out of three fails.