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)

Another good documentation Lord of the io_uring


  • Linux 5.5+ (5.7+ recommended)
  • Python 3.6+


  • liburing 0.7.0

Install, update & uninstall (Alpha)

Use pip to install, upgrade & uninstall Python wrapper:

pip install liburing

pip install --upgrade liburing

pip uninstall liburing

Install directly from GitHub:

pip install --upgrade git+

To find out all the functions and definitions:

import liburing


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 absolute path and in bytes.
    _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.

Files for liburing, version 2020.7.13
Filename, size File type Python version Upload date Hashes
Filename, size liburing-2020.7.13.tar.gz (31.2 kB) File type Source Python version None Upload date Hashes View

Supported by

Pingdom Pingdom Monitoring Google Google Object Storage and Download Analytics Sentry Sentry Error logging AWS AWS Cloud computing DataDog DataDog Monitoring Fastly Fastly CDN DigiCert DigiCert EV certificate StatusPage StatusPage Status page