Commit 60c4bdb14c
Changed files (5)
lib
lib/std/Io/net/HostName.zig
@@ -539,7 +539,7 @@ pub const ResolvConf = struct {
.search_buffer = undefined,
.search_len = 0,
.ndots = 1,
- .timeout = 5,
+ .timeout = .seconds(5),
.attempts = 2,
};
@@ -589,7 +589,7 @@ pub const ResolvConf = struct {
switch (std.meta.stringToEnum(Option, name) orelse continue) {
.ndots => rc.ndots = @min(value, 15),
.attempts => rc.attempts = @min(value, 10),
- .timeout => rc.timeout = @min(value, 60),
+ .timeout => rc.timeout = .seconds(@min(value, 60)),
}
},
.nameserver => {
@@ -638,14 +638,15 @@ pub const ResolvConf = struct {
const socket = s: {
if (any_ip6) ip6: {
const ip6_addr: IpAddress = .{ .ip6 = .unspecified(0) };
- const socket = ip6_addr.bind(io, .{ .ip6_only = true }) catch |err| switch (err) {
- error.AddressFamilyNotSupported => break :ip6,
+ const socket = ip6_addr.bind(io, .{ .ip6_only = true, .mode = .dgram }) catch |err| switch (err) {
+ error.AddressFamilyUnsupported => break :ip6,
+ else => |e| return e,
};
break :s socket;
}
any_ip6 = false;
const ip4_addr: IpAddress = .{ .ip4 = .unspecified(0) };
- const socket = try ip4_addr.bind(io, .{});
+ const socket = try ip4_addr.bind(io, .{ .mode = .dgram });
break :s socket;
};
defer socket.close();
lib/std/Io/net.zig
@@ -6,26 +6,41 @@ const assert = std.debug.assert;
pub const HostName = @import("net/HostName.zig");
-pub const ListenError = std.net.Address.ListenError || Io.Cancelable;
-
-pub const BindError = std.net.Address.BindError || Io.Cancelable;
-
-pub const ListenOptions = struct {
- /// How many connections the kernel will accept on the application's behalf.
- /// If more than this many connections pool in the kernel, clients will start
- /// seeing "Connection refused".
- kernel_backlog: u31 = 128,
- /// Sets SO_REUSEADDR and SO_REUSEPORT on POSIX.
- /// Sets SO_REUSEADDR on Windows, which is roughly equivalent.
- reuse_address: bool = false,
- force_nonblocking: bool = false,
-};
-
-pub const BindOptions = struct {
- /// The socket is restricted to sending and receiving IPv6 packets only.
- /// In this case, an IPv4 and an IPv6 application can bind to a single port
- /// at the same time.
- ip6_only: bool = false,
+/// Source of truth: Internet Assigned Numbers Authority (IANA)
+pub const Protocol = enum(u32) {
+ hopopts = 0,
+ icmp = 1,
+ igmp = 2,
+ ipip = 4,
+ tcp = 6,
+ egp = 8,
+ pup = 12,
+ udp = 17,
+ idp = 22,
+ tp = 29,
+ dccp = 33,
+ ipv6 = 41,
+ routing = 43,
+ fragment = 44,
+ rsvp = 46,
+ gre = 47,
+ esp = 50,
+ ah = 51,
+ icmpv6 = 58,
+ none = 59,
+ dstopts = 60,
+ mtp = 92,
+ beetph = 94,
+ encap = 98,
+ pim = 103,
+ comp = 108,
+ sctp = 132,
+ mh = 135,
+ udplite = 136,
+ mpls = 137,
+ ethernet = 143,
+ raw = 255,
+ mptcp = 262,
};
pub const IpAddress = union(enum) {
@@ -132,11 +147,69 @@ pub const IpAddress = union(enum) {
};
}
+ pub const ListenError = error{
+ /// The address is already taken. Can occur when bound port is 0 but
+ /// all ephemeral ports are already in use.
+ AddressInUse,
+ /// A nonexistent interface was requested or the requested address was not local.
+ AddressUnavailable,
+ /// The local network interface used to reach the destination is offline.
+ NetworkSubsystemDown,
+ /// Insufficient memory or other resource internal to the operating system.
+ SystemResources,
+ /// Per-process limit on the number of open file descriptors has been reached.
+ ProcessFdQuotaExceeded,
+ /// System-wide limit on the total number of open files has been reached.
+ SystemFdQuotaExceeded,
+ /// The requested address family (IPv4 or IPv6) is not supported by the operating system.
+ AddressFamilyUnsupported,
+ } || Io.UnexpectedError || Io.Cancelable;
+
+ pub const ListenOptions = struct {
+ /// How many connections the kernel will accept on the application's behalf.
+ /// If more than this many connections pool in the kernel, clients will start
+ /// seeing "Connection refused".
+ kernel_backlog: u31 = 128,
+ /// Sets SO_REUSEADDR and SO_REUSEPORT on POSIX.
+ /// Sets SO_REUSEADDR on Windows, which is roughly equivalent.
+ reuse_address: bool = false,
+ };
+
/// Waits for a TCP connection. When using this API, `bind` does not need
/// to be called. The returned `Server` has an open `stream`.
pub fn listen(address: IpAddress, io: Io, options: ListenOptions) ListenError!Server {
- return io.vtable.listen(io.userdata, address, options);
- }
+ return io.vtable.tcpListen(io.userdata, address, options);
+ }
+
+ pub const BindError = error{
+ /// The address is already taken. Can occur when bound port is 0 but
+ /// all ephemeral ports are already in use.
+ AddressInUse,
+ /// A nonexistent interface was requested or the requested address was not local.
+ AddressUnavailable,
+ /// The address is not valid for the address family of socket.
+ AddressFamilyUnsupported,
+ /// Insufficient memory or other resource internal to the operating system.
+ SystemResources,
+ /// The local network interface used to reach the destination is offline.
+ NetworkSubsystemDown,
+ ProtocolUnsupportedBySystem,
+ ProtocolUnsupportedByAddressFamily,
+ /// Per-process limit on the number of open file descriptors has been reached.
+ ProcessFdQuotaExceeded,
+ /// System-wide limit on the total number of open files has been reached.
+ SystemFdQuotaExceeded,
+ SocketModeUnsupported,
+ } || Io.UnexpectedError || Io.Cancelable;
+
+ pub const BindOptions = struct {
+ /// The socket is restricted to sending and receiving IPv6 packets only.
+ /// In this case, an IPv4 and an IPv6 application can bind to a single port
+ /// at the same time.
+ ip6_only: bool = false,
+ mode: Socket.Mode,
+ protocol: ?Protocol = null,
+ };
/// Associates an address with a `Socket` which can be used to receive UDP
/// packets and other kinds of non-streaming messages. See `listen` for a
@@ -145,7 +218,7 @@ pub const IpAddress = union(enum) {
/// One bound `Socket` can be used to receive messages from multiple
/// different addresses.
pub fn bind(address: IpAddress, io: Io, options: BindOptions) BindError!Socket {
- return io.vtable.bind(io.userdata, address, options);
+ return io.vtable.ipBind(io.userdata, address, options);
}
};
@@ -255,7 +328,7 @@ pub const Ip6Address = struct {
pub fn fromIp4(ip4: Ip4Address) Ip6Address {
const b = &ip4.bytes;
return .{
- .bytes = .{ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0xff, 0xff, b[0], b[1], b[2], b[3] },
+ .bytes = .{ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0xff, 0xff, b[0], b[1], b[2], b[3] },
.port = ip4.port,
};
}
@@ -682,7 +755,25 @@ pub const Interface = struct {
pub const Socket = struct {
handle: Handle,
/// Contains the resolved ephemeral port number if requested.
- bind_address: IpAddress,
+ address: IpAddress,
+
+ pub const Mode = enum {
+ /// Provides sequenced, reliable, two-way, connection-based byte
+ /// streams. An out-of-band data transmission mechanism may be
+ /// supported.
+ stream,
+ /// Supports datagrams (connectionless, unreliable messages of a fixed
+ /// maximum length).
+ dgram,
+ /// Provides a sequenced, reliable, two-way connection-based data
+ /// transmission path for datagrams of fixed maximum length; a consumer
+ /// is required to read an entire packet with each input system call.
+ seqpacket,
+ /// Provides raw network protocol access.
+ raw,
+ /// Provides a reliable datagram layer that does not guarantee ordering.
+ rdm,
+ };
/// Underlying platform-defined type which may or may not be
/// interchangeable with a file system file descriptor.
@@ -691,8 +782,48 @@ pub const Socket = struct {
else => std.posix.fd_t,
};
- pub fn close(s: Socket, io: Io) void {
- return io.vtable.netClose(io.userdata, s);
+ pub fn close(s: *Socket, io: Io) void {
+ io.vtable.netClose(io.userdata, s.handle);
+ s.handle = undefined;
+ }
+
+ 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 is not transmitted.
+ MessageTooBig,
+ /// 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.
+ NetworkSubsystemDown,
+ /// The destination address is not listening. Can still occur for
+ /// connectionless messages.
+ ConnectionRefused,
+ /// Operating system or protocol does not support the address family.
+ AddressFamilyUnsupported,
+ } || Io.UnexpectedError || Io.Cancelable;
+
+ /// Transfers `data` to `dest`, connectionless.
+ pub fn send(s: *const Socket, io: Io, dest: *const IpAddress, data: []const u8) SendError!void {
+ return io.vtable.netSend(io.userdata, s.handle, dest, data);
+ }
+
+ pub const ReceiveError = error{} || Io.Cancelable;
+
+ /// Transfers `data` from `source`, connectionless.
+ ///
+ /// Returned slice has same pointer as `buffer` with possibly shorter length.
+ pub fn receive(s: *const Socket, io: Io, source: *const IpAddress, buffer: []u8) ReceiveError![]u8 {
+ const n = try io.vtable.netReceive(io.userdata, s.handle, source, buffer);
+ return buffer[0..n];
}
};
@@ -784,11 +915,6 @@ pub const Stream = struct {
pub const Server = struct {
socket: Socket,
- pub const Connection = struct {
- stream: Stream,
- address: IpAddress,
- };
-
pub fn deinit(s: *Server, io: Io) void {
s.socket.close(io);
s.* = undefined;
@@ -796,9 +922,8 @@ pub const Server = struct {
pub const AcceptError = std.posix.AcceptError || Io.Cancelable;
- /// Blocks until a client connects to the server. The returned `Connection` has
- /// an open stream.
- pub fn accept(s: *Server, io: Io) AcceptError!Connection {
+ /// Blocks until a client connects to the server.
+ pub fn accept(s: *Server, io: Io) AcceptError!Stream {
return io.vtable.accept(io, s);
}
};
lib/std/Io/Threaded.zig
@@ -124,8 +124,19 @@ pub fn io(pool: *Pool) Io {
.now = now,
.sleep = sleep,
- .listen = listen,
- .accept = accept,
+ .listen = switch (builtin.os.tag) {
+ .windows => @panic("TODO"),
+ else => listenPosix,
+ },
+ .accept = switch (builtin.os.tag) {
+ .windows => @panic("TODO"),
+ else => acceptPosix,
+ },
+ .ipBind = switch (builtin.os.tag) {
+ .windows => @panic("TODO"),
+ else => ipBindPosix,
+ },
+ .netClose = netClose,
.netRead = switch (builtin.os.tag) {
.windows => @panic("TODO"),
else => netReadPosix,
@@ -134,7 +145,8 @@ pub fn io(pool: *Pool) Io {
.windows => @panic("TODO"),
else => netWritePosix,
},
- .netClose = netClose,
+ .netSend = netSend,
+ .netReceive = netReceive,
.netInterfaceNameResolve = netInterfaceNameResolve,
.netInterfaceName = netInterfaceName,
},
@@ -460,7 +472,7 @@ fn asyncDetached(
fn await(
userdata: ?*anyopaque,
- any_future: *std.Io.AnyFuture,
+ any_future: *Io.AnyFuture,
result: []u8,
result_alignment: std.mem.Alignment,
) void {
@@ -984,59 +996,228 @@ fn select(userdata: ?*anyopaque, futures: []const *Io.AnyFuture) usize {
return result.?;
}
-fn listen(userdata: ?*anyopaque, address: Io.net.IpAddress, options: Io.net.ListenOptions) Io.net.ListenError!Io.net.Server {
+fn listenPosix(
+ userdata: ?*anyopaque,
+ address: Io.net.IpAddress,
+ options: Io.net.IpAddress.ListenOptions,
+) Io.net.IpAddress.ListenError!Io.net.Server {
const pool: *Pool = @ptrCast(@alignCast(userdata));
- try pool.checkCancel();
-
- const nonblock: u32 = if (options.force_nonblocking) posix.SOCK.NONBLOCK else 0;
- const sock_flags = posix.SOCK.STREAM | posix.SOCK.CLOEXEC | nonblock;
- const proto: u32 = posix.IPPROTO.TCP;
- const family = posixAddressFamily(address);
- const sockfd = try posix.socket(family, sock_flags, proto);
- const stream: std.net.Stream = .{ .handle = sockfd };
- errdefer stream.close();
+ const family = posixAddressFamily(&address);
+ const protocol: u32 = posix.IPPROTO.TCP;
+ const socket_fd = while (true) {
+ try pool.checkCancel();
+ const flags: u32 = posix.SOCK.STREAM | if (socket_flags_unsupported) 0 else posix.SOCK.CLOEXEC;
+ const socket_rc = posix.system.socket(family, flags, protocol);
+ switch (posix.errno(socket_rc)) {
+ .SUCCESS => {
+ const fd: posix.fd_t = @intCast(socket_rc);
+ errdefer posix.close(fd);
+ if (socket_flags_unsupported) while (true) {
+ try pool.checkCancel();
+ switch (posix.errno(posix.system.fcntl(fd, posix.F.SETFD, posix.FD_CLOEXEC))) {
+ .SUCCESS => break,
+ .INTR => continue,
+ else => |err| return posix.unexpectedErrno(err),
+ }
+ };
+ break fd;
+ },
+ .INTR => continue,
+ .AFNOSUPPORT => return error.AddressFamilyUnsupported,
+ .MFILE => return error.ProcessFdQuotaExceeded,
+ .NFILE => return error.SystemFdQuotaExceeded,
+ .NOBUFS => return error.SystemResources,
+ .NOMEM => return error.SystemResources,
+ else => |err| return posix.unexpectedErrno(err),
+ }
+ };
+ errdefer posix.close(socket_fd);
if (options.reuse_address) {
- try posix.setsockopt(
- sockfd,
- posix.SOL.SOCKET,
- posix.SO.REUSEADDR,
- &std.mem.toBytes(@as(c_int, 1)),
- );
- if (@hasDecl(posix.SO, "REUSEPORT") and family != posix.AF.UNIX) {
- try posix.setsockopt(
- sockfd,
- posix.SOL.SOCKET,
- posix.SO.REUSEPORT,
- &std.mem.toBytes(@as(c_int, 1)),
- );
- }
+ try setSocketOption(pool, socket_fd, posix.SOL.SOCKET, posix.SO.REUSEADDR, 1);
+ if (@hasDecl(posix.SO, "REUSEPORT"))
+ try setSocketOption(pool, socket_fd, posix.SOL.SOCKET, posix.SO.REUSEPORT, 1);
}
var storage: PosixAddress = undefined;
var socklen = addressToPosix(address, &storage);
- try posix.bind(sockfd, &storage.any, socklen);
- try posix.listen(sockfd, options.kernel_backlog);
- try posix.getsockname(sockfd, &storage.any, &socklen);
+ try posixBind(pool, socket_fd, &storage.any, socklen);
+
+ while (true) {
+ try pool.checkCancel();
+ switch (posix.errno(posix.system.listen(socket_fd, options.kernel_backlog))) {
+ .SUCCESS => break,
+ .ADDRINUSE => return error.AddressInUse,
+ .BADF => |err| return errnoBug(err),
+ else => |err| return posix.unexpectedErrno(err),
+ }
+ }
+
+ try posixGetSockName(pool, socket_fd, &storage.any, &socklen);
return .{
- .listen_address = addressFromPosix(&storage),
- .stream = .{ .handle = stream.handle },
+ .socket = .{
+ .handle = socket_fd,
+ .address = addressFromPosix(&storage),
+ },
};
}
-fn accept(userdata: ?*anyopaque, server: *Io.net.Server) Io.net.Server.AcceptError!Io.net.Server.Connection {
+fn posixBind(pool: *Pool, socket_fd: posix.socket_t, addr: *const posix.sockaddr, addr_len: posix.socklen_t) !void {
+ while (true) {
+ try pool.checkCancel();
+ switch (posix.errno(posix.system.bind(socket_fd, addr, addr_len))) {
+ .SUCCESS => break,
+ .INTR => continue,
+ .ADDRINUSE => return error.AddressInUse,
+ .BADF => |err| return errnoBug(err), // always a race condition if this error is returned
+ .INVAL => |err| return errnoBug(err), // invalid parameters
+ .NOTSOCK => |err| return errnoBug(err), // invalid `sockfd`
+ .AFNOSUPPORT => return error.AddressFamilyUnsupported,
+ .ADDRNOTAVAIL => return error.AddressUnavailable,
+ .FAULT => |err| return errnoBug(err), // invalid `addr` pointer
+ .NOMEM => return error.SystemResources,
+ else => |err| return posix.unexpectedErrno(err),
+ }
+ }
+}
+
+fn posixGetSockName(pool: *Pool, socket_fd: posix.fd_t, addr: *posix.sockaddr, addr_len: *posix.socklen_t) !void {
+ while (true) {
+ try pool.checkCancel();
+ switch (posix.errno(posix.system.getsockname(socket_fd, addr, addr_len))) {
+ .SUCCESS => break,
+ .INTR => continue,
+ .BADF => |err| return errnoBug(err), // always a race condition
+ .FAULT => |err| return errnoBug(err),
+ .INVAL => |err| return errnoBug(err), // invalid parameters
+ .NOTSOCK => |err| return errnoBug(err), // always a race condition
+ .NOBUFS => return error.SystemResources,
+ else => |err| return posix.unexpectedErrno(err),
+ }
+ }
+}
+
+fn setSocketOption(pool: *Pool, fd: posix.fd_t, level: i32, opt_name: u32, option: u32) !void {
+ const o: []const u8 = @ptrCast(&option);
+ while (true) {
+ try pool.checkCancel();
+ switch (posix.errno(posix.system.setsockopt(fd, level, opt_name, o.ptr, @intCast(o.len)))) {
+ .SUCCESS => return,
+ .INTR => continue,
+ .BADF => |err| return errnoBug(err), // always a race condition
+ .NOTSOCK => |err| return errnoBug(err), // always a race condition
+ .INVAL => |err| return errnoBug(err),
+ .FAULT => |err| return errnoBug(err),
+ else => |err| return posix.unexpectedErrno(err),
+ }
+ }
+}
+
+fn ipBindPosix(
+ userdata: ?*anyopaque,
+ address: Io.net.IpAddress,
+ options: Io.net.IpAddress.BindOptions,
+) Io.net.IpAddress.BindError!Io.net.Socket {
const pool: *Pool = @ptrCast(@alignCast(userdata));
- try pool.checkCancel();
+ const mode = posixSocketMode(options.mode);
+ const family = posixAddressFamily(&address);
+ const protocol = posixProtocol(options.protocol);
+ const socket_fd = while (true) {
+ try pool.checkCancel();
+ const flags: u32 = mode | if (socket_flags_unsupported) 0 else posix.SOCK.CLOEXEC;
+ const socket_rc = posix.system.socket(family, flags, protocol);
+ switch (posix.errno(socket_rc)) {
+ .SUCCESS => {
+ const fd: posix.fd_t = @intCast(socket_rc);
+ errdefer posix.close(fd);
+ if (socket_flags_unsupported) while (true) {
+ try pool.checkCancel();
+ switch (posix.errno(posix.system.fcntl(fd, posix.F.SETFD, posix.FD_CLOEXEC))) {
+ .SUCCESS => break,
+ .INTR => continue,
+ else => |err| return posix.unexpectedErrno(err),
+ }
+ };
+ break fd;
+ },
+ .INTR => continue,
+ .AFNOSUPPORT => return error.AddressFamilyUnsupported,
+ .INVAL => return error.ProtocolUnsupportedBySystem,
+ .MFILE => return error.ProcessFdQuotaExceeded,
+ .NFILE => return error.SystemFdQuotaExceeded,
+ .NOBUFS => return error.SystemResources,
+ .NOMEM => return error.SystemResources,
+ .PROTONOSUPPORT => return error.ProtocolUnsupportedByAddressFamily,
+ .PROTOTYPE => return error.SocketModeUnsupported,
+ else => |err| return posix.unexpectedErrno(err),
+ }
+ };
+
+ if (options.ip6_only) {
+ try setSocketOption(pool, socket_fd, posix.IPPROTO.IPV6, posix.IPV6.V6ONLY, 0);
+ }
var storage: PosixAddress = undefined;
- var addr_len: posix.socklen_t = @sizeOf(PosixAddress);
- const fd = try posix.accept(server.stream.handle, &storage.any, &addr_len, posix.SOCK.CLOEXEC);
+ var socklen = addressToPosix(address, &storage);
+ try posixBind(pool, socket_fd, &storage.any, socklen);
+ try posixGetSockName(pool, socket_fd, &storage.any, &socklen);
return .{
- .stream = .{ .handle = fd },
+ .handle = socket_fd,
.address = addressFromPosix(&storage),
};
}
+const socket_flags_unsupported = builtin.os.tag.isDarwin() or native_os == .haiku; // 💩💩
+const have_accept4 = !socket_flags_unsupported;
+
+fn acceptPosix(userdata: ?*anyopaque, server: *Io.net.Server) Io.net.Server.AcceptError!Io.net.Stream {
+ const pool: *Pool = @ptrCast(@alignCast(userdata));
+ const listen_fd = server.socket.handle;
+ var storage: PosixAddress = undefined;
+ var addr_len: posix.socklen_t = @sizeOf(PosixAddress);
+ const fd = while (true) {
+ try pool.checkCancel();
+ const rc = if (have_accept4)
+ posix.system.accept4(listen_fd, &storage.any, &addr_len, posix.SOCK.CLOEXEC)
+ else
+ posix.system.accept(listen_fd, &storage.any, &addr_len);
+ switch (posix.errno(rc)) {
+ .SUCCESS => {
+ const fd: posix.fd_t = @intCast(rc);
+ errdefer posix.close(fd);
+ if (!have_accept4) while (true) {
+ try pool.checkCancel();
+ switch (posix.errno(posix.system.fcntl(fd, posix.F.SETFD, posix.FD_CLOEXEC))) {
+ .SUCCESS => break,
+ .INTR => continue,
+ else => |err| return posix.unexpectedErrno(err),
+ }
+ };
+ break fd;
+ },
+ .INTR => continue,
+ .AGAIN => |err| return errnoBug(err),
+ .BADF => |err| return errnoBug(err), // always a race condition
+ .CONNABORTED => return error.ConnectionAborted,
+ .FAULT => |err| return errnoBug(err),
+ .INVAL => return error.SocketNotListening,
+ .NOTSOCK => |err| return errnoBug(err),
+ .MFILE => return error.ProcessFdQuotaExceeded,
+ .NFILE => return error.SystemFdQuotaExceeded,
+ .NOBUFS => return error.SystemResources,
+ .NOMEM => return error.SystemResources,
+ .OPNOTSUPP => |err| return errnoBug(err),
+ .PROTO => return error.ProtocolFailure,
+ .PERM => return error.BlockedByFirewall,
+ else => |err| return posix.unexpectedErrno(err),
+ }
+ };
+ return .{ .socket = .{
+ .handle = fd,
+ .address = addressFromPosix(&storage),
+ } };
+}
+
fn netReadPosix(userdata: ?*anyopaque, stream: Io.net.Stream, data: [][]u8) Io.net.Stream.Reader.Error!usize {
const pool: *Pool = @ptrCast(@alignCast(userdata));
try pool.checkCancel();
@@ -1052,11 +1233,41 @@ fn netReadPosix(userdata: ?*anyopaque, stream: Io.net.Stream, data: [][]u8) Io.n
}
const dest = iovecs_buffer[0..i];
assert(dest[0].len > 0);
- const n = try posix.readv(stream.handle, dest);
+ const n = try posix.readv(stream.socket.handle, dest);
if (n == 0) return error.EndOfStream;
return n;
}
+fn netSend(
+ userdata: ?*anyopaque,
+ handle: Io.net.Socket.Handle,
+ address: Io.net.IpAddress,
+ data: []const u8,
+) Io.net.Socket.SendError!void {
+ const pool: *Pool = @ptrCast(@alignCast(userdata));
+ try pool.checkCancel();
+
+ _ = handle;
+ _ = address;
+ _ = data;
+ @panic("TODO");
+}
+
+fn netReceive(
+ userdata: ?*anyopaque,
+ handle: Io.net.Socket.Handle,
+ address: Io.net.IpAddress,
+ buffer: []u8,
+) Io.net.Socket.ReceiveError!void {
+ const pool: *Pool = @ptrCast(@alignCast(userdata));
+ try pool.checkCancel();
+
+ _ = handle;
+ _ = address;
+ _ = buffer;
+ @panic("TODO");
+}
+
fn netWritePosix(
userdata: ?*anyopaque,
stream: Io.net.Stream,
@@ -1106,7 +1317,7 @@ fn netWritePosix(
},
};
const flags = posix.MSG.NOSIGNAL;
- return posix.sendmsg(stream.handle, &msg, flags);
+ return posix.sendmsg(stream.socket.handle, &msg, flags);
}
fn addBuf(v: []posix.iovec_const, i: *@FieldType(posix.msghdr_const, "iovlen"), bytes: []const u8) void {
@@ -1117,11 +1328,13 @@ fn addBuf(v: []posix.iovec_const, i: *@FieldType(posix.msghdr_const, "iovlen"),
i.* += 1;
}
-fn netClose(userdata: ?*anyopaque, stream: Io.net.Stream) void {
+fn netClose(userdata: ?*anyopaque, handle: Io.net.Socket.Handle) void {
const pool: *Pool = @ptrCast(@alignCast(userdata));
_ = pool;
- const net_stream: std.net.Stream = .{ .handle = stream.handle };
- return net_stream.close();
+ switch (native_os) {
+ .windows => windows.closesocket(handle) catch recoverableOsBugDetected(),
+ else => posix.close(handle),
+ }
}
fn netInterfaceNameResolve(
@@ -1153,13 +1366,13 @@ fn netInterfaceNameResolve(
try pool.checkCancel();
switch (posix.errno(posix.system.ioctl(sock_fd, posix.SIOCGIFINDEX, @intFromPtr(&ifr)))) {
.SUCCESS => return .{ .index = @bitCast(ifr.ifru.ivalue) },
- .INVAL => |err| return badErrno(err), // Bad parameters.
- .NOTTY => |err| return badErrno(err),
- .NXIO => |err| return badErrno(err),
- .BADF => |err| return badErrno(err), // Always a race condition.
- .FAULT => |err| return badErrno(err), // Bad pointer parameter.
+ .INVAL => |err| return errnoBug(err), // Bad parameters.
+ .NOTTY => |err| return errnoBug(err),
+ .NXIO => |err| return errnoBug(err),
+ .BADF => |err| return errnoBug(err), // Always a race condition.
+ .FAULT => |err| return errnoBug(err), // Bad pointer parameter.
.INTR => continue,
- .IO => |err| return badErrno(err), // sock_fd is not a file descriptor
+ .IO => |err| return errnoBug(err), // sock_fd is not a file descriptor
.NODEV => return error.InterfaceNotFound,
else => |err| return posix.unexpectedErrno(err),
}
@@ -1207,8 +1420,8 @@ const PosixAddress = extern union {
in6: posix.sockaddr.in6,
};
-fn posixAddressFamily(a: Io.net.IpAddress) posix.sa_family_t {
- return switch (a) {
+fn posixAddressFamily(a: *const Io.net.IpAddress) posix.sa_family_t {
+ return switch (a.*) {
.ip4 => posix.AF.INET,
.ip6 => posix.AF.INET6,
};
@@ -1267,9 +1480,27 @@ fn address6ToPosix(a: Io.net.Ip6Address) posix.sockaddr.in6 {
};
}
-fn badErrno(err: posix.E) Io.UnexpectedError {
+fn errnoBug(err: posix.E) Io.UnexpectedError {
switch (builtin.mode) {
.Debug => std.debug.panic("programmer bug caused syscall error: {t}", .{err}),
else => return error.Unexpected,
}
}
+
+fn posixSocketMode(mode: Io.net.Socket.Mode) u32 {
+ return switch (mode) {
+ .stream => posix.SOCK.STREAM,
+ .dgram => posix.SOCK.DGRAM,
+ .seqpacket => posix.SOCK.SEQPACKET,
+ .raw => posix.SOCK.RAW,
+ .rdm => posix.SOCK.RDM,
+ };
+}
+
+fn posixProtocol(protocol: ?Io.net.Protocol) u32 {
+ return @intFromEnum(protocol orelse return 0);
+}
+
+fn recoverableOsBugDetected() void {
+ if (builtin.mode == .Debug) unreachable;
+}
lib/std/Io.zig
@@ -663,13 +663,14 @@ pub const VTable = struct {
now: *const fn (?*anyopaque, clockid: std.posix.clockid_t) ClockGetTimeError!Timestamp,
sleep: *const fn (?*anyopaque, clockid: std.posix.clockid_t, deadline: Deadline) SleepError!void,
- listen: *const fn (?*anyopaque, address: net.IpAddress, options: net.ListenOptions) net.ListenError!net.Server,
- bind: *const fn (?*anyopaque, address: net.IpAddress, options: net.BindOptions) net.BindError!net.Socket,
- accept: *const fn (?*anyopaque, server: *net.Server) net.Server.AcceptError!net.Server.Connection,
- netSend: *const fn (?*anyopaque, address: net.IpAddress, data: []const []const u8) net.SendError!void,
+ listen: *const fn (?*anyopaque, address: net.IpAddress, options: net.IpAddress.ListenOptions) net.IpAddress.ListenError!net.Server,
+ accept: *const fn (?*anyopaque, server: *net.Server) net.Server.AcceptError!net.Stream,
+ ipBind: *const fn (?*anyopaque, address: net.IpAddress, options: net.IpAddress.BindOptions) net.IpAddress.BindError!net.Socket,
+ netSend: *const fn (?*anyopaque, handle: net.Socket.Handle, address: net.IpAddress, data: []const u8) net.Socket.SendError!void,
+ netReceive: *const fn (?*anyopaque, handle: net.Socket.Handle, address: net.IpAddress, buffer: []u8) net.Socket.ReceiveError!void,
netRead: *const fn (?*anyopaque, src: net.Stream, data: [][]u8) net.Stream.Reader.Error!usize,
netWrite: *const fn (?*anyopaque, dest: net.Stream, header: []const u8, data: []const []const u8, splat: usize) net.Stream.Writer.Error!usize,
- netClose: *const fn (?*anyopaque, socket: net.Socket) void,
+ netClose: *const fn (?*anyopaque, handle: net.Socket.Handle) void,
netInterfaceNameResolve: *const fn (?*anyopaque, *const net.Interface.Name) net.Interface.Name.ResolveError!net.Interface,
netInterfaceName: *const fn (?*anyopaque, net.Interface) net.Interface.NameError!net.Interface.Name,
};
@@ -711,6 +712,10 @@ pub const Duration = struct {
pub fn ms(x: u64) Duration {
return .{ .nanoseconds = @as(i96, x) * std.time.ns_per_ms };
}
+
+ pub fn seconds(x: u64) Duration {
+ return .{ .nanoseconds = @as(i96, x) * std.time.ns_per_s };
+ }
};
pub const Deadline = union(enum) {
duration: Duration,
lib/std/net.zig
@@ -2384,6 +2384,7 @@ pub const Stream = struct {
}
};
+/// A bound, listening TCP socket, ready to accept new connections.
pub const Server = struct {
listen_address: Address,
stream: Stream,