diff --git a/lib/std/Io.zig b/lib/std/Io.zig index ec6afd24f9553e853a1eee5897b6b0f5dd363ff2..ee1519b7d9806407e2b4e6315553fb7840808eb8 100644 --- a/lib/std/Io.zig +++ b/lib/std/Io.zig @@ -235,7 +235,6 @@ pub const VTable = struct { netListenUnix: *const fn (?*anyopaque, *const net.UnixAddress, net.UnixAddress.ListenOptions) net.UnixAddress.ListenError!net.Socket.Handle, netConnectUnix: *const fn (?*anyopaque, *const net.UnixAddress) net.UnixAddress.ConnectError!net.Socket.Handle, netSocketCreatePair: *const fn (?*anyopaque, net.Socket.CreatePairOptions) net.Socket.CreatePairError![2]net.Socket, - netSend: *const fn (?*anyopaque, net.Socket.Handle, []net.OutgoingMessage, net.SendFlags) struct { ?net.Socket.SendError, usize }, netWrite: *const fn (?*anyopaque, dest: net.Socket.Handle, header: []const u8, data: []const []const u8, splat: usize) net.Stream.Writer.Error!usize, netWriteFile: *const fn (?*anyopaque, net.Socket.Handle, header: []const u8, *Io.File.Reader, Io.Limit) net.Stream.Writer.WriteFileError!usize, netClose: *const fn (?*anyopaque, sockets: []const net.Socket) void, @@ -252,6 +251,7 @@ pub const Operation = union(enum) { /// other systems this tag is unreachable. device_io_control: DeviceIoControl, net_receive: NetReceive, + net_send: NetSend, net_read: NetRead, pub const Tag = @typeInfo(Operation).@"union".tag_type.?; @@ -378,6 +378,49 @@ pub const Operation = union(enum) { pub const Result = struct { ?net.Socket.ReceiveError, usize }; }; + pub const NetSend = struct { + socket_handle: net.Socket.Handle, + messages: []net.OutgoingMessage, + flags: net.SendFlags, + + pub const Error = error{ + /// The socket type requires that message be sent atomically, and the + /// size of the message to be sent made this impossible. The message + /// was not transmitted, or was partially transmitted. + MessageOversize, + /// The output queue for a network interface was full. This generally indicates that the + /// interface has stopped sending, but may be caused by transient congestion. (Normally, + /// this does not occur in Linux. Packets are just silently dropped when a device queue + /// overflows.) + /// + /// This is also caused when there is not enough kernel memory available. + SystemResources, + /// No route to network. + NetworkUnreachable, + /// Network reached but no route to host. + HostUnreachable, + /// The local network interface used to reach the destination is offline. + NetworkDown, + /// The destination address is not listening. Can still occur for + /// connectionless messages. + ConnectionRefused, + /// Operating system or protocol does not support the address family. + AddressFamilyUnsupported, + /// Another TCP Fast Open is already in progress. + FastOpenAlreadyInProgress, + /// Network session was unexpectedly closed by recipient. + ConnectionResetByPeer, + /// Local end has been shut down on a connection-oriented socket, or + /// the socket was never connected. + SocketUnconnected, + /// An attempt was made to send to a network/broadcast address as + /// though it was a unicast address. + AccessDenied, + } || Io.UnexpectedError; + + pub const Result = struct { ?net.Socket.SendError, usize }; + }; + pub const NetRead = struct { socket_handle: net.Socket.Handle, data: [][]u8, @@ -2712,7 +2755,6 @@ pub const failing: std.Io = .{ .netListenUnix = failingNetListenUnix, .netConnectUnix = failingNetConnectUnix, .netSocketCreatePair = failingNetSocketCreatePair, - .netSend = failingNetSend, .netWrite = failingNetWrite, .netWriteFile = failingNetWriteFile, .netClose = unreachableNetClose, @@ -2860,6 +2902,7 @@ pub fn failingOperate(userdata: ?*anyopaque, operation: Operation) Cancelable!Op .file_write_streaming => .{ .file_write_streaming = error.InputOutput }, .device_io_control => unreachable, .net_receive => .{ .net_receive = .{ error.NetworkDown, 0 } }, + .net_send => .{ .net_send = .{ error.NetworkDown, 0 } }, .net_read => .{ .net_read = error.NetworkDown }, }; } @@ -3456,14 +3499,6 @@ pub fn failingNetSocketCreatePair(userdata: ?*anyopaque, options: net.Socket.Cre return error.OperationUnsupported; } -pub fn failingNetSend(userdata: ?*anyopaque, handle: net.Socket.Handle, messages: []net.OutgoingMessage, flags: net.SendFlags) struct { ?net.Socket.SendError, usize } { - _ = userdata; - _ = handle; - _ = messages; - _ = flags; - return .{ error.NetworkDown, 0 }; -} - pub fn failingNetWrite(userdata: ?*anyopaque, dest: net.Socket.Handle, header: []const u8, data: []const []const u8, splat: usize) net.Stream.Writer.Error!usize { _ = userdata; _ = dest; diff --git a/lib/std/Io/Dispatch.zig b/lib/std/Io/Dispatch.zig index e342a0a9f9606bfd6a8dff6cc17e900fba8d2129..c7ac107a90e0abaa09ea6c8a02dff2197888ecb1 100644 --- a/lib/std/Io/Dispatch.zig +++ b/lib/std/Io/Dispatch.zig @@ -458,7 +458,6 @@ pub fn io(ev: *Evented) Io { .netListenUnix = netListenUnixUnavailable, .netConnectUnix = netConnectUnixUnavailable, .netSocketCreatePair = netSocketCreatePairUnavailable, - .netSend = netSendUnavailable, .netWrite = netWriteUnavailable, .netWriteFile = netWriteFileUnavailable, .netClose = netClose, @@ -1713,6 +1712,7 @@ fn operate(userdata: ?*anyopaque, operation: Io.Operation) Io.Cancelable!Io.Oper }, .device_io_control => |*o| return .{ .device_io_control = try deviceIoControl(o) }, .net_receive => @panic("TODO implement net_receive operation"), + .net_send => @panic("TODO implement net_send operation"), .net_read => @panic("TODO implement net_read operation"), } } @@ -4866,20 +4866,6 @@ fn netSocketCreatePairUnavailable( return error.OperationUnsupported; } -fn netSendUnavailable( - userdata: ?*anyopaque, - handle: net.Socket.Handle, - messages: []net.OutgoingMessage, - flags: net.SendFlags, -) struct { ?net.Socket.SendError, usize } { - const ev: *Evented = @ptrCast(@alignCast(userdata)); - _ = ev; - _ = handle; - _ = messages; - _ = flags; - return .{ error.NetworkDown, 0 }; -} - fn netWriteUnavailable( userdata: ?*anyopaque, handle: net.Socket.Handle, diff --git a/lib/std/Io/Threaded.zig b/lib/std/Io/Threaded.zig index 555ff219d308cba7647a791bf770b28b6895c6d1..c40f2a922b11bf077d2d8b1a8f6d5fe20f5160a4 100644 --- a/lib/std/Io/Threaded.zig +++ b/lib/std/Io/Threaded.zig @@ -1951,10 +1951,6 @@ pub fn io(t: *Threaded) Io { else => netWritePosix, }, .netWriteFile = netWriteFile, - .netSend = switch (native_os) { - .windows => netSendWindows, - else => netSendPosix, - }, .netInterfaceNameResolve = netInterfaceNameResolve, .netInterfaceName = netInterfaceName, .netLookup = netLookup, @@ -2565,6 +2561,25 @@ fn operate(userdata: ?*anyopaque, operation: Io.Operation) Io.Cancelable!Io.Oper }; break :o .{ null, 1 }; } }, + .net_send => |*o| return .{ + .net_send = o: { + if (!have_networking) break :o .{ error.NetworkDown, 0 }; + if (is_windows) break :o netSendWindows(t, o.socket_handle, o.messages, o.flags); + const send_err, const sent = netSendPosix(t, o.socket_handle, o.messages, o.flags, false); + if (send_err) |err| switch (err) { + error.Canceled => |e| if (sent == 0) { + return e; + } else { + // Leave the `error.Canceled` for later, but don't try to send any more messages. + recancelInner(); + break :o .{ null, sent }; + }, + error.WouldBlock => unreachable, + else => |e| break :o .{ e, sent }, + }; + break :o .{ null, sent }; + }, + }, .net_read => |o| return .{ .net_read = netRead(o.socket_handle, o.data) catch |err| switch (err) { error.Canceled => |e| return e, @@ -2626,6 +2641,14 @@ fn batchAwaitAsync(userdata: ?*anyopaque, b: *Io.Batch) Io.Cancelable!void { }; poll_len += 1; }, + .net_send => |*o| { + poll_buffer[poll_len] = .{ + .fd = o.socket_handle, + .events = posix.POLL.OUT | posix.POLL.ERR, + .revents = 0, + }; + poll_len += 1; + }, .net_read => |o| { poll_buffer[poll_len] = .{ .fd = o.socket_handle, @@ -2811,6 +2834,35 @@ fn batchAwaitConcurrent(userdata: ?*anyopaque, b: *Io.Batch, timeout: Io.Timeout storage.* = .{ .completion = .{ .node = .{ .next = .none }, .result = result } }; b.completed.tail = index; }, + .net_send => |*o| nb: { + const result: Io.Operation.Result = .{ + .net_send = o: { + const send_err, const sent = netSendPosix(t, o.socket_handle, o.messages, o.flags, true); + if (send_err) |err| switch (err) { + error.Canceled => |e| if (sent == 0) { + return e; + } else { + // Leave the `error.Canceled` for later, but don't try to send any more messages. + recancelInner(); + break :o .{ null, sent }; + }, + error.WouldBlock => { + if (sent != 0) break :o .{ null, sent }; + try poll_storage.add(o.socket_handle, posix.POLL.OUT | posix.POLL.ERR); + break :nb; + }, + else => |e| break :o .{ e, sent }, + }; + break :o .{ null, sent }; + }, + }; + switch (b.completed.tail) { + .none => b.completed.head = index, + else => |tail_index| b.storage[tail_index.toIndex()].completion.node.next = index, + } + storage.* = .{ .completion = .{ .node = .{ .next = .none }, .result = result } }; + b.completed.tail = index; + }, .net_read => |o| try poll_storage.add(o.socket_handle, posix.POLL.IN | posix.POLL.ERR), } index = submission.node.next; @@ -3007,6 +3059,7 @@ fn batchApc( .file_write_streaming => .{ .file_write_streaming = ntWriteFileResult(iosb) }, .device_io_control => .{ .device_io_control = iosb.* }, .net_receive => unreachable, + .net_send => unreachable, .net_read => unreachable, }; storage.* = .{ .completion = .{ .node = .{ .next = .none }, .result = result } }; @@ -3216,6 +3269,13 @@ fn batchDrainSubmittedWindows(t: *Threaded, b: *Io.Batch, concurrency: bool) (Io .net_receive = netReceiveWindows(t, o.socket_handle, o.message_buffer, o.data_buffer, o.flags), }); }, + .net_send => |*o| { + // TODO integrate with overlapped I/O or equivalent to avoid this error + if (concurrency) return error.ConcurrencyUnavailable; + batchCompleteBlockingWindows(b, operation_userdata, .{ + .net_send = netSendWindows(t, o.socket_handle, o.messages, o.flags), + }); + }, .net_read => |*o| { // TODO integrate with overlapped I/O or equivalent to avoid this error if (concurrency) return error.ConcurrencyUnavailable; @@ -12866,13 +12926,13 @@ fn netReadWindows(socket_handle: net.Socket.Handle, data: [][]u8) net.Stream.Rea } fn netSendPosix( - userdata: ?*anyopaque, + t: *Threaded, socket_handle: net.Socket.Handle, messages: []net.OutgoingMessage, flags: net.SendFlags, -) struct { ?net.Socket.SendError, usize } { + nonblocking: bool, +) struct { ?(net.Socket.SendError || error{WouldBlock}), usize } { if (!have_networking) return .{ error.NetworkDown, 0 }; - const t: *Threaded = @ptrCast(@alignCast(userdata)); const posix_flags: u32 = @as(u32, if (@hasDecl(posix.MSG, "CONFIRM") and flags.confirm) posix.MSG.CONFIRM else 0) | @@ -12880,6 +12940,7 @@ fn netSendPosix( @as(u32, if (@hasDecl(posix.MSG, "EOR") and flags.eor) posix.MSG.EOR else 0) | @as(u32, if (@hasDecl(posix.MSG, "OOB") and flags.oob) posix.MSG.OOB else 0) | @as(u32, if (@hasDecl(posix.MSG, "FASTOPEN") and flags.fastopen) posix.MSG.FASTOPEN else 0) | + @as(u32, if (@hasDecl(posix.MSG, "DONTWAIT") and nonblocking) posix.MSG.DONTWAIT else 0) | posix.MSG.NOSIGNAL; var i: usize = 0; @@ -12895,13 +12956,12 @@ fn netSendPosix( } fn netSendWindows( - userdata: ?*anyopaque, + t: *Threaded, socket_handle: net.Socket.Handle, messages: []net.OutgoingMessage, flags: net.SendFlags, ) struct { ?net.Socket.SendError, usize } { if (!have_networking) return .{ error.NetworkDown, 0 }; - const t: *Threaded = @ptrCast(@alignCast(userdata)); for (messages, 0..) |*m, i| { t.netSendOneWindows(socket_handle, m, flags) catch |err| return .{ err, i }; } @@ -12953,7 +13013,7 @@ fn netSendOnePosix( socket_handle: net.Socket.Handle, message: *net.OutgoingMessage, flags: u32, -) net.Socket.SendError!void { +) (net.Socket.SendError || error{WouldBlock})!void { _ = t; var addr: PosixAddress = undefined; var iovec: posix.iovec_const = .{ .base = @constCast(message.data_ptr), .len = message.data_len }; @@ -12981,6 +13041,7 @@ fn netSendOnePosix( continue; }, .ACCES => return syscall.fail(error.AccessDenied), + .AGAIN => return syscall.fail(error.WouldBlock), .ALREADY => return syscall.fail(error.FastOpenAlreadyInProgress), .CONNRESET => return syscall.fail(error.ConnectionResetByPeer), .MSGSIZE => return syscall.fail(error.MessageOversize), @@ -13008,7 +13069,7 @@ fn netSendManyPosix( socket_handle: net.Socket.Handle, messages: []net.OutgoingMessage, flags: u32, -) net.Socket.SendError!usize { +) (net.Socket.SendError || error{WouldBlock})!usize { var msg_buffer: [64]posix.system.mmsghdr = undefined; var addr_buffer: [msg_buffer.len]PosixAddress = undefined; var iovecs_buffer: [msg_buffer.len]posix.iovec = undefined; @@ -13051,6 +13112,7 @@ fn netSendManyPosix( continue; }, .ACCES => return syscall.fail(error.AccessDenied), + .AGAIN => return syscall.fail(error.WouldBlock), .ALREADY => return syscall.fail(error.FastOpenAlreadyInProgress), .CONNRESET => return syscall.fail(error.ConnectionResetByPeer), .MSGSIZE => return syscall.fail(error.MessageOversize), @@ -13063,7 +13125,6 @@ fn netSendManyPosix( .NOTCONN => return syscall.fail(error.SocketUnconnected), .NETDOWN => return syscall.fail(error.NetworkDown), - .AGAIN => |err| return syscall.errnoBug(err), .BADF => |err| return syscall.errnoBug(err), // File descriptor used after closed. .DESTADDRREQ => |err| return syscall.errnoBug(err), // The socket is not connection-mode, and no peer address is set. .FAULT => |err| return syscall.errnoBug(err), // An invalid user space address was specified for an argument. @@ -14609,7 +14670,7 @@ fn lookupDns( message_i += 1; } } - _ = netSendPosix(t, socket.handle, message_buffer[0..message_i], .{}); + _ = netSendPosix(t, socket.handle, message_buffer[0..message_i], .{}, false); } const timeout: Io.Timeout = .{ .deadline = .{ @@ -14657,7 +14718,7 @@ fn lookupDns( .data_ptr = query.ptr, .data_len = query.len, }; - _ = netSendPosix(t, socket.handle, (&retry_message)[0..1], .{}); + _ = netSendPosix(t, socket.handle, (&retry_message)[0..1], .{}, false); continue; }, else => continue, diff --git a/lib/std/Io/Uring.zig b/lib/std/Io/Uring.zig index 86f5bee4bf69329a21257fe6be0946fb0eb28153..6250bebfccef782d9a682f1bbbb5d4e23a6d010d 100644 --- a/lib/std/Io/Uring.zig +++ b/lib/std/Io/Uring.zig @@ -778,7 +778,6 @@ pub fn io(ev: *Evented) Io { .netListenUnix = netListenUnixUnavailable, .netConnectUnix = netConnectUnixUnavailable, .netSocketCreatePair = netSocketCreatePairUnavailable, - .netSend = netSendUnavailable, .netWrite = netWriteUnavailable, .netWriteFile = netWriteFileUnavailable, .netClose = netClose, @@ -2107,6 +2106,12 @@ fn operate(userdata: ?*anyopaque, operation: Io.Operation) Io.Cancelable!Io.Oper }; }, }, + .net_send => |o| .{ + .net_send = r: { + _ = o; + break :r .{ error.NetworkDown, 0 }; // TODO + }, + }, .net_read => |o| .{ .net_read = r: { _ = o; @@ -2400,6 +2405,10 @@ fn batchDrainSubmitted( _ = o; @panic("TODO implement batchDrainSubmitted for net_receive"); }, + .net_send => |o| { + _ = o; + @panic("TODO implement batchDrainSubmitted for net_send"); + }, .net_read => |o| { _ = o; @panic("TODO implement batchDrainSubmitted for net_read"); @@ -2505,6 +2514,7 @@ fn batchDrainReady(batch: *Io.Batch) Io.Timeout.Error!void { }, .device_io_control => unreachable, .net_receive => @panic("TODO"), + .net_send => @panic("TODO"), .net_read => @panic("TODO"), })) |result| { switch (batch.completed.tail) { @@ -5054,20 +5064,6 @@ fn netSocketCreatePairUnavailable( return error.OperationUnsupported; } -fn netSendUnavailable( - userdata: ?*anyopaque, - handle: net.Socket.Handle, - messages: []net.OutgoingMessage, - flags: net.SendFlags, -) struct { ?net.Socket.SendError, usize } { - const ev: *Evented = @ptrCast(@alignCast(userdata)); - _ = ev; - _ = handle; - _ = messages; - _ = flags; - return .{ error.NetworkDown, 0 }; -} - fn netReceive( ev: *Evented, cancel_region: *CancelRegion, diff --git a/lib/std/Io/net.zig b/lib/std/Io/net.zig index c778dcde66f4b773a3d3f95f698c628eacb3f5f1..0c2e571ed0a080c511ded2a82f544268b309a7d9 100644 --- a/lib/std/Io/net.zig +++ b/lib/std/Io/net.zig @@ -1088,52 +1088,73 @@ pub const Socket = struct { io.vtable.netClose(io.userdata, sockets); } - pub const SendError = error{ - /// The socket type requires that message be sent atomically, and the - /// size of the message to be sent made this impossible. The message - /// was not transmitted, or was partially transmitted. - MessageOversize, - /// The output queue for a network interface was full. This generally indicates that the - /// interface has stopped sending, but may be caused by transient congestion. (Normally, - /// this does not occur in Linux. Packets are just silently dropped when a device queue - /// overflows.) - /// - /// This is also caused when there is not enough kernel memory available. - SystemResources, - /// No route to network. - NetworkUnreachable, - /// Network reached but no route to host. - HostUnreachable, - /// The local network interface used to reach the destination is offline. - NetworkDown, - /// The destination address is not listening. Can still occur for - /// connectionless messages. - ConnectionRefused, - /// Operating system or protocol does not support the address family. - AddressFamilyUnsupported, - /// Another TCP Fast Open is already in progress. - FastOpenAlreadyInProgress, - /// Network session was unexpectedly closed by recipient. - ConnectionResetByPeer, - /// Local end has been shut down on a connection-oriented socket, or - /// the socket was never connected. - SocketUnconnected, - /// An attempt was made to send to a network/broadcast address as - /// though it was a unicast address. - AccessDenied, - } || Io.UnexpectedError || Io.Cancelable; + pub const SendError = Io.Operation.NetSend.Error || Io.Cancelable; /// Transfers `data` to `dest`, connectionless, in one packet. pub fn send(s: *const Socket, io: Io, dest: *const IpAddress, data: []const u8) SendError!void { var message: OutgoingMessage = .{ .address = dest, .data_ptr = data.ptr, .data_len = data.len }; - const err, const n = io.vtable.netSend(io.userdata, s.handle, (&message)[0..1], .{}); - if (n != 1) return err.?; + const maybe_err, const count = (try io.operate(.{ .net_send = .{ + .socket_handle = s.handle, + .messages = (&message)[0..1], + .flags = .{}, + } })).net_send; + if (maybe_err) |err| { + assert(count == 0); + return err; + } else { + assert(count == 1); + } if (message.data_len != data.len) return error.MessageOversize; } + pub const SendTimeoutError = SendError || Io.Timeout.Error || Io.ConcurrentError; + + pub fn sendTimeout( + s: *const Socket, + io: Io, + dest: *const IpAddress, + data: []const u8, + timeout: Io.Timeout, + ) SendTimeoutError!void { + var message: OutgoingMessage = .{ .address = dest, .data_ptr = data.ptr, .data_len = data.len }; + const maybe_err, const count = (try io.operateTimeout(.{ .net_send = .{ + .socket_handle = s.handle, + .messages = (&message)[0..1], + .flags = .{}, + } }, timeout)).net_send; + if (maybe_err) |err| return err; + assert(1 == count); + if (message.data_len != data.len) return error.MessageOversize; + } + + /// Deprecated; use `sendManyTimeout` with a timeout of `.none`. + /// + /// If this function returns an error, some (but not all) of `messages` may + /// still have been sent. This condition is not reported by this function, + /// but is reported by `sendManyTimeout`. pub fn sendMany(s: *const Socket, io: Io, messages: []OutgoingMessage, flags: SendFlags) SendError!void { - const err, const n = io.vtable.netSend(io.userdata, s.handle, messages, flags); - if (n != messages.len) return err.?; + const result = try io.operate(.{ .net_send = .{ + .socket_handle = s.handle, + .messages = messages, + .flags = flags, + } }); + const maybe_send_err, _ = result.net_send; + return maybe_send_err orelse {}; + } + + pub fn sendManyTimeout( + s: *const Socket, + io: Io, + messages: []OutgoingMessage, + flags: SendFlags, + timeout: Io.Timeout, + ) struct { ?SendTimeoutError, usize } { + const result = io.operateTimeout(.{ .net_send = .{ + .socket_handle = s.handle, + .messages = messages, + .flags = flags, + } }, timeout) catch |err| return .{ err, 0 }; + return result.net_send; } pub const ReceiveError = Io.Operation.NetReceive.Error || Io.Cancelable; diff --git a/lib/std/Io/net/test.zig b/lib/std/Io/net/test.zig index e40352899217ca822b2b70d22e3413910c088c62..038269b30b3d7dcf927038bd6719357569eddd53 100644 --- a/lib/std/Io/net/test.zig +++ b/lib/std/Io/net/test.zig @@ -394,7 +394,7 @@ test "UDP send and receive" { try testing.expectEqualStrings(&send_data, received.data); } -test "UDP send and receiveTimeout" { +test "UDP sendTimeout and receiveTimeout" { const io = testing.io; const localhost: net.IpAddress = .{ .ip4 = .loopback(0) }; @@ -406,15 +406,16 @@ test "UDP send and receiveTimeout" { const send_sock = try localhost.bind(io, .{ .mode = .dgram }); defer send_sock.close(io); + const six_hours: Io.Timeout = .{ .duration = .{ .clock = .awake, .raw = .fromSeconds(21600) } }; + const send_data: [3]u8 = .{ '1', '2', '3' }; - try send_sock.send(io, &recv_sock.address, &send_data); - - const timeo: Io.Timeout = .{ .duration = .{ .clock = .awake, .raw = .fromSeconds(10) } }; - var recv_buf: [4]u8 = undefined; - const received = recv_sock.receiveTimeout(io, &recv_buf, timeo) catch |err| switch (err) { + send_sock.sendTimeout(io, &recv_sock.address, &send_data, six_hours) catch |err| switch (err) { error.ConcurrencyUnavailable => return error.SkipZigTest, else => |e| return e, }; + + var recv_buf: [4]u8 = undefined; + const received = try recv_sock.receiveTimeout(io, &recv_buf, six_hours); try testing.expect(received.from.eql(&send_sock.address)); try testing.expectEqualStrings(&send_data, received.data); @@ -422,7 +423,7 @@ test "UDP send and receiveTimeout" { try testing.expectError(error.Timeout, recv_sock.receiveTimeout(io, &recv_buf, short)); } -test "UDP sendMany 1 recvManyTimeout 2" { +test "UDP sendMany 2 and receive 2" { const io = testing.io; const localhost: net.IpAddress = .{ .ip4 = .loopback(0) }; @@ -434,26 +435,119 @@ test "UDP sendMany 1 recvManyTimeout 2" { const send_sock = try localhost.bind(io, .{ .mode = .dgram }); defer send_sock.close(io); + const send_data: [3]u8 = .{ '1', '2', '3' }; + var send_msgs: [2]Io.net.OutgoingMessage = @splat(.{ + .address = &recv_sock.address, + .data_ptr = &send_data, + .data_len = 3, + }); + // note sendMany is deprecated, but should remain tested until removed + try send_sock.sendMany(io, &send_msgs, .{}); + try testing.expectEqual(3, send_msgs[0].data_len); + try testing.expectEqual(3, send_msgs[1].data_len); + + var recv_buf: [4]u8 = undefined; + + { + const first = try recv_sock.receive(io, &recv_buf); + try testing.expect(first.from.eql(&send_sock.address)); + try testing.expectEqualStrings(&send_data, first.data); + } + + { + const second = try recv_sock.receive(io, &recv_buf); + try testing.expect(second.from.eql(&send_sock.address)); + try testing.expectEqualStrings(&send_data, second.data); + } +} + +test "UDP sendManyTimeout 1 recvManyTimeout 2" { + const io = testing.io; + const localhost: net.IpAddress = .{ .ip4 = .loopback(0) }; + + const recv_sock = localhost.bind(io, .{ .mode = .dgram }) catch |err| switch (err) { + error.NetworkDown => return error.SkipZigTest, + else => |e| return e, + }; + defer recv_sock.close(io); + const send_sock = try localhost.bind(io, .{ .mode = .dgram }); + defer send_sock.close(io); + + const six_hours: Io.Timeout = .{ .duration = .{ .clock = .awake, .raw = .fromSeconds(21600) } }; const send_data: [3]u8 = .{ '1', '2', '3' }; var send_msg: Io.net.OutgoingMessage = .{ .address = &recv_sock.address, .data_ptr = &send_data, .data_len = 3, }; - try send_sock.sendMany(io, (&send_msg)[0..1], .{}); + + const maybe_send_err, const send_count = send_sock.sendManyTimeout(io, (&send_msg)[0..1], .{}, six_hours); + if (maybe_send_err) |err| switch (err) { + error.ConcurrencyUnavailable => return error.SkipZigTest, + else => |e| return e, + }; + try testing.expectEqual(1, send_count); try testing.expectEqual(3, send_msg.data_len); - // This should not wait 10 seconds for the absent second message, it should - // complete as soon as the first one arrives + // This should complete as soon as the first message arrives, and not stall + // for the timeout waiting on the second one. var recv_msgs: [2]net.IncomingMessage = @splat(.init); var recv_buf: [10]u8 = undefined; - const timeo: Io.Timeout = .{ .duration = .{ .clock = .awake, .raw = .fromSeconds(10) } }; - const maybe_recv_err, const recv_count = recv_sock.receiveManyTimeout(io, &recv_msgs, &recv_buf, .{}, timeo); - if (maybe_recv_err) |err| switch (err) { - error.ConcurrencyUnavailable => return error.SkipZigTest, - else => |e| return e, - }; + const maybe_recv_err, const recv_count = recv_sock.receiveManyTimeout(io, &recv_msgs, &recv_buf, .{}, six_hours); + if (maybe_recv_err) |err| return err; try testing.expectEqual(1, recv_count); try testing.expect(recv_msgs[0].from.eql(&send_sock.address)); try testing.expectEqualStrings(&send_data, recv_msgs[0].data); } + +fn testUdpSender(io: Io, send_sock: Io.net.Socket, send_data: []const u8, dest: Io.net.IpAddress) !void { + try io.sleep(.fromMilliseconds(10), .boot); + try send_sock.send(io, &dest, send_data); + try send_sock.send(io, &dest, send_data); + try io.sleep(.fromMilliseconds(10), .boot); + try send_sock.send(io, &dest, send_data); +} + +test "UDP concurrency and timeouts" { + const io = testing.io; + const localhost: net.IpAddress = .{ .ip4 = .loopback(0) }; + + const recv_sock = localhost.bind(io, .{ .mode = .dgram }) catch |err| switch (err) { + error.NetworkDown => return error.SkipZigTest, + else => |e| return e, + }; + defer recv_sock.close(io); + const send_sock = try localhost.bind(io, .{ .mode = .dgram }); + defer send_sock.close(io); + + const send_data: [3]u8 = .{ '1', '2', '3' }; + var sender = io.async(testUdpSender, .{ io, send_sock, &send_data, recv_sock.address }); + defer sender.cancel(io) catch {}; + + // Because the sender is async (may execute serially or concurrently) and + // it has some 10ms timing gaps, it's likely that there will be a variety + // of random behaviors (total iterations, messages per iteration) on the + // receive end of things related to the target and runner conditions. This + // should still suceed so long as all 3 packets arrive in reasonable time. + const six_hours: Io.Timeout = .{ .duration = .{ .clock = .awake, .raw = .fromSeconds(21600) } }; + var received: usize = 0; + for (0..3) |_| { + var recv_msgs: [3]net.IncomingMessage = @splat(.init); + var recv_buf: [9]u8 = undefined; + const maybe_recv_err, const recv_count = recv_sock.receiveManyTimeout(io, &recv_msgs, &recv_buf, .{}, six_hours); + if (maybe_recv_err) |err| switch (err) { + error.ConcurrencyUnavailable => return error.SkipZigTest, + else => |e| return e, + }; + received += recv_count; + try testing.expect(received <= 3); + for (0..recv_count) |i| { + const msg = recv_msgs[i]; + try testing.expect(msg.from.eql(&send_sock.address)); + try testing.expectEqualStrings(&send_data, msg.data); + } + if (received == 3) break; + } + try testing.expectEqual(3, received); + try sender.await(io); // ensure sender didn't fail +}