lesson 193 Async Python · advanced

Cancellation and timeouts

A slow request shouldn't hang a program forever. Cancelling a task raises CancelledError inside it at its next await; timeouts cancel for you; finally blocks clean up either way.

You'll be able to

Cancel a task with task.cancel() and explain where CancelledError appears
Limit an await with asyncio.timeout or asyncio.wait_for and handle TimeoutError
Release resources in finally, and never swallow CancelledError

The idea

task.cancel() asks a task to stop: at the task's next await, asyncio raises `asyncio.CancelledError` inside it. The task can clean up (finally, or except CancelledError: followed by raise), and then it ends as *cancelled*. Awaiting a cancelled task raises CancelledError in the awaiting code too.

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

example-1.py · Stopping a price feed

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 stream_prices(log):
    try:
        while True:
            await asyncio.sleep(1)
            log.append("tick")
    except asyncio.CancelledError:
        log.append("cancelled")
        raise
    finally:
        log.append("closed feed")

async def main():
    loop = asyncio.get_running_loop()
    log = []
    task = asyncio.create_task(stream_prices(log))
    await asyncio.sleep(2.5)
    task.cancel()
    try:
        await task
    except asyncio.CancelledError:
        pass
    print(log, task.cancelled(), loop.time())

run(main())

Usually you don't cancel by hand; you set a time limit. async with asyncio.timeout(2): (Python 3.11+) cancels whatever is awaited inside the block after 2 seconds and turns that into `TimeoutError` at the async with. await asyncio.wait_for(coro, timeout=10) does the same for one awaitable. And asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED) returns as soon as one task is done, with the finished and still-pending tasks as two sets, so you can cancel the rest (with timeout= instead, it returns whatever finished in time).

example-2.py · Giving up, and taking the first answer

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 slow_report():
    await asyncio.sleep(5)
    return "report"

async def download(region, seconds):
    await asyncio.sleep(seconds)
    return region

async def main():
    loop = asyncio.get_running_loop()
    try:
        async with asyncio.timeout(2):
            await slow_report()
    except TimeoutError:
        print("gave up at", loop.time())
    print(await asyncio.wait_for(slow_report(), timeout=10), loop.time())
    tasks = {asyncio.create_task(download(r, s)) for r, s in [("eu", 0.4), ("us", 0.1), ("asia", 0.3)]}
    done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
    print([t.result() for t in done], len(pending))
    for t in pending:
        t.cancel()

run(main())

When to use it: every network call needs a time limit; when the first of several answers is enough, cancel the rest; when the user closes a page, cancel the work it started. Mistakes to avoid: catching CancelledError and not re-raising it (the task looks finished but cancellation never worked); except BaseException: or a bare except: around awaits, which swallow it too; cleanup that isn't in finally; and forgetting to cancel tasks you no longer need.

Review

task.cancel() raises CancelledError inside the task at its next await; awaiting it raises it again.
asyncio.timeout(s) and wait_for(..., timeout=s) cancel slow work and raise TimeoutError; asyncio.wait can stop at the first result or a deadline.
Clean up in finally; never swallow CancelledError.
Next: limiting how many tasks run at once, with semaphores.

Practise this