authorgravatar for joran@ronomon.comJoran Dirk Greef <joran@ronomon.com> 2020-10-04 13:01:41+02:00
committergravatar for joran@ronomon.comJoran Dirk Greef <joran@ronomon.com> 2020-10-04 13:01:41+02:00
log69a55fc560dab477222f2ad104050b443647faa5
treea0ee9403ed326445b075aee1803ecb26303a6890
parente32c7d06e5c51ff88856c8f48f6fb4fcdf564d17

Allow for advanced non-sequential SQE allocation schemes

Decouples SQE queueing and SQE prepping methods to allow for non-sequential SQE allocation schemes as suggested by @daurnimator. Adds essential SQE prepping methods from liburing to reduce boilerplate. Removes non-essential .link_with_next_sqe() and .use_registered_fd().

1 files changed, 244 insertions(+), 113 deletions(-)

lib/std/os/linux/io_uring.zig+244-113
...@@ -135,13 +135,13 @@ pub const IO_Uring = struct {...@@ -135,13 +135,13 @@ pub const IO_Uring = struct {
135 self.fd = -1;135 self.fd = -1;
136 }136 }
137137
138 /// Returns a pointer to a zeroed SQE, or an error if the submission queue is full.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.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.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,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 better142 /// 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.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, except zeroes for safety.144 /// Matches the implementation of io_uring_get_sqe() in liburing.
145 pub fn get_sqe(self: *IO_Uring) !*io_uring_sqe {145 pub fn get_sqe(self: *IO_Uring) !*io_uring_sqe {
146 const head = @atomicLoad(u32, self.sq.head, .Acquire);146 const head = @atomicLoad(u32, self.sq.head, .Acquire);
147 // Remember that these head and tail offsets wrap around every four billion operations.147 // Remember that these head and tail offsets wrap around every four billion operations.
...@@ -150,8 +150,6 @@ pub const IO_Uring = struct {...@@ -150,8 +150,6 @@ pub const IO_Uring = struct {
150 if (next -% head > self.sq.sqes.len) return error.SubmissionQueueFull;150 if (next -% head > self.sq.sqes.len) return error.SubmissionQueueFull;
151 var sqe = &self.sq.sqes[self.sq.sqe_tail & self.sq.mask];151 var sqe = &self.sq.sqes[self.sq.sqe_tail & self.sq.mask];
152 self.sq.sqe_tail = next;152 self.sq.sqe_tail = next;
153 // We zero the SQE slot here in a single place, rather than in many `queue_` methods.
154 @memset(@ptrCast([*]u8, sqe), 0, @sizeOf(io_uring_sqe));
155 return sqe;153 return sqe;
156 }154 }
157155
...@@ -336,29 +334,6 @@ pub const IO_Uring = struct {...@@ -336,29 +334,6 @@ pub const IO_Uring = struct {
336 }334 }
337 }335 }
338336
339 /// Queues (but does not submit) an SQE to perform an `accept4(2)` on a socket.
340 /// Returns a pointer to the SQE.
341 pub fn queue_accept(
342 self: *IO_Uring,
343 user_data: u64,
344 fd: os.fd_t,
345 addr: *os.sockaddr,
346 addrlen: *os.socklen_t,
347 accept_flags: u32
348 ) !*io_uring_sqe {
349 // "sqe->fd is the file descriptor, sqe->addr holds a pointer to struct sockaddr,
350 // sqe->addr2 holds a pointer to socklen_t, and finally sqe->accept_flags holds the flags
351 // for accept(4)." - https://lwn.net/ml/linux-block/20191025173037.13486-1-axboe@kernel.dk/
352 const sqe = try self.get_sqe();
353 sqe.opcode = .ACCEPT;
354 sqe.fd = fd;
355 sqe.off = @ptrToInt(addrlen); // `addr2` is a newer union member that maps to `off`.
356 sqe.addr = @ptrToInt(addr);
357 sqe.user_data = user_data;
358 sqe.rw_flags = accept_flags;
359 return sqe;
360 }
361
362 /// Queues (but does not submit) an SQE to perform an `fsync(2)`.337 /// Queues (but does not submit) an SQE to perform an `fsync(2)`.
363 /// Returns a pointer to the SQE so that you can further modify the SQE for advanced use cases.338 /// Returns a pointer to the SQE so that you can further modify the SQE for advanced use cases.
364 /// For example, for `fdatasync()` you can set `IORING_FSYNC_DATASYNC` in the SQE's `rw_flags`.339 /// For example, for `fdatasync()` you can set `IORING_FSYNC_DATASYNC` in the SQE's `rw_flags`.
...@@ -368,11 +343,9 @@ pub const IO_Uring = struct {...@@ -368,11 +343,9 @@ pub const IO_Uring = struct {
368 /// apply to the write, since the fsync may complete before the write is issued to the disk.343 /// apply to the write, since the fsync may complete before the write is issued to the disk.
369 /// You should preferably use `link_with_next_sqe()` on a write's SQE to link it with an fsync,344 /// You should preferably use `link_with_next_sqe()` on a write's SQE to link it with an fsync,
370 /// or else insert a full write barrier using `drain_previous_sqes()` when queueing an fsync.345 /// or else insert a full write barrier using `drain_previous_sqes()` when queueing an fsync.
371 pub fn queue_fsync(self: *IO_Uring, user_data: u64, fd: os.fd_t, flags: u32) !*io_uring_sqe {346 pub fn fsync(self: *IO_Uring, user_data: u64, fd: os.fd_t, flags: u32) !*io_uring_sqe {
372 const sqe = try self.get_sqe();347 const sqe = try self.get_sqe();
373 sqe.opcode = .FSYNC;348 io_uring_prep_fsync(sqe, fd, flags);
374 sqe.fd = fd;
375 sqe.rw_flags = flags;
376 sqe.user_data = user_data;349 sqe.user_data = user_data;
377 return sqe;350 return sqe;
378 }351 }
...@@ -382,16 +355,16 @@ pub const IO_Uring = struct {...@@ -382,16 +355,16 @@ pub const IO_Uring = struct {
382 /// A no-op is more useful than may appear at first glance.355 /// A no-op is more useful than may appear at first glance.
383 /// For example, you could call `drain_previous_sqes()` on the returned SQE, to use the no-op to356 /// For example, you could call `drain_previous_sqes()` on the returned SQE, to use the no-op to
384 /// know when the ring is idle before acting on a kill signal.357 /// know when the ring is idle before acting on a kill signal.
385 pub fn queue_nop(self: *IO_Uring, user_data: u64) !*io_uring_sqe {358 pub fn nop(self: *IO_Uring, user_data: u64) !*io_uring_sqe {
386 const sqe = try self.get_sqe();359 const sqe = try self.get_sqe();
387 sqe.opcode = .NOP;360 io_uring_prep_nop(sqe);
388 sqe.user_data = user_data;361 sqe.user_data = user_data;
389 return sqe;362 return sqe;
390 }363 }
391364
392 /// Queues (but does not submit) an SQE to perform a `read(2)`.365 /// Queues (but does not submit) an SQE to perform a `read(2)`.
393 /// Returns a pointer to the SQE.366 /// Returns a pointer to the SQE.
394 pub fn queue_read(367 pub fn read(
395 self: *IO_Uring,368 self: *IO_Uring,
396 user_data: u64,369 user_data: u64,
397 fd: os.fd_t,370 fd: os.fd_t,
...@@ -399,18 +372,14 @@ pub const IO_Uring = struct {...@@ -399,18 +372,14 @@ pub const IO_Uring = struct {
399 offset: u64372 offset: u64
400 ) !*io_uring_sqe {373 ) !*io_uring_sqe {
401 const sqe = try self.get_sqe();374 const sqe = try self.get_sqe();
402 sqe.opcode = .READ;375 io_uring_prep_read(sqe, fd, buffer, offset);
403 sqe.fd = fd;
404 sqe.off = offset;
405 sqe.addr = @ptrToInt(buffer.ptr);
406 sqe.len = @intCast(u32, buffer.len);
407 sqe.user_data = user_data;376 sqe.user_data = user_data;
408 return sqe;377 return sqe;
409 }378 }
410379
411 /// Queues (but does not submit) an SQE to perform a `write(2)`.380 /// Queues (but does not submit) an SQE to perform a `write(2)`.
412 /// Returns a pointer to the SQE.381 /// Returns a pointer to the SQE.
413 pub fn queue_write(382 pub fn write(
414 self: *IO_Uring,383 self: *IO_Uring,
415 user_data: u64,384 user_data: u64,
416 fd: os.fd_t,385 fd: os.fd_t,
...@@ -418,11 +387,7 @@ pub const IO_Uring = struct {...@@ -418,11 +387,7 @@ pub const IO_Uring = struct {
418 offset: u64387 offset: u64
419 ) !*io_uring_sqe {388 ) !*io_uring_sqe {
420 const sqe = try self.get_sqe();389 const sqe = try self.get_sqe();
421 sqe.opcode = .WRITE;390 io_uring_prep_write(sqe, fd, buffer, offset);
422 sqe.fd = fd;
423 sqe.off = offset;
424 sqe.addr = @ptrToInt(buffer.ptr);
425 sqe.len = @intCast(u32, buffer.len);
426 sqe.user_data = user_data;391 sqe.user_data = user_data;
427 return sqe;392 return sqe;
428 }393 }
...@@ -431,7 +396,7 @@ pub const IO_Uring = struct {...@@ -431,7 +396,7 @@ pub const IO_Uring = struct {
431 /// Returns a pointer to the SQE so that you can further modify the SQE for advanced use cases.396 /// Returns a pointer to the SQE so that you can further modify the SQE for advanced use cases.
432 /// For example, if you want to do a `preadv2()` then set `rw_flags` on the returned SQE.397 /// For example, if you want to do a `preadv2()` then set `rw_flags` on the returned SQE.
433 /// See https://linux.die.net/man/2/preadv.398 /// See https://linux.die.net/man/2/preadv.
434 pub fn queue_readv(399 pub fn readv(
435 self: *IO_Uring,400 self: *IO_Uring,
436 user_data: u64,401 user_data: u64,
437 fd: os.fd_t,402 fd: os.fd_t,
...@@ -439,11 +404,7 @@ pub const IO_Uring = struct {...@@ -439,11 +404,7 @@ pub const IO_Uring = struct {
439 offset: u64404 offset: u64
440 ) !*io_uring_sqe {405 ) !*io_uring_sqe {
441 const sqe = try self.get_sqe();406 const sqe = try self.get_sqe();
442 sqe.opcode = .READV;407 io_uring_prep_readv(sqe, fd, iovecs, offset);
443 sqe.fd = fd;
444 sqe.off = offset;
445 sqe.addr = @ptrToInt(iovecs.ptr);
446 sqe.len = @intCast(u32, iovecs.len);
447 sqe.user_data = user_data;408 sqe.user_data = user_data;
448 return sqe;409 return sqe;
449 }410 }
...@@ -452,7 +413,7 @@ pub const IO_Uring = struct {...@@ -452,7 +413,7 @@ pub const IO_Uring = struct {
452 /// Returns a pointer to the SQE so that you can further modify the SQE for advanced use cases.413 /// Returns a pointer to the SQE so that you can further modify the SQE for advanced use cases.
453 /// For example, if you want to do a `pwritev2()` then set `rw_flags` on the returned SQE.414 /// For example, if you want to do a `pwritev2()` then set `rw_flags` on the returned SQE.
454 /// See https://linux.die.net/man/2/pwritev.415 /// See https://linux.die.net/man/2/pwritev.
455 pub fn queue_writev(416 pub fn writev(
456 self: *IO_Uring,417 self: *IO_Uring,
457 user_data: u64,418 user_data: u64,
458 fd: os.fd_t,419 fd: os.fd_t,
...@@ -460,25 +421,25 @@ pub const IO_Uring = struct {...@@ -460,25 +421,25 @@ pub const IO_Uring = struct {
460 offset: u64421 offset: u64
461 ) !*io_uring_sqe {422 ) !*io_uring_sqe {
462 const sqe = try self.get_sqe();423 const sqe = try self.get_sqe();
463 sqe.opcode = .WRITEV;424 io_uring_prep_writev(sqe, fd, iovecs, offset);
464 sqe.fd = fd;
465 sqe.off = offset;
466 sqe.addr = @ptrToInt(iovecs.ptr);
467 sqe.len = @intCast(u32, iovecs.len);
468 sqe.user_data = user_data;425 sqe.user_data = user_data;
469 return sqe;426 return sqe;
470 }427 }
471428
472 /// The next SQE will not be started until this one completes.429 /// Queues (but does not submit) an SQE to perform an `accept4(2)` on a socket.
473 /// This can be used to chain causally dependent SQEs, and the chain can be arbitrarily long.430 /// Returns a pointer to the SQE.
474 /// The tail of the chain is denoted by the first SQE that does not have this flag set.431 pub fn accept(
475 /// This flag has no effect on previous SQEs, nor does it impact SQEs outside the chain.432 self: *IO_Uring,
476 /// This means that multiple chains can be executing in parallel, along with individual SQEs.433 user_data: u64,
477 /// Only members inside the chain are serialized.434 fd: os.fd_t,
478 /// A chain will be broken if any SQE in the chain ends in error, where any unexpected result is435 addr: *os.sockaddr,
479 /// considered an error. For example, a short read will terminate the remainder of the chain.436 addrlen: *os.socklen_t,
480 pub fn link_with_next_sqe(self: *IO_Uring, sqe: *io_uring_sqe) void {437 flags: u32
481 sqe.flags |= linux.IOSQE_IO_LINK;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;
482 }443 }
483 444
484 /// Like `link_with_next_sqe()` but stronger.445 /// Like `link_with_next_sqe()` but stronger.
...@@ -538,11 +499,6 @@ pub const IO_Uring = struct {...@@ -538,11 +499,6 @@ pub const IO_Uring = struct {
538 }499 }
539 }500 }
540501
541 /// Changes the semantics of the SQE's `fd` to refer to a pre-registered file descriptor.
542 pub fn use_registered_fd(self: *IO_Uring, sqe: *io_uring_sqe) void {
543 sqe.flags |= linux.IOSQE_FIXED_FILE;
544 }
545
546 /// Unregisters all registered file descriptors previously associated with the ring.502 /// Unregisters all registered file descriptors previously associated with the ring.
547 pub fn unregister_files(self: *IO_Uring) !void {503 pub fn unregister_files(self: *IO_Uring) !void {
548 assert(self.fd >= 0);504 assert(self.fd >= 0);
...@@ -563,8 +519,8 @@ pub const SubmissionQueue = struct {...@@ -563,8 +519,8 @@ pub const SubmissionQueue = struct {
563 dropped: *u32,519 dropped: *u32,
564 array: []u32,520 array: []u32,
565 sqes: []io_uring_sqe,521 sqes: []io_uring_sqe,
566 mmap: []align(std.mem.page_size) u8,522 mmap: []align(mem.page_size) u8,
567 mmap_sqes: []align(std.mem.page_size) u8,523 mmap_sqes: []align(mem.page_size) u8,
568524
569 // We use `sqe_head` and `sqe_tail` in the same way as liburing:525 // We use `sqe_head` and `sqe_tail` in the same way as liburing:
570 // We increment `sqe_tail` (but not `tail`) for each call to `get_sqe()`.526 // We increment `sqe_tail` (but not `tail`) for each call to `get_sqe()`.
...@@ -666,7 +622,156 @@ pub const CompletionQueue = struct {...@@ -666,7 +622,156 @@ pub const CompletionQueue = struct {
666 }622 }
667};623};
668624
669test "structs and offsets" {625pub fn io_uring_prep_nop(sqe: *io_uring_sqe) void {
626 sqe.* = .{
627 .opcode = .NOP,
628 .flags = 0,
629 .ioprio = 0,
630 .fd = 0,
631 .off = 0,
632 .addr = 0,
633 .len = 0,
634 .rw_flags = 0,
635 .user_data = 0,
636 .buf_index = 0,
637 .personality = 0,
638 .splice_fd_in = 0,
639 .__pad2 = [2]u64{ 0, 0 }
640 };
641}
642
643pub fn io_uring_prep_fsync(sqe: *io_uring_sqe, fd: os.fd_t, flags: u32) void {
644 sqe.* = .{
645 .opcode = .FSYNC,
646 .flags = 0,
647 .ioprio = 0,
648 .fd = fd,
649 .off = 0,
650 .addr = 0,
651 .len = 0,
652 .rw_flags = flags,
653 .user_data = 0,
654 .buf_index = 0,
655 .personality = 0,
656 .splice_fd_in = 0,
657 .__pad2 = [2]u64{ 0, 0 }
658 };
659}
660
661pub fn io_uring_prep_rw(
662 op: linux.IORING_OP,
663 sqe: *io_uring_sqe,
664 fd: os.fd_t,
665 addr: anytype,
666 len: usize,
667 offset: u64
668) void {
669 sqe.* = .{
670 .opcode = op,
671 .flags = 0,
672 .ioprio = 0,
673 .fd = fd,
674 .off = offset,
675 .addr = @ptrToInt(addr),
676 .len = @intCast(u32, len),
677 .rw_flags = 0,
678 .user_data = 0,
679 .buf_index = 0,
680 .personality = 0,
681 .splice_fd_in = 0,
682 .__pad2 = [2]u64{ 0, 0 }
683 };
684}
685
686pub fn io_uring_prep_read(sqe: *io_uring_sqe, fd: os.fd_t, buffer: []u8, offset: u64) void {
687 io_uring_prep_rw(.READ, sqe, fd, buffer.ptr, buffer.len, offset);
688}
689
690pub fn io_uring_prep_write(sqe: *io_uring_sqe, fd: os.fd_t, buffer: []const u8, offset: u64) void {
691 io_uring_prep_rw(.WRITE, sqe, fd, buffer.ptr, buffer.len, offset);
692}
693
694pub fn io_uring_prep_readv(
695 sqe: *io_uring_sqe,
696 fd: os.fd_t,
697 iovecs: []const os.iovec,
698 offset: u64
699) void {
700 io_uring_prep_rw(.READV, sqe, fd, iovecs.ptr, iovecs.len, offset);
701}
702
703pub fn io_uring_prep_writev(
704 sqe: *io_uring_sqe,
705 fd: os.fd_t,
706 iovecs: []const os.iovec_const,
707 offset: u64
708) void {
709 io_uring_prep_rw(.WRITEV, sqe, fd, iovecs.ptr, iovecs.len, offset);
710}
711
712pub fn io_uring_prep_accept(
713 sqe: *io_uring_sqe,
714 fd: os.fd_t,
715 addr: *os.sockaddr,
716 addrlen: *os.socklen_t,
717 flags: u32
718) void {
719 // `addr` holds a pointer to `sockaddr`, and `addr2` holds a pointer to socklen_t`.
720 // `addr2` maps to `sqe.off` (u64) instead of `sqe.len` (which is only a u32).
721 io_uring_prep_rw(.ACCEPT, sqe, fd, addr, 0, @ptrToInt(addrlen));
722 sqe.rw_flags = flags;
723}
724
725pub fn io_uring_prep_connect(
726 sqe: *io_uring_sqe,
727 fd: os.fd_t,
728 addr: *const os.sockaddr,
729 addrlen: os.socklen_t
730) void {
731 // `addrlen` maps to `sqe.off` (u64) instead of `sqe.len` (which is only a u32).
732 io_uring_prep_rw(.CONNECT, sqe, fd, addr, 0, addrlen);
733}
734
735pub fn io_uring_prep_recv(sqe: *io_uring_sqe, fd: os.fd_t, buffer: []u8, flags: u32) void {
736 io_uring_prep_rw(.RECV, sqe, fd, buffer.ptr, buffer.len, 0);
737 sqe.rw_flags = flags;
738}
739
740pub fn io_uring_prep_send(sqe: *io_uring_sqe, fd: os.fd_t, buffer: []const u8, flags: u32) void {
741 io_uring_prep_rw(.SEND, sqe, fd, buffer.ptr, buffer.len, 0);
742 sqe.rw_flags = flags;
743}
744
745pub fn io_uring_prep_openat(
746 sqe: *io_uring_sqe,
747 fd: os.fd_t,
748 path: [*:0]const u8,
749 flags: u32,
750 mode: os.mode_t
751) void {
752 io_uring_prep_rw(.OPENAT, sqe, fd, path, mode, 0);
753 sqe.rw_flags = flags;
754}
755
756pub fn io_uring_prep_close(sqe: *io_uring_sqe, fd: os.fd_t) void {
757 sqe.* = .{
758 .opcode = .CLOSE,
759 .flags = 0,
760 .ioprio = 0,
761 .fd = fd,
762 .off = 0,
763 .addr = 0,
764 .len = 0,
765 .rw_flags = 0,
766 .user_data = 0,
767 .buf_index = 0,
768 .personality = 0,
769 .splice_fd_in = 0,
770 .__pad2 = [2]u64{ 0, 0 }
771 };
772}
773
774test "structs/offsets/entries" {
670 if (builtin.os.tag != .linux) return error.SkipZigTest;775 if (builtin.os.tag != .linux) return error.SkipZigTest;
671776
672 testing.expectEqual(@as(usize, 120), @sizeOf(io_uring_params));777 testing.expectEqual(@as(usize, 120), @sizeOf(io_uring_params));
...@@ -681,7 +786,7 @@ test "structs and offsets" {...@@ -681,7 +786,7 @@ test "structs and offsets" {
681 testing.expectError(error.EntriesNotPowerOfTwo, IO_Uring.init(3, 0));786 testing.expectError(error.EntriesNotPowerOfTwo, IO_Uring.init(3, 0));
682}787}
683788
684test "queue_nop" {789test "nop" {
685 if (builtin.os.tag != .linux) return error.SkipZigTest;790 if (builtin.os.tag != .linux) return error.SkipZigTest;
686791
687 var ring = IO_Uring.init(1, 0) catch |err| switch (err) {792 var ring = IO_Uring.init(1, 0) catch |err| switch (err) {
...@@ -694,7 +799,7 @@ test "queue_nop" {...@@ -694,7 +799,7 @@ test "queue_nop" {
694 testing.expectEqual(@as(os.fd_t, -1), ring.fd);799 testing.expectEqual(@as(os.fd_t, -1), ring.fd);
695 }800 }
696801
697 var sqe = try ring.queue_nop(@intCast(u64, 0xaaaaaaaa));802 var sqe = try ring.nop(0xaaaaaaaa);
698 testing.expectEqual(io_uring_sqe {803 testing.expectEqual(io_uring_sqe {
699 .opcode = .NOP,804 .opcode = .NOP,
700 .flags = 0,805 .flags = 0,
...@@ -704,7 +809,7 @@ test "queue_nop" {...@@ -704,7 +809,7 @@ test "queue_nop" {
704 .addr = 0,809 .addr = 0,
705 .len = 0,810 .len = 0,
706 .rw_flags = 0,811 .rw_flags = 0,
707 .user_data = @intCast(u64, 0xaaaaaaaa),812 .user_data = 0xaaaaaaaa,
708 .buf_index = 0,813 .buf_index = 0,
709 .personality = 0,814 .personality = 0,
710 .splice_fd_in = 0,815 .splice_fd_in = 0,
...@@ -733,9 +838,8 @@ test "queue_nop" {...@@ -733,9 +838,8 @@ test "queue_nop" {
733 testing.expectEqual(@as(u32, 1), ring.cq.head.*);838 testing.expectEqual(@as(u32, 1), ring.cq.head.*);
734 testing.expectEqual(@as(u32, 0), ring.cq_ready());839 testing.expectEqual(@as(u32, 0), ring.cq_ready());
735840
736 var sqe_barrier = try ring.queue_nop(@intCast(u64, 0xbbbbbbbb));841 var sqe_barrier = try ring.nop(0xbbbbbbbb);
737 ring.drain_previous_sqes(sqe_barrier);842 sqe_barrier.flags |= linux.IOSQE_IO_DRAIN;
738 testing.expectEqual(@as(u8, linux.IOSQE_IO_DRAIN), sqe_barrier.flags);
739 testing.expectEqual(@as(u32, 1), try ring.submit());843 testing.expectEqual(@as(u32, 1), try ring.submit());
740 testing.expectEqual(io_uring_cqe {844 testing.expectEqual(io_uring_cqe {
741 .user_data = 0xbbbbbbbb,845 .user_data = 0xbbbbbbbb,
...@@ -748,7 +852,7 @@ test "queue_nop" {...@@ -748,7 +852,7 @@ test "queue_nop" {
748 testing.expectEqual(@as(u32, 2), ring.cq.head.*);852 testing.expectEqual(@as(u32, 2), ring.cq.head.*);
749}853}
750854
751test "queue_readv" {855test "readv" {
752 if (builtin.os.tag != .linux) return error.SkipZigTest;856 if (builtin.os.tag != .linux) return error.SkipZigTest;
753857
754 var ring = IO_Uring.init(1, 0) catch |err| switch (err) {858 var ring = IO_Uring.init(1, 0) catch |err| switch (err) {
...@@ -774,11 +878,11 @@ test "queue_readv" {...@@ -774,11 +878,11 @@ test "queue_readv" {
774878
775 var buffer = [_]u8{42} ** 128;879 var buffer = [_]u8{42} ** 128;
776 var iovecs = [_]os.iovec{ os.iovec { .iov_base = &buffer, .iov_len = buffer.len } };880 var iovecs = [_]os.iovec{ os.iovec { .iov_base = &buffer, .iov_len = buffer.len } };
777 var sqe = try ring.queue_readv(0xcccccccc, fd_index, iovecs[0..], 0);881 var sqe = try ring.readv(0xcccccccc, fd_index, iovecs[0..], 0);
778 ring.use_registered_fd(sqe);882 testing.expectEqual(linux.IORING_OP.READV, sqe.opcode);
779 testing.expectEqual(@as(u8, linux.IOSQE_FIXED_FILE), sqe.flags);883 sqe.flags |= linux.IOSQE_FIXED_FILE;
780884
781 testing.expectError(error.SubmissionQueueFull, ring.queue_nop(0));885 testing.expectError(error.SubmissionQueueFull, ring.nop(0));
782 testing.expectEqual(@as(u32, 1), try ring.submit());886 testing.expectEqual(@as(u32, 1), try ring.submit());
783 testing.expectEqual(linux.io_uring_cqe {887 testing.expectEqual(linux.io_uring_cqe {
784 .user_data = 0xcccccccc,888 .user_data = 0xcccccccc,
...@@ -790,52 +894,75 @@ test "queue_readv" {...@@ -790,52 +894,75 @@ test "queue_readv" {
790 try ring.unregister_files();894 try ring.unregister_files();
791}895}
792896
793test "queue_writev/queue_fsync" {897test "writev/fsync/readv" {
794 if (builtin.os.tag != .linux) return error.SkipZigTest;898 if (builtin.os.tag != .linux) return error.SkipZigTest;
795899
796 var ring = IO_Uring.init(2, 0) catch |err| switch (err) {900 var ring = IO_Uring.init(4, 0) catch |err| switch (err) {
797 error.SystemOutdated => return error.SkipZigTest,901 error.SystemOutdated => return error.SkipZigTest,
798 error.PermissionDenied => return error.SkipZigTest,902 error.PermissionDenied => return error.SkipZigTest,
799 else => return err903 else => return err
800 };904 };
801 defer ring.deinit();905 defer ring.deinit();
802 906
803 const path = "test_io_uring_queue_writev";907 const path = "test_io_uring_writev_fsync_readv";
804 const file = try std.fs.cwd().createFile(path, .{ .truncate = true });908 const file = try std.fs.cwd().createFile(path, .{ .read = true, .truncate = true });
805 defer file.close();909 defer file.close();
806 defer std.fs.cwd().deleteFile(path) catch {};910 defer std.fs.cwd().deleteFile(path) catch {};
807 const fd = file.handle;911 const fd = file.handle;
808912
809 var buffer = [_]u8{42} ** 128;913 var buffer_write = [_]u8{42} ** 128;
810 var iovecs = [_]os.iovec_const {914 var iovecs_write = [_]os.iovec_const {
811 os.iovec_const { .iov_base = &buffer, .iov_len = buffer.len }915 os.iovec_const { .iov_base = &buffer_write, .iov_len = buffer_write.len }
916 };
917 var buffer_read = [_]u8{0} ** 128;
918 var iovecs_read = [_]os.iovec {
919 os.iovec { .iov_base = &buffer_read, .iov_len = buffer_read.len }
812 };920 };
813 var sqe_writev = try ring.queue_writev(0xdddddddd, fd, iovecs[0..], 0);921
814 ring.link_with_next_sqe(sqe_writev);922 var sqe_writev = try ring.writev(0xdddddddd, fd, iovecs_write[0..], 17);
815 testing.expectEqual(@as(u8, linux.IOSQE_IO_LINK), sqe_writev.flags);923 testing.expectEqual(linux.IORING_OP.WRITEV, sqe_writev.opcode);
924 testing.expectEqual(@as(u64, 17), sqe_writev.off);
925 sqe_writev.flags |= linux.IOSQE_IO_LINK;
816 926
817 var sqe_fsync = try ring.queue_fsync(0xeeeeeeee, fd, 0);927 var sqe_fsync = try ring.fsync(0xeeeeeeee, fd, 0);
928 testing.expectEqual(linux.IORING_OP.FSYNC, sqe_fsync.opcode);
818 testing.expectEqual(fd, sqe_fsync.fd);929 testing.expectEqual(fd, sqe_fsync.fd);
930 sqe_fsync.flags |= linux.IOSQE_IO_LINK;
931
932 var sqe_readv = try ring.readv(0xffffffff, fd, iovecs_read[0..], 17);
933 testing.expectEqual(linux.IORING_OP.READV, sqe_readv.opcode);
934 testing.expectEqual(@as(u64, 17), sqe_readv.off);
819935
820 testing.expectEqual(@as(u32, 2), ring.sq_ready());936 testing.expectEqual(@as(u32, 3), ring.sq_ready());
821 testing.expectEqual(@as(u32, 2), try ring.submit_and_wait(2));937 testing.expectEqual(@as(u32, 3), try ring.submit_and_wait(3));
822 testing.expectEqual(@as(u32, 0), ring.sq_ready());938 testing.expectEqual(@as(u32, 0), ring.sq_ready());
823 testing.expectEqual(@as(u32, 2), ring.cq_ready());939 testing.expectEqual(@as(u32, 3), ring.cq_ready());
940
824 testing.expectEqual(linux.io_uring_cqe {941 testing.expectEqual(linux.io_uring_cqe {
825 .user_data = 0xdddddddd,942 .user_data = 0xdddddddd,
826 .res = buffer.len,943 .res = buffer_write.len,
827 .flags = 0,944 .flags = 0,
828 }, try ring.copy_cqe());945 }, try ring.copy_cqe());
829 testing.expectEqual(@as(u32, 1), ring.cq_ready());946 testing.expectEqual(@as(u32, 2), ring.cq_ready());
947
830 testing.expectEqual(linux.io_uring_cqe {948 testing.expectEqual(linux.io_uring_cqe {
831 .user_data = 0xeeeeeeee,949 .user_data = 0xeeeeeeee,
832 .res = 0,950 .res = 0,
833 .flags = 0,951 .flags = 0,
834 }, try ring.copy_cqe());952 }, try ring.copy_cqe());
953 testing.expectEqual(@as(u32, 1), ring.cq_ready());
954
955 testing.expectEqual(linux.io_uring_cqe {
956 .user_data = 0xffffffff,
957 .res = buffer_read.len,
958 .flags = 0,
959 }, try ring.copy_cqe());
835 testing.expectEqual(@as(u32, 0), ring.cq_ready());960 testing.expectEqual(@as(u32, 0), ring.cq_ready());
961
962 testing.expectEqualSlices(u8, buffer_write[0..], buffer_read[0..]);
836}963}
837964
838test "queue_write/queue_read" {965test "write/read" {
839 if (builtin.os.tag != .linux) return error.SkipZigTest;966 if (builtin.os.tag != .linux) return error.SkipZigTest;
840967
841 var ring = IO_Uring.init(2, 0) catch |err| switch (err) {968 var ring = IO_Uring.init(2, 0) catch |err| switch (err) {
...@@ -845,7 +972,7 @@ test "queue_write/queue_read" {...@@ -845,7 +972,7 @@ test "queue_write/queue_read" {
845 };972 };
846 defer ring.deinit();973 defer ring.deinit();
847 974
848 const path = "test_io_uring_queue_write";975 const path = "test_io_uring_write_read";
849 const file = try std.fs.cwd().createFile(path, .{ .read = true, .truncate = true });976 const file = try std.fs.cwd().createFile(path, .{ .read = true, .truncate = true });
850 defer file.close();977 defer file.close();
851 defer std.fs.cwd().deleteFile(path) catch {};978 defer std.fs.cwd().deleteFile(path) catch {};
...@@ -853,26 +980,30 @@ test "queue_write/queue_read" {...@@ -853,26 +980,30 @@ test "queue_write/queue_read" {
853980
854 var buffer_write = [_]u8{97} ** 20;981 var buffer_write = [_]u8{97} ** 20;
855 var buffer_read = [_]u8{98} ** 20;982 var buffer_read = [_]u8{98} ** 20;
856 var sqe_write = try ring.queue_write(123, fd, buffer_write[0..], 10);983 var sqe_write = try ring.write(123, fd, buffer_write[0..], 10);
857 ring.link_with_next_sqe(sqe_write);984 testing.expectEqual(linux.IORING_OP.WRITE, sqe_write.opcode);
858 var sqe_read = try ring.queue_read(456, fd, buffer_read[0..], 10);985 testing.expectEqual(@as(u64, 10), sqe_write.off);
986 sqe_write.flags |= linux.IOSQE_IO_LINK;
987 var sqe_read = try ring.read(456, fd, buffer_read[0..], 10);
988 testing.expectEqual(linux.IORING_OP.READ, sqe_read.opcode);
989 testing.expectEqual(@as(u64, 10), sqe_read.off);
859 testing.expectEqual(@as(u32, 2), try ring.submit());990 testing.expectEqual(@as(u32, 2), try ring.submit());
860991
861 var cqe1 = try ring.copy_cqe();992 var cqe_write = try ring.copy_cqe();
862 var cqe2 = try ring.copy_cqe();993 var cqe_read = try ring.copy_cqe();
863 // Prior to Linux Kernel 5.6 this is the only way to test for read/write support:994 // Prior to Linux Kernel 5.6 this is the only way to test for read/write support:
864 // https://lwn.net/Articles/809820/995 // https://lwn.net/Articles/809820/
865 if (cqe1.res == -linux.EINVAL) return error.SkipZigTest;996 if (cqe_write.res == -linux.EINVAL) return error.SkipZigTest;
866 if (cqe2.res == -linux.EINVAL) return error.SkipZigTest;997 if (cqe_read.res == -linux.EINVAL) return error.SkipZigTest;
867 testing.expectEqual(linux.io_uring_cqe {998 testing.expectEqual(linux.io_uring_cqe {
868 .user_data = 123,999 .user_data = 123,
869 .res = buffer_write.len,1000 .res = buffer_write.len,
870 .flags = 0,1001 .flags = 0,
871 }, cqe1);1002 }, cqe_write);
872 testing.expectEqual(linux.io_uring_cqe {1003 testing.expectEqual(linux.io_uring_cqe {
873 .user_data = 456,1004 .user_data = 456,
874 .res = buffer_read.len,1005 .res = buffer_read.len,
875 .flags = 0,1006 .flags = 0,
876 }, cqe2);1007 }, cqe_read);
877 testing.expectEqualSlices(u8, buffer_write[0..], buffer_read[0..]);1008 testing.expectEqualSlices(u8, buffer_write[0..], buffer_read[0..]);
878}1009}