feature. See also
. The project being documented here (as the example) is the Zig library itself.
File
Code
const IoUring = @This();
const builtin = @import("builtin");
const is_linux = builtin.os.tag == .linux;
const std = @import("../../std.zig");
const Io = std.Io;
const Allocator = std.mem.Allocator;
const assert = std.debug.assert;
const posix = std.posix;
const linux = std.os.linux;
const testing = std.testing;
const page_size_min = std.heap.page_size_min;
const createSocketTestHarness = @import("IoUring/test.zig").createSocketTestHarness;
fd: linux.fd_t = -1,
sq: SubmissionQueue,
cq: CompletionQueue,
flags: u32,
features: u32,
pub fn init(entries: u16, flags: u32) !IoUring {
var params = std.mem.zeroInit(linux.io_uring_params, .{
.flags = flags,
.sq_thread_idle = 1000,
});
return try IoUring.init_params(entries, ¶ms);
}
pub fn init_params(entries: u16, p: *linux.io_uring_params) !IoUring {
if (entries == 0) return error.EntriesZero;
if (!std.math.isPowerOfTwo(entries)) return error.EntriesNotPowerOfTwo;
assert(p.sq_entries == 0);
assert(p.cq_entries == 0 or p.flags & linux.IORING_SETUP_CQSIZE != 0);
assert(p.features == 0);
assert(p.wq_fd == 0 or p.flags & linux.IORING_SETUP_ATTACH_WQ != 0);
assert(p.resv[0] == 0);
assert(p.resv[1] == 0);
assert(p.resv[2] == 0);
const res = linux.io_uring_setup(entries, p);
switch (linux.errno(res)) {
.SUCCESS => {},
.FAULT => return error.ParamsOutsideAccessibleAddressSpace,
// entries out of bounds, IORING_SETUP_SQ_AFF was specified without IORING_SETUP_SQPOLL,
// or IORING_SETUP_CQSIZE was specified but linux.io_uring_params.cq_entries was invalid:
.INVAL => return error.ArgumentsInvalid,
.MFILE => return error.ProcessFdQuotaExceeded,
.NFILE => return error.SystemFdQuotaExceeded,
.NOMEM => return error.SystemResources,
// or a container seccomp policy prohibits io_uring syscalls:
.PERM => return error.PermissionDenied,
.NOSYS => return error.SystemOutdated,
else => |errno| return posix.unexpectedErrno(errno),
}
const fd = @as(linux.fd_t, @intCast(res));
assert(fd >= 0);
errdefer _ = linux.close(fd);
// This is not an optional feature for us... if the kernel does it, we have to do it.
// The thinking on this by the kernel developers was that both the submission and the
// completion queue rings have sizes just over a power of two, but the submission queue ring
// is significantly smaller with u32 slots. By bundling both in a single mmap, the kernel
// gets the submission queue ring for free.
// See https://patchwork.kernel.org/patch/11115257 for the kernel patch.
// We do not support the double mmap() done before 5.4, because we want to keep the
// init/deinit mmap paths simple and because io_uring has had many bug fixes even since 5.4.
if ((p.features & linux.IORING_FEAT_SINGLE_MMAP) == 0) {
return error.SystemOutdated;
}
assert(p.sq_entries != 0);
assert(p.cq_entries != 0);
assert(p.cq_entries >= p.sq_entries);
// The completion queue shares the mmap with the submission queue, so pass `sq` there too.
var sq = try SubmissionQueue.init(fd, p.*);
errdefer sq.deinit();
var cq = try CompletionQueue.init(fd, p.*, sq);
errdefer cq.deinit();
assert(sq.head.* == 0);
assert(sq.tail.* == 0);
assert(sq.mask == p.sq_entries - 1);
assert(sq.dropped.* == 0);
assert(sq.array.len == p.sq_entries);
assert(sq.sqes.len == p.sq_entries);
assert(sq.sqe_head == 0);
assert(sq.sqe_tail == 0);
assert(cq.head.* == 0);
assert(cq.tail.* == 0);
assert(cq.mask == p.cq_entries - 1);
assert(cq.overflow.* == 0);
assert(cq.cqes.len == p.cq_entries);
return IoUring{
.fd = fd,
.sq = sq,
.cq = cq,
.flags = p.flags,
.features = p.features,
};
}
pub fn deinit(self: *IoUring) void {
assert(self.fd >= 0);
self.cq.deinit();
self.sq.deinit();
_ = linux.close(self.fd);
self.fd = -1;
}
pub fn get_sqe(self: *IoUring) !*linux.io_uring_sqe {
const head = @atomicLoad(u32, self.sq.head, .acquire);
// We must therefore use wrapping addition and subtraction to avoid a runtime crash.
const next = self.sq.sqe_tail +% 1;
if (next -% head > self.sq.sqes.len) return error.SubmissionQueueFull;
const sqe = &self.sq.sqes[self.sq.sqe_tail & self.sq.mask];
self.sq.sqe_tail = next;
return sqe;
}
pub fn submit(self: *IoUring) !u32 {
return self.submit_and_wait(0);
}
pub fn submit_and_wait(self: *IoUring, wait_nr: u32) !u32 {
const submitted = self.flush_sq();
var flags: u32 = 0;
if (self.sq_ring_needs_enter(&flags) or wait_nr > 0) {
if (wait_nr > 0 or (self.flags & linux.IORING_SETUP_IOPOLL) != 0) {
flags |= linux.IORING_ENTER_GETEVENTS;
}
return try self.enter(submitted, wait_nr, flags);
}
return submitted;
}
pub fn enter(self: *IoUring, to_submit: u32, min_complete: u32, flags: u32) !u32 {
assert(self.fd >= 0);
const res = linux.io_uring_enter(self.fd, to_submit, min_complete, flags, null);
switch (linux.errno(res)) {
.SUCCESS => {},
// The application should wait for some completions and try again:
.AGAIN => return error.SystemResources,
.BADF => return error.FileDescriptorInvalid,
// See io_uring_register(2) for how to enable the ring.
.BADFD => return error.FileDescriptorInBadState,
// The application should wait for some completions and try again:
.BUSY => return error.CompletionQueueOvercommitted,
.INVAL => return error.SubmissionQueueEntryInvalid,
// or IORING_OP_WRITE_FIXED was specified but no buffers were registered, or the range
// described by `addr` and `len` is not within the buffer registered at `buf_index`:
.FAULT => return error.BufferInvalid,
.NXIO => return error.RingShuttingDown,
// or the opcode is valid but not supported by this kernel (more likely):
.OPNOTSUPP => return error.OpcodeNotSupported,
// and IORING_SETUP_DEFER_TASKRUN is set, but the submitting thread is not the thread
// that initially created or enabled the io_uring associated with fd.
.EXIST => return error.InvalidThread,
// This can happen while waiting for events with IORING_ENTER_GETEVENTS:
.INTR => return error.SignalInterrupt,
else => |errno| return posix.unexpectedErrno(errno),
}
return @as(u32, @intCast(res));
}
pub fn flush_sq(self: *IoUring) u32 {
if (self.sq.sqe_head != self.sq.sqe_tail) {
const to_submit = self.sq.sqe_tail -% self.sq.sqe_head;
var tail = self.sq.tail.*;
var i: usize = 0;
while (i < to_submit) : (i += 1) {
self.sq.array[tail & self.sq.mask] = self.sq.sqe_head & self.sq.mask;
tail +%= 1;
self.sq.sqe_head +%= 1;
}
@atomicStore(u32, self.sq.tail, tail, .release);
}
return self.sq_ready();
}
pub fn sq_ring_needs_enter(self: *IoUring, flags: *u32) bool {
assert(flags.* == 0);
if ((self.flags & linux.IORING_SETUP_SQPOLL) == 0) return true;
if ((@atomicLoad(u32, self.sq.flags, .unordered) & linux.IORING_SQ_NEED_WAKEUP) != 0) {
flags.* |= linux.IORING_ENTER_SQ_WAKEUP;
return true;
}
return false;
}
pub fn sq_ready(self: *IoUring) u32 {
// see https://github.com/axboe/liburing/issues/92.
return self.sq.sqe_tail -% @atomicLoad(u32, self.sq.head, .acquire);
}
pub fn cq_ready(self: *IoUring) u32 {
return @atomicLoad(u32, self.cq.tail, .acquire) -% self.cq.head.*;
}
pub fn copy_cqes(self: *IoUring, cqes: []linux.io_uring_cqe, wait_nr: u32) !u32 {
const count = self.copy_cqes_ready(cqes);
if (count > 0) return count;
if (self.cq_ring_needs_flush() or wait_nr > 0) {
_ = try self.enter(0, wait_nr, linux.IORING_ENTER_GETEVENTS);
return self.copy_cqes_ready(cqes);
}
return 0;
}
fn copy_cqes_ready(self: *IoUring, cqes: []linux.io_uring_cqe) u32 {
const ready = self.cq_ready();
const count = @min(cqes.len, ready);
const head = self.cq.head.* & self.cq.mask;
const n = @min(self.cq.cqes.len - head, count);
@memcpy(cqes[0..n], self.cq.cqes[head..][0..n]);
if (count > n) {
const w = count - n;
@memcpy(cqes[n..][0..w], self.cq.cqes[0..w]);
}
self.cq_advance(count);
return count;
}
pub fn copy_cqe(ring: *IoUring) !linux.io_uring_cqe {
var cqes: [1]linux.io_uring_cqe = undefined;
while (true) {
const count = try ring.copy_cqes(&cqes, 1);
if (count > 0) return cqes[0];
}
}
pub fn cq_ring_needs_flush(self: *IoUring) bool {
return (@atomicLoad(u32, self.sq.flags, .unordered) & linux.IORING_SQ_CQ_OVERFLOW) != 0;
}
pub fn cqe_seen(self: *IoUring, cqe: *linux.io_uring_cqe) void {
_ = cqe;
self.cq_advance(1);
}
pub fn cq_advance(self: *IoUring, count: u32) void {
if (count > 0) {
@atomicStore(u32, self.cq.head, self.cq.head.* +% count, .release);
}
}
pub fn fsync(self: *IoUring, user_data: u64, fd: linux.fd_t, flags: u32) !*linux.io_uring_sqe {
const sqe = try self.get_sqe();
sqe.prep_fsync(fd, flags);
sqe.user_data = user_data;
return sqe;
}
pub fn nop(self: *IoUring, user_data: u64) !*linux.io_uring_sqe {
const sqe = try self.get_sqe();
sqe.prep_nop();
sqe.user_data = user_data;
return sqe;
}
pub const ReadBuffer = union(enum) {
buffer: []u8,
iovecs: []const posix.iovec,
buffer_selection: struct {
group_id: u16,
len: usize,
},
};
pub fn read(
self: *IoUring,
user_data: u64,
fd: linux.fd_t,
buffer: ReadBuffer,
offset: u64,
) !*linux.io_uring_sqe {
const sqe = try self.get_sqe();
switch (buffer) {
.buffer => |slice| sqe.prep_read(fd, slice, offset),
.iovecs => |vecs| sqe.prep_readv(fd, vecs, offset),
.buffer_selection => |selection| {
sqe.prep_rw(.READ, fd, 0, selection.len, offset);
sqe.flags |= linux.IOSQE_BUFFER_SELECT;
sqe.buf_index = selection.group_id;
},
}
sqe.user_data = user_data;
return sqe;
}
pub fn write(
self: *IoUring,
user_data: u64,
fd: linux.fd_t,
buffer: []const u8,
offset: u64,
) !*linux.io_uring_sqe {
const sqe = try self.get_sqe();
sqe.prep_write(fd, buffer, offset);
sqe.user_data = user_data;
return sqe;
}
pub fn splice(self: *IoUring, user_data: u64, fd_in: linux.fd_t, off_in: u64, fd_out: linux.fd_t, off_out: u64, len: usize) !*linux.io_uring_sqe {
const sqe = try self.get_sqe();
sqe.prep_splice(fd_in, off_in, fd_out, off_out, len);
sqe.user_data = user_data;
return sqe;
}
pub fn read_fixed(
self: *IoUring,
user_data: u64,
fd: linux.fd_t,
buffer: *posix.iovec,
offset: u64,
buffer_index: u16,
) !*linux.io_uring_sqe {
const sqe = try self.get_sqe();
sqe.prep_read_fixed(fd, buffer, offset, buffer_index);
sqe.user_data = user_data;
return sqe;
}
pub fn writev(
self: *IoUring,
user_data: u64,
fd: linux.fd_t,
iovecs: []const posix.iovec_const,
offset: u64,
) !*linux.io_uring_sqe {
const sqe = try self.get_sqe();
sqe.prep_writev(fd, iovecs, offset);
sqe.user_data = user_data;
return sqe;
}
pub fn write_fixed(
self: *IoUring,
user_data: u64,
fd: linux.fd_t,
buffer: *posix.iovec,
offset: u64,
buffer_index: u16,
) !*linux.io_uring_sqe {
const sqe = try self.get_sqe();
sqe.prep_write_fixed(fd, buffer, offset, buffer_index);
sqe.user_data = user_data;
return sqe;
}
pub fn accept(
self: *IoUring,
user_data: u64,
fd: linux.fd_t,
addr: ?*posix.sockaddr,
addrlen: ?*posix.socklen_t,
flags: u32,
) !*linux.io_uring_sqe {
const sqe = try self.get_sqe();
sqe.prep_accept(fd, addr, addrlen, flags);
sqe.user_data = user_data;
return sqe;
}
pub fn accept_multishot(
self: *IoUring,
user_data: u64,
fd: linux.fd_t,
addr: ?*posix.sockaddr,
addrlen: ?*posix.socklen_t,
flags: u32,
) !*linux.io_uring_sqe {
const sqe = try self.get_sqe();
sqe.prep_multishot_accept(fd, addr, addrlen, flags);
sqe.user_data = user_data;
return sqe;
}
pub fn accept_direct(
self: *IoUring,
user_data: u64,
fd: linux.fd_t,
addr: ?*posix.sockaddr,
addrlen: ?*posix.socklen_t,
flags: u32,
) !*linux.io_uring_sqe {
const sqe = try self.get_sqe();
sqe.prep_accept_direct(fd, addr, addrlen, flags, linux.IORING_FILE_INDEX_ALLOC);
sqe.user_data = user_data;
return sqe;
}
pub fn accept_multishot_direct(
self: *IoUring,
user_data: u64,
fd: linux.fd_t,
addr: ?*posix.sockaddr,
addrlen: ?*posix.socklen_t,
flags: u32,
) !*linux.io_uring_sqe {
const sqe = try self.get_sqe();
sqe.prep_multishot_accept_direct(fd, addr, addrlen, flags);
sqe.user_data = user_data;
return sqe;
}
pub fn connect(
self: *IoUring,
user_data: u64,
fd: linux.fd_t,
addr: *const posix.sockaddr,
addrlen: posix.socklen_t,
) !*linux.io_uring_sqe {
const sqe = try self.get_sqe();
sqe.prep_connect(fd, addr, addrlen);
sqe.user_data = user_data;
return sqe;
}
pub fn epoll_ctl(
self: *IoUring,
user_data: u64,
epfd: linux.fd_t,
fd: linux.fd_t,
op: u32,
ev: ?*linux.epoll_event,
) !*linux.io_uring_sqe {
const sqe = try self.get_sqe();
sqe.prep_epoll_ctl(epfd, fd, op, ev);
sqe.user_data = user_data;
return sqe;
}
pub const RecvBuffer = union(enum) {
buffer: []u8,
buffer_selection: struct {
group_id: u16,
len: usize,
},
};
pub fn recv(
self: *IoUring,
user_data: u64,
fd: linux.fd_t,
buffer: RecvBuffer,
flags: u32,
) !*linux.io_uring_sqe {
const sqe = try self.get_sqe();
switch (buffer) {
.buffer => |slice| sqe.prep_recv(fd, slice, flags),
.buffer_selection => |selection| {
sqe.prep_rw(.RECV, fd, 0, selection.len, 0);
sqe.rw_flags = flags;
sqe.flags |= linux.IOSQE_BUFFER_SELECT;
sqe.buf_index = selection.group_id;
},
}
sqe.user_data = user_data;
return sqe;
}
pub fn send(
self: *IoUring,
user_data: u64,
fd: linux.fd_t,
buffer: []const u8,
flags: u32,
) !*linux.io_uring_sqe {
const sqe = try self.get_sqe();
sqe.prep_send(fd, buffer, flags);
sqe.user_data = user_data;
return sqe;
}
pub fn send_zc(
self: *IoUring,
user_data: u64,
fd: linux.fd_t,
buffer: []const u8,
send_flags: u32,
zc_flags: u16,
) !*linux.io_uring_sqe {
const sqe = try self.get_sqe();
sqe.prep_send_zc(fd, buffer, send_flags, zc_flags);
sqe.user_data = user_data;
return sqe;
}
pub fn send_zc_fixed(
self: *IoUring,
user_data: u64,
fd: linux.fd_t,
buffer: []const u8,
send_flags: u32,
zc_flags: u16,
buf_index: u16,
) !*linux.io_uring_sqe {
const sqe = try self.get_sqe();
sqe.prep_send_zc_fixed(fd, buffer, send_flags, zc_flags, buf_index);
sqe.user_data = user_data;
return sqe;
}
pub fn recvmsg(
self: *IoUring,
user_data: u64,
fd: linux.fd_t,
msg: *linux.msghdr,
flags: u32,
) !*linux.io_uring_sqe {
const sqe = try self.get_sqe();
sqe.prep_recvmsg(fd, msg, flags);
sqe.user_data = user_data;
return sqe;
}
pub fn sendmsg(
self: *IoUring,
user_data: u64,
fd: linux.fd_t,
msg: *const linux.msghdr_const,
flags: u32,
) !*linux.io_uring_sqe {
const sqe = try self.get_sqe();
sqe.prep_sendmsg(fd, msg, flags);
sqe.user_data = user_data;
return sqe;
}
pub fn sendmsg_zc(
self: *IoUring,
user_data: u64,
fd: linux.fd_t,
msg: *const linux.msghdr_const,
flags: u32,
) !*linux.io_uring_sqe {
const sqe = try self.get_sqe();
sqe.prep_sendmsg_zc(fd, msg, flags);
sqe.user_data = user_data;
return sqe;
}
pub fn openat(
self: *IoUring,
user_data: u64,
fd: linux.fd_t,
path: [*:0]const u8,
flags: linux.O,
mode: posix.mode_t,
) !*linux.io_uring_sqe {
const sqe = try self.get_sqe();
sqe.prep_openat(fd, path, flags, mode);
sqe.user_data = user_data;
return sqe;
}
pub fn openat_direct(
self: *IoUring,
user_data: u64,
fd: linux.fd_t,
path: [*:0]const u8,
flags: linux.O,
mode: posix.mode_t,
file_index: u32,
) !*linux.io_uring_sqe {
const sqe = try self.get_sqe();
sqe.prep_openat_direct(fd, path, flags, mode, file_index);
sqe.user_data = user_data;
return sqe;
}
pub fn close(self: *IoUring, user_data: u64, fd: linux.fd_t) !*linux.io_uring_sqe {
const sqe = try self.get_sqe();
sqe.prep_close(fd);
sqe.user_data = user_data;
return sqe;
}
pub fn close_direct(self: *IoUring, user_data: u64, file_index: u32) !*linux.io_uring_sqe {
const sqe = try self.get_sqe();
sqe.prep_close_direct(file_index);
sqe.user_data = user_data;
return sqe;
}
pub fn timeout(
self: *IoUring,
user_data: u64,
ts: *const linux.kernel_timespec,
count: u32,
flags: u32,
) !*linux.io_uring_sqe {
const sqe = try self.get_sqe();
sqe.prep_timeout(ts, count, flags);
sqe.user_data = user_data;
return sqe;
}
pub fn timeout_remove(
self: *IoUring,
user_data: u64,
timeout_user_data: u64,
flags: u32,
) !*linux.io_uring_sqe {
const sqe = try self.get_sqe();
sqe.prep_timeout_remove(timeout_user_data, flags);
sqe.user_data = user_data;
return sqe;
}
pub fn link_timeout(
self: *IoUring,
user_data: u64,
ts: *const linux.kernel_timespec,
flags: u32,
) !*linux.io_uring_sqe {
const sqe = try self.get_sqe();
sqe.prep_link_timeout(ts, flags);
sqe.user_data = user_data;
return sqe;
}
pub fn poll_add(
self: *IoUring,
user_data: u64,
fd: linux.fd_t,
poll_mask: u32,
) !*linux.io_uring_sqe {
const sqe = try self.get_sqe();
sqe.prep_poll_add(fd, poll_mask);
sqe.user_data = user_data;
return sqe;
}
pub fn poll_remove(
self: *IoUring,
user_data: u64,
target_user_data: u64,
) !*linux.io_uring_sqe {
const sqe = try self.get_sqe();
sqe.prep_poll_remove(target_user_data);
sqe.user_data = user_data;
return sqe;
}
pub fn poll_update(
self: *IoUring,
user_data: u64,
old_user_data: u64,
new_user_data: u64,
poll_mask: u32,
flags: u32,
) !*linux.io_uring_sqe {
const sqe = try self.get_sqe();
sqe.prep_poll_update(old_user_data, new_user_data, poll_mask, flags);
sqe.user_data = user_data;
return sqe;
}
pub fn fallocate(
self: *IoUring,
user_data: u64,
fd: linux.fd_t,
mode: i32,
offset: u64,
len: u64,
) !*linux.io_uring_sqe {
const sqe = try self.get_sqe();
sqe.prep_fallocate(fd, mode, offset, len);
sqe.user_data = user_data;
return sqe;
}
pub fn statx(
self: *IoUring,
user_data: u64,
fd: linux.fd_t,
path: [:0]const u8,
flags: u32,
mask: linux.STATX,
buf: *linux.Statx,
) !*linux.io_uring_sqe {
const sqe = try self.get_sqe();
sqe.prep_statx(fd, path, flags, mask, buf);
sqe.user_data = user_data;
return sqe;
}
pub fn cancel(
self: *IoUring,
user_data: u64,
cancel_user_data: u64,
flags: u32,
) !*linux.io_uring_sqe {
const sqe = try self.get_sqe();
sqe.prep_cancel(cancel_user_data, flags);
sqe.user_data = user_data;
return sqe;
}
pub fn shutdown(
self: *IoUring,
user_data: u64,
sockfd: posix.socket_t,
how: u32,
) !*linux.io_uring_sqe {
const sqe = try self.get_sqe();
sqe.prep_shutdown(sockfd, how);
sqe.user_data = user_data;
return sqe;
}
pub fn renameat(
self: *IoUring,
user_data: u64,
old_dir_fd: linux.fd_t,
old_path: [*:0]const u8,
new_dir_fd: linux.fd_t,
new_path: [*:0]const u8,
flags: u32,
) !*linux.io_uring_sqe {
const sqe = try self.get_sqe();
sqe.prep_renameat(old_dir_fd, old_path, new_dir_fd, new_path, flags);
sqe.user_data = user_data;
return sqe;
}
pub fn unlinkat(
self: *IoUring,
user_data: u64,
dir_fd: linux.fd_t,
path: [*:0]const u8,
flags: u32,
) !*linux.io_uring_sqe {
const sqe = try self.get_sqe();
sqe.prep_unlinkat(dir_fd, path, flags);
sqe.user_data = user_data;
return sqe;
}
pub fn mkdirat(
self: *IoUring,
user_data: u64,
dir_fd: linux.fd_t,
path: [*:0]const u8,
mode: posix.mode_t,
) !*linux.io_uring_sqe {
const sqe = try self.get_sqe();
sqe.prep_mkdirat(dir_fd, path, mode);
sqe.user_data = user_data;
return sqe;
}
pub fn symlinkat(
self: *IoUring,
user_data: u64,
target: [*:0]const u8,
new_dir_fd: linux.fd_t,
link_path: [*:0]const u8,
) !*linux.io_uring_sqe {
const sqe = try self.get_sqe();
sqe.prep_symlinkat(target, new_dir_fd, link_path);
sqe.user_data = user_data;
return sqe;
}
pub fn linkat(
self: *IoUring,
user_data: u64,
old_dir_fd: linux.fd_t,
old_path: [*:0]const u8,
new_dir_fd: linux.fd_t,
new_path: [*:0]const u8,
flags: u32,
) !*linux.io_uring_sqe {
const sqe = try self.get_sqe();
sqe.prep_linkat(old_dir_fd, old_path, new_dir_fd, new_path, flags);
sqe.user_data = user_data;
return sqe;
}
pub fn provide_buffers(
self: *IoUring,
user_data: u64,
buffers: [*]u8,
buffer_size: usize,
buffers_count: usize,
group_id: usize,
buffer_id: usize,
) !*linux.io_uring_sqe {
const sqe = try self.get_sqe();
sqe.prep_provide_buffers(buffers, buffer_size, buffers_count, group_id, buffer_id);
sqe.user_data = user_data;
return sqe;
}
pub fn remove_buffers(
self: *IoUring,
user_data: u64,
buffers_count: usize,
group_id: usize,
) !*linux.io_uring_sqe {
const sqe = try self.get_sqe();
sqe.prep_remove_buffers(buffers_count, group_id);
sqe.user_data = user_data;
return sqe;
}
pub fn waitid(
self: *IoUring,
user_data: u64,
id_type: linux.P,
id: i32,
infop: *linux.siginfo_t,
options: u32,
flags: u32,
) !*linux.io_uring_sqe {
const sqe = try self.get_sqe();
sqe.prep_waitid(id_type, id, infop, options, flags);
sqe.user_data = user_data;
return sqe;
}
pub fn register_files(self: *IoUring, fds: []const linux.fd_t) !void {
assert(self.fd >= 0);
const res = linux.io_uring_register(
self.fd,
.REGISTER_FILES,
@as(*const anyopaque, @ptrCast(fds.ptr)),
@as(u32, @intCast(fds.len)),
);
try handle_registration_result(res);
}
pub fn register_files_update(self: *IoUring, offset: u32, fds: []const linux.fd_t) !void {
assert(self.fd >= 0);
const FilesUpdate = extern struct {
offset: u32,
resv: u32,
fds: u64 align(8),
};
var update = FilesUpdate{
.offset = offset,
.resv = @as(u32, 0),
.fds = @as(u64, @intFromPtr(fds.ptr)),
};
const res = linux.io_uring_register(
self.fd,
.REGISTER_FILES_UPDATE,
@as(*const anyopaque, @ptrCast(&update)),
@as(u32, @intCast(fds.len)),
);
try handle_registration_result(res);
}
pub fn register_files_sparse(self: *IoUring, nr_files: u32) !void {
assert(self.fd >= 0);
const reg = &linux.io_uring_rsrc_register{
.nr = nr_files,
.flags = linux.IORING_RSRC_REGISTER_SPARSE,
.resv2 = 0,
.data = 0,
.tags = 0,
};
const res = linux.io_uring_register(
self.fd,
.REGISTER_FILES2,
@ptrCast(reg),
@as(u32, @sizeOf(linux.io_uring_rsrc_register)),
);
return handle_registration_result(res);
}
// Available since 6.0
pub fn register_file_alloc_range(self: *IoUring, offset: u32, len: u32) !void {
assert(self.fd >= 0);
const range = &linux.io_uring_file_index_range{
.off = offset,
.len = len,
.resv = 0,
};
const res = linux.io_uring_register(
self.fd,
.REGISTER_FILE_ALLOC_RANGE,
@ptrCast(range),
@as(u32, @sizeOf(linux.io_uring_file_index_range)),
);
return handle_registration_result(res);
}
pub fn register_eventfd(self: *IoUring, fd: linux.fd_t) !void {
assert(self.fd >= 0);
const res = linux.io_uring_register(
self.fd,
.REGISTER_EVENTFD,
@as(*const anyopaque, @ptrCast(&fd)),
1,
);
try handle_registration_result(res);
}
pub fn register_eventfd_async(self: *IoUring, fd: linux.fd_t) !void {
assert(self.fd >= 0);
const res = linux.io_uring_register(
self.fd,
.REGISTER_EVENTFD_ASYNC,
@as(*const anyopaque, @ptrCast(&fd)),
1,
);
try handle_registration_result(res);
}
pub fn unregister_eventfd(self: *IoUring) !void {
assert(self.fd >= 0);
const res = linux.io_uring_register(
self.fd,
.UNREGISTER_EVENTFD,
null,
0,
);
try handle_registration_result(res);
}
pub fn register_napi(self: *IoUring, napi: *linux.io_uring_napi) !void {
assert(self.fd >= 0);
const res = linux.io_uring_register(self.fd, .REGISTER_NAPI, napi, 1);
try handle_registration_result(res);
}
pub fn unregister_napi(self: *IoUring, napi: *linux.io_uring_napi) !void {
assert(self.fd >= 0);
const res = linux.io_uring_register(self.fd, .UNREGISTER_NAPI, napi, 1);
try handle_registration_result(res);
}
pub fn register_buffers(self: *IoUring, buffers: []const posix.iovec) !void {
assert(self.fd >= 0);
const res = linux.io_uring_register(
self.fd,
.REGISTER_BUFFERS,
buffers.ptr,
@as(u32, @intCast(buffers.len)),
);
try handle_registration_result(res);
}
pub fn unregister_buffers(self: *IoUring) !void {
assert(self.fd >= 0);
const res = linux.io_uring_register(self.fd, .UNREGISTER_BUFFERS, null, 0);
switch (linux.errno(res)) {
.SUCCESS => {},
.NXIO => return error.BuffersNotRegistered,
else => |errno| return posix.unexpectedErrno(errno),
}
}
pub fn get_probe(self: *IoUring) !linux.io_uring_probe {
var probe = std.mem.zeroInit(linux.io_uring_probe, .{});
const res = linux.io_uring_register(self.fd, .REGISTER_PROBE, &probe, probe.ops.len);
try handle_register_buf_ring_result(res);
return probe;
}
fn handle_registration_result(res: usize) !void {
switch (linux.errno(res)) {
.SUCCESS => {},
.BADF => return error.FileDescriptorInvalid,
.BUSY => return error.FilesAlreadyRegistered,
.INVAL => return error.FilesEmpty,
// user is allowed to have according to the per-user RLIMIT_NOFILE resource limit and
// the CAP_SYS_RESOURCE capability is not set, or `nr_args` exceeds the maximum allowed
// for a fixed file set (older kernels have a limit of 1024 files vs 64K files):
.MFILE => return error.UserFdQuotaExceeded,
// resource limit but tried to lock more memory than the limit permitted (not enforced
// when the process is privileged with CAP_IPC_LOCK):
.NOMEM => return error.SystemResources,
.NXIO => return error.RingShuttingDownOrAlreadyRegisteringFiles,
else => |errno| return posix.unexpectedErrno(errno),
}
}
pub fn unregister_files(self: *IoUring) !void {
assert(self.fd >= 0);
const res = linux.io_uring_register(self.fd, .UNREGISTER_FILES, null, 0);
switch (linux.errno(res)) {
.SUCCESS => {},
.NXIO => return error.FilesNotRegistered,
else => |errno| return posix.unexpectedErrno(errno),
}
}
pub fn socket(
self: *IoUring,
user_data: u64,
domain: u32,
socket_type: u32,
protocol: u32,
flags: u32,
) !*linux.io_uring_sqe {
const sqe = try self.get_sqe();
sqe.prep_socket(domain, socket_type, protocol, flags);
sqe.user_data = user_data;
return sqe;
}
pub fn socket_direct(
self: *IoUring,
user_data: u64,
domain: u32,
socket_type: u32,
protocol: u32,
flags: u32,
file_index: u32,
) !*linux.io_uring_sqe {
const sqe = try self.get_sqe();
sqe.prep_socket_direct(domain, socket_type, protocol, flags, file_index);
sqe.user_data = user_data;
return sqe;
}
pub fn socket_direct_alloc(
self: *IoUring,
user_data: u64,
domain: u32,
socket_type: u32,
protocol: u32,
flags: u32,
) !*linux.io_uring_sqe {
const sqe = try self.get_sqe();
sqe.prep_socket_direct_alloc(domain, socket_type, protocol, flags);
sqe.user_data = user_data;
return sqe;
}
pub fn bind(
self: *IoUring,
user_data: u64,
fd: linux.fd_t,
addr: *const posix.sockaddr,
addrlen: posix.socklen_t,
flags: u32,
) !*linux.io_uring_sqe {
const sqe = try self.get_sqe();
sqe.prep_bind(fd, addr, addrlen, flags);
sqe.user_data = user_data;
return sqe;
}
pub fn listen(
self: *IoUring,
user_data: u64,
fd: linux.fd_t,
backlog: usize,
flags: u32,
) !*linux.io_uring_sqe {
const sqe = try self.get_sqe();
sqe.prep_listen(fd, backlog, flags);
sqe.user_data = user_data;
return sqe;
}
pub fn cmd_sock(
self: *IoUring,
user_data: u64,
cmd_op: linux.IO_URING_SOCKET_OP,
fd: linux.fd_t,
level: u32,
optname: u32,
optval: u64,
optlen: u32,
) !*linux.io_uring_sqe {
const sqe = try self.get_sqe();
sqe.prep_cmd_sock(cmd_op, fd, level, optname, optval, optlen);
sqe.user_data = user_data;
return sqe;
}
pub fn setsockopt(
self: *IoUring,
user_data: u64,
fd: linux.fd_t,
level: u32,
optname: u32,
opt: []const u8,
) !*linux.io_uring_sqe {
return try self.cmd_sock(
user_data,
.SETSOCKOPT,
fd,
level,
optname,
@intFromPtr(opt.ptr),
@intCast(opt.len),
);
}
pub fn getsockopt(
self: *IoUring,
user_data: u64,
fd: linux.fd_t,
level: u32,
optname: u32,
opt: []u8,
) !*linux.io_uring_sqe {
return try self.cmd_sock(
user_data,
.GETSOCKOPT,
fd,
level,
optname,
@intFromPtr(opt.ptr),
@intCast(opt.len),
);
}
pub const SubmissionQueue = struct {
head: *u32,
tail: *u32,
mask: u32,
flags: *u32,
dropped: *u32,
array: []u32,
sqes: []linux.io_uring_sqe,
mmap: []align(page_size_min) u8,
mmap_sqes: []align(page_size_min) u8,
// We increment `sqe_tail` (but not `tail`) for each call to `get_sqe()`.
// We then set `tail` to `sqe_tail` once, only when these events are actually submitted.
// This allows us to amortize the cost of the @atomicStore to `tail` across multiple SQEs.
sqe_head: u32 = 0,
sqe_tail: u32 = 0,
pub fn init(fd: linux.fd_t, p: linux.io_uring_params) !SubmissionQueue {
assert(fd >= 0);
assert((p.features & linux.IORING_FEAT_SINGLE_MMAP) != 0);
const size = @max(
p.sq_off.array + p.sq_entries * @sizeOf(u32),
p.cq_off.cqes + p.cq_entries * @sizeOf(linux.io_uring_cqe),
);
const mmap = try posix.mmap(
null,
size,
.{ .READ = true, .WRITE = true },
.{ .TYPE = .SHARED, .POPULATE = true },
fd,
linux.IORING_OFF_SQ_RING,
);
errdefer posix.munmap(mmap);
assert(mmap.len == size);
// application to preallocate static linux.io_uring_sqe entries and then replay them when needed.
const size_sqes = p.sq_entries * @sizeOf(linux.io_uring_sqe);
const mmap_sqes = try posix.mmap(
null,
size_sqes,
.{ .READ = true, .WRITE = true },
.{ .TYPE = .SHARED, .POPULATE = true },
fd,
linux.IORING_OFF_SQES,
);
errdefer posix.munmap(mmap_sqes);
assert(mmap_sqes.len == size_sqes);
const array: [*]u32 = @ptrCast(@alignCast(&mmap[p.sq_off.array]));
const sqes: [*]linux.io_uring_sqe = @ptrCast(@alignCast(&mmap_sqes[0]));
// see https://github.com/torvalds/linux/blob/v5.8/fs/io_uring.c#L7843-L7844.
assert(p.sq_entries == @as(*u32, @ptrCast(@alignCast(&mmap[p.sq_off.ring_entries]))).*);
return SubmissionQueue{
.head = @ptrCast(@alignCast(&mmap[p.sq_off.head])),
.tail = @ptrCast(@alignCast(&mmap[p.sq_off.tail])),
.mask = @as(*u32, @ptrCast(@alignCast(&mmap[p.sq_off.ring_mask]))).*,
.flags = @ptrCast(@alignCast(&mmap[p.sq_off.flags])),
.dropped = @ptrCast(@alignCast(&mmap[p.sq_off.dropped])),
.array = array[0..p.sq_entries],
.sqes = sqes[0..p.sq_entries],
.mmap = mmap,
.mmap_sqes = mmap_sqes,
};
}
pub fn deinit(self: *SubmissionQueue) void {
posix.munmap(self.mmap_sqes);
posix.munmap(self.mmap);
}
};
pub const CompletionQueue = struct {
head: *u32,
tail: *u32,
mask: u32,
overflow: *u32,
cqes: []linux.io_uring_cqe,
pub fn init(fd: linux.fd_t, p: linux.io_uring_params, sq: SubmissionQueue) !CompletionQueue {
assert(fd >= 0);
assert((p.features & linux.IORING_FEAT_SINGLE_MMAP) != 0);
const mmap = sq.mmap;
const cqes: [*]linux.io_uring_cqe = @ptrCast(@alignCast(&mmap[p.cq_off.cqes]));
assert(p.cq_entries == @as(*u32, @ptrCast(@alignCast(&mmap[p.cq_off.ring_entries]))).*);
return CompletionQueue{
.head = @ptrCast(@alignCast(&mmap[p.cq_off.head])),
.tail = @ptrCast(@alignCast(&mmap[p.cq_off.tail])),
.mask = @as(*u32, @ptrCast(@alignCast(&mmap[p.cq_off.ring_mask]))).*,
.overflow = @ptrCast(@alignCast(&mmap[p.cq_off.overflow])),
.cqes = cqes[0..p.cq_entries],
};
}
pub fn deinit(self: *CompletionQueue) void {
_ = self;
// Here for symmetry with the submission queue, and for any future feature support.
}
};
pub const BufferGroup = struct {
ring: *IoUring,
br: *align(page_size_min) linux.io_uring_buf_ring,
buffers: []u8,
buffer_size: u32,
buffers_count: u16,
heads: []u32,
group_id: u16,
pub fn init(
ring: *IoUring,
allocator: Allocator,
group_id: u16,
buffer_size: u32,
buffers_count: u16,
) !BufferGroup {
const buffers = try allocator.alloc(u8, buffer_size * buffers_count);
errdefer allocator.free(buffers);
const heads = try allocator.alloc(u32, buffers_count);
errdefer allocator.free(heads);
const br = try setup_buf_ring(ring.fd, buffers_count, group_id, .{ .inc = true });
buf_ring_init(br);
const mask = buf_ring_mask(buffers_count);
var i: u16 = 0;
while (i < buffers_count) : (i += 1) {
const pos = buffer_size * i;
const buf = buffers[pos .. pos + buffer_size];
heads[i] = 0;
buf_ring_add(br, buf, i, mask, i);
}
buf_ring_advance(br, buffers_count);
return BufferGroup{
.ring = ring,
.group_id = group_id,
.br = br,
.buffers = buffers,
.heads = heads,
.buffer_size = buffer_size,
.buffers_count = buffers_count,
};
}
pub fn deinit(self: *BufferGroup, allocator: Allocator) void {
free_buf_ring(self.ring.fd, self.br, self.buffers_count, self.group_id);
allocator.free(self.buffers);
allocator.free(self.heads);
}
pub fn recv(self: *BufferGroup, user_data: u64, fd: linux.fd_t, flags: u32) !*linux.io_uring_sqe {
var sqe = try self.ring.get_sqe();
sqe.prep_rw(.RECV, fd, 0, 0, 0);
sqe.rw_flags = flags;
sqe.flags |= linux.IOSQE_BUFFER_SELECT;
sqe.buf_index = self.group_id;
sqe.user_data = user_data;
return sqe;
}
pub fn recv_multishot(self: *BufferGroup, user_data: u64, fd: linux.fd_t, flags: u32) !*linux.io_uring_sqe {
var sqe = try self.recv(user_data, fd, flags);
sqe.ioprio |= linux.IORING_RECV_MULTISHOT;
return sqe;
}
pub fn get_by_id(self: *BufferGroup, buffer_id: u16) []u8 {
const pos = self.buffer_size * buffer_id;
return self.buffers[pos .. pos + self.buffer_size][self.heads[buffer_id]..];
}
pub fn get(self: *BufferGroup, cqe: linux.io_uring_cqe) ![]u8 {
const buffer_id = try cqe.buffer_id();
const used_len = @as(usize, @intCast(cqe.res));
return self.get_by_id(buffer_id)[0..used_len];
}
pub fn put(self: *BufferGroup, cqe: linux.io_uring_cqe) !void {
const buffer_id = try cqe.buffer_id();
if (cqe.flags & linux.IORING_CQE_F_BUF_MORE == linux.IORING_CQE_F_BUF_MORE) {
const used_len = @as(u32, @intCast(cqe.res));
self.heads[buffer_id] += used_len;
return;
}
self.heads[buffer_id] = 0;
const mask = buf_ring_mask(self.buffers_count);
buf_ring_add(self.br, self.get_by_id(buffer_id), buffer_id, mask, 0);
buf_ring_advance(self.br, 1);
}
};
pub fn setup_buf_ring(
fd: linux.fd_t,
entries: u16,
group_id: u16,
flags: linux.io_uring_buf_reg.Flags,
) !*align(page_size_min) linux.io_uring_buf_ring {
if (entries == 0 or entries > 1 << 15) return error.EntriesNotInRange;
if (!std.math.isPowerOfTwo(entries)) return error.EntriesNotPowerOfTwo;
const mmap_size = @as(usize, entries) * @sizeOf(linux.io_uring_buf);
const mmap = try posix.mmap(
null,
mmap_size,
.{ .READ = true, .WRITE = true },
.{ .TYPE = .PRIVATE, .ANONYMOUS = true },
-1,
0,
);
errdefer posix.munmap(mmap);
assert(mmap.len == mmap_size);
const br: *align(page_size_min) linux.io_uring_buf_ring = @ptrCast(mmap.ptr);
try register_buf_ring(fd, @intFromPtr(br), entries, group_id, flags);
return br;
}
fn register_buf_ring(
fd: linux.fd_t,
addr: u64,
entries: u32,
group_id: u16,
flags: linux.io_uring_buf_reg.Flags,
) !void {
var reg = std.mem.zeroInit(linux.io_uring_buf_reg, .{
.ring_addr = addr,
.ring_entries = entries,
.bgid = group_id,
.flags = flags,
});
var res = linux.io_uring_register(fd, .REGISTER_PBUF_RING, @as(*const anyopaque, @ptrCast(®)), 1);
if (linux.errno(res) == .INVAL and reg.flags.inc) {
// It is available since kernel 6.12. returns INVAL on older.
reg.flags.inc = false;
res = linux.io_uring_register(fd, .REGISTER_PBUF_RING, @as(*const anyopaque, @ptrCast(®)), 1);
}
try handle_register_buf_ring_result(res);
}
fn unregister_buf_ring(fd: linux.fd_t, group_id: u16) !void {
var reg = std.mem.zeroInit(linux.io_uring_buf_reg, .{
.bgid = group_id,
});
const res = linux.io_uring_register(
fd,
.UNREGISTER_PBUF_RING,
@as(*const anyopaque, @ptrCast(®)),
1,
);
try handle_register_buf_ring_result(res);
}
fn handle_register_buf_ring_result(res: usize) !void {
switch (linux.errno(res)) {
.SUCCESS => {},
.INVAL => return error.ArgumentsInvalid,
else => |errno| return posix.unexpectedErrno(errno),
}
}
pub fn free_buf_ring(fd: linux.fd_t, br: *align(page_size_min) linux.io_uring_buf_ring, entries: u32, group_id: u16) void {
unregister_buf_ring(fd, group_id) catch {};
var mmap: []align(page_size_min) u8 = undefined;
mmap.ptr = @ptrCast(br);
mmap.len = entries * @sizeOf(linux.io_uring_buf);
posix.munmap(mmap);
}
pub fn buf_ring_init(br: *linux.io_uring_buf_ring) void {
br.tail = 0;
}
pub fn buf_ring_mask(entries: u16) u16 {
return entries - 1;
}
pub fn buf_ring_add(
br: *linux.io_uring_buf_ring,
buffer: []u8,
buffer_id: u16,
mask: u16,
buffer_offset: u16,
) void {
const bufs: [*]linux.io_uring_buf = @ptrCast(br);
const buf: *linux.io_uring_buf = &bufs[(br.tail +% buffer_offset) & mask];
buf.addr = @intFromPtr(buffer.ptr);
buf.len = @intCast(buffer.len);
buf.bid = buffer_id;
}
pub fn buf_ring_advance(br: *linux.io_uring_buf_ring, count: u16) void {
const tail: u16 = br.tail +% count;
@atomicStore(u16, &br.tail, tail, .release);
}
test BufferGroup {
if (builtin.target.cpu.arch.isPowerPC()) return;
if (!is_linux) return error.SkipZigTest;
const io = testing.io;
_ = io;
var ring = IoUring.init(16, 0) catch |err| switch (err) {
error.SystemOutdated => return error.SkipZigTest,
error.PermissionDenied => return error.SkipZigTest,
else => return err,
};
defer ring.deinit();
const group_id: u16 = 1;
const buffers_count: u16 = 1;
const buffer_size: usize = 128;
var buf_grp = BufferGroup.init(
&ring,
testing.allocator,
group_id,
buffer_size,
buffers_count,
) catch |err| switch (err) {
error.ArgumentsInvalid => return error.SkipZigTest,
else => return err,
};
defer buf_grp.deinit(testing.allocator);
const fds = try createSocketTestHarness(&ring);
defer fds.close();
const data = [_]u8{ 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0xa, 0xb, 0xc, 0xd, 0xe };
{
_ = try ring.send(1, fds.client, data[0..], 0);
const submitted = try ring.submit();
try testing.expectEqual(1, submitted);
const cqe_send = try ring.copy_cqe();
if (cqe_send.err() == .INVAL) return error.SkipZigTest;
try testing.expectEqual(linux.io_uring_cqe{ .user_data = 1, .res = data.len, .flags = 0 }, cqe_send);
}
{
_ = try buf_grp.recv(2, fds.server, 0);
const submitted = try ring.submit();
try testing.expectEqual(1, submitted);
const cqe = try ring.copy_cqe();
try testing.expectEqual(2, cqe.user_data);
try testing.expect(cqe.res >= 0);
try testing.expectEqual(posix.E.SUCCESS, cqe.err());
try testing.expectEqual(data.len, @as(usize, @intCast(cqe.res)));
// Get buffer from pool
const buf = try buf_grp.get(cqe);
try testing.expectEqualSlices(u8, &data, buf);
try buf_grp.put(cqe);
}
}
test {
if (builtin.target.cpu.arch.isPowerPC()) return;
if (is_linux) _ = @import("IoUring/test.zig");
}