diff --git a/lib/std/Io.zig b/lib/std/Io.zig index b0354cb595ae51e74f3a7f2a178f0a730c662acb..cb18bb8c6e67f64d0658c15910a3c326cc825b65 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, - 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, netShutdown: *const fn (?*anyopaque, handle: net.Socket.Handle, how: net.ShutdownHow) net.ShutdownError!void, @@ -253,6 +252,7 @@ pub const Operation = union(enum) { net_receive: NetReceive, net_send: NetSend, net_read: NetRead, + net_write: NetWrite, pub const Tag = @typeInfo(Operation).@"union".tag_type.?; @@ -438,6 +438,43 @@ pub const Operation = union(enum) { pub const Result = Error!usize; }; + pub const NetWrite = struct { + socket_handle: net.Socket.Handle, + header: []const u8 = &.{}, + data: []const []const u8, + splat: usize = 1, + + pub const Error = error{ + /// Another TCP Fast Open is already in progress. + FastOpenAlreadyInProgress, + /// Network session was unexpectedly closed by recipient. + ConnectionResetByPeer, + /// 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 down. + NetworkDown, + /// The destination address is not listening. + ConnectionRefused, + /// The passed address didn't have the correct address family in its sa_family field. + AddressFamilyUnsupported, + /// Local end has been shut down on a connection-oriented socket, or + /// the socket was never connected. + SocketUnconnected, + SocketNotBound, + } || Io.UnexpectedError; + + pub const Result = Error!usize; + }; + pub const Result = Result: { const operation_info = @typeInfo(Operation).@"union"; const operation_count = operation_info.field_names.len; @@ -2771,7 +2808,6 @@ pub const failing: std.Io = .{ .netListenUnix = failingNetListenUnix, .netConnectUnix = failingNetConnectUnix, .netSocketCreatePair = failingNetSocketCreatePair, - .netWrite = failingNetWrite, .netWriteFile = failingNetWriteFile, .netClose = unreachableNetClose, .netShutdown = failingNetShutdown, @@ -2920,6 +2956,7 @@ pub fn failingOperate(userdata: ?*anyopaque, operation: Operation) Cancelable!Op .net_receive => .{ .net_receive = .{ error.NetworkDown, 0 } }, .net_send => .{ .net_send = .{ error.NetworkDown, 0 } }, .net_read => .{ .net_read = error.NetworkDown }, + .net_write => .{ .net_write = error.NetworkDown }, }; } @@ -3515,15 +3552,6 @@ pub fn failingNetSocketCreatePair(userdata: ?*anyopaque, options: net.Socket.Cre return error.OperationUnsupported; } -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; - _ = header; - _ = data; - _ = splat; - return error.NetworkDown; -} - pub fn failingNetWriteFile(userdata: ?*anyopaque, handle: net.Socket.Handle, header: []const u8, file_reader: *Io.File.Reader, limit: Io.Limit) net.Stream.Writer.WriteFileError!usize { _ = userdata; _ = handle; diff --git a/lib/std/Io/Dispatch.zig b/lib/std/Io/Dispatch.zig index c7ac107a90e0abaa09ea6c8a02dff2197888ecb1..6628c73b381d040a071c1137f675027e4e980152 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, - .netWrite = netWriteUnavailable, .netWriteFile = netWriteFileUnavailable, .netClose = netClose, .netShutdown = netShutdownUnavailable, @@ -1714,6 +1713,7 @@ fn operate(userdata: ?*anyopaque, operation: Io.Operation) Io.Cancelable!Io.Oper .net_receive => @panic("TODO implement net_receive operation"), .net_send => @panic("TODO implement net_send operation"), .net_read => @panic("TODO implement net_read operation"), + .net_write => @panic("TODO implement net_write operation"), } } @@ -2136,6 +2136,7 @@ fn batchDrainSubmitted( .device_io_control => {}, .net_receive => @panic("TODO implement batched net_receive"), .net_read => @panic("TODO implement batched net_read"), + .net_write => @panic("TODO implement batched net_write"), }; if (concurrency) return error.ConcurrencyUnavailable; break :result try operate(ev, storage.submission.operation); @@ -2196,6 +2197,7 @@ fn batchSourceEvent(context: ?*anyopaque) callconv(.c) void { .device_io_control => unreachable, .net_receive => @panic("TODO implement batched net_receive"), .net_read => @panic("TODO implement batched net_read"), + .net_write => @panic("TODO implement batched net_write"), }; switch (pending.node.prev) { @@ -4866,22 +4868,6 @@ fn netSocketCreatePairUnavailable( return error.OperationUnsupported; } -fn netWriteUnavailable( - userdata: ?*anyopaque, - handle: net.Socket.Handle, - header: []const u8, - data: []const []const u8, - splat: usize, -) net.Stream.Writer.Error!usize { - const ev: *Evented = @ptrCast(@alignCast(userdata)); - _ = ev; - _ = handle; - _ = header; - _ = data; - _ = splat; - return error.NetworkDown; -} - fn netWriteFileUnavailable( userdata: ?*anyopaque, socket_handle: net.Socket.Handle, diff --git a/lib/std/Io/Kqueue.zig b/lib/std/Io/Kqueue.zig index ea8cbb6cad024be2fe987cc3296bd3e70006aab9..8f8be91d59bd3f3131ad5c08955da9de09b90e26 100644 --- a/lib/std/Io/Kqueue.zig +++ b/lib/std/Io/Kqueue.zig @@ -650,10 +650,8 @@ pub fn io(k: *Kqueue) Io { .netBindIp = netBindIp, .netConnectIp = netConnectIp, .netConnectUnix = netConnectUnix, - .netClose = netClose, .netShutdown = netShutdown, .netRead = netRead, - .netWrite = netWrite, .netSend = netSend, .netReceive = netReceive, .netInterfaceNameResolve = netInterfaceNameResolve, @@ -1270,23 +1268,6 @@ fn netRead(userdata: ?*anyopaque, fd: net.Socket.Handle, data: [][]u8) net.Strea } } -fn netWrite(userdata: ?*anyopaque, dest: net.Socket.Handle, header: []const u8, data: []const []const u8, splat: usize) net.Stream.Writer.Error!usize { - const k: *Kqueue = @ptrCast(@alignCast(userdata)); - _ = k; - _ = dest; - _ = header; - _ = data; - _ = splat; - @panic("TODO"); -} - -fn netClose(userdata: ?*anyopaque, sockets: []const net.Socket) void { - const k: *Kqueue = @ptrCast(@alignCast(userdata)); - _ = k; - _ = sockets; - @panic("TODO"); -} - fn netShutdown(userdata: ?*anyopaque, handle: net.Socket.Handle, how: net.ShutdownHow) net.ShutdownError!void { const k: *Kqueue = @ptrCast(@alignCast(userdata)); _ = k; diff --git a/lib/std/Io/Threaded.zig b/lib/std/Io/Threaded.zig index 5ed8b097ea8ebe6b1fc88ff47e785f2be73ffa74..4b92964ee12a3bd8d2096df1b6db32f3b0889bbd 100644 --- a/lib/std/Io/Threaded.zig +++ b/lib/std/Io/Threaded.zig @@ -1946,10 +1946,6 @@ pub fn io(t: *Threaded) Io { .windows => netShutdownWindows, else => netShutdownPosix, }, - .netWrite = switch (native_os) { - .windows => netWriteWindows, - else => netWritePosix, - }, .netWriteFile = netWriteFile, .netInterfaceNameResolve = netInterfaceNameResolve, .netInterfaceName = netInterfaceName, @@ -2586,6 +2582,15 @@ fn operate(userdata: ?*anyopaque, operation: Io.Operation) Io.Cancelable!Io.Oper else => |e| e, }, }, + .net_write => |o| return .{ + .net_write = (if (is_windows) + netWriteWindows(o.socket_handle, o.header, o.data, o.splat) + else + netWritePosix(o.socket_handle, o.header, o.data, o.splat)) catch |err| switch (err) { + error.Canceled => |e| return e, + else => |e| e, + }, + }, } } @@ -2657,6 +2662,14 @@ fn batchAwaitAsync(userdata: ?*anyopaque, b: *Io.Batch) Io.Cancelable!void { }; poll_len += 1; }, + .net_write => |o| { + poll_buffer[poll_len] = .{ + .fd = o.socket_handle, + .events = posix.POLL.OUT | posix.POLL.ERR, + .revents = 0, + }; + poll_len += 1; + }, } index = submission.node.next; } @@ -2864,6 +2877,7 @@ fn batchAwaitConcurrent(userdata: ?*anyopaque, b: *Io.Batch, timeout: Io.Timeout b.completed.tail = index; }, .net_read => |o| try poll_storage.add(o.socket_handle, posix.POLL.IN | posix.POLL.ERR), + .net_write => |o| try poll_storage.add(o.socket_handle, posix.POLL.OUT | posix.POLL.ERR), } index = submission.node.next; } @@ -3061,6 +3075,7 @@ fn batchApc( .net_receive => unreachable, .net_send => unreachable, .net_read => unreachable, + .net_write => unreachable, }; storage.* = .{ .completion = .{ .node = .{ .next = .none }, .result = result } }; }, @@ -3286,6 +3301,16 @@ fn batchDrainSubmittedWindows(t: *Threaded, b: *Io.Batch, concurrency: bool) (Io }, }); }, + .net_write => |*o| { + // TODO integrate with overlapped I/O or equivalent to avoid this error + if (concurrency) return error.ConcurrencyUnavailable; + batchCompleteBlockingWindows(b, operation_userdata, .{ + .net_write = netWriteWindows(o.socket_handle, o.header, o.data, o.splat) catch |err| switch (err) { + error.Canceled => |e| return e, + else => |e| e, + }, + }); + }, } index = submission.node.next; } @@ -13283,15 +13308,12 @@ fn netReceiveOneWindows( } fn netWritePosix( - userdata: ?*anyopaque, fd: net.Socket.Handle, header: []const u8, data: []const []const u8, splat: usize, ) net.Stream.Writer.Error!usize { if (!have_networking) return error.NetworkDown; - const t: *Threaded = @ptrCast(@alignCast(userdata)); - _ = t; var iovecs: [max_iovecs_len]posix.iovec_const = undefined; var msg: posix.msghdr_const = .{ @@ -13377,15 +13399,12 @@ fn netWritePosix( } fn netWriteWindows( - userdata: ?*anyopaque, handle: net.Socket.Handle, header: []const u8, data: []const []const u8, splat: usize, ) net.Stream.Writer.Error!usize { if (!have_networking) return error.NetworkDown; - const t: *Threaded = @ptrCast(@alignCast(userdata)); - _ = t; var iovecs: [max_iovecs_len]windows.AFD.WSABUF(.@"const") = undefined; var len: u32 = 0; diff --git a/lib/std/Io/Uring.zig b/lib/std/Io/Uring.zig index 6250bebfccef782d9a682f1bbbb5d4e23a6d010d..4aace9849cfb1a7c7a636d1ec348f664358c19c4 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, - .netWrite = netWriteUnavailable, .netWriteFile = netWriteFileUnavailable, .netClose = netClose, .netShutdown = netShutdown, @@ -2118,6 +2117,7 @@ fn operate(userdata: ?*anyopaque, operation: Io.Operation) Io.Cancelable!Io.Oper break :r error.NetworkDown; // TODO }, }, + .net_write => @panic("TODO implement net_write operation"), }; } @@ -2413,6 +2413,10 @@ fn batchDrainSubmitted( _ = o; @panic("TODO implement batchDrainSubmitted for net_read"); }, + .net_write => |o| { + _ = o; + @panic("TODO implement batchDrainSubmitted for net_write"); + }, })) |result| { switch (batch.completed.tail) { .none => batch.completed.head = index, @@ -2516,6 +2520,7 @@ fn batchDrainReady(batch: *Io.Batch) Io.Timeout.Error!void { .net_receive => @panic("TODO"), .net_send => @panic("TODO"), .net_read => @panic("TODO"), + .net_write => @panic("TODO"), })) |result| { switch (batch.completed.tail) { .none => batch.completed.head = index, @@ -5153,22 +5158,6 @@ fn netReceive( } } -fn netWriteUnavailable( - userdata: ?*anyopaque, - handle: net.Socket.Handle, - header: []const u8, - data: []const []const u8, - splat: usize, -) net.Stream.Writer.Error!usize { - const ev: *Evented = @ptrCast(@alignCast(userdata)); - _ = ev; - _ = handle; - _ = header; - _ = data; - _ = splat; - return error.NetworkDown; -} - fn netWriteFileUnavailable( userdata: ?*anyopaque, socket_handle: net.Socket.Handle, diff --git a/lib/std/Io/net.zig b/lib/std/Io/net.zig index 0c2e571ed0a080c511ded2a82f544268b309a7d9..779ad28a8eaae233782bc4b7181cb566ebd1ffd6 100644 --- a/lib/std/Io/net.zig +++ b/lib/std/Io/net.zig @@ -1348,33 +1348,7 @@ pub const Stream = struct { err: ?Error = null, write_file_err: ?WriteFileError = null, - pub const Error = error{ - /// Another TCP Fast Open is already in progress. - FastOpenAlreadyInProgress, - /// Network session was unexpectedly closed by recipient. - ConnectionResetByPeer, - /// 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 down. - NetworkDown, - /// The destination address is not listening. - ConnectionRefused, - /// The passed address didn't have the correct address family in its sa_family field. - AddressFamilyUnsupported, - /// Local end has been shut down on a connection-oriented socket, or - /// the socket was never connected. - SocketUnconnected, - SocketNotBound, - } || Io.UnexpectedError || Io.Cancelable; + pub const Error = Io.Operation.NetWrite.Error || Io.Cancelable; pub const WriteFileError = Error || error{ /// The `Io` implementation cannot offer a more efficient @@ -1407,7 +1381,16 @@ pub const Stream = struct { const io = w.io; const buffered = io_w.buffered(); const handle = w.stream.socket.handle; - const n = io.vtable.netWrite(io.userdata, handle, buffered, data, splat) catch |err| { + const result = io.operate(.{ .net_write = .{ + .socket_handle = handle, + .header = buffered, + .data = data, + .splat = splat, + } }) catch |err| { + w.err = err; + return error.WriteFailed; + }; + const n = result.net_write catch |err| { w.err = err; return error.WriteFailed; };