overhead-of-python-asyncio.py

import asyncio
from time import process_time as time


async def anoop() -> None:
    pass


def noop() -> None:
    pass


async def create_task_await(count: int = 100) -> float:
    start = time()
    [await anoop() for _ in range(count)]
    return time() - start


async def create_task_async(count: int = 100) -> float:
    start = time()
    await asyncio.wait([asyncio.create_task(anoop()) for _ in range(count)])
    return time() - start


async def create_task_gather(count: int = 100) -> float:
    start = time()
    await asyncio.gather(*(anoop() for _ in range(count)))
    return time() - start


async def create_task_group(count: int = 100) -> float:
    start = time()
    async with asyncio.TaskGroup() as tg:
        [tg.create_task(anoop()) for _ in range(count)]
    return time() - start


def create_task_sync(count: int = 100) -> float:
    start = time()
    [noop() for _ in range(count)]
    return time() - start


def bench_sync() -> None:
    for count in range (100_000, 1_000_000 + 1, 100_000):
        elapsed = create_task_sync(count)
        try:
            rate = 1 / (elapsed / count)
        except ZeroDivisionError:
            rate = float("inf")
        print(f"{count:,} tasks\t{rate:0,.0f} tasks per/s")


async def bench_async() -> None:
    for count in range (100_000, 1_000_000 + 1, 100_000):
        elapsed = await create_task_async(count)
        try:
            rate = 1 / (elapsed / count)
        except ZeroDivisionError:
            rate = float("inf")
        print(f"{count:,} tasks\t{rate:0,.0f} tasks per/s")


async def bench_await() -> None:
    for count in range (100_000, 1_000_000 + 1, 100_000):
        elapsed = await create_task_await(count)
        try:
            rate = 1 / (elapsed / count)
        except ZeroDivisionError:
            rate = float("inf")
        print(f"{count:,} tasks\t{rate:0,.0f} tasks per/s")


async def bench_gather() -> None:
    for count in range (100_000, 1_000_000 + 1, 100_000):
        elapsed = await create_task_gather(count)
        try:
            rate = 1 / (elapsed / count)
        except ZeroDivisionError:
            rate = float("inf")
        print(f"{count:,} tasks\t{rate:0,.0f} tasks per/s")


async def bench_group() -> None:
    for count in range (100_000, 1_000_000 + 1, 100_000):
        elapsed = await create_task_group(count)
        try:
            rate = 1 / (elapsed / count)
        except ZeroDivisionError:
            rate = float("inf")
        print(f"{count:,} tasks\t{rate:0,.0f} tasks per/s")


def main() -> None:
    print("Synchronous")
    bench_sync()

    print()
    print("await")
    asyncio.get_event_loop().run_until_complete(bench_await())

    print()
    print("asyncio.create_task()")
    asyncio.get_event_loop().run_until_complete(bench_async())

    print()
    print("asyncio.gather()")
    asyncio.get_event_loop().run_until_complete(bench_gather())

    print()
    print("asyncio.TaskGroup.create_task()")
    asyncio.get_event_loop().run_until_complete(bench_group())


if __name__ == "__main__":
    main()