liburing 2021.3.10

Requires

• Linux 5.1+ (5.11+ recommended)

• Python 3.6+

Includes

• liburing 2.0

Install, update & uninstall (Alpha)

Use pip to install, upgrade & uninstall Python wrapper:

python3 -m pip install --user liburing

python3 -m pip install --user --upgrade liburing

python3 -m pip uninstall liburing

Install directly from GitHub:

python3 -m pip install --user --upgrade git+https://github.com/YoSTEALTH/Liburing

To find out all the functions and definitions:

import liburing

help(liburing)

Find out which io_uring operations is supported by the kernel:

import liburing

probe = liburing.probe()
print(probe)

Simple File Example

import os
import os.path
from liburing import *


def open(ring, cqes, path, flags, mode=0o660, dir_fd=-1):
    # file `path` must be in bytes and as absolute path if no `dir_fd` is provided.
    _path = os.path.abspath(path).encode()

    sqe = io_uring_get_sqe(ring)  # sqe(submission queue entry)
    io_uring_prep_openat(sqe, dir_fd, _path, flags, mode)
    return _submit_and_wait(ring, cqes)  # returns fd


def write(ring, cqes, fd, data, offset=0):
    buffer = bytearray(data)
    iov = iovec(buffer)

    sqe = io_uring_get_sqe(ring)
    io_uring_prep_write(sqe, fd, iov[0].iov_base, iov[0].iov_len, offset)
    return _submit_and_wait(ring, cqes)  # returns length(s) of bytes written


def read(ring, cqes, fd, length, offset=0):
    buffer = bytearray(length)
    iov = iovec(buffer)

    sqe = io_uring_get_sqe(ring)
    io_uring_prep_read(sqe, fd, iov[0].iov_base, iov[0].iov_len, offset)
    read_length = _submit_and_wait(ring, cqes)  # get actual length of file read.
    return buffer[:read_length]


def close(ring, cqes, fd):
    sqe = io_uring_get_sqe(ring)
    io_uring_prep_close(sqe, fd)
    _submit_and_wait(ring, cqes)  # no error means success!


def _submit_and_wait(ring, cqes):
    io_uring_submit(ring)  # submit entry
    io_uring_wait_cqe(ring, cqes)  # wait for entry to finish
    cqe = cqes[0]  # cqe(completion queue entry)
    result = trap_error(cqe.res)  # auto raise appropriate exception if failed
    # note `cqe.res` returns results, if ``< 0`` its an error, if ``>= 0`` its the value

    # done with current entry so clear it from completion queue.
    io_uring_cqe_seen(ring, cqe)
    return result  # type: int


def main():
    ring = io_uring()
    cqes = io_uring_cqes()
    try:
        io_uring_queue_init(8, ring, 0)

        fd = open(ring, cqes, '/tmp/liburing-test-file.txt', os.O_CREAT | os.O_RDWR)
        print('fd:', fd)

        length = write(ring, cqes, fd, b'hello world')
        print('wrote:', length)

        content = read(ring, cqes, fd, length)
        print('read:', content)

        close(ring, cqes, fd)
        print('closed.')
    finally:
        io_uring_queue_exit(ring)


if __name__ == '__main__':
    main()

License

Free, Public Domain (CC0). Read more

TODO

• create more test

• Development Status :: 4 - Beta

• create example

• Development Status :: 5 - Production/Stable