authorgravatar for igor.anic@gmail.comIgor Anić <igor.anic@gmail.com> 2023-11-10 16:44:18+01:00
committergravatar for igor.anic@gmail.comIgor Anić <igor.anic@gmail.com> 2023-11-10 16:44:18+01:00
log087ee497d430d30bd9c7a2cdfdfc3d2654d105d3
treebfd7bf5d00f66d3e2803efd1ebbfd7bf2bc8154c
parent6b9f7e26c9e43700225b7c2479305015df75a2b9

io_uring: add zero-copy send operation

`send_zc` tries to avoid making intermediate copies of data. Zerocopy execution is not guaranteed and may fall back to copying. The flags field of the first struct io_uring_cqe may likely contain IORING_CQE_F_MORE , which means that there will be a second completion event / notification for the request, with the user_data field set to the same value. The user must not modify the data buffer until the notification is posted. The first cqe follows the usual rules and so its res field will contain the number of bytes sent or a negative error code. The notification's res field will be set to zero and the flags field will contain IORING_CQE_F_NOTIF. The two step model is needed because the kernel may hold on to buffers for a long time, e.g. waiting for a TCP ACK, and having a separate cqe for request completions allows userspace to push more data without extra delays. Note, notifications are only responsible for controlling the lifetime of the buffers, and as such don't mean anything about whether the data has atually been sent out or received by the other end. Even errored requests may generate a notification, and the user must check for IORING_CQE_F_MORE rather than relying on the result. Available since kernel 6.0. References: https://man7.org/linux/man-pages/man3/io_uring_prep_send_zc.3.html https://man7.org/linux/man-pages/man2/io_uring_enter.2.html

1 files changed, 126 insertions(+), 0 deletions(-)

lib/std/os/linux/io_uring.zig+126
...@@ -606,6 +606,39 @@ pub const IO_Uring = struct {...@@ -606,6 +606,39 @@ pub const IO_Uring = struct {
606 return sqe;606 return sqe;
607 }607 }
608608
609 /// Queues (but does not submit) an SQE to perform an async zerocopy `send(2)`.
610 /// Returns a pointer to the SQE.
611 pub fn send_zc(
612 self: *IO_Uring,
613 user_data: u64,
614 fd: os.fd_t,
615 buffer: []const u8,
616 send_flags: u32,
617 zc_flags: u16,
618 ) !*linux.io_uring_sqe {
619 const sqe = try self.get_sqe();
620 io_uring_prep_send_zc(sqe, fd, buffer, send_flags, zc_flags);
621 sqe.user_data = user_data;
622 return sqe;
623 }
624
625 /// Queues (but does not submit) an SQE to perform an async zerocopy `send(2)`.
626 /// Returns a pointer to the SQE.
627 pub fn send_zc_fixed(
628 self: *IO_Uring,
629 user_data: u64,
630 fd: os.fd_t,
631 buffer: []const u8,
632 send_flags: u32,
633 zc_flags: u16,
634 buf_index: u16,
635 ) !*linux.io_uring_sqe {
636 const sqe = try self.get_sqe();
637 io_uring_prep_send_zc_fixed(sqe, fd, buffer, send_flags, zc_flags, buf_index);
638 sqe.user_data = user_data;
639 return sqe;
640 }
641
609 /// Queues (but does not submit) an SQE to perform a `recvmsg(2)`.642 /// Queues (but does not submit) an SQE to perform a `recvmsg(2)`.
610 /// Returns a pointer to the SQE.643 /// Returns a pointer to the SQE.
611 pub fn recvmsg(644 pub fn recvmsg(
...@@ -636,6 +669,21 @@ pub const IO_Uring = struct {...@@ -636,6 +669,21 @@ pub const IO_Uring = struct {
636 return sqe;669 return sqe;
637 }670 }
638671
672 /// Queues (but does not submit) an SQE to perform an async zerocopy `sendmsg(2)`.
673 /// Returns a pointer to the SQE.
674 pub fn sendmsg_zc(
675 self: *IO_Uring,
676 user_data: u64,
677 fd: os.fd_t,
678 msg: *const os.msghdr_const,
679 flags: u32,
680 ) !*linux.io_uring_sqe {
681 const sqe = try self.get_sqe();
682 io_uring_prep_sendmsg_zc(sqe, fd, msg, flags);
683 sqe.user_data = user_data;
684 return sqe;
685 }
686
639 /// Queues (but does not submit) an SQE to perform an `openat(2)`.687 /// Queues (but does not submit) an SQE to perform an `openat(2)`.
640 /// Returns a pointer to the SQE.688 /// Returns a pointer to the SQE.
641 pub fn openat(689 pub fn openat(
...@@ -1373,6 +1421,28 @@ pub fn io_uring_prep_send(sqe: *linux.io_uring_sqe, fd: os.fd_t, buffer: []const...@@ -1373,6 +1421,28 @@ pub fn io_uring_prep_send(sqe: *linux.io_uring_sqe, fd: os.fd_t, buffer: []const
1373 sqe.rw_flags = flags;1421 sqe.rw_flags = flags;
1374}1422}
13751423
1424pub fn io_uring_prep_send_zc(sqe: *linux.io_uring_sqe, fd: os.fd_t, buffer: []const u8, flags: u32, zc_flags: u16) void {
1425 io_uring_prep_rw(.SEND_ZC, sqe, fd, @intFromPtr(buffer.ptr), buffer.len, 0);
1426 sqe.rw_flags = flags;
1427 sqe.ioprio = zc_flags;
1428}
1429
1430pub fn io_uring_prep_send_zc_fixed(sqe: *linux.io_uring_sqe, fd: os.fd_t, buffer: []const u8, flags: u32, zc_flags: u16, buf_index: u16) void {
1431 io_uring_prep_send_zc(sqe, fd, buffer, flags, zc_flags);
1432 sqe.ioprio |= linux.IORING_RECVSEND_FIXED_BUF;
1433 sqe.buf_index = buf_index;
1434}
1435
1436pub fn io_uring_prep_sendmsg_zc(
1437 sqe: *linux.io_uring_sqe,
1438 fd: os.fd_t,
1439 msg: *const os.msghdr_const,
1440 flags: u32,
1441) void {
1442 io_uring_prep_sendmsg(sqe, fd, msg, flags);
1443 sqe.opcode = .SENDMSG_ZC;
1444}
1445
1376pub fn io_uring_prep_recvmsg(1446pub fn io_uring_prep_recvmsg(
1377 sqe: *linux.io_uring_sqe,1447 sqe: *linux.io_uring_sqe,
1378 fd: os.fd_t,1448 fd: os.fd_t,
...@@ -3491,3 +3561,59 @@ test "accept multishot" {...@@ -3491,3 +3561,59 @@ test "accept multishot" {
3491 os.closeSocket(client);3561 os.closeSocket(client);
3492 }3562 }
3493}3563}
3564
3565test "accept/connect/send_zc/recv" {
3566 if (builtin.os.tag != .linux) return error.SkipZigTest;
3567
3568 var ring = IO_Uring.init(16, 0) catch |err| switch (err) {
3569 error.SystemOutdated => return error.SkipZigTest,
3570 error.PermissionDenied => return error.SkipZigTest,
3571 else => return err,
3572 };
3573 defer ring.deinit();
3574
3575 const socket_test_harness = try createSocketTestHarness(&ring);
3576 defer socket_test_harness.close();
3577
3578 const buffer_send = [_]u8{ 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0xa, 0xb, 0xc, 0xd, 0xe };
3579 var buffer_recv = [_]u8{0} ** 10;
3580
3581 // zero-copy send
3582 const send = try ring.send_zc(0xeeeeeeee, socket_test_harness.client, buffer_send[0..], 0, 0);
3583 send.flags |= linux.IOSQE_IO_LINK;
3584 _ = try ring.recv(0xffffffff, socket_test_harness.server, .{ .buffer = buffer_recv[0..] }, 0);
3585 try testing.expectEqual(@as(u32, 2), try ring.submit());
3586
3587 // First completion of zero-copy send.
3588 // IORING_CQE_F_MORE, means that there
3589 // will be a second completion event / notification for the
3590 // request, with the user_data field set to the same value.
3591 // buffer_send must be keep alive until second cqe.
3592 var cqe_send = try ring.copy_cqe();
3593 if (cqe_send.err() == .INVAL) return error.SkipZigTest;
3594 try testing.expectEqual(linux.io_uring_cqe{
3595 .user_data = 0xeeeeeeee,
3596 .res = buffer_send.len,
3597 .flags = linux.IORING_CQE_F_MORE,
3598 }, cqe_send);
3599
3600 const cqe_recv = try ring.copy_cqe();
3601 if (cqe_recv.err() == .INVAL) return error.SkipZigTest;
3602 try testing.expectEqual(linux.io_uring_cqe{
3603 .user_data = 0xffffffff,
3604 .res = buffer_recv.len,
3605 .flags = cqe_recv.flags & linux.IORING_CQE_F_SOCK_NONEMPTY,
3606 }, cqe_recv);
3607
3608 try testing.expectEqualSlices(u8, buffer_send[0..buffer_recv.len], buffer_recv[0..]);
3609
3610 // Second completion of zero-copy send.
3611 // IORING_CQE_F_NOTIF in flags signals that kernel is done with send_buffer
3612 cqe_send = try ring.copy_cqe();
3613 if (cqe_send.err() == .INVAL) return error.SkipZigTest;
3614 try testing.expectEqual(linux.io_uring_cqe{
3615 .user_data = 0xeeeeeeee,
3616 .res = 0,
3617 .flags = linux.IORING_CQE_F_NOTIF,
3618 }, cqe_send);
3619}