Linux AIO comparison in Python

experiments.py

#!/usr/bin/env python3.7
# coding: UTF-8

import asyncio
import sys
from time import sleep

import aiofiles
import linux_aio
from aiofile import AIOFile
from libaio import AIOBLOCK_MODE_READ, AIOBlock, AIOContext
from pathlib import Path
from typing import Dict, TextIO, Tuple

TARGETS = (
    Path('/home/bhyoo/.bashrc'),
    Path('/home/bhyoo/.zcompdump'),
    Path('/home/bhyoo/.zshrc'),
    Path('/home/bhyoo/.gitconfig'),
    Path('/home/bhyoo/.viminfo'),
    Path('/home/bhyoo/.zsh_history'),
)

INTERVAL = 1
TOT_NUM = 1000


def bench_linux_aio():
    targets: Dict[Path, Tuple[TextIO, bytearray]] = dict()

    for path in TARGETS:
        targets[path] = (path.open(), bytearray(64))

    with linux_aio.AIOContext(64) as ctx:
        aio_blocks: Tuple[linux_aio.AIOBlock, ...] = tuple(
                linux_aio.AIOBlock(fp.fileno(), linux_aio.IOCBCMD.PREAD, buf)
                for fp, buf in targets.values()
        )

        for _ in range(TOT_NUM):
            ctx.submit(*aio_blocks)
            sleep(INTERVAL / 1000)
            for event in ctx.get_events(len(targets), len(targets)):
                pass

        for fd, buf in targets.values():
            fd.close()


def bench_libaio():
    targets: Dict[Path, Tuple[TextIO, bytearray]] = dict()

    for path in TARGETS:
        targets[path] = (path.open(), bytearray(64))

    with AIOContext(64) as aio_ctx:
        aio_blocks: Tuple[AIOBlock, ...] = tuple(
                AIOBlock(AIOBLOCK_MODE_READ, fp, (buf,))
                for fp, buf in targets.values()
        )

        for _ in range(TOT_NUM):
            aio_ctx.submit(aio_blocks)
            sleep(INTERVAL / 1000)
            for a, b, c in aio_ctx.getEvents(len(targets), len(targets)):  # type: AIOBlock, int, int
                pass

        for fd, buf in targets.values():
            fd.close()

        aio_ctx.close()


async def bench_aiofiles():
    targets = await asyncio.gather(*(aiofiles.open(path) for path in TARGETS))

    for _ in range(TOT_NUM):
        for target in targets:
            ret = await target.read(64)
            await target.seek(0)
        await asyncio.sleep(INTERVAL / 1000)

    for target in targets:
        target.close()


async def bench_aiofile():
    targets: Tuple[AIOFile, ...] = await asyncio.gather(*(AIOFile(path) for path in TARGETS))

    for _ in range(TOT_NUM):
        for target in targets:
            ret = await target.read(64, 0)
        await asyncio.sleep(INTERVAL / 1000)

    for target in targets:
        await target.close()


def bench_open():
    targets: Tuple[TextIO, ...] = tuple(open(path) for path in TARGETS)

    for _ in range(TOT_NUM):
        for target in targets:
            ret = target.read(64)
            target.seek(0)
        sleep(INTERVAL / 1000)

    for target in targets:
        target.close()


if __name__ == '__main__':
    if len(sys.argv) is not 2:
        print('choose libaio, aiofiles, aiofile, open, linux_aio', file=sys.stderr)
        exit(1)

    lib = sys.argv[1]

    if lib == 'libaio':
        bench_libaio()
    elif lib == 'aiofiles':
        asyncio.run(bench_aiofiles())
    elif lib == 'aiofile':
        asyncio.run(bench_aiofile())
    elif lib == 'open':
        bench_open()
    elif lib == 'linux_aio':
        bench_linux_aio()
    else:
        print('choose libaio, aiofiles, aiofile, open, linux_aio', file=sys.stderr)