asyncio timeouts are cursed

timeouts.py

import asyncio
import time
import random
from collections.abc import Callable


def roll(dice: int = 1, sides: int = 6) -> int:
    total = 0
    for _ in range(dice):
        total += random.randint(1, sides)
    return total


async def call[**P, T](f: Callable[P, T], *args: P.args, **kwargs: P.kwargs) -> T:
    return f(*args, **kwargs)


def run_sync(num_dice: int) -> float:
    start = time.monotonic()
    roll(num_dice)
    elapsed = time.monotonic() - start
    return elapsed


async def main():
    num_dice = 10_000_000

    # Run the work sync to establish how long we expect the work to take, then
    # use half of that time as our timeout value.
    timeout = run_sync(num_dice) * 0.5

    # Now run the work "async", but give it half of the amount of time that the
    # work takes to do. What does the timeout mechanism do?
    start = time.monotonic()
    try:
        coro = call(roll, num_dice)
        await asyncio.wait_for(coro, timeout)
        elapsed = time.monotonic() - start
        print(
            f"ASYNC Succeeded after an elapsed time of {elapsed} using a timeout of {timeout}"
        )
    except asyncio.TimeoutError as e:
        elapsed = time.monotonic() - start
        print(f"Timed out after {elapsed} with error: {e}")


asyncio.run(main())

the output:

ASYNC Succeeded after an elapsed time of 1.7850587499560788 using a timeout of 0.8907289164781105