authorgravatar for joran@ronomon.comJoran Dirk Greef <joran@ronomon.com> 2020-09-16 15:07:13+02:00
committergravatar for joran@ronomon.comJoran Dirk Greef <joran@ronomon.com> 2020-09-16 15:07:13+02:00
logb340bbfc1d0fd52d94799fa8d3b016178b46d6ba
treefa8b75011eff57f6e9a1ef60f7c0349f80cb98f6
parent281fc10ec5aa8052490b9f951e0ceed1b7008ff4

std: add io_uring library

This brings io_uring helper methods to Zig for kernels >= 5.4. We follow liburing's design decisions so that anyone who is comfortable with liburing (https://unixism.net/loti/ref-liburing/index.html) will feel at home. Thanks to @daurnimator for the first draft. Refs: https://github.com/ziglang/zig/pull/3083 Signed-off-by: Joran Dirk Greef <joran@coil.com>

2 files changed, 827 insertions(+), 0 deletions(-)

lib/std/io_uring.zig created+826
...@@ -0,0 +1,826 @@
1const builtin = @import("builtin");
2const std = @import("std");
3const assert = std.debug.assert;
4const os = std.os;
5const linux = os.linux;
6const mem = std.mem;
7const net = std.net;
8const testing = std.testing;
9
10pub const io_uring_params = linux.io_uring_params;
11pub const io_uring_cqe = linux.io_uring_cqe;
12
13// TODO Update linux.zig's definition of linux.io_uring_sqe:
14// linux.io_uring_sqe uses numbered unions, i.e. `union1` etc. that are not future-proof and need to
15// be re-numbered whenever new unions are interposed by the kernel. Furthermore, Zig's unions do not
16// support assignment by any union member directly as in C, without going through the union, so the
17// kernel adding new unions would also break existing Zig code.
18// We therefore use a flat struct without unions to avoid these two issues.
19// Pending https://github.com/ziglang/zig/issues/6349.
20pub const io_uring_sqe = extern struct {
21 opcode: linux.IORING_OP,
22 flags: u8 = 0,
23 ioprio: u16 = 0,
24 fd: i32 = 0,
25 off: u64 = 0,
26 addr: u64 = 0,
27 len: u32 = 0,
28 opflags: u32 = 0,
29 user_data: u64 = 0,
30 buffer: u16 = 0,
31 personality: u16 = 0,
32 splice_fd_in: i32 = 0,
33 options: [2]u64 = [2]u64{ 0, 0 }
34};
35
36// TODO Add to zig/std/os/bits/linux.zig:
37const IORING_SQ_CQ_OVERFLOW = 1 << 1;
38
39comptime {
40 assert(@sizeOf(io_uring_params) == 120);
41 assert(@sizeOf(io_uring_sqe) == 64);
42 assert(@sizeOf(io_uring_cqe) == 16);
43
44 assert(linux.IORING_OFF_SQ_RING == 0);
45 assert(linux.IORING_OFF_CQ_RING == 0x8000000);
46 assert(linux.IORING_OFF_SQES == 0x10000000);
47}
48
49pub const IO_Uring = struct {
50 fd: i32 = -1,
51 sq: SubmissionQueue,
52 cq: CompletionQueue,
53 flags: u32,
54
55 /// A friendly way to setup an io_uring, with default io_uring_params.
56 /// `entries` must be a power of two between 1 and 4096, although the kernel will make the final
57 /// call on how many entries the submission and completion queues will ultimately have,
58 /// see https://github.com/torvalds/linux/blob/v5.8/fs/io_uring.c#L8027-L8050.
59 /// Matches the interface of io_uring_queue_init() in liburing.
60 pub fn init(entries: u32, flags: u32) !IO_Uring {
61 var params = io_uring_params {
62 .sq_entries = 0,
63 .cq_entries = 0,
64 .flags = flags,
65 .sq_thread_cpu = 0,
66 .sq_thread_idle = 1000,
67 .features = 0,
68 .wq_fd = 0,
69 .resv = [_]u32{0} ** 3,
70 .sq_off = undefined,
71 .cq_off = undefined,
72 };
73 // The kernel will zero the memory of the sq_off and cq_off structs in io_uring_create(),
74 // see https://github.com/torvalds/linux/blob/v5.8/fs/io_uring.c#L7986-L8002.
75 return try IO_Uring.init_params(entries, &params);
76 }
77
78 /// A powerful way to setup an io_uring, if you want to tweak io_uring_params such as submission
79 /// queue thread cpu affinity or thread idle timeout (the kernel and our default is 1 second).
80 /// `params` is passed by reference because the kernel needs to modify the parameters.
81 /// You may only set the `flags`, `sq_thread_cpu` and `sq_thread_idle` parameters.
82 /// Every other parameter belongs to the kernel and must be zeroed.
83 /// Matches the interface of io_uring_queue_init_params() in liburing.
84 pub fn init_params(entries: u32, p: *io_uring_params) !IO_Uring {
85 assert(entries >= 1 and entries <= 4096 and std.math.isPowerOfTwo(entries));
86 assert(p.*.sq_entries == 0);
87 assert(p.*.cq_entries == 0);
88 assert(p.*.features == 0);
89 assert(p.*.wq_fd == 0);
90 assert(p.*.resv[0] == 0);
91 assert(p.*.resv[1] == 0);
92 assert(p.*.resv[2] == 0);
93
94 const res = linux.io_uring_setup(entries, p);
95 try check_errno(res);
96 const fd = @intCast(i32, res);
97 assert(fd >= 0);
98 errdefer os.close(fd);
99
100 // Kernel versions 5.4 and up use only one mmap() for the submission and completion queues.
101 // This is not an optional feature for us... if the kernel does it, we have to do it.
102 // The thinking on this by the kernel developers was that both the submission and the
103 // completion queue rings have sizes just over a power of two, but the submission queue ring
104 // is significantly smaller with u32 slots. By bundling both in a single mmap, the kernel
105 // gets the submission queue ring for free.
106 // See https://patchwork.kernel.org/patch/11115257 for the kernel patch.
107 // We do not support the double mmap() done before 5.4, because we want to keep the
108 // init/deinit mmap paths simple and because io_uring has had many bug fixes even since 5.4.
109 if ((p.*.features & linux.IORING_FEAT_SINGLE_MMAP) == 0) {
110 return error.IO_UringKernelNotSupported;
111 }
112
113 // Check that the kernel has actually set params and that "impossible is nothing".
114 assert(p.*.sq_entries != 0);
115 assert(p.*.cq_entries != 0);
116 assert(p.*.cq_entries >= p.*.sq_entries);
117
118 // From here on, we only need to read from params, so pass `p` by value for convenience.
119 // The completion queue shares the mmap with the submission queue, so pass `sq` there too.
120 var sq = try SubmissionQueue.init(fd, p.*);
121 errdefer sq.deinit();
122 var cq = try CompletionQueue.init(fd, p.*, sq);
123 errdefer cq.deinit();
124
125 // Check that our starting state is as we expect.
126 assert(sq.head.* == 0);
127 assert(sq.tail.* == 0);
128 assert(sq.mask.* == p.*.sq_entries - 1);
129 // Allow flags.* to be non-zero, since the kernel may set IORING_SQ_NEED_WAKEUP at any time.
130 assert(sq.dropped.* == 0);
131 assert(sq.array.len == p.*.sq_entries);
132 assert(sq.sqes.len == p.*.sq_entries);
133 assert(sq.sqe_head == 0);
134 assert(sq.sqe_tail == 0);
135
136 assert(cq.head.* == 0);
137 assert(cq.tail.* == 0);
138 assert(cq.mask.* == p.*.cq_entries - 1);
139 assert(cq.overflow.* == 0);
140 assert(cq.cqes.len == p.*.cq_entries);
141
142 // Alles in Ordnung!
143 return IO_Uring {
144 .fd = fd,
145 .sq = sq,
146 .cq = cq,
147 .flags = p.*.flags
148 };
149 }
150
151 pub fn deinit(self: *IO_Uring) void {
152 assert(self.fd >= 0);
153 // The mmaps depend on the fd, so the order of these calls is important:
154 self.cq.deinit();
155 self.sq.deinit();
156 os.close(self.fd);
157 self.fd = -1;
158 }
159
160 /// Returns a vacant SQE, or an error if the submission queue is full.
161 /// We follow the implementation (and atomics) of liburing's `io_uring_get_sqe()` exactly.
162 /// However, instead of a null we return an error to force safe handling.
163 /// Any situation where the submission queue is full tends more towards a control flow error,
164 /// and the null return in liburing is more a C idiom than anything else, for lack of a better
165 /// alternative. In Zig, we have first-class error handling... so let's use it.
166 /// Matches the implementation of io_uring_get_sqe() in liburing.
167 pub fn get_sqe(self: *IO_Uring) !*io_uring_sqe {
168 const head = @atomicLoad(u32, self.sq.head, .Acquire);
169 // Remember that these head and tail offsets wrap around every four billion operations.
170 // We must therefore use wrapping addition and subtraction to avoid a runtime crash.
171 const next = self.sq.sqe_tail +% 1;
172 if (next -% head > self.sq.sqes.len) return error.IO_UringSubmissionQueueFull;
173 var sqe = &self.sq.sqes[self.sq.sqe_tail & self.sq.mask.*];
174 self.sq.sqe_tail = next;
175 return sqe;
176 }
177
178 /// Submits the SQEs acquired via get_sqe() to the kernel. You can call this once after you have
179 /// called get_sqe() multiple times to setup multiple I/O requests.
180 /// Returns the number of SQEs submitted.
181 /// Matches the implementation of io_uring_submit() in liburing.
182 pub fn submit(self: *IO_Uring) !u32 {
183 return self.submit_and_wait(0);
184 }
185
186 /// Like submit(), but allows waiting for events as well.
187 /// Returns the number of SQEs submitted.
188 /// Matches the implementation of io_uring_submit_and_wait() in liburing.
189 pub fn submit_and_wait(self: *IO_Uring, wait_nr: u32) !u32 {
190 var submitted = self.flush_sq();
191 var flags: u32 = 0;
192 if (self.sq_ring_needs_enter(submitted, &flags) or wait_nr > 0) {
193 if (wait_nr > 0 or (self.flags & linux.IORING_SETUP_IOPOLL) > 0) {
194 flags |= linux.IORING_ENTER_GETEVENTS;
195 }
196 return try self.enter(submitted, wait_nr, flags);
197 }
198 return submitted;
199 }
200
201 // Tell the kernel we have submitted SQEs and/or want to wait for CQEs.
202 // Returns the number of SQEs submitted.
203 fn enter(self: *IO_Uring, to_submit: u32, min_complete: u32, flags: u32) !u32 {
204 assert(self.fd >= 0);
205 const res = linux.io_uring_enter(self.fd, to_submit, min_complete, flags, null);
206 try check_errno(res);
207 return @truncate(u32, res);
208 }
209
210 // Sync internal state with kernel ring state on the SQ side.
211 // Returns the number of all pending events in the SQ ring, for the shared ring.
212 // This return value includes previously flushed SQEs, as per liburing.
213 // The reasoning for this is to suggest that an io_uring_enter() call is needed rather than not.
214 // Matches the implementation of __io_uring_flush_sq() in liburing.
215 fn flush_sq(self: *IO_Uring) u32 {
216 if (self.sq.sqe_head != self.sq.sqe_tail) {
217 // Fill in SQEs that we have queued up, adding them to the kernel ring.
218 const to_submit = self.sq.sqe_tail -% self.sq.sqe_head;
219 const mask = self.sq.mask.*;
220 var tail = self.sq.tail.*;
221 var i: usize = 0;
222 while (i < to_submit) : (i += 1) {
223 self.sq.array[tail & mask] = self.sq.sqe_head & mask;
224 tail +%= 1;
225 self.sq.sqe_head +%= 1;
226 }
227 // Ensure that the kernel can actually see the SQE updates when it sees the tail update.
228 @atomicStore(u32, self.sq.tail, tail, .Release);
229 }
230 return self.sq_ready();
231 }
232
233 /// Returns true if we are not using an SQ thread (thus nobody submits but us),
234 /// or if IORING_SQ_NEED_WAKEUP is set and the SQ thread must be explicitly awakened.
235 /// For the latter case, we set the SQ thread wakeup flag.
236 /// Matches the implementation of sq_ring_needs_enter() in liburing.
237 fn sq_ring_needs_enter(self: *IO_Uring, submitted: u32, flags: *u32) bool {
238 assert(flags.* == 0);
239 if ((self.flags & linux.IORING_SETUP_SQPOLL) == 0 and submitted > 0) return true;
240 if ((@atomicLoad(u32, self.sq.flags, .Unordered) & linux.IORING_SQ_NEED_WAKEUP) > 0) {
241 flags.* |= linux.IORING_ENTER_SQ_WAKEUP;
242 return true;
243 }
244 return false;
245 }
246
247 /// Returns the number of flushed and unflushed SQEs pending in the submission queue.
248 /// In other words, this is the number of SQEs in the submission queue, i.e. its length.
249 /// These are SQEs that the kernel is yet to consume.
250 /// Matches the implementation of io_uring_sq_ready in liburing.
251 pub fn sq_ready(self: *IO_Uring) u32 {
252 // Always use the shared ring state (i.e. head and not sqe_head) to avoid going out of sync,
253 // see https://github.com/axboe/liburing/issues/92.
254 return self.sq.sqe_tail -% @atomicLoad(u32, self.sq.head, .Acquire);
255 }
256
257 /// Returns the number of CQEs in the completion queue, i.e. its length.
258 /// These are CQEs that the application is yet to consume.
259 /// Matches the implementation of io_uring_cq_ready in liburing.
260 pub fn cq_ready(self: *IO_Uring) u32 {
261 return @atomicLoad(u32, self.cq.tail, .Acquire) -% self.cq.head.*;
262 }
263
264 /// Copies as many CQEs as are ready, and that can fit into the destination `cqes` slice.
265 /// If none are available, enters into the kernel to wait for at most `wait_nr` CQEs.
266 /// Returns the number of CQEs copied, advancing the CQ ring.
267 /// Provides all the wait/peek methods found in liburing, but with batching and a single method.
268 /// The rationale for copying CQEs rather than copying pointers is that pointers are 8 bytes
269 /// whereas CQEs are not much more at only 16 bytes, and this provides a safer faster interface.
270 /// Safer, because you no longer need to call cqe_seen(), avoiding idempotency bugs.
271 /// Faster, because we can now amortize the atomic store release to `cq.head` across the batch.
272 /// See https://github.com/axboe/liburing/issues/103#issuecomment-686665007.
273 /// Matches the implementation of io_uring_peek_batch_cqe() in liburing, but supports waiting.
274 pub fn copy_cqes(self: *IO_Uring, cqes: []io_uring_cqe, wait_nr: u32) !u32 {
275 const count = self.copy_cqes_ready(cqes, wait_nr);
276 if (count > 0) return count;
277 if (self.cq_ring_needs_flush() or wait_nr > 0) {
278 _ = try self.enter(0, wait_nr, linux.IORING_ENTER_GETEVENTS);
279 return self.copy_cqes_ready(cqes, wait_nr);
280 }
281 return 0;
282 }
283
284 fn copy_cqes_ready(self: *IO_Uring, cqes: []io_uring_cqe, wait_nr: u32) u32 {
285 const ready = self.cq_ready();
286 const count = std.math.min(cqes.len, ready);
287 const mask = self.cq.mask.*;
288 var head = self.cq.head.*;
289 var tail = head +% count;
290 // TODO Optimize this by using 1 or 2 memcpy's (if the tail wraps) rather than a loop.
291 var i: usize = 0;
292 // Do not use "less-than" operator since head and tail may wrap:
293 while (head != tail) {
294 cqes[i] = self.cq.cqes[head & mask]; // Copy struct by value.
295 head +%= 1;
296 i += 1;
297 }
298 self.cq_advance(count);
299 return count;
300 }
301
302 /// Returns a copy of an I/O completion, waiting for it if necessary, and advancing the CQ ring.
303 /// A convenience method for `copy_cqes()` for when you don't need to batch or peek.
304 pub fn copy_cqe(ring: *IO_Uring) !io_uring_cqe {
305 var cqes: [1]io_uring_cqe = undefined;
306 const count = try ring.copy_cqes(&cqes, 1);
307 assert(count == 1);
308 return cqes[0];
309 }
310
311 // Matches the implementation of cq_ring_needs_flush() in liburing.
312 fn cq_ring_needs_flush(self: *IO_Uring) bool {
313 return (@atomicLoad(u32, self.sq.flags, .Unordered) & IORING_SQ_CQ_OVERFLOW) > 0;
314 }
315
316 /// For advanced use cases only that implement custom completion queue methods.
317 /// If you use copy_cqes() or copy_cqe() you must not call cqe_seen() or cq_advance().
318 /// Must be called exactly once after a zero-copy CQE has been processed by your application.
319 /// Not idempotent, calling more than once will result in other CQEs being lost.
320 /// Matches the implementation of cqe_seen() in liburing.
321 pub fn cqe_seen(self: *IO_Uring, cqe: *io_uring_cqe) void {
322 self.cq_advance(1);
323 }
324
325 /// For advanced use cases only that implement custom completion queue methods.
326 /// Matches the implementation of cq_advance() in liburing.
327 pub fn cq_advance(self: *IO_Uring, count: u32) void {
328 if (count > 0) {
329 // Ensure the kernel only sees the new head value after the CQEs have been read.
330 @atomicStore(u32, self.cq.head, self.cq.head.* +% count, .Release);
331 }
332 }
333
334 /// Queues (but does not submit) an SQE to perform an `accept4(2)` on a socket.
335 /// Returns a pointer to the SQE.
336 pub fn queue_accept(
337 self: *IO_Uring,
338 user_data: u64,
339 fd: os.fd_t,
340 addr: *os.sockaddr,
341 addrlen: *os.socklen_t,
342 accept_flags: u32
343 ) !*io_uring_sqe {
344 // "sqe->fd is the file descriptor, sqe->addr holds a pointer to struct sockaddr,
345 // sqe->addr2 holds a pointer to socklen_t, and finally sqe->accept_flags holds the flags
346 // for accept(4)." - https://lwn.net/ml/linux-block/20191025173037.13486-1-axboe@kernel.dk/
347 const sqe = try self.get_sqe();
348 sqe.* = .{
349 .opcode = .ACCEPT,
350 .fd = fd,
351 .off = @ptrToInt(addrlen), // `addr2` is a newer union member that maps to `off`.
352 .addr = @ptrToInt(addr),
353 .user_data = user_data,
354 .opflags = accept_flags
355 };
356 return sqe;
357 }
358
359 /// Queues (but does not submit) an SQE to perform an `fsync(2)`.
360 /// Returns a pointer to the SQE so that you can further modify the SQE for advanced use cases.
361 /// For example, for `fdatasync()` you can set `IORING_FSYNC_DATASYNC` in the SQE's `opflags`.
362 /// N.B. While SQEs are initiated in the order in which they appear in the submission queue,
363 /// operations execute in parallel and completions are unordered. Therefore, an application that
364 /// submits a write followed by an fsync in the submission queue cannot expect the fsync to
365 /// apply to the write, since the fsync may complete before the write is issued to the disk.
366 /// You should preferably use `link_with_next_sqe()` on a write's SQE to link it with an fsync,
367 /// or else insert a full write barrier using `drain_previous_sqes()` when queueing an fsync.
368 pub fn queue_fsync(self: *IO_Uring, user_data: u64, fd: os.fd_t) !*io_uring_sqe {
369 const sqe = try self.get_sqe();
370 sqe.* = .{
371 .opcode = .FSYNC,
372 .fd = fd,
373 .user_data = user_data
374 };
375 return sqe;
376 }
377
378 /// Queues (but does not submit) an SQE to perform a no-op.
379 /// Returns a pointer to the SQE so that you can further modify the SQE for advanced use cases.
380 /// A no-op is more useful than may appear at first glance.
381 /// For example, you could call `drain_previous_sqes()` on the returned SQE, to use the no-op to
382 /// know when the ring is idle before acting on a kill signal.
383 pub fn queue_nop(self: *IO_Uring, user_data: u64) !*io_uring_sqe {
384 const sqe = try self.get_sqe();
385 sqe.* = .{
386 .opcode = .NOP,
387 .user_data = user_data
388 };
389 return sqe;
390 }
391
392 /// Queues (but does not submit) an SQE to perform a `read(2)`.
393 /// Returns a pointer to the SQE.
394 pub fn queue_read(
395 self: *IO_Uring,
396 user_data: u64,
397 fd: os.fd_t,
398 buffer: []u8,
399 offset: u64
400 ) !*io_uring_sqe {
401 const sqe = try self.get_sqe();
402 sqe.* = .{
403 .opcode = .READ,
404 .fd = fd,
405 .off = offset,
406 .addr = @ptrToInt(buffer.ptr),
407 .len = @truncate(u32, buffer.len),
408 .user_data = user_data
409 };
410 return sqe;
411 }
412
413 /// Queues (but does not submit) an SQE to perform a `write(2)`.
414 /// Returns a pointer to the SQE.
415 pub fn queue_write(
416 self: *IO_Uring,
417 user_data: u64,
418 fd: os.fd_t,
419 buffer: []const u8,
420 offset: u64
421 ) !*io_uring_sqe {
422 const sqe = try self.get_sqe();
423 sqe.* = .{
424 .opcode = .WRITE,
425 .fd = fd,
426 .off = offset,
427 .addr = @ptrToInt(buffer.ptr),
428 .len = @truncate(u32, buffer.len),
429 .user_data = user_data
430 };
431 return sqe;
432 }
433
434 /// Queues (but does not submit) an SQE to perform a `preadv()`.
435 /// Returns a pointer to the SQE so that you can further modify the SQE for advanced use cases.
436 /// For example, if you want to do a `preadv2()` then set `opflags` on the returned SQE.
437 /// See https://linux.die.net/man/2/preadv.
438 pub fn queue_readv(
439 self: *IO_Uring,
440 user_data: u64,
441 fd: os.fd_t,
442 iovecs: []const os.iovec,
443 offset: u64
444 ) !*io_uring_sqe {
445 const sqe = try self.get_sqe();
446 sqe.* = .{
447 .opcode = .READV,
448 .fd = fd,
449 .off = offset,
450 .addr = @ptrToInt(iovecs.ptr),
451 .len = @truncate(u32, iovecs.len),
452 .user_data = user_data
453 };
454 return sqe;
455 }
456
457 /// Queues (but does not submit) an SQE to perform a `pwritev()`.
458 /// Returns a pointer to the SQE so that you can further modify the SQE for advanced use cases.
459 /// For example, if you want to do a `pwritev2()` then set `opflags` on the returned SQE.
460 /// See https://linux.die.net/man/2/pwritev.
461 pub fn queue_writev(
462 self: *IO_Uring,
463 user_data: u64,
464 fd: os.fd_t,
465 iovecs: []const os.iovec_const,
466 offset: u64
467 ) !*io_uring_sqe {
468 const sqe = try self.get_sqe();
469 sqe.* = .{
470 .opcode = .WRITEV,
471 .fd = fd,
472 .off = offset,
473 .addr = @ptrToInt(iovecs.ptr),
474 .len = @truncate(u32, iovecs.len),
475 .user_data = user_data
476 };
477 return sqe;
478 }
479
480 /// The next SQE will not be started until this one completes.
481 /// This can be used to chain causally dependent SQEs, and the chain can be arbitrarily long.
482 /// The tail of the chain is denoted by the first SQE that does not have this flag set.
483 /// This flag has no effect on previous SQEs, nor does it impact SQEs outside the chain.
484 /// This means that multiple chains can be executing in parallel, along with individual SQEs.
485 /// Only members inside the chain are serialized.
486 /// A chain will be broken if any SQE in the chain ends in error, where any unexpected result is
487 /// considered an error. For example, a short read will terminate the remainder of the chain.
488 pub fn link_with_next_sqe(self: *IO_Uring, sqe: *io_uring_sqe) void {
489 sqe.*.flags |= linux.IOSQE_IO_LINK;
490 }
491
492 /// Like `link_with_next_sqe()` but stronger.
493 /// For when you don't want the chain to fail in the event of a completion result error.
494 /// For example, you may know that some commands will fail and may want the chain to continue.
495 /// Hard links are resilient to completion results, but are not resilient to submission errors.
496 pub fn hardlink_with_next_sqe(self: *IO_Uring, sqe: *io_uring_sqe) void {
497 sqe.*.flags |= linux.IOSQE_IO_HARDLINK;
498 }
499
500 /// This creates a full pipeline barrier in the submission queue.
501 /// This SQE will not be started until previous SQEs complete.
502 /// Subsequent SQEs will not be started until this SQE completes.
503 /// In other words, this stalls the entire submission queue.
504 /// You should first consider using link_with_next_sqe() for more granular SQE sequence control.
505 pub fn drain_previous_sqes(self: *IO_Uring, sqe: *io_uring_sqe) void {
506 sqe.*.flags |= linux.IOSQE_IO_DRAIN;
507 }
508
509 /// Registers an array of file descriptors.
510 /// Every time a file descriptor is put in an SQE and submitted to the kernel, the kernel must
511 /// retrieve a reference to the file, and once I/O has completed the file reference must be
512 /// dropped. The atomic nature of this file reference can be a slowdown for high IOPS workloads.
513 /// This slowdown can be avoided by pre-registering file descriptors.
514 /// To refer to a registered file descriptor, IOSQE_FIXED_FILE must be set in the SQE's flags,
515 /// and the SQE's fd must be set to the index of the file descriptor in the registered array.
516 /// Registering file descriptors will wait for the ring to idle.
517 /// Files are automatically unregistered by the kernel when the ring is torn down.
518 /// An application need unregister only if it wants to register a new array of file descriptors.
519 pub fn register_files(self: *IO_Uring, fds: []const i32) !void {
520 assert(self.fd >= 0);
521 const res = linux.io_uring_register(
522 self.fd,
523 .REGISTER_FILES,
524 fds.ptr,
525 @truncate(u32, fds.len)
526 );
527 try check_errno(res);
528 }
529
530 /// Changes the semantics of the SQE's `fd` to refer to a pre-registered file descriptor.
531 pub fn use_registered_fd(self: *IO_Uring, sqe: *io_uring_sqe) void {
532 sqe.*.flags |= linux.IOSQE_FIXED_FILE;
533 }
534
535 /// Unregisters all registered file descriptors previously associated with the ring.
536 pub fn unregister_files(self: *IO_Uring) !void {
537 assert(self.fd >= 0);
538 const res = linux.io_uring_register(self.fd, .UNREGISTER_FILES, null, 0);
539 try check_errno(res);
540 }
541};
542
543
544pub const SubmissionQueue = struct {
545 head: *u32,
546 tail: *u32,
547 mask: *u32,
548 flags: *u32,
549 dropped: *u32,
550 array: []u32,
551 sqes: []io_uring_sqe,
552 mmap: []align(std.mem.page_size) u8,
553 mmap_sqes: []align(std.mem.page_size) u8,
554
555 // We use `sqe_head` and `sqe_tail` in the same way as liburing:
556 // We increment `sqe_tail` (but not `tail`) for each call to `get_sqe()`.
557 // We then set `tail` to `sqe_tail` once, only when these events are actually submitted.
558 // This allows us to amortize the cost of the @atomicStore to `tail` across multiple SQEs.
559 sqe_head: u32 = 0,
560 sqe_tail: u32 = 0,
561
562 pub fn init(fd: i32, p: io_uring_params) !SubmissionQueue {
563 assert(fd >= 0);
564 assert((p.features & linux.IORING_FEAT_SINGLE_MMAP) > 0);
565 const size = std.math.max(
566 p.sq_off.array + p.sq_entries * @sizeOf(u32),
567 p.cq_off.cqes + p.cq_entries * @sizeOf(io_uring_cqe)
568 );
569 const mmap = try os.mmap(
570 null,
571 size,
572 os.PROT_READ | os.PROT_WRITE,
573 os.MAP_SHARED | os.MAP_POPULATE,
574 fd,
575 linux.IORING_OFF_SQ_RING,
576 );
577 errdefer os.munmap(mmap);
578 assert(mmap.len == size);
579
580 // The motivation for the `sqes` and `array` indirection is to make it possible for the
581 // application to preallocate static io_uring_sqe entries and then replay them when needed.
582 const size_sqes = p.sq_entries * @sizeOf(io_uring_sqe);
583 const mmap_sqes = try os.mmap(
584 null,
585 size_sqes,
586 os.PROT_READ | os.PROT_WRITE,
587 os.MAP_SHARED | os.MAP_POPULATE,
588 fd,
589 linux.IORING_OFF_SQES,
590 );
591 errdefer os.munmap(mmap_sqes);
592 assert(mmap_sqes.len == size_sqes);
593
594 const array = @ptrCast([*]u32, @alignCast(@alignOf(u32), &mmap[p.sq_off.array]));
595 const sqes = @ptrCast([*]io_uring_sqe, @alignCast(@alignOf(io_uring_sqe), &mmap_sqes[0]));
596 // We expect the kernel copies p.sq_entries to the u32 pointed to by p.sq_off.ring_entries,
597 // see https://github.com/torvalds/linux/blob/v5.8/fs/io_uring.c#L7843-L7844.
598 assert(
599 p.sq_entries ==
600 @ptrCast(*u32, @alignCast(@alignOf(u32), &mmap[p.sq_off.ring_entries])).*
601 );
602 return SubmissionQueue {
603 .head = @ptrCast(*u32, @alignCast(@alignOf(u32), &mmap[p.sq_off.head])),
604 .tail = @ptrCast(*u32, @alignCast(@alignOf(u32), &mmap[p.sq_off.tail])),
605 .mask = @ptrCast(*u32, @alignCast(@alignOf(u32), &mmap[p.sq_off.ring_mask])),
606 .flags = @ptrCast(*u32, @alignCast(@alignOf(u32), &mmap[p.sq_off.flags])),
607 .dropped = @ptrCast(*u32, @alignCast(@alignOf(u32), &mmap[p.sq_off.dropped])),
608 .array = array[0..p.sq_entries],
609 .sqes = sqes[0..p.sq_entries],
610 .mmap = mmap,
611 .mmap_sqes = mmap_sqes
612 };
613 }
614
615 pub fn deinit(self: *SubmissionQueue) void {
616 os.munmap(self.mmap_sqes);
617 os.munmap(self.mmap);
618 }
619};
620
621pub const CompletionQueue = struct {
622 head: *u32,
623 tail: *u32,
624 mask: *u32,
625 overflow: *u32,
626 cqes: []io_uring_cqe,
627
628 pub fn init(fd: i32, p: io_uring_params, sq: SubmissionQueue) !CompletionQueue {
629 assert(fd >= 0);
630 assert((p.features & linux.IORING_FEAT_SINGLE_MMAP) > 0);
631 const mmap = sq.mmap;
632 const cqes = @ptrCast(
633 [*]io_uring_cqe,
634 @alignCast(@alignOf(io_uring_cqe), &mmap[p.cq_off.cqes])
635 );
636 assert(
637 p.cq_entries ==
638 @ptrCast(*u32, @alignCast(@alignOf(u32), &mmap[p.cq_off.ring_entries])).*
639 );
640 return CompletionQueue {
641 .head = @ptrCast(*u32, @alignCast(@alignOf(u32), &mmap[p.cq_off.head])),
642 .tail = @ptrCast(*u32, @alignCast(@alignOf(u32), &mmap[p.cq_off.tail])),
643 .mask = @ptrCast(*u32, @alignCast(@alignOf(u32), &mmap[p.cq_off.ring_mask])),
644 .overflow = @ptrCast(*u32, @alignCast(@alignOf(u32), &mmap[p.cq_off.overflow])),
645 .cqes = cqes[0..p.cq_entries]
646 };
647 }
648
649 pub fn deinit(self: *CompletionQueue) void {
650 // A no-op since we now share the mmap with the submission queue.
651 // Here for symmetry with the submission queue, and for any future feature support.
652 }
653};
654
655inline fn check_errno(res: usize) !void {
656 const errno = linux.getErrno(res);
657 if (errno != 0) return os.unexpectedErrno(errno);
658}
659
660test "queue_nop" {
661 if (builtin.os.tag != .linux) return error.SkipZigTest;
662
663 var ring = try IO_Uring.init(1, 0);
664 defer {
665 ring.deinit();
666 testing.expectEqual(@as(i32, -1), ring.fd);
667 }
668
669 var sqe = try ring.queue_nop(@intCast(u64, 0xaaaaaaaa));
670 testing.expectEqual(io_uring_sqe {
671 .opcode = .NOP,
672 .flags = 0,
673 .ioprio = 0,
674 .fd = 0,
675 .off = 0,
676 .addr = 0,
677 .len = 0,
678 .opflags = 0,
679 .user_data = @intCast(u64, 0xaaaaaaaa),
680 .buffer = 0,
681 .personality = 0,
682 .splice_fd_in = 0,
683 .options = [2]u64{ 0, 0 }
684 }, sqe.*);
685
686 testing.expectEqual(@as(u32, 0), ring.sq.sqe_head);
687 testing.expectEqual(@as(u32, 1), ring.sq.sqe_tail);
688 testing.expectEqual(@as(u32, 0), ring.sq.tail.*);
689 testing.expectEqual(@as(u32, 0), ring.cq.head.*);
690 testing.expectEqual(@as(u32, 1), ring.sq_ready());
691 testing.expectEqual(@as(u32, 0), ring.cq_ready());
692
693 testing.expectEqual(@as(u32, 1), try ring.submit());
694 testing.expectEqual(@as(u32, 1), ring.sq.sqe_head);
695 testing.expectEqual(@as(u32, 1), ring.sq.sqe_tail);
696 testing.expectEqual(@as(u32, 1), ring.sq.tail.*);
697 testing.expectEqual(@as(u32, 0), ring.cq.head.*);
698 testing.expectEqual(@as(u32, 0), ring.sq_ready());
699
700 testing.expectEqual(io_uring_cqe {
701 .user_data = 0xaaaaaaaa,
702 .res = 0,
703 .flags = 0
704 }, try ring.copy_cqe());
705 testing.expectEqual(@as(u32, 1), ring.cq.head.*);
706 testing.expectEqual(@as(u32, 0), ring.cq_ready());
707
708 var sqe_barrier = try ring.queue_nop(@intCast(u64, 0xbbbbbbbb));
709 ring.drain_previous_sqes(sqe_barrier);
710 testing.expectEqual(@as(u8, linux.IOSQE_IO_DRAIN), sqe_barrier.*.flags);
711 testing.expectEqual(@as(u32, 1), try ring.submit());
712 testing.expectEqual(io_uring_cqe {
713 .user_data = 0xbbbbbbbb,
714 .res = 0,
715 .flags = 0
716 }, try ring.copy_cqe());
717 testing.expectEqual(@as(u32, 2), ring.sq.sqe_head);
718 testing.expectEqual(@as(u32, 2), ring.sq.sqe_tail);
719 testing.expectEqual(@as(u32, 2), ring.sq.tail.*);
720 testing.expectEqual(@as(u32, 2), ring.cq.head.*);
721}
722
723test "queue_readv" {
724 if (builtin.os.tag != .linux) return error.SkipZigTest;
725
726 var ring = try IO_Uring.init(1, 0);
727 defer ring.deinit();
728
729 const fd = try os.openZ("/dev/zero", os.O_RDONLY | os.O_CLOEXEC, 0);
730 defer os.close(fd);
731
732 var registered_fds = [_]i32{-1} ** 10;
733 const fd_index = 9;
734 registered_fds[fd_index] = fd;
735 try ring.register_files(registered_fds[0..]);
736
737 var buffer = [_]u8{42} ** 128;
738 var iovecs = [_]os.iovec{ os.iovec { .iov_base = &buffer, .iov_len = buffer.len } };
739 var sqe = try ring.queue_readv(0xcccccccc, fd_index, iovecs[0..], 0);
740 ring.use_registered_fd(sqe);
741 testing.expectEqual(@as(u8, linux.IOSQE_FIXED_FILE), sqe.*.flags);
742
743 testing.expectError(error.IO_UringSubmissionQueueFull, ring.queue_nop(0));
744 testing.expectEqual(@as(u32, 1), try ring.submit());
745 testing.expectEqual(linux.io_uring_cqe {
746 .user_data = 0xcccccccc,
747 .res = buffer.len,
748 .flags = 0,
749 }, try ring.copy_cqe());
750 testing.expectEqualSlices(u8, &([_]u8{0} ** buffer.len), buffer[0..]);
751
752 try ring.unregister_files();
753}
754
755test "queue_writev/queue_fsync" {
756 if (builtin.os.tag != .linux) return error.SkipZigTest;
757
758 var ring = try IO_Uring.init(2, 0);
759 defer ring.deinit();
760
761 const path = "test_io_uring_queue_writev";
762 const file = try std.fs.cwd().createFile(path, .{ .truncate = true });
763 defer file.close();
764 defer std.fs.cwd().deleteFile(path) catch {};
765 const fd = file.handle;
766
767 var buffer = [_]u8{42} ** 128;
768 var iovecs = [_]os.iovec_const {
769 os.iovec_const { .iov_base = &buffer, .iov_len = buffer.len }
770 };
771 var sqe_writev = try ring.queue_writev(0xdddddddd, fd, iovecs[0..], 0);
772 ring.link_with_next_sqe(sqe_writev);
773 testing.expectEqual(@as(u8, linux.IOSQE_IO_LINK), sqe_writev.*.flags);
774
775 var sqe_fsync = try ring.queue_fsync(0xeeeeeeee, fd);
776 testing.expectEqual(fd, sqe_fsync.*.fd);
777
778 testing.expectEqual(@as(u32, 2), ring.sq_ready());
779 testing.expectEqual(@as(u32, 2), try ring.submit_and_wait(2));
780 testing.expectEqual(@as(u32, 0), ring.sq_ready());
781 testing.expectEqual(@as(u32, 2), ring.cq_ready());
782 testing.expectEqual(linux.io_uring_cqe {
783 .user_data = 0xdddddddd,
784 .res = buffer.len,
785 .flags = 0,
786 }, try ring.copy_cqe());
787 testing.expectEqual(@as(u32, 1), ring.cq_ready());
788 testing.expectEqual(linux.io_uring_cqe {
789 .user_data = 0xeeeeeeee,
790 .res = 0,
791 .flags = 0,
792 }, try ring.copy_cqe());
793 testing.expectEqual(@as(u32, 0), ring.cq_ready());
794}
795
796test "queue_write/queue_read" {
797 if (builtin.os.tag != .linux) return error.SkipZigTest;
798 // This test may require newer kernel versions.
799
800 var ring = try IO_Uring.init(2, 0);
801 defer ring.deinit();
802
803 const path = "test_io_uring_queue_write";
804 const file = try std.fs.cwd().createFile(path, .{ .read = true, .truncate = true });
805 defer file.close();
806 defer std.fs.cwd().deleteFile(path) catch {};
807 const fd = file.handle;
808
809 var buffer_write = [_]u8{97} ** 20;
810 var buffer_read = [_]u8{98} ** 20;
811 var sqe_write = try ring.queue_write(123, fd, buffer_write[0..], 10);
812 ring.link_with_next_sqe(sqe_write);
813 var sqe_read = try ring.queue_read(456, fd, buffer_read[0..], 10);
814 testing.expectEqual(@as(u32, 2), try ring.submit());
815 testing.expectEqual(linux.io_uring_cqe {
816 .user_data = 123,
817 .res = buffer_write.len,
818 .flags = 0,
819 }, try ring.copy_cqe());
820 testing.expectEqual(linux.io_uring_cqe {
821 .user_data = 456,
822 .res = buffer_read.len,
823 .flags = 0,
824 }, try ring.copy_cqe());
825 testing.expectEqualSlices(u8, buffer_write[0..], buffer_read[0..]);
826}
lib/std/std.zig+1
...@@ -65,6 +65,7 @@ pub const hash_map = @import("hash_map.zig");...@@ -65,6 +65,7 @@ pub const hash_map = @import("hash_map.zig");
65pub const heap = @import("heap.zig");65pub const heap = @import("heap.zig");
66pub const http = @import("http.zig");66pub const http = @import("http.zig");
67pub const io = @import("io.zig");67pub const io = @import("io.zig");
68pub const io_uring = @import("io_uring.zig");
68pub const json = @import("json.zig");69pub const json = @import("json.zig");
69pub const log = @import("log.zig");70pub const log = @import("log.zig");
70pub const macho = @import("macho.zig");71pub const macho = @import("macho.zig");