aiohttp.StreamReader benchmark

aiostreamsbench.py

import time
import functools
import asyncio
from aiohttp import StreamReader


def make_async(fn):
    @functools.wraps(fn)
    def async(*args, **kwargs):
        f = asyncio.futures.Future()
        f.set_result(fn(*args, **kwargs))
        return f
    return async


TIMES = 20


def timeit(fn, n=TIMES):
    @functools.wraps(fn)
    def wrapper(*args, **kwargs):
        results = []
        for _ in range(n):
            start = time.time()
            yield from asyncio.tasks.async(fn(*args, **kwargs))
            results.append(time.time() - start)
        print('>>>', fn.__name__, sorted(results)[n // 2])

    return wrapper


chunk_size = 8000
test_bin = b'1234567890' * (chunk_size // 10)
test_bin_times = 10000
large_test_bin = test_bin * test_bin_times


def get_reader(chunked):
    reader = StreamReader()
    if chunked:
        for _ in range(test_bin_times):
            reader.feed_data(test_bin)
    else:
        reader.feed_data(large_test_bin)
    reader.feed_eof()
    return reader

# ------------------------


@timeit
@make_async
def test_stream_feed(chunked):
    get_reader(chunked)


@timeit
def test_stream_read_all(chunked):
    reader = get_reader(chunked)
    yield from reader.read()


@timeit
def test_stream_read_less(chunked):
    reader = get_reader(chunked)
    while (yield from reader.read(chunk_size // 2)):
        pass


@timeit
def test_stream_read_any(chunked):
    reader = get_reader(chunked)
    while (yield from reader.readany()):
        pass


@timeit
def test_stream_read_exactly(chunked):
    reader = get_reader(chunked)
    while not reader.at_eof():
        yield from reader.readexactly(chunk_size * 4)


if __name__ == '__main__':
    loop = asyncio.get_event_loop()

    loop.run_until_complete(test_stream_feed(False))
    loop.run_until_complete(test_stream_feed(True))
    loop.run_until_complete(test_stream_read_all(False))
    loop.run_until_complete(test_stream_read_all(True))
    loop.run_until_complete(test_stream_read_less(False))
    loop.run_until_complete(test_stream_read_less(True))
    loop.run_until_complete(test_stream_read_any(False))
    loop.run_until_complete(test_stream_read_any(True))
    loop.run_until_complete(test_stream_read_exactly(False))
    loop.run_until_complete(test_stream_read_exactly(True))
    time.sleep(5)