benchmarking trio.Queue

queue_bench.py

import timeit

import trio


class Stop(Exception):
    pass


async def benchmark1(qsize):
    async def queue_putter(q):
        for i in range(10000):
            await q.put(i)
        raise Stop

    async def queue_getter(q):
        async for _ in q:
            pass

    q = trio.Queue(qsize)
    try:
        async with trio.open_nursery() as nursery:
            nursery.start_soon(queue_getter, q)
            nursery.start_soon(queue_putter, q)
    except Stop:
        pass


async def benchmark2(qsize):
    q = trio.Queue(qsize)
    for i in range(qsize):
        q.put_nowait(i)


def run_benchmark(afn, *args):
    timer = timeit.Timer(
        f'trio.run(afn, *args)',
        globals={'trio': trio, 'afn': afn, 'args': args},
    )
    best = min(timer.repeat(repeat=3, number=10))
    print(f'{afn.__name__}{args}; loops: 10, best of 3: {best}')


if __name__ == '__main__':
    print("Benchmark 1 (1 putter task, 1 getter task)")
    try:
        trio.Queue(0)
    except Exception:
        pass
    else:
        run_benchmark(benchmark1, 0)
    run_benchmark(benchmark1, 1)
    run_benchmark(benchmark1, 10)
    run_benchmark(benchmark1, 100)

    print("Benchmark 2 (put_nowait only)")
    run_benchmark(benchmark2, 100)
    run_benchmark(benchmark2, 10000)
    run_benchmark(benchmark2, 100000)