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
Requires
- Linux 5.5+ (5.7+ recommended)
- Python 3.6+
Includes
- 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+https://github.com/YoSTEALTH/Liburing
To find out all the functions and definitions:
import liburing help(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() 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
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
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 |