| ... | ... | @@ -0,0 +1,1218 @@ |
| 1 | // SPDX-License-Identifier: MIT |
| 2 | // Copyright (c) 2015-2020 Zig Contributors |
| 3 | // This file is part of [zig](https://ziglang.org/), which is MIT licensed. |
| 4 | // The MIT license requires this copyright notice to be included in all copies |
| 5 | // and substantial portions of the software. |
| 6 | const std = @import("../../std.zig"); |
| 7 | const assert = std.debug.assert; |
| 8 | const builtin = std.builtin; |
| 9 | const mem = std.mem; |
| 10 | const net = std.net; |
| 11 | const os = std.os; |
| 12 | const linux = os.linux; |
| 13 | const testing = std.testing; |
| 14 | |
| 15 | const io_uring_params = linux.io_uring_params; |
| 16 | const io_uring_sqe = linux.io_uring_sqe; |
| 17 | const io_uring_cqe = linux.io_uring_cqe; |
| 18 | |
| 19 | pub const IO_Uring = struct { |
| 20 | fd: os.fd_t = -1, |
| 21 | sq: SubmissionQueue, |
| 22 | cq: CompletionQueue, |
| 23 | flags: u32, |
| 24 | features: u32, |
| 25 | |
| 26 | /// A friendly way to setup an io_uring, with default io_uring_params. |
| 27 | /// `entries` must be a power of two between 1 and 4096, although the kernel will make the final |
| 28 | /// call on how many entries the submission and completion queues will ultimately have, |
| 29 | /// see https://github.com/torvalds/linux/blob/v5.8/fs/io_uring.c#L8027-L8050. |
| 30 | /// Matches the interface of io_uring_queue_init() in liburing. |
| 31 | pub fn init(entries: u12, flags: u32) !IO_Uring { |
| 32 | var params = mem.zeroInit(io_uring_params, .{ |
| 33 | .flags = flags, |
| 34 | .sq_thread_idle = 1000 |
| 35 | }); |
| 36 | return try IO_Uring.init_params(entries, ¶ms); |
| 37 | } |
| 38 | |
| 39 | /// A powerful way to setup an io_uring, if you want to tweak io_uring_params such as submission |
| 40 | /// queue thread cpu affinity or thread idle timeout (the kernel and our default is 1 second). |
| 41 | /// `params` is passed by reference because the kernel needs to modify the parameters. |
| 42 | /// You may only set the `flags`, `sq_thread_cpu` and `sq_thread_idle` parameters. |
| 43 | /// Every other parameter belongs to the kernel and must be zeroed. |
| 44 | /// Matches the interface of io_uring_queue_init_params() in liburing. |
| 45 | pub fn init_params(entries: u12, p: *io_uring_params) !IO_Uring { |
| 46 | if (entries == 0) return error.EntriesZero; |
| 47 | if (!std.math.isPowerOfTwo(entries)) return error.EntriesNotPowerOfTwo; |
| 48 | |
| 49 | assert(p.sq_entries == 0); |
| 50 | assert(p.cq_entries == 0); |
| 51 | assert(p.features == 0); |
| 52 | assert(p.wq_fd == 0); |
| 53 | assert(p.resv[0] == 0); |
| 54 | assert(p.resv[1] == 0); |
| 55 | assert(p.resv[2] == 0); |
| 56 | |
| 57 | const res = linux.io_uring_setup(entries, p); |
| 58 | switch (linux.getErrno(res)) { |
| 59 | 0 => {}, |
| 60 | linux.EFAULT => return error.ParamsOutsideAccessibleAddressSpace, |
| 61 | // The resv array contains non-zero data, p.flags contains an unsupported flag, |
| 62 | // entries out of bounds, IORING_SETUP_SQ_AFF was specified without IORING_SETUP_SQPOLL, |
| 63 | // or IORING_SETUP_CQSIZE was specified but io_uring_params.cq_entries was invalid: |
| 64 | linux.EINVAL => return error.ArgumentsInvalid, |
| 65 | linux.EMFILE => return error.ProcessFdQuotaExceeded, |
| 66 | linux.ENFILE => return error.SystemFdQuotaExceeded, |
| 67 | linux.ENOMEM => return error.SystemResources, |
| 68 | // IORING_SETUP_SQPOLL was specified but effective user ID lacks sufficient privileges, |
| 69 | // or a container seccomp policy prohibits io_uring syscalls: |
| 70 | linux.EPERM => return error.PermissionDenied, |
| 71 | linux.ENOSYS => return error.SystemOutdated, |
| 72 | else => |errno| return os.unexpectedErrno(errno) |
| 73 | } |
| 74 | const fd = @intCast(os.fd_t, res); |
| 75 | assert(fd >= 0); |
| 76 | errdefer os.close(fd); |
| 77 | |
| 78 | // Kernel versions 5.4 and up use only one mmap() for the submission and completion queues. |
| 79 | // This is not an optional feature for us... if the kernel does it, we have to do it. |
| 80 | // The thinking on this by the kernel developers was that both the submission and the |
| 81 | // completion queue rings have sizes just over a power of two, but the submission queue ring |
| 82 | // is significantly smaller with u32 slots. By bundling both in a single mmap, the kernel |
| 83 | // gets the submission queue ring for free. |
| 84 | // See https://patchwork.kernel.org/patch/11115257 for the kernel patch. |
| 85 | // We do not support the double mmap() done before 5.4, because we want to keep the |
| 86 | // init/deinit mmap paths simple and because io_uring has had many bug fixes even since 5.4. |
| 87 | if ((p.features & linux.IORING_FEAT_SINGLE_MMAP) == 0) { |
| 88 | return error.SystemOutdated; |
| 89 | } |
| 90 | |
| 91 | // Check that the kernel has actually set params and that "impossible is nothing". |
| 92 | assert(p.sq_entries != 0); |
| 93 | assert(p.cq_entries != 0); |
| 94 | assert(p.cq_entries >= p.sq_entries); |
| 95 | |
| 96 | // From here on, we only need to read from params, so pass `p` by value as immutable. |
| 97 | // The completion queue shares the mmap with the submission queue, so pass `sq` there too. |
| 98 | var sq = try SubmissionQueue.init(fd, p.*); |
| 99 | errdefer sq.deinit(); |
| 100 | var cq = try CompletionQueue.init(fd, p.*, sq); |
| 101 | errdefer cq.deinit(); |
| 102 | |
| 103 | // Check that our starting state is as we expect. |
| 104 | assert(sq.head.* == 0); |
| 105 | assert(sq.tail.* == 0); |
| 106 | assert(sq.mask == p.sq_entries - 1); |
| 107 | // Allow flags.* to be non-zero, since the kernel may set IORING_SQ_NEED_WAKEUP at any time. |
| 108 | assert(sq.dropped.* == 0); |
| 109 | assert(sq.array.len == p.sq_entries); |
| 110 | assert(sq.sqes.len == p.sq_entries); |
| 111 | assert(sq.sqe_head == 0); |
| 112 | assert(sq.sqe_tail == 0); |
| 113 | |
| 114 | assert(cq.head.* == 0); |
| 115 | assert(cq.tail.* == 0); |
| 116 | assert(cq.mask == p.cq_entries - 1); |
| 117 | assert(cq.overflow.* == 0); |
| 118 | assert(cq.cqes.len == p.cq_entries); |
| 119 | |
| 120 | return IO_Uring { |
| 121 | .fd = fd, |
| 122 | .sq = sq, |
| 123 | .cq = cq, |
| 124 | .flags = p.flags, |
| 125 | .features = p.features |
| 126 | }; |
| 127 | } |
| 128 | |
| 129 | pub fn deinit(self: *IO_Uring) void { |
| 130 | assert(self.fd >= 0); |
| 131 | // The mmaps depend on the fd, so the order of these calls is important: |
| 132 | self.cq.deinit(); |
| 133 | self.sq.deinit(); |
| 134 | os.close(self.fd); |
| 135 | self.fd = -1; |
| 136 | } |
| 137 | |
| 138 | /// Returns a pointer to a vacant SQE, or an error if the submission queue is full. |
| 139 | /// We follow the implementation (and atomics) of liburing's `io_uring_get_sqe()` exactly. |
| 140 | /// However, instead of a null we return an error to force safe handling. |
| 141 | /// Any situation where the submission queue is full tends more towards a control flow error, |
| 142 | /// and the null return in liburing is more a C idiom than anything else, for lack of a better |
| 143 | /// alternative. In Zig, we have first-class error handling... so let's use it. |
| 144 | /// Matches the implementation of io_uring_get_sqe() in liburing. |
| 145 | pub fn get_sqe(self: *IO_Uring) !*io_uring_sqe { |
| 146 | const head = @atomicLoad(u32, self.sq.head, .Acquire); |
| 147 | // Remember that these head and tail offsets wrap around every four billion operations. |
| 148 | // We must therefore use wrapping addition and subtraction to avoid a runtime crash. |
| 149 | const next = self.sq.sqe_tail +% 1; |
| 150 | if (next -% head > self.sq.sqes.len) return error.SubmissionQueueFull; |
| 151 | var sqe = &self.sq.sqes[self.sq.sqe_tail & self.sq.mask]; |
| 152 | self.sq.sqe_tail = next; |
| 153 | return sqe; |
| 154 | } |
| 155 | |
| 156 | /// Submits the SQEs acquired via get_sqe() to the kernel. You can call this once after you have |
| 157 | /// called get_sqe() multiple times to setup multiple I/O requests. |
| 158 | /// Returns the number of SQEs submitted. |
| 159 | /// Matches the implementation of io_uring_submit() in liburing. |
| 160 | pub fn submit(self: *IO_Uring) !u32 { |
| 161 | return self.submit_and_wait(0); |
| 162 | } |
| 163 | |
| 164 | /// Like submit(), but allows waiting for events as well. |
| 165 | /// Returns the number of SQEs submitted. |
| 166 | /// Matches the implementation of io_uring_submit_and_wait() in liburing. |
| 167 | pub fn submit_and_wait(self: *IO_Uring, wait_nr: u32) !u32 { |
| 168 | var submitted = self.flush_sq(); |
| 169 | var flags: u32 = 0; |
| 170 | if (self.sq_ring_needs_enter(submitted, &flags) or wait_nr > 0) { |
| 171 | if (wait_nr > 0 or (self.flags & linux.IORING_SETUP_IOPOLL) != 0) { |
| 172 | flags |= linux.IORING_ENTER_GETEVENTS; |
| 173 | } |
| 174 | return try self.enter(submitted, wait_nr, flags); |
| 175 | } |
| 176 | return submitted; |
| 177 | } |
| 178 | |
| 179 | /// Tell the kernel we have submitted SQEs and/or want to wait for CQEs. |
| 180 | /// Returns the number of SQEs submitted. |
| 181 | pub fn enter(self: *IO_Uring, to_submit: u32, min_complete: u32, flags: u32) !u32 { |
| 182 | assert(self.fd >= 0); |
| 183 | const res = linux.io_uring_enter(self.fd, to_submit, min_complete, flags, null); |
| 184 | switch (linux.getErrno(res)) { |
| 185 | 0 => {}, |
| 186 | // The kernel was unable to allocate memory or ran out of resources for the request. |
| 187 | // The application should wait for some completions and try again: |
| 188 | linux.EAGAIN => return error.SystemResources, |
| 189 | // The SQE `fd` is invalid, or IOSQE_FIXED_FILE was set but no files were registered: |
| 190 | linux.EBADF => return error.FileDescriptorInvalid, |
| 191 | // The file descriptor is valid, but the ring is not in the right state. |
| 192 | // See io_uring_register(2) for how to enable the ring. |
| 193 | linux.EBADFD => return error.FileDescriptorInBadState, |
| 194 | // The application attempted to overcommit the number of requests it can have pending. |
| 195 | // The application should wait for some completions and try again: |
| 196 | linux.EBUSY => return error.CompletionQueueOvercommitted, |
| 197 | // The SQE is invalid, or valid but the ring was setup with IORING_SETUP_IOPOLL: |
| 198 | linux.EINVAL => return error.SubmissionQueueEntryInvalid, |
| 199 | // The buffer is outside the process' accessible address space, or IORING_OP_READ_FIXED |
| 200 | // or IORING_OP_WRITE_FIXED was specified but no buffers were registered, or the range |
| 201 | // described by `addr` and `len` is not within the buffer registered at `buf_index`: |
| 202 | linux.EFAULT => return error.BufferInvalid, |
| 203 | linux.ENXIO => return error.RingShuttingDown, |
| 204 | // The kernel believes our `self.fd` does not refer to an io_uring instance, |
| 205 | // or the opcode is valid but not supported by this kernel (more likely): |
| 206 | linux.EOPNOTSUPP => return error.OpcodeNotSupported, |
| 207 | // The operation was interrupted by a delivery of a signal before it could complete. |
| 208 | // This can happen while waiting for events with IORING_ENTER_GETEVENTS: |
| 209 | linux.EINTR => return error.SignalInterrupt, |
| 210 | else => |errno| return os.unexpectedErrno(errno) |
| 211 | } |
| 212 | return @intCast(u32, res); |
| 213 | } |
| 214 | |
| 215 | /// Sync internal state with kernel ring state on the SQ side. |
| 216 | /// Returns the number of all pending events in the SQ ring, for the shared ring. |
| 217 | /// This return value includes previously flushed SQEs, as per liburing. |
| 218 | /// The rationale is to suggest that an io_uring_enter() call is needed rather than not. |
| 219 | /// Matches the implementation of __io_uring_flush_sq() in liburing. |
| 220 | pub fn flush_sq(self: *IO_Uring) u32 { |
| 221 | if (self.sq.sqe_head != self.sq.sqe_tail) { |
| 222 | // Fill in SQEs that we have queued up, adding them to the kernel ring. |
| 223 | const to_submit = self.sq.sqe_tail -% self.sq.sqe_head; |
| 224 | var tail = self.sq.tail.*; |
| 225 | var i: usize = 0; |
| 226 | while (i < to_submit) : (i += 1) { |
| 227 | self.sq.array[tail & self.sq.mask] = self.sq.sqe_head & self.sq.mask; |
| 228 | tail +%= 1; |
| 229 | self.sq.sqe_head +%= 1; |
| 230 | } |
| 231 | // Ensure that the kernel can actually see the SQE updates when it sees the tail update. |
| 232 | @atomicStore(u32, self.sq.tail, tail, .Release); |
| 233 | } |
| 234 | return self.sq_ready(); |
| 235 | } |
| 236 | |
| 237 | /// Returns true if we are not using an SQ thread (thus nobody submits but us), |
| 238 | /// or if IORING_SQ_NEED_WAKEUP is set and the SQ thread must be explicitly awakened. |
| 239 | /// For the latter case, we set the SQ thread wakeup flag. |
| 240 | /// Matches the implementation of sq_ring_needs_enter() in liburing. |
| 241 | pub fn sq_ring_needs_enter(self: *IO_Uring, submitted: u32, flags: *u32) bool { |
| 242 | assert(flags.* == 0); |
| 243 | if ((self.flags & linux.IORING_SETUP_SQPOLL) == 0 and submitted > 0) return true; |
| 244 | if ((@atomicLoad(u32, self.sq.flags, .Unordered) & linux.IORING_SQ_NEED_WAKEUP) != 0) { |
| 245 | flags.* |= linux.IORING_ENTER_SQ_WAKEUP; |
| 246 | return true; |
| 247 | } |
| 248 | return false; |
| 249 | } |
| 250 | |
| 251 | /// Returns the number of flushed and unflushed SQEs pending in the submission queue. |
| 252 | /// In other words, this is the number of SQEs in the submission queue, i.e. its length. |
| 253 | /// These are SQEs that the kernel is yet to consume. |
| 254 | /// Matches the implementation of io_uring_sq_ready in liburing. |
| 255 | pub fn sq_ready(self: *IO_Uring) u32 { |
| 256 | // Always use the shared ring state (i.e. head and not sqe_head) to avoid going out of sync, |
| 257 | // see https://github.com/axboe/liburing/issues/92. |
| 258 | return self.sq.sqe_tail -% @atomicLoad(u32, self.sq.head, .Acquire); |
| 259 | } |
| 260 | |
| 261 | /// Returns the number of CQEs in the completion queue, i.e. its length. |
| 262 | /// These are CQEs that the application is yet to consume. |
| 263 | /// Matches the implementation of io_uring_cq_ready in liburing. |
| 264 | pub fn cq_ready(self: *IO_Uring) u32 { |
| 265 | return @atomicLoad(u32, self.cq.tail, .Acquire) -% self.cq.head.*; |
| 266 | } |
| 267 | |
| 268 | /// Copies as many CQEs as are ready, and that can fit into the destination `cqes` slice. |
| 269 | /// If none are available, enters into the kernel to wait for at most `wait_nr` CQEs. |
| 270 | /// Returns the number of CQEs copied, advancing the CQ ring. |
| 271 | /// Provides all the wait/peek methods found in liburing, but with batching and a single method. |
| 272 | /// The rationale for copying CQEs rather than copying pointers is that pointers are 8 bytes |
| 273 | /// whereas CQEs are not much more at only 16 bytes, and this provides a safer faster interface. |
| 274 | /// Safer, because you no longer need to call cqe_seen(), avoiding idempotency bugs. |
| 275 | /// Faster, because we can now amortize the atomic store release to `cq.head` across the batch. |
| 276 | /// See https://github.com/axboe/liburing/issues/103#issuecomment-686665007. |
| 277 | /// Matches the implementation of io_uring_peek_batch_cqe() in liburing, but supports waiting. |
| 278 | pub fn copy_cqes(self: *IO_Uring, cqes: []io_uring_cqe, wait_nr: u32) !u32 { |
| 279 | const count = self.copy_cqes_ready(cqes, wait_nr); |
| 280 | if (count > 0) return count; |
| 281 | if (self.cq_ring_needs_flush() or wait_nr > 0) { |
| 282 | _ = try self.enter(0, wait_nr, linux.IORING_ENTER_GETEVENTS); |
| 283 | return self.copy_cqes_ready(cqes, wait_nr); |
| 284 | } |
| 285 | return 0; |
| 286 | } |
| 287 | |
| 288 | fn copy_cqes_ready(self: *IO_Uring, cqes: []io_uring_cqe, wait_nr: u32) u32 { |
| 289 | const ready = self.cq_ready(); |
| 290 | const count = std.math.min(cqes.len, ready); |
| 291 | var head = self.cq.head.*; |
| 292 | var tail = head +% count; |
| 293 | // TODO Optimize this by using 1 or 2 memcpy's (if the tail wraps) rather than a loop. |
| 294 | var i: usize = 0; |
| 295 | // Do not use "less-than" operator since head and tail may wrap: |
| 296 | while (head != tail) { |
| 297 | cqes[i] = self.cq.cqes[head & self.cq.mask]; // Copy struct by value. |
| 298 | head +%= 1; |
| 299 | i += 1; |
| 300 | } |
| 301 | self.cq_advance(count); |
| 302 | return count; |
| 303 | } |
| 304 | |
| 305 | /// Returns a copy of an I/O completion, waiting for it if necessary, and advancing the CQ ring. |
| 306 | /// A convenience method for `copy_cqes()` for when you don't need to batch or peek. |
| 307 | pub fn copy_cqe(ring: *IO_Uring) !io_uring_cqe { |
| 308 | var cqes: [1]io_uring_cqe = undefined; |
| 309 | const count = try ring.copy_cqes(&cqes, 1); |
| 310 | assert(count == 1); |
| 311 | return cqes[0]; |
| 312 | } |
| 313 | |
| 314 | /// Matches the implementation of cq_ring_needs_flush() in liburing. |
| 315 | pub fn cq_ring_needs_flush(self: *IO_Uring) bool { |
| 316 | return (@atomicLoad(u32, self.sq.flags, .Unordered) & linux.IORING_SQ_CQ_OVERFLOW) != 0; |
| 317 | } |
| 318 | |
| 319 | /// For advanced use cases only that implement custom completion queue methods. |
| 320 | /// If you use copy_cqes() or copy_cqe() you must not call cqe_seen() or cq_advance(). |
| 321 | /// Must be called exactly once after a zero-copy CQE has been processed by your application. |
| 322 | /// Not idempotent, calling more than once will result in other CQEs being lost. |
| 323 | /// Matches the implementation of cqe_seen() in liburing. |
| 324 | pub fn cqe_seen(self: *IO_Uring, cqe: *io_uring_cqe) void { |
| 325 | self.cq_advance(1); |
| 326 | } |
| 327 | |
| 328 | /// For advanced use cases only that implement custom completion queue methods. |
| 329 | /// Matches the implementation of cq_advance() in liburing. |
| 330 | pub fn cq_advance(self: *IO_Uring, count: u32) void { |
| 331 | if (count > 0) { |
| 332 | // Ensure the kernel only sees the new head value after the CQEs have been read. |
| 333 | @atomicStore(u32, self.cq.head, self.cq.head.* +% count, .Release); |
| 334 | } |
| 335 | } |
| 336 | |
| 337 | /// Queues (but does not submit) an SQE to perform an `fsync(2)`. |
| 338 | /// Returns a pointer to the SQE so that you can further modify the SQE for advanced use cases. |
| 339 | /// For example, for `fdatasync()` you can set `IORING_FSYNC_DATASYNC` in the SQE's `rw_flags`. |
| 340 | /// N.B. While SQEs are initiated in the order in which they appear in the submission queue, |
| 341 | /// operations execute in parallel and completions are unordered. Therefore, an application that |
| 342 | /// submits a write followed by an fsync in the submission queue cannot expect the fsync to |
| 343 | /// apply to the write, since the fsync may complete before the write is issued to the disk. |
| 344 | /// You should preferably use `link_with_next_sqe()` on a write's SQE to link it with an fsync, |
| 345 | /// or else insert a full write barrier using `drain_previous_sqes()` when queueing an fsync. |
| 346 | pub fn fsync(self: *IO_Uring, user_data: u64, fd: os.fd_t, flags: u32) !*io_uring_sqe { |
| 347 | const sqe = try self.get_sqe(); |
| 348 | io_uring_prep_fsync(sqe, fd, flags); |
| 349 | sqe.user_data = user_data; |
| 350 | return sqe; |
| 351 | } |
| 352 | |
| 353 | /// Queues (but does not submit) an SQE to perform a no-op. |
| 354 | /// Returns a pointer to the SQE so that you can further modify the SQE for advanced use cases. |
| 355 | /// A no-op is more useful than may appear at first glance. |
| 356 | /// For example, you could call `drain_previous_sqes()` on the returned SQE, to use the no-op to |
| 357 | /// know when the ring is idle before acting on a kill signal. |
| 358 | pub fn nop(self: *IO_Uring, user_data: u64) !*io_uring_sqe { |
| 359 | const sqe = try self.get_sqe(); |
| 360 | io_uring_prep_nop(sqe); |
| 361 | sqe.user_data = user_data; |
| 362 | return sqe; |
| 363 | } |
| 364 | |
| 365 | /// Queues (but does not submit) an SQE to perform a `read(2)`. |
| 366 | /// Returns a pointer to the SQE. |
| 367 | pub fn read( |
| 368 | self: *IO_Uring, |
| 369 | user_data: u64, |
| 370 | fd: os.fd_t, |
| 371 | buffer: []u8, |
| 372 | offset: u64 |
| 373 | ) !*io_uring_sqe { |
| 374 | const sqe = try self.get_sqe(); |
| 375 | io_uring_prep_read(sqe, fd, buffer, offset); |
| 376 | sqe.user_data = user_data; |
| 377 | return sqe; |
| 378 | } |
| 379 | |
| 380 | /// Queues (but does not submit) an SQE to perform a `write(2)`. |
| 381 | /// Returns a pointer to the SQE. |
| 382 | pub fn write( |
| 383 | self: *IO_Uring, |
| 384 | user_data: u64, |
| 385 | fd: os.fd_t, |
| 386 | buffer: []const u8, |
| 387 | offset: u64 |
| 388 | ) !*io_uring_sqe { |
| 389 | const sqe = try self.get_sqe(); |
| 390 | io_uring_prep_write(sqe, fd, buffer, offset); |
| 391 | sqe.user_data = user_data; |
| 392 | return sqe; |
| 393 | } |
| 394 | |
| 395 | /// Queues (but does not submit) an SQE to perform a `preadv()`. |
| 396 | /// Returns a pointer to the SQE so that you can further modify the SQE for advanced use cases. |
| 397 | /// For example, if you want to do a `preadv2()` then set `rw_flags` on the returned SQE. |
| 398 | /// See https://linux.die.net/man/2/preadv. |
| 399 | pub fn readv( |
| 400 | self: *IO_Uring, |
| 401 | user_data: u64, |
| 402 | fd: os.fd_t, |
| 403 | iovecs: []const os.iovec, |
| 404 | offset: u64 |
| 405 | ) !*io_uring_sqe { |
| 406 | const sqe = try self.get_sqe(); |
| 407 | io_uring_prep_readv(sqe, fd, iovecs, offset); |
| 408 | sqe.user_data = user_data; |
| 409 | return sqe; |
| 410 | } |
| 411 | |
| 412 | /// Queues (but does not submit) an SQE to perform a `pwritev()`. |
| 413 | /// Returns a pointer to the SQE so that you can further modify the SQE for advanced use cases. |
| 414 | /// For example, if you want to do a `pwritev2()` then set `rw_flags` on the returned SQE. |
| 415 | /// See https://linux.die.net/man/2/pwritev. |
| 416 | pub fn writev( |
| 417 | self: *IO_Uring, |
| 418 | user_data: u64, |
| 419 | fd: os.fd_t, |
| 420 | iovecs: []const os.iovec_const, |
| 421 | offset: u64 |
| 422 | ) !*io_uring_sqe { |
| 423 | const sqe = try self.get_sqe(); |
| 424 | io_uring_prep_writev(sqe, fd, iovecs, offset); |
| 425 | sqe.user_data = user_data; |
| 426 | return sqe; |
| 427 | } |
| 428 | |
| 429 | /// Queues (but does not submit) an SQE to perform an `accept4(2)` on a socket. |
| 430 | /// Returns a pointer to the SQE. |
| 431 | pub fn accept( |
| 432 | self: *IO_Uring, |
| 433 | user_data: u64, |
| 434 | fd: os.fd_t, |
| 435 | addr: *os.sockaddr, |
| 436 | addrlen: *os.socklen_t, |
| 437 | flags: u32 |
| 438 | ) !*io_uring_sqe { |
| 439 | const sqe = try self.get_sqe(); |
| 440 | io_uring_prep_accept(sqe, fd, addr, addrlen, flags); |
| 441 | sqe.user_data = user_data; |
| 442 | return sqe; |
| 443 | } |
| 444 | |
| 445 | /// Queue (but does not submit) an SQE to perform a `connect(2)` on a socket. |
| 446 | /// Returns a pointer to the SQE. |
| 447 | pub fn connect( |
| 448 | self: *IO_Uring, |
| 449 | user_data: u64, |
| 450 | fd: os.fd_t, |
| 451 | addr: *const os.sockaddr, |
| 452 | addrlen: os.socklen_t |
| 453 | ) !*io_uring_sqe { |
| 454 | const sqe = try self.get_sqe(); |
| 455 | io_uring_prep_connect(sqe, fd, addr, addrlen); |
| 456 | sqe.user_data = user_data; |
| 457 | return sqe; |
| 458 | } |
| 459 | |
| 460 | /// Queues (but does not submit) an SQE to perform a `recv(2)`. |
| 461 | /// Returns a pointer to the SQE. |
| 462 | pub fn recv( |
| 463 | self: *IO_Uring, |
| 464 | user_data: u64, |
| 465 | fd: os.fd_t, |
| 466 | buffer: []u8, |
| 467 | flags: u32 |
| 468 | ) !*io_uring_sqe { |
| 469 | const sqe = try self.get_sqe(); |
| 470 | io_uring_prep_recv(sqe, fd, buffer, flags); |
| 471 | sqe.user_data = user_data; |
| 472 | return sqe; |
| 473 | } |
| 474 | |
| 475 | /// Queues (but does not submit) an SQE to perform a `send(2)`. |
| 476 | /// Returns a pointer to the SQE. |
| 477 | pub fn send( |
| 478 | self: *IO_Uring, |
| 479 | user_data: u64, |
| 480 | fd: os.fd_t, |
| 481 | buffer: []const u8, |
| 482 | flags: u32 |
| 483 | ) !*io_uring_sqe { |
| 484 | const sqe = try self.get_sqe(); |
| 485 | io_uring_prep_send(sqe, fd, buffer, flags); |
| 486 | sqe.user_data = user_data; |
| 487 | return sqe; |
| 488 | } |
| 489 | |
| 490 | /// Queues (but does not submit) an SQE to perform an `openat(2)`. |
| 491 | /// Returns a pointer to the SQE. |
| 492 | pub fn openat( |
| 493 | self: *IO_Uring, |
| 494 | user_data: u64, |
| 495 | fd: os.fd_t, |
| 496 | path: [*:0]const u8, |
| 497 | flags: u32, |
| 498 | mode: os.mode_t |
| 499 | ) !*io_uring_sqe { |
| 500 | const sqe = try self.get_sqe(); |
| 501 | io_uring_prep_openat(sqe, fd, path, flags, mode); |
| 502 | sqe.user_data = user_data; |
| 503 | return sqe; |
| 504 | } |
| 505 | |
| 506 | /// Queues (but does not submit) an SQE to perform a `close(2)`. |
| 507 | /// Returns a pointer to the SQE. |
| 508 | pub fn close(self: *IO_Uring, user_data: u64, fd: os.fd_t) !*io_uring_sqe { |
| 509 | const sqe = try self.get_sqe(); |
| 510 | io_uring_prep_close(sqe, fd); |
| 511 | sqe.user_data = user_data; |
| 512 | return sqe; |
| 513 | } |
| 514 | |
| 515 | /// Registers an array of file descriptors. |
| 516 | /// Every time a file descriptor is put in an SQE and submitted to the kernel, the kernel must |
| 517 | /// retrieve a reference to the file, and once I/O has completed the file reference must be |
| 518 | /// dropped. The atomic nature of this file reference can be a slowdown for high IOPS workloads. |
| 519 | /// This slowdown can be avoided by pre-registering file descriptors. |
| 520 | /// To refer to a registered file descriptor, IOSQE_FIXED_FILE must be set in the SQE's flags, |
| 521 | /// and the SQE's fd must be set to the index of the file descriptor in the registered array. |
| 522 | /// Registering file descriptors will wait for the ring to idle. |
| 523 | /// Files are automatically unregistered by the kernel when the ring is torn down. |
| 524 | /// An application need unregister only if it wants to register a new array of file descriptors. |
| 525 | pub fn register_files(self: *IO_Uring, fds: []const os.fd_t) !void { |
| 526 | assert(self.fd >= 0); |
| 527 | comptime assert(@sizeOf(os.fd_t) == @sizeOf(c_int)); |
| 528 | const res = linux.io_uring_register( |
| 529 | self.fd, |
| 530 | .REGISTER_FILES, |
| 531 | @ptrCast(*const c_void, fds.ptr), |
| 532 | @intCast(u32, fds.len) |
| 533 | ); |
| 534 | switch (linux.getErrno(res)) { |
| 535 | 0 => {}, |
| 536 | // One or more fds in the array are invalid, or the kernel does not support sparse sets: |
| 537 | linux.EBADF => return error.FileDescriptorInvalid, |
| 538 | linux.EBUSY => return error.FilesAlreadyRegistered, |
| 539 | linux.EINVAL => return error.FilesEmpty, |
| 540 | // Adding `nr_args` file references would exceed the maximum allowed number of files the |
| 541 | // user is allowed to have according to the per-user RLIMIT_NOFILE resource limit and |
| 542 | // the CAP_SYS_RESOURCE capability is not set, or `nr_args` exceeds the maximum allowed |
| 543 | // for a fixed file set (older kernels have a limit of 1024 files vs 64K files): |
| 544 | linux.EMFILE => return error.UserFdQuotaExceeded, |
| 545 | // Insufficient kernel resources, or the caller had a non-zero RLIMIT_MEMLOCK soft |
| 546 | // resource limit but tried to lock more memory than the limit permitted (not enforced |
| 547 | // when the process is privileged with CAP_IPC_LOCK): |
| 548 | linux.ENOMEM => return error.SystemResources, |
| 549 | // Attempt to register files on a ring already registering files or being torn down: |
| 550 | linux.ENXIO => return error.RingShuttingDownOrAlreadyRegisteringFiles, |
| 551 | else => |errno| return os.unexpectedErrno(errno) |
| 552 | } |
| 553 | } |
| 554 | |
| 555 | /// Unregisters all registered file descriptors previously associated with the ring. |
| 556 | pub fn unregister_files(self: *IO_Uring) !void { |
| 557 | assert(self.fd >= 0); |
| 558 | const res = linux.io_uring_register(self.fd, .UNREGISTER_FILES, null, 0); |
| 559 | switch (linux.getErrno(res)) { |
| 560 | 0 => {}, |
| 561 | linux.ENXIO => return error.FilesNotRegistered, |
| 562 | else => |errno| return os.unexpectedErrno(errno) |
| 563 | } |
| 564 | } |
| 565 | }; |
| 566 | |
| 567 | pub const SubmissionQueue = struct { |
| 568 | head: *u32, |
| 569 | tail: *u32, |
| 570 | mask: u32, |
| 571 | flags: *u32, |
| 572 | dropped: *u32, |
| 573 | array: []u32, |
| 574 | sqes: []io_uring_sqe, |
| 575 | mmap: []align(mem.page_size) u8, |
| 576 | mmap_sqes: []align(mem.page_size) u8, |
| 577 | |
| 578 | // We use `sqe_head` and `sqe_tail` in the same way as liburing: |
| 579 | // We increment `sqe_tail` (but not `tail`) for each call to `get_sqe()`. |
| 580 | // We then set `tail` to `sqe_tail` once, only when these events are actually submitted. |
| 581 | // This allows us to amortize the cost of the @atomicStore to `tail` across multiple SQEs. |
| 582 | sqe_head: u32 = 0, |
| 583 | sqe_tail: u32 = 0, |
| 584 | |
| 585 | pub fn init(fd: os.fd_t, p: io_uring_params) !SubmissionQueue { |
| 586 | assert(fd >= 0); |
| 587 | assert((p.features & linux.IORING_FEAT_SINGLE_MMAP) != 0); |
| 588 | const size = std.math.max( |
| 589 | p.sq_off.array + p.sq_entries * @sizeOf(u32), |
| 590 | p.cq_off.cqes + p.cq_entries * @sizeOf(io_uring_cqe) |
| 591 | ); |
| 592 | const mmap = try os.mmap( |
| 593 | null, |
| 594 | size, |
| 595 | os.PROT_READ | os.PROT_WRITE, |
| 596 | os.MAP_SHARED | os.MAP_POPULATE, |
| 597 | fd, |
| 598 | linux.IORING_OFF_SQ_RING, |
| 599 | ); |
| 600 | errdefer os.munmap(mmap); |
| 601 | assert(mmap.len == size); |
| 602 | |
| 603 | // The motivation for the `sqes` and `array` indirection is to make it possible for the |
| 604 | // application to preallocate static io_uring_sqe entries and then replay them when needed. |
| 605 | const size_sqes = p.sq_entries * @sizeOf(io_uring_sqe); |
| 606 | const mmap_sqes = try os.mmap( |
| 607 | null, |
| 608 | size_sqes, |
| 609 | os.PROT_READ | os.PROT_WRITE, |
| 610 | os.MAP_SHARED | os.MAP_POPULATE, |
| 611 | fd, |
| 612 | linux.IORING_OFF_SQES, |
| 613 | ); |
| 614 | errdefer os.munmap(mmap_sqes); |
| 615 | assert(mmap_sqes.len == size_sqes); |
| 616 | |
| 617 | const array = @ptrCast([*]u32, @alignCast(@alignOf(u32), &mmap[p.sq_off.array])); |
| 618 | const sqes = @ptrCast([*]io_uring_sqe, @alignCast(@alignOf(io_uring_sqe), &mmap_sqes[0])); |
| 619 | // We expect the kernel copies p.sq_entries to the u32 pointed to by p.sq_off.ring_entries, |
| 620 | // see https://github.com/torvalds/linux/blob/v5.8/fs/io_uring.c#L7843-L7844. |
| 621 | assert( |
| 622 | p.sq_entries == |
| 623 | @ptrCast(*u32, @alignCast(@alignOf(u32), &mmap[p.sq_off.ring_entries])).* |
| 624 | ); |
| 625 | return SubmissionQueue { |
| 626 | .head = @ptrCast(*u32, @alignCast(@alignOf(u32), &mmap[p.sq_off.head])), |
| 627 | .tail = @ptrCast(*u32, @alignCast(@alignOf(u32), &mmap[p.sq_off.tail])), |
| 628 | .mask = @ptrCast(*u32, @alignCast(@alignOf(u32), &mmap[p.sq_off.ring_mask])).*, |
| 629 | .flags = @ptrCast(*u32, @alignCast(@alignOf(u32), &mmap[p.sq_off.flags])), |
| 630 | .dropped = @ptrCast(*u32, @alignCast(@alignOf(u32), &mmap[p.sq_off.dropped])), |
| 631 | .array = array[0..p.sq_entries], |
| 632 | .sqes = sqes[0..p.sq_entries], |
| 633 | .mmap = mmap, |
| 634 | .mmap_sqes = mmap_sqes |
| 635 | }; |
| 636 | } |
| 637 | |
| 638 | pub fn deinit(self: *SubmissionQueue) void { |
| 639 | os.munmap(self.mmap_sqes); |
| 640 | os.munmap(self.mmap); |
| 641 | } |
| 642 | }; |
| 643 | |
| 644 | pub const CompletionQueue = struct { |
| 645 | head: *u32, |
| 646 | tail: *u32, |
| 647 | mask: u32, |
| 648 | overflow: *u32, |
| 649 | cqes: []io_uring_cqe, |
| 650 | |
| 651 | pub fn init(fd: os.fd_t, p: io_uring_params, sq: SubmissionQueue) !CompletionQueue { |
| 652 | assert(fd >= 0); |
| 653 | assert((p.features & linux.IORING_FEAT_SINGLE_MMAP) != 0); |
| 654 | const mmap = sq.mmap; |
| 655 | const cqes = @ptrCast( |
| 656 | [*]io_uring_cqe, |
| 657 | @alignCast(@alignOf(io_uring_cqe), &mmap[p.cq_off.cqes]) |
| 658 | ); |
| 659 | assert( |
| 660 | p.cq_entries == |
| 661 | @ptrCast(*u32, @alignCast(@alignOf(u32), &mmap[p.cq_off.ring_entries])).* |
| 662 | ); |
| 663 | return CompletionQueue { |
| 664 | .head = @ptrCast(*u32, @alignCast(@alignOf(u32), &mmap[p.cq_off.head])), |
| 665 | .tail = @ptrCast(*u32, @alignCast(@alignOf(u32), &mmap[p.cq_off.tail])), |
| 666 | .mask = @ptrCast(*u32, @alignCast(@alignOf(u32), &mmap[p.cq_off.ring_mask])).*, |
| 667 | .overflow = @ptrCast(*u32, @alignCast(@alignOf(u32), &mmap[p.cq_off.overflow])), |
| 668 | .cqes = cqes[0..p.cq_entries] |
| 669 | }; |
| 670 | } |
| 671 | |
| 672 | pub fn deinit(self: *CompletionQueue) void { |
| 673 | // A no-op since we now share the mmap with the submission queue. |
| 674 | // Here for symmetry with the submission queue, and for any future feature support. |
| 675 | } |
| 676 | }; |
| 677 | |
| 678 | pub fn io_uring_prep_nop(sqe: *io_uring_sqe) void { |
| 679 | sqe.* = .{ |
| 680 | .opcode = .NOP, |
| 681 | .flags = 0, |
| 682 | .ioprio = 0, |
| 683 | .fd = 0, |
| 684 | .off = 0, |
| 685 | .addr = 0, |
| 686 | .len = 0, |
| 687 | .rw_flags = 0, |
| 688 | .user_data = 0, |
| 689 | .buf_index = 0, |
| 690 | .personality = 0, |
| 691 | .splice_fd_in = 0, |
| 692 | .__pad2 = [2]u64{ 0, 0 } |
| 693 | }; |
| 694 | } |
| 695 | |
| 696 | pub fn io_uring_prep_fsync(sqe: *io_uring_sqe, fd: os.fd_t, flags: u32) void { |
| 697 | sqe.* = .{ |
| 698 | .opcode = .FSYNC, |
| 699 | .flags = 0, |
| 700 | .ioprio = 0, |
| 701 | .fd = fd, |
| 702 | .off = 0, |
| 703 | .addr = 0, |
| 704 | .len = 0, |
| 705 | .rw_flags = flags, |
| 706 | .user_data = 0, |
| 707 | .buf_index = 0, |
| 708 | .personality = 0, |
| 709 | .splice_fd_in = 0, |
| 710 | .__pad2 = [2]u64{ 0, 0 } |
| 711 | }; |
| 712 | } |
| 713 | |
| 714 | pub fn io_uring_prep_rw( |
| 715 | op: linux.IORING_OP, |
| 716 | sqe: *io_uring_sqe, |
| 717 | fd: os.fd_t, |
| 718 | addr: anytype, |
| 719 | len: usize, |
| 720 | offset: u64 |
| 721 | ) void { |
| 722 | sqe.* = .{ |
| 723 | .opcode = op, |
| 724 | .flags = 0, |
| 725 | .ioprio = 0, |
| 726 | .fd = fd, |
| 727 | .off = offset, |
| 728 | .addr = @ptrToInt(addr), |
| 729 | .len = @intCast(u32, len), |
| 730 | .rw_flags = 0, |
| 731 | .user_data = 0, |
| 732 | .buf_index = 0, |
| 733 | .personality = 0, |
| 734 | .splice_fd_in = 0, |
| 735 | .__pad2 = [2]u64{ 0, 0 } |
| 736 | }; |
| 737 | } |
| 738 | |
| 739 | pub fn io_uring_prep_read(sqe: *io_uring_sqe, fd: os.fd_t, buffer: []u8, offset: u64) void { |
| 740 | io_uring_prep_rw(.READ, sqe, fd, buffer.ptr, buffer.len, offset); |
| 741 | } |
| 742 | |
| 743 | pub fn io_uring_prep_write(sqe: *io_uring_sqe, fd: os.fd_t, buffer: []const u8, offset: u64) void { |
| 744 | io_uring_prep_rw(.WRITE, sqe, fd, buffer.ptr, buffer.len, offset); |
| 745 | } |
| 746 | |
| 747 | pub fn io_uring_prep_readv( |
| 748 | sqe: *io_uring_sqe, |
| 749 | fd: os.fd_t, |
| 750 | iovecs: []const os.iovec, |
| 751 | offset: u64 |
| 752 | ) void { |
| 753 | io_uring_prep_rw(.READV, sqe, fd, iovecs.ptr, iovecs.len, offset); |
| 754 | } |
| 755 | |
| 756 | pub fn io_uring_prep_writev( |
| 757 | sqe: *io_uring_sqe, |
| 758 | fd: os.fd_t, |
| 759 | iovecs: []const os.iovec_const, |
| 760 | offset: u64 |
| 761 | ) void { |
| 762 | io_uring_prep_rw(.WRITEV, sqe, fd, iovecs.ptr, iovecs.len, offset); |
| 763 | } |
| 764 | |
| 765 | pub fn io_uring_prep_accept( |
| 766 | sqe: *io_uring_sqe, |
| 767 | fd: os.fd_t, |
| 768 | addr: *os.sockaddr, |
| 769 | addrlen: *os.socklen_t, |
| 770 | flags: u32 |
| 771 | ) void { |
| 772 | // `addr` holds a pointer to `sockaddr`, and `addr2` holds a pointer to socklen_t`. |
| 773 | // `addr2` maps to `sqe.off` (u64) instead of `sqe.len` (which is only a u32). |
| 774 | io_uring_prep_rw(.ACCEPT, sqe, fd, addr, 0, @ptrToInt(addrlen)); |
| 775 | sqe.rw_flags = flags; |
| 776 | } |
| 777 | |
| 778 | pub fn io_uring_prep_connect( |
| 779 | sqe: *io_uring_sqe, |
| 780 | fd: os.fd_t, |
| 781 | addr: *const os.sockaddr, |
| 782 | addrlen: os.socklen_t |
| 783 | ) void { |
| 784 | // `addrlen` maps to `sqe.off` (u64) instead of `sqe.len` (which is only a u32). |
| 785 | io_uring_prep_rw(.CONNECT, sqe, fd, addr, 0, addrlen); |
| 786 | } |
| 787 | |
| 788 | pub fn io_uring_prep_recv(sqe: *io_uring_sqe, fd: os.fd_t, buffer: []u8, flags: u32) void { |
| 789 | io_uring_prep_rw(.RECV, sqe, fd, buffer.ptr, buffer.len, 0); |
| 790 | sqe.rw_flags = flags; |
| 791 | } |
| 792 | |
| 793 | pub fn io_uring_prep_send(sqe: *io_uring_sqe, fd: os.fd_t, buffer: []const u8, flags: u32) void { |
| 794 | io_uring_prep_rw(.SEND, sqe, fd, buffer.ptr, buffer.len, 0); |
| 795 | sqe.rw_flags = flags; |
| 796 | } |
| 797 | |
| 798 | pub fn io_uring_prep_openat( |
| 799 | sqe: *io_uring_sqe, |
| 800 | fd: os.fd_t, |
| 801 | path: [*:0]const u8, |
| 802 | flags: u32, |
| 803 | mode: os.mode_t |
| 804 | ) void { |
| 805 | io_uring_prep_rw(.OPENAT, sqe, fd, path, mode, 0); |
| 806 | sqe.rw_flags = flags; |
| 807 | } |
| 808 | |
| 809 | pub fn io_uring_prep_close(sqe: *io_uring_sqe, fd: os.fd_t) void { |
| 810 | sqe.* = .{ |
| 811 | .opcode = .CLOSE, |
| 812 | .flags = 0, |
| 813 | .ioprio = 0, |
| 814 | .fd = fd, |
| 815 | .off = 0, |
| 816 | .addr = 0, |
| 817 | .len = 0, |
| 818 | .rw_flags = 0, |
| 819 | .user_data = 0, |
| 820 | .buf_index = 0, |
| 821 | .personality = 0, |
| 822 | .splice_fd_in = 0, |
| 823 | .__pad2 = [2]u64{ 0, 0 } |
| 824 | }; |
| 825 | } |
| 826 | |
| 827 | test "structs/offsets/entries" { |
| 828 | if (builtin.os.tag != .linux) return error.SkipZigTest; |
| 829 | |
| 830 | testing.expectEqual(@as(usize, 120), @sizeOf(io_uring_params)); |
| 831 | testing.expectEqual(@as(usize, 64), @sizeOf(io_uring_sqe)); |
| 832 | testing.expectEqual(@as(usize, 16), @sizeOf(io_uring_cqe)); |
| 833 | |
| 834 | testing.expectEqual(0, linux.IORING_OFF_SQ_RING); |
| 835 | testing.expectEqual(0x8000000, linux.IORING_OFF_CQ_RING); |
| 836 | testing.expectEqual(0x10000000, linux.IORING_OFF_SQES); |
| 837 | |
| 838 | testing.expectError(error.EntriesZero, IO_Uring.init(0, 0)); |
| 839 | testing.expectError(error.EntriesNotPowerOfTwo, IO_Uring.init(3, 0)); |
| 840 | } |
| 841 | |
| 842 | test "nop" { |
| 843 | if (builtin.os.tag != .linux) return error.SkipZigTest; |
| 844 | |
| 845 | var ring = IO_Uring.init(1, 0) catch |err| switch (err) { |
| 846 | error.SystemOutdated => return error.SkipZigTest, |
| 847 | error.PermissionDenied => return error.SkipZigTest, |
| 848 | else => return err |
| 849 | }; |
| 850 | defer { |
| 851 | ring.deinit(); |
| 852 | testing.expectEqual(@as(os.fd_t, -1), ring.fd); |
| 853 | } |
| 854 | |
| 855 | const sqe = try ring.nop(0xaaaaaaaa); |
| 856 | testing.expectEqual(io_uring_sqe { |
| 857 | .opcode = .NOP, |
| 858 | .flags = 0, |
| 859 | .ioprio = 0, |
| 860 | .fd = 0, |
| 861 | .off = 0, |
| 862 | .addr = 0, |
| 863 | .len = 0, |
| 864 | .rw_flags = 0, |
| 865 | .user_data = 0xaaaaaaaa, |
| 866 | .buf_index = 0, |
| 867 | .personality = 0, |
| 868 | .splice_fd_in = 0, |
| 869 | .__pad2 = [2]u64{ 0, 0 } |
| 870 | }, sqe.*); |
| 871 | |
| 872 | testing.expectEqual(@as(u32, 0), ring.sq.sqe_head); |
| 873 | testing.expectEqual(@as(u32, 1), ring.sq.sqe_tail); |
| 874 | testing.expectEqual(@as(u32, 0), ring.sq.tail.*); |
| 875 | testing.expectEqual(@as(u32, 0), ring.cq.head.*); |
| 876 | testing.expectEqual(@as(u32, 1), ring.sq_ready()); |
| 877 | testing.expectEqual(@as(u32, 0), ring.cq_ready()); |
| 878 | |
| 879 | testing.expectEqual(@as(u32, 1), try ring.submit()); |
| 880 | testing.expectEqual(@as(u32, 1), ring.sq.sqe_head); |
| 881 | testing.expectEqual(@as(u32, 1), ring.sq.sqe_tail); |
| 882 | testing.expectEqual(@as(u32, 1), ring.sq.tail.*); |
| 883 | testing.expectEqual(@as(u32, 0), ring.cq.head.*); |
| 884 | testing.expectEqual(@as(u32, 0), ring.sq_ready()); |
| 885 | |
| 886 | testing.expectEqual(io_uring_cqe { |
| 887 | .user_data = 0xaaaaaaaa, |
| 888 | .res = 0, |
| 889 | .flags = 0 |
| 890 | }, try ring.copy_cqe()); |
| 891 | testing.expectEqual(@as(u32, 1), ring.cq.head.*); |
| 892 | testing.expectEqual(@as(u32, 0), ring.cq_ready()); |
| 893 | |
| 894 | const sqe_barrier = try ring.nop(0xbbbbbbbb); |
| 895 | sqe_barrier.flags |= linux.IOSQE_IO_DRAIN; |
| 896 | testing.expectEqual(@as(u32, 1), try ring.submit()); |
| 897 | testing.expectEqual(io_uring_cqe { |
| 898 | .user_data = 0xbbbbbbbb, |
| 899 | .res = 0, |
| 900 | .flags = 0 |
| 901 | }, try ring.copy_cqe()); |
| 902 | testing.expectEqual(@as(u32, 2), ring.sq.sqe_head); |
| 903 | testing.expectEqual(@as(u32, 2), ring.sq.sqe_tail); |
| 904 | testing.expectEqual(@as(u32, 2), ring.sq.tail.*); |
| 905 | testing.expectEqual(@as(u32, 2), ring.cq.head.*); |
| 906 | } |
| 907 | |
| 908 | test "readv" { |
| 909 | if (builtin.os.tag != .linux) return error.SkipZigTest; |
| 910 | |
| 911 | var ring = IO_Uring.init(1, 0) catch |err| switch (err) { |
| 912 | error.SystemOutdated => return error.SkipZigTest, |
| 913 | error.PermissionDenied => return error.SkipZigTest, |
| 914 | else => return err |
| 915 | }; |
| 916 | defer ring.deinit(); |
| 917 | |
| 918 | const fd = try os.openZ("/dev/zero", os.O_RDONLY | os.O_CLOEXEC, 0); |
| 919 | defer os.close(fd); |
| 920 | |
| 921 | // Linux Kernel 5.4 supports IORING_REGISTER_FILES but not sparse fd sets (i.e. an fd of -1). |
| 922 | // Linux Kernel 5.5 adds support for sparse fd sets. |
| 923 | // Compare: |
| 924 | // https://github.com/torvalds/linux/blob/v5.4/fs/io_uring.c#L3119-L3124 vs |
| 925 | // https://github.com/torvalds/linux/blob/v5.8/fs/io_uring.c#L6687-L6691 |
| 926 | // We therefore avoid stressing sparse fd sets here: |
| 927 | var registered_fds = [_]os.fd_t{0} ** 1; |
| 928 | const fd_index = 0; |
| 929 | registered_fds[fd_index] = fd; |
| 930 | try ring.register_files(registered_fds[0..]); |
| 931 | |
| 932 | var buffer = [_]u8{42} ** 128; |
| 933 | var iovecs = [_]os.iovec{ os.iovec { .iov_base = &buffer, .iov_len = buffer.len } }; |
| 934 | const sqe = try ring.readv(0xcccccccc, fd_index, iovecs[0..], 0); |
| 935 | testing.expectEqual(linux.IORING_OP.READV, sqe.opcode); |
| 936 | sqe.flags |= linux.IOSQE_FIXED_FILE; |
| 937 | |
| 938 | testing.expectError(error.SubmissionQueueFull, ring.nop(0)); |
| 939 | testing.expectEqual(@as(u32, 1), try ring.submit()); |
| 940 | testing.expectEqual(linux.io_uring_cqe { |
| 941 | .user_data = 0xcccccccc, |
| 942 | .res = buffer.len, |
| 943 | .flags = 0, |
| 944 | }, try ring.copy_cqe()); |
| 945 | testing.expectEqualSlices(u8, &([_]u8{0} ** buffer.len), buffer[0..]); |
| 946 | |
| 947 | try ring.unregister_files(); |
| 948 | } |
| 949 | |
| 950 | test "writev/fsync/readv" { |
| 951 | if (builtin.os.tag != .linux) return error.SkipZigTest; |
| 952 | |
| 953 | var ring = IO_Uring.init(4, 0) catch |err| switch (err) { |
| 954 | error.SystemOutdated => return error.SkipZigTest, |
| 955 | error.PermissionDenied => return error.SkipZigTest, |
| 956 | else => return err |
| 957 | }; |
| 958 | defer ring.deinit(); |
| 959 | |
| 960 | const path = "test_io_uring_writev_fsync_readv"; |
| 961 | const file = try std.fs.cwd().createFile(path, .{ .read = true, .truncate = true }); |
| 962 | defer file.close(); |
| 963 | defer std.fs.cwd().deleteFile(path) catch {}; |
| 964 | const fd = file.handle; |
| 965 | |
| 966 | const buffer_write = [_]u8{42} ** 128; |
| 967 | const iovecs_write = [_]os.iovec_const { |
| 968 | os.iovec_const { .iov_base = &buffer_write, .iov_len = buffer_write.len } |
| 969 | }; |
| 970 | var buffer_read = [_]u8{0} ** 128; |
| 971 | var iovecs_read = [_]os.iovec { |
| 972 | os.iovec { .iov_base = &buffer_read, .iov_len = buffer_read.len } |
| 973 | }; |
| 974 | |
| 975 | const sqe_writev = try ring.writev(0xdddddddd, fd, iovecs_write[0..], 17); |
| 976 | testing.expectEqual(linux.IORING_OP.WRITEV, sqe_writev.opcode); |
| 977 | testing.expectEqual(@as(u64, 17), sqe_writev.off); |
| 978 | sqe_writev.flags |= linux.IOSQE_IO_LINK; |
| 979 | |
| 980 | const sqe_fsync = try ring.fsync(0xeeeeeeee, fd, 0); |
| 981 | testing.expectEqual(linux.IORING_OP.FSYNC, sqe_fsync.opcode); |
| 982 | testing.expectEqual(fd, sqe_fsync.fd); |
| 983 | sqe_fsync.flags |= linux.IOSQE_IO_LINK; |
| 984 | |
| 985 | const sqe_readv = try ring.readv(0xffffffff, fd, iovecs_read[0..], 17); |
| 986 | testing.expectEqual(linux.IORING_OP.READV, sqe_readv.opcode); |
| 987 | testing.expectEqual(@as(u64, 17), sqe_readv.off); |
| 988 | |
| 989 | testing.expectEqual(@as(u32, 3), ring.sq_ready()); |
| 990 | testing.expectEqual(@as(u32, 3), try ring.submit_and_wait(3)); |
| 991 | testing.expectEqual(@as(u32, 0), ring.sq_ready()); |
| 992 | testing.expectEqual(@as(u32, 3), ring.cq_ready()); |
| 993 | |
| 994 | testing.expectEqual(linux.io_uring_cqe { |
| 995 | .user_data = 0xdddddddd, |
| 996 | .res = buffer_write.len, |
| 997 | .flags = 0, |
| 998 | }, try ring.copy_cqe()); |
| 999 | testing.expectEqual(@as(u32, 2), ring.cq_ready()); |
| 1000 | |
| 1001 | testing.expectEqual(linux.io_uring_cqe { |
| 1002 | .user_data = 0xeeeeeeee, |
| 1003 | .res = 0, |
| 1004 | .flags = 0, |
| 1005 | }, try ring.copy_cqe()); |
| 1006 | testing.expectEqual(@as(u32, 1), ring.cq_ready()); |
| 1007 | |
| 1008 | testing.expectEqual(linux.io_uring_cqe { |
| 1009 | .user_data = 0xffffffff, |
| 1010 | .res = buffer_read.len, |
| 1011 | .flags = 0, |
| 1012 | }, try ring.copy_cqe()); |
| 1013 | testing.expectEqual(@as(u32, 0), ring.cq_ready()); |
| 1014 | |
| 1015 | testing.expectEqualSlices(u8, buffer_write[0..], buffer_read[0..]); |
| 1016 | } |
| 1017 | |
| 1018 | test "write/read" { |
| 1019 | if (builtin.os.tag != .linux) return error.SkipZigTest; |
| 1020 | |
| 1021 | var ring = IO_Uring.init(2, 0) catch |err| switch (err) { |
| 1022 | error.SystemOutdated => return error.SkipZigTest, |
| 1023 | error.PermissionDenied => return error.SkipZigTest, |
| 1024 | else => return err |
| 1025 | }; |
| 1026 | defer ring.deinit(); |
| 1027 | |
| 1028 | const path = "test_io_uring_write_read"; |
| 1029 | const file = try std.fs.cwd().createFile(path, .{ .read = true, .truncate = true }); |
| 1030 | defer file.close(); |
| 1031 | defer std.fs.cwd().deleteFile(path) catch {}; |
| 1032 | const fd = file.handle; |
| 1033 | |
| 1034 | const buffer_write = [_]u8{97} ** 20; |
| 1035 | var buffer_read = [_]u8{98} ** 20; |
| 1036 | const sqe_write = try ring.write(0x11111111, fd, buffer_write[0..], 10); |
| 1037 | testing.expectEqual(linux.IORING_OP.WRITE, sqe_write.opcode); |
| 1038 | testing.expectEqual(@as(u64, 10), sqe_write.off); |
| 1039 | sqe_write.flags |= linux.IOSQE_IO_LINK; |
| 1040 | const sqe_read = try ring.read(0x22222222, fd, buffer_read[0..], 10); |
| 1041 | testing.expectEqual(linux.IORING_OP.READ, sqe_read.opcode); |
| 1042 | testing.expectEqual(@as(u64, 10), sqe_read.off); |
| 1043 | testing.expectEqual(@as(u32, 2), try ring.submit()); |
| 1044 | |
| 1045 | const cqe_write = try ring.copy_cqe(); |
| 1046 | const cqe_read = try ring.copy_cqe(); |
| 1047 | // Prior to Linux Kernel 5.6 this is the only way to test for read/write support: |
| 1048 | // https://lwn.net/Articles/809820/ |
| 1049 | if (cqe_write.res == -linux.EINVAL) return error.SkipZigTest; |
| 1050 | if (cqe_read.res == -linux.EINVAL) return error.SkipZigTest; |
| 1051 | testing.expectEqual(linux.io_uring_cqe { |
| 1052 | .user_data = 0x11111111, |
| 1053 | .res = buffer_write.len, |
| 1054 | .flags = 0, |
| 1055 | }, cqe_write); |
| 1056 | testing.expectEqual(linux.io_uring_cqe { |
| 1057 | .user_data = 0x22222222, |
| 1058 | .res = buffer_read.len, |
| 1059 | .flags = 0, |
| 1060 | }, cqe_read); |
| 1061 | testing.expectEqualSlices(u8, buffer_write[0..], buffer_read[0..]); |
| 1062 | } |
| 1063 | |
| 1064 | test "openat" { |
| 1065 | if (builtin.os.tag != .linux) return error.SkipZigTest; |
| 1066 | |
| 1067 | var ring = IO_Uring.init(1, 0) catch |err| switch (err) { |
| 1068 | error.SystemOutdated => return error.SkipZigTest, |
| 1069 | error.PermissionDenied => return error.SkipZigTest, |
| 1070 | else => return err |
| 1071 | }; |
| 1072 | defer ring.deinit(); |
| 1073 | |
| 1074 | const path = "test_io_uring_openat"; |
| 1075 | defer std.fs.cwd().deleteFile(path) catch {}; |
| 1076 | |
| 1077 | const flags: u32 = os.O_CLOEXEC | os.O_RDWR | os.O_CREAT; |
| 1078 | const mode: os.mode_t = 0o666; |
| 1079 | const sqe_openat = try ring.openat(0x33333333, linux.AT_FDCWD, path, flags, mode); |
| 1080 | testing.expectEqual(io_uring_sqe { |
| 1081 | .opcode = .OPENAT, |
| 1082 | .flags = 0, |
| 1083 | .ioprio = 0, |
| 1084 | .fd = linux.AT_FDCWD, |
| 1085 | .off = 0, |
| 1086 | .addr = @ptrToInt(path), |
| 1087 | .len = mode, |
| 1088 | .rw_flags = flags, |
| 1089 | .user_data = 0x33333333, |
| 1090 | .buf_index = 0, |
| 1091 | .personality = 0, |
| 1092 | .splice_fd_in = 0, |
| 1093 | .__pad2 = [2]u64{ 0, 0 } |
| 1094 | }, sqe_openat.*); |
| 1095 | testing.expectEqual(@as(u32, 1), try ring.submit()); |
| 1096 | |
| 1097 | const cqe_openat = try ring.copy_cqe(); |
| 1098 | testing.expectEqual(@as(u64, 0x33333333), cqe_openat.user_data); |
| 1099 | if (cqe_openat.res == -linux.EINVAL) return error.SkipZigTest; |
| 1100 | // AT_FDCWD is not fully supported before kernel 5.6: |
| 1101 | // See https://lore.kernel.org/io-uring/20200207155039.12819-1-axboe@kernel.dk/T/ |
| 1102 | // We use IORING_FEAT_RW_CUR_POS to know if we are pre-5.6 since that feature was added in 5.6. |
| 1103 | if (cqe_openat.res == -linux.EBADF and (ring.features & linux.IORING_FEAT_RW_CUR_POS) == 0) { |
| 1104 | return error.SkipZigTest; |
| 1105 | } |
| 1106 | if (cqe_openat.res <= 0) std.debug.print("\ncqe_openat.res={}\n", .{ cqe_openat.res }); |
| 1107 | testing.expect(cqe_openat.res > 0); |
| 1108 | testing.expectEqual(@as(u32, 0), cqe_openat.flags); |
| 1109 | |
| 1110 | os.close(cqe_openat.res); |
| 1111 | } |
| 1112 | |
| 1113 | test "close" { |
| 1114 | if (builtin.os.tag != .linux) return error.SkipZigTest; |
| 1115 | |
| 1116 | var ring = IO_Uring.init(1, 0) catch |err| switch (err) { |
| 1117 | error.SystemOutdated => return error.SkipZigTest, |
| 1118 | error.PermissionDenied => return error.SkipZigTest, |
| 1119 | else => return err |
| 1120 | }; |
| 1121 | defer ring.deinit(); |
| 1122 | |
| 1123 | const path = "test_io_uring_close"; |
| 1124 | const file = try std.fs.cwd().createFile(path, .{}); |
| 1125 | errdefer file.close(); |
| 1126 | defer std.fs.cwd().deleteFile(path) catch {}; |
| 1127 | |
| 1128 | const sqe_close = try ring.close(0x44444444, file.handle); |
| 1129 | testing.expectEqual(linux.IORING_OP.CLOSE, sqe_close.opcode); |
| 1130 | testing.expectEqual(file.handle, sqe_close.fd); |
| 1131 | testing.expectEqual(@as(u32, 1), try ring.submit()); |
| 1132 | |
| 1133 | const cqe_close = try ring.copy_cqe(); |
| 1134 | if (cqe_close.res == -linux.EINVAL) return error.SkipZigTest; |
| 1135 | testing.expectEqual(linux.io_uring_cqe { |
| 1136 | .user_data = 0x44444444, |
| 1137 | .res = 0, |
| 1138 | .flags = 0, |
| 1139 | }, cqe_close); |
| 1140 | } |
| 1141 | |
| 1142 | test "accept/connect/send/recv" { |
| 1143 | if (builtin.os.tag != .linux) return error.SkipZigTest; |
| 1144 | |
| 1145 | var ring = IO_Uring.init(16, 0) catch |err| switch (err) { |
| 1146 | error.SystemOutdated => return error.SkipZigTest, |
| 1147 | error.PermissionDenied => return error.SkipZigTest, |
| 1148 | else => return err |
| 1149 | }; |
| 1150 | defer ring.deinit(); |
| 1151 | |
| 1152 | const address = try net.Address.parseIp4("127.0.0.1", 3131); |
| 1153 | const kernel_backlog = 1; |
| 1154 | const server = try os.socket(address.any.family, os.SOCK_STREAM | os.SOCK_CLOEXEC, 0); |
| 1155 | defer os.close(server); |
| 1156 | try os.setsockopt(server, os.SOL_SOCKET, os.SO_REUSEADDR, &mem.toBytes(@as(c_int, 1))); |
| 1157 | try os.bind(server, &address.any, address.getOsSockLen()); |
| 1158 | try os.listen(server, kernel_backlog); |
| 1159 | |
| 1160 | const buffer_send = [_]u8{ 1,0,1,0,1,0,1,0,1,0 }; |
| 1161 | var buffer_recv = [_]u8{ 0,1,0,1,0 }; |
| 1162 | |
| 1163 | var accept_addr: os.sockaddr = undefined; |
| 1164 | var accept_addr_len: os.socklen_t = @sizeOf(@TypeOf(accept_addr)); |
| 1165 | const accept = try ring.accept(0xaaaaaaaa, server, &accept_addr, &accept_addr_len, 0); |
| 1166 | testing.expectEqual(@as(u32, 1), try ring.submit()); |
| 1167 | |
| 1168 | const client = try os.socket(address.any.family, os.SOCK_STREAM | os.SOCK_CLOEXEC, 0); |
| 1169 | defer os.close(client); |
| 1170 | const connect = try ring.connect(0xcccccccc, client, &address.any, address.getOsSockLen()); |
| 1171 | testing.expectEqual(@as(u32, 1), try ring.submit()); |
| 1172 | |
| 1173 | var cqe_accept = try ring.copy_cqe(); |
| 1174 | if (cqe_accept.res == -linux.EINVAL) return error.SkipZigTest; |
| 1175 | var cqe_connect = try ring.copy_cqe(); |
| 1176 | if (cqe_connect.res == -linux.EINVAL) return error.SkipZigTest; |
| 1177 | |
| 1178 | // The accept/connect CQEs may arrive in any order, the connect CQE will sometimes come first: |
| 1179 | if (cqe_accept.user_data == 0xcccccccc and cqe_connect.user_data == 0xaaaaaaaa) { |
| 1180 | const a = cqe_accept; |
| 1181 | const b = cqe_connect; |
| 1182 | cqe_accept = b; |
| 1183 | cqe_connect = a; |
| 1184 | } |
| 1185 | |
| 1186 | testing.expectEqual(@as(u64, 0xaaaaaaaa), cqe_accept.user_data); |
| 1187 | if (cqe_accept.res <= 0) std.debug.print("\ncqe_accept.res={}\n", .{ cqe_accept.res }); |
| 1188 | testing.expect(cqe_accept.res > 0); |
| 1189 | testing.expectEqual(@as(u32, 0), cqe_accept.flags); |
| 1190 | testing.expectEqual(linux.io_uring_cqe { |
| 1191 | .user_data = 0xcccccccc, |
| 1192 | .res = 0, |
| 1193 | .flags = 0, |
| 1194 | }, cqe_connect); |
| 1195 | |
| 1196 | const send = try ring.send(0xeeeeeeee, client, buffer_send[0..], 0); |
| 1197 | send.flags |= linux.IOSQE_IO_LINK; |
| 1198 | const recv = try ring.recv(0xffffffff, cqe_accept.res, buffer_recv[0..], 0); |
| 1199 | testing.expectEqual(@as(u32, 2), try ring.submit()); |
| 1200 | |
| 1201 | const cqe_send = try ring.copy_cqe(); |
| 1202 | if (cqe_send.res == -linux.EINVAL) return error.SkipZigTest; |
| 1203 | testing.expectEqual(linux.io_uring_cqe { |
| 1204 | .user_data = 0xeeeeeeee, |
| 1205 | .res = buffer_send.len, |
| 1206 | .flags = 0, |
| 1207 | }, cqe_send); |
| 1208 | |
| 1209 | const cqe_recv = try ring.copy_cqe(); |
| 1210 | if (cqe_recv.res == -linux.EINVAL) return error.SkipZigTest; |
| 1211 | testing.expectEqual(linux.io_uring_cqe { |
| 1212 | .user_data = 0xffffffff, |
| 1213 | .res = buffer_recv.len, |
| 1214 | .flags = 0, |
| 1215 | }, cqe_recv); |
| 1216 | |
| 1217 | testing.expectEqualSlices(u8, buffer_send[0..buffer_recv.len], buffer_recv[0..]); |
| 1218 | } |
Andrew Kelley <andrew@ziglang.org> 2020-10-29 18:20:38-04:00
GitHub <noreply@github.com> 2020-10-29 18:20:38-04:00
Signed by PGP key 4AEE18F83AFDEB23
lib