authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-10-02 16:59:22-07:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-10-02 18:21:24-07:00
loga5d8dcc025cc17466627c48bd3645b466b7ba0a5
treecfe83822ac94f6430884ff66359badac9870d637
parent8d03ec6766fef7833057daa27b264c808bb6c2a2

std.Io.Threaded.netReceive: recvmsg first, then poll

Calling recvmsg first means no poll syscall needed when messages are already in the operating system queue. Empirically, this happens when repeating a DNS query that has been already been made recently. In such case, poll() is never called!

1 files changed, 76 insertions(+), 77 deletions(-)

lib/std/Io/Threaded.zig+76-77
...@@ -1431,8 +1431,8 @@ fn netReceive(...@@ -1431,8 +1431,8 @@ fn netReceive(
1431 // the split vectors though because reducing the buffer size might make1431 // the split vectors though because reducing the buffer size might make
1432 // some messages unreceivable.1432 // some messages unreceivable.
14331433
1434 // So the strategy instead is to use poll with timeout and then non-blocking1434 // So the strategy instead is to use non-blocking recvmsg calls, calling
1435 // recvmsg calls.1435 // poll() with timeout if the first one returns EAGAIN.
1436 const posix_flags: u32 =1436 const posix_flags: u32 =
1437 @as(u32, if (flags.oob) posix.MSG.OOB else 0) |1437 @as(u32, if (flags.oob) posix.MSG.OOB else 0) |
1438 @as(u32, if (flags.peek) posix.MSG.PEEK else 0) |1438 @as(u32, if (flags.peek) posix.MSG.PEEK else 0) |
...@@ -1449,93 +1449,92 @@ fn netReceive(...@@ -1449,93 +1449,92 @@ fn netReceive(
1449 var message_i: usize = 0;1449 var message_i: usize = 0;
1450 var data_i: usize = 0;1450 var data_i: usize = 0;
14511451
1452 // TODO: recvmsg first, then poll if EAGAIN. saves syscall in case the messages are already queued.
1453
1454 const deadline = timeout.toDeadline(pool.io()) catch |err| return .{ err, message_i };1452 const deadline = timeout.toDeadline(pool.io()) catch |err| return .{ err, message_i };
14551453
1456 poll: while (true) {1454 recv: while (true) {
1457 pool.checkCancel() catch |err| return .{ err, message_i };1455 pool.checkCancel() catch |err| return .{ err, message_i };
14581456
1459 if (message_i > 0 or message_buffer.len - message_i == 0) return .{ null, message_i };1457 if (message_buffer.len - message_i == 0) return .{ null, message_i };
14601458 const message = &message_buffer[message_i];
1461 const max_poll_ms = std.math.maxInt(u31);1459 const remaining_data_buffer = data_buffer[data_i..];
1462 const timeout_ms: u31 = if (deadline) |d| t: {1460 var storage: PosixAddress = undefined;
1463 const duration = d.durationFromNow(pool.io()) catch |err| return .{ err, message_i };1461 var iov: posix.iovec = .{ .base = remaining_data_buffer.ptr, .len = remaining_data_buffer.len };
1464 if (duration.nanoseconds <= 0) return .{ error.Timeout, message_i };1462 var msg: posix.msghdr = .{
1465 break :t @intCast(@min(max_poll_ms, duration.toMilliseconds()));1463 .name = &storage.any,
1466 } else max_poll_ms;1464 .namelen = @sizeOf(PosixAddress),
1465 .iov = (&iov)[0..1],
1466 .iovlen = 1,
1467 .control = message.control.ptr,
1468 .controllen = message.control.len,
1469 .flags = undefined,
1470 };
14671471
1468 const poll_rc = posix.system.poll(&poll_fds, poll_fds.len, timeout_ms);1472 const recv_rc = posix.system.recvmsg(handle, &msg, posix_flags);
1469 switch (posix.errno(poll_rc)) {1473 switch (posix.errno(recv_rc)) {
1470 .SUCCESS => {1474 .SUCCESS => {
1471 if (poll_rc == 0) {1475 const data = remaining_data_buffer[0..@intCast(recv_rc)];
1472 // Possibly spurious timeout.1476 data_i += data.len;
1473 if (deadline == null) continue;1477 message.* = .{
1474 return .{ error.Timeout, message_i };1478 .from = addressFromPosix(&storage),
1475 }1479 .data = data,
14761480 .control = if (msg.control) |ptr| @as([*]u8, @ptrCast(ptr))[0..msg.controllen] else message.control,
1477 // Proceed to recvmsg.1481 .flags = .{
1478 while (true) {1482 .eor = (msg.flags & posix.MSG.EOR) != 0,
1479 pool.checkCancel() catch |err| return .{ err, message_i };1483 .trunc = (msg.flags & posix.MSG.TRUNC) != 0,
14801484 .ctrunc = (msg.flags & posix.MSG.CTRUNC) != 0,
1481 const message = &message_buffer[message_i];1485 .oob = (msg.flags & posix.MSG.OOB) != 0,
1482 const remaining_data_buffer = data_buffer[data_i..];1486 .errqueue = (msg.flags & posix.MSG.ERRQUEUE) != 0,
1483 var storage: PosixAddress = undefined;1487 },
1484 var iov: posix.iovec = .{ .base = remaining_data_buffer.ptr, .len = remaining_data_buffer.len };1488 };
1485 var msg: posix.msghdr = .{1489 message_i += 1;
1486 .name = &storage.any,1490 continue;
1487 .namelen = @sizeOf(PosixAddress),1491 },
1488 .iov = (&iov)[0..1],1492 .AGAIN => while (true) {
1489 .iovlen = 1,1493 pool.checkCancel() catch |err| return .{ err, message_i };
1490 .control = message.control.ptr,1494 if (message_i != 0) return .{ null, message_i };
1491 .controllen = message.control.len,1495
1492 .flags = undefined,1496 const max_poll_ms = std.math.maxInt(u31);
1493 };1497 const timeout_ms: u31 = if (deadline) |d| t: {
14941498 const duration = d.durationFromNow(pool.io()) catch |err| return .{ err, message_i };
1495 const rc = posix.system.recvmsg(handle, &msg, posix_flags);1499 if (duration.nanoseconds <= 0) return .{ error.Timeout, message_i };
1496 switch (posix.errno(rc)) {1500 break :t @intCast(@min(max_poll_ms, duration.toMilliseconds()));
1497 .SUCCESS => {1501 } else max_poll_ms;
1498 const data = remaining_data_buffer[0..@intCast(rc)];1502
1499 data_i += data.len;1503 const poll_rc = posix.system.poll(&poll_fds, poll_fds.len, timeout_ms);
1500 message.* = .{1504 switch (posix.errno(poll_rc)) {
1501 .from = addressFromPosix(&storage),1505 .SUCCESS => {
1502 .data = data,1506 if (poll_rc == 0) {
1503 .control = if (msg.control) |ptr| @as([*]u8, @ptrCast(ptr))[0..msg.controllen] else message.control,1507 // Although spurious timeouts are OK, when no deadline
1504 .flags = .{1508 // is passed we must not return `error.Timeout`.
1505 .eor = (msg.flags & posix.MSG.EOR) != 0,1509 if (deadline == null) continue;
1506 .trunc = (msg.flags & posix.MSG.TRUNC) != 0,1510 return .{ error.Timeout, message_i };
1507 .ctrunc = (msg.flags & posix.MSG.CTRUNC) != 0,1511 }
1508 .oob = (msg.flags & posix.MSG.OOB) != 0,1512 continue :recv;
1509 .errqueue = (msg.flags & posix.MSG.ERRQUEUE) != 0,1513 },
1510 },1514 .INTR => continue,
1511 };1515
1512 message_i += 1;1516 .FAULT => |err| return .{ errnoBug(err), message_i },
1513 continue;1517 .INVAL => |err| return .{ errnoBug(err), message_i },
1514 },1518 .NOMEM => return .{ error.SystemResources, message_i },
1515 .AGAIN => continue :poll,1519 else => |err| return .{ posix.unexpectedErrno(err), message_i },
1516 .BADF => |err| return .{ errnoBug(err), message_i },
1517 .NFILE => return .{ error.SystemFdQuotaExceeded, message_i },
1518 .MFILE => return .{ error.ProcessFdQuotaExceeded, message_i },
1519 .INTR => continue,
1520 .FAULT => |err| return .{ errnoBug(err), message_i },
1521 .INVAL => |err| return .{ errnoBug(err), message_i },
1522 .NOBUFS => return .{ error.SystemResources, message_i },
1523 .NOMEM => return .{ error.SystemResources, message_i },
1524 .NOTCONN => return .{ error.SocketUnconnected, message_i },
1525 .NOTSOCK => |err| return .{ errnoBug(err), message_i },
1526 .MSGSIZE => return .{ error.MessageOversize, message_i },
1527 .PIPE => return .{ error.SocketUnconnected, message_i },
1528 .OPNOTSUPP => |err| return .{ errnoBug(err), message_i },
1529 .CONNRESET => return .{ error.ConnectionResetByPeer, message_i },
1530 .NETDOWN => return .{ error.NetworkDown, message_i },
1531 else => |err| return .{ posix.unexpectedErrno(err), message_i },
1532 }
1533 }1520 }
1534 },1521 },
1535 .INTR => continue,1522 .INTR => continue,
1523
1524 .BADF => |err| return .{ errnoBug(err), message_i },
1525 .NFILE => return .{ error.SystemFdQuotaExceeded, message_i },
1526 .MFILE => return .{ error.ProcessFdQuotaExceeded, message_i },
1536 .FAULT => |err| return .{ errnoBug(err), message_i },1527 .FAULT => |err| return .{ errnoBug(err), message_i },
1537 .INVAL => |err| return .{ errnoBug(err), message_i },1528 .INVAL => |err| return .{ errnoBug(err), message_i },
1529 .NOBUFS => return .{ error.SystemResources, message_i },
1538 .NOMEM => return .{ error.SystemResources, message_i },1530 .NOMEM => return .{ error.SystemResources, message_i },
1531 .NOTCONN => return .{ error.SocketUnconnected, message_i },
1532 .NOTSOCK => |err| return .{ errnoBug(err), message_i },
1533 .MSGSIZE => return .{ error.MessageOversize, message_i },
1534 .PIPE => return .{ error.SocketUnconnected, message_i },
1535 .OPNOTSUPP => |err| return .{ errnoBug(err), message_i },
1536 .CONNRESET => return .{ error.ConnectionResetByPeer, message_i },
1537 .NETDOWN => return .{ error.NetworkDown, message_i },
1539 else => |err| return .{ posix.unexpectedErrno(err), message_i },1538 else => |err| return .{ posix.unexpectedErrno(err), message_i },
1540 }1539 }
1541 }1540 }