lesson 192 Async Python · advanced

Tasks: running coroutines concurrently

A task schedules a coroutine to run while you do something else. Start several tasks and their waits overlap: three downloads take as long as the slowest, not the sum.

You'll be able to

Start tasks with asyncio.create_task and wait for them with await or asyncio.gather
Use asyncio.TaskGroup so that a failure cancels the group's other tasks
Collect errors with gather(..., return_exceptions=True) instead of losing results

The idea

await coro() runs one coroutine to the end before continuing. To overlap independent work, wrap each coroutine in a task: asyncio.create_task(coro()) schedules it on the loop right away and returns a Task you can await later. While one task waits, the loop runs the others. await asyncio.gather(a, b, c) waits for several coroutines or tasks and returns their results in the order given, whatever order they finished in.

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

example-1.py · Three downloads at once

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 download(name, seconds):
    await asyncio.sleep(seconds)
    return name

async def main():
    loop = asyncio.get_running_loop()
    tasks = [asyncio.create_task(download(n, s)) for n, s in [("orders.csv", 0.3), ("customers.csv", 0.1), ("products.csv", 0.2)]]
    print(await asyncio.gather(*tasks), round(loop.time(), 2))
    async with asyncio.TaskGroup() as group:
        a = group.create_task(download("reviews.csv", 0.25))
        b = group.create_task(download("sellers.csv", 0.05))
    print(a.result(), b.result(), round(loop.time(), 2))

run(main())

Errors need a decision. By default gather raises the first exception and you lose the other results (those tasks keep running unless you cancel them). With return_exceptions=True it returns the exception objects in place of results, so you can report each failure. A TaskGroup takes the strict view: if one task fails, it cancels the rest and raises an ExceptionGroup.

example-2.py · One export fails

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 export(name):
    await asyncio.sleep(0.1)
    if name == "payments.csv":
        raise ConnectionError("reset by peer")
    return name

async def main():
    results = await asyncio.gather(export("orders.csv"), export("payments.csv"), return_exceptions=True)
    for r in results:
        print(type(r).__name__, r)
    try:
        await asyncio.gather(export("orders.csv"), export("payments.csv"))
    except ConnectionError as e:
        print("gather raised", e)

run(main())

When to use what: gather for a known list of independent awaits; TaskGroup when the tasks belong together and one failure should stop the rest; create_task alone for background work you await later. Keep a reference to every task you create (the loop holds only a weak one, so an unreferenced task can disappear). Mistakes to avoid: awaiting each coroutine in a loop (serial again); creating tasks and never awaiting them (errors vanish); and expecting results in completion order from gather (they are in argument order).

Review

asyncio.create_task(coro()) schedules a coroutine; gather and TaskGroup wait for several.
Independent waits overlap: total time ≈ the slowest, not the sum.
gather(..., return_exceptions=True) keeps every outcome; a TaskGroup cancels the rest on failure.
Next: stopping tasks deliberately, cancellation and timeouts.

Practise this