Skip to main content

This is a Python + CFFI wrapper around Liburing C library, which is a helper to setup and tear-down io_uring instances.

Project description

This is a Python + CFFI wrapper around Liburing C library, which is a helper to setup and tear-down io_uring instances.

Read how to use Liburing (pdf)

Good documentation Lord of the io_uring


  • Linux 5.1+ (5.11+ recommended)

  • Python 3.6+


  • 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+

To find out all the functions and definitions:

import liburing


Find out which io_uring operations is supported by the kernel:

import liburing

probe = liburing.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()
        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)

if __name__ == '__main__':


Free, Public Domain (CC0). Read more


  • create more test

  • Development Status :: 4 - Beta

  • create example

  • Development Status :: 5 - Production/Stable

Project details

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

liburing-2021.3.10.tar.gz (35.5 kB view hashes)

Uploaded source

Supported by

AWS AWS Cloud computing Datadog Datadog Monitoring Fastly Fastly CDN Google Google Object Storage and Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page