authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-07-19 23:36:20-07:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2025-10-02 16:30:59-07:00
log07cc4077fba5701ef582e17c795f9f5214915179
tree73921260ce07a499622d7c0c6ea1eda5b897a9ee
parentf680c095bd96b2d739c7a1257fb3bff9abbac14a

add some networking

3 files changed, 684 insertions(+), 8 deletions(-)

lib/std/Io.zig+7
...@@ -559,6 +559,7 @@ const Io = @This();...@@ -559,6 +559,7 @@ const Io = @This();
559559
560pub const EventLoop = @import("Io/EventLoop.zig");560pub const EventLoop = @import("Io/EventLoop.zig");
561pub const ThreadPool = @import("Io/ThreadPool.zig");561pub const ThreadPool = @import("Io/ThreadPool.zig");
562pub const net = @import("Io/net.zig");
562563
563userdata: ?*anyopaque,564userdata: ?*anyopaque,
564vtable: *const VTable,565vtable: *const VTable,
...@@ -656,6 +657,12 @@ pub const VTable = struct {...@@ -656,6 +657,12 @@ pub const VTable = struct {
656657
657 now: *const fn (?*anyopaque, clockid: std.posix.clockid_t) ClockGetTimeError!Timestamp,658 now: *const fn (?*anyopaque, clockid: std.posix.clockid_t) ClockGetTimeError!Timestamp,
658 sleep: *const fn (?*anyopaque, clockid: std.posix.clockid_t, deadline: Deadline) SleepError!void,659 sleep: *const fn (?*anyopaque, clockid: std.posix.clockid_t, deadline: Deadline) SleepError!void,
660
661 listen: *const fn (?*anyopaque, address: net.IpAddress, options: net.ListenOptions) net.ListenError!net.Server,
662 accept: *const fn (?*anyopaque, server: *net.Server) net.Server.AcceptError!net.Server.Connection,
663 netRead: *const fn (?*anyopaque, src: net.Stream, dest: *Io.Writer, limit: Io.Limit) net.Stream.Reader.Error!usize,
664 netWrite: *const fn (?*anyopaque, dest: net.Stream, header: []const u8, data: []const []const u8, splat: usize) net.Stream.Writer.Error!usize,
665 netClose: *const fn (?*anyopaque, stream: net.Stream) void,
659};666};
660667
661pub const Cancelable = error{668pub const Cancelable = error{
lib/std/Io/ThreadPool.zig+227-8
...@@ -3,6 +3,7 @@ const std = @import("../std.zig");...@@ -3,6 +3,7 @@ const std = @import("../std.zig");
3const Allocator = std.mem.Allocator;3const Allocator = std.mem.Allocator;
4const assert = std.debug.assert;4const assert = std.debug.assert;
5const WaitGroup = std.Thread.WaitGroup;5const WaitGroup = std.Thread.WaitGroup;
6const posix = std.posix;
6const Io = std.Io;7const Io = std.Io;
7const Pool = @This();8const Pool = @This();
89
...@@ -19,6 +20,9 @@ parallel_count: usize,...@@ -19,6 +20,9 @@ parallel_count: usize,
1920
20threadlocal var current_closure: ?*AsyncClosure = null;21threadlocal var current_closure: ?*AsyncClosure = null;
2122
23const max_iovecs_len = 8;
24const splat_buffer_size = 64;
25
22pub const Runnable = struct {26pub const Runnable = struct {
23 start: Start,27 start: Start,
24 node: std.SinglyLinkedList.Node = .{},28 node: std.SinglyLinkedList.Node = .{},
...@@ -107,6 +111,18 @@ pub fn io(pool: *Pool) Io {...@@ -107,6 +111,18 @@ pub fn io(pool: *Pool) Io {
107111
108 .now = now,112 .now = now,
109 .sleep = sleep,113 .sleep = sleep,
114
115 .listen = listen,
116 .accept = accept,
117 .netRead = switch (builtin.os.tag) {
118 .windows => @panic("TODO"),
119 else => netReadPosix,
120 },
121 .netWrite = switch (builtin.os.tag) {
122 .windows => @panic("TODO"),
123 else => netWritePosix,
124 },
125 .netClose = netClose,
110 },126 },
111 };127 };
112}128}
...@@ -461,7 +477,7 @@ fn cancel(...@@ -461,7 +477,7 @@ fn cancel(
461 .linux => _ = std.os.linux.tgkill(477 .linux => _ = std.os.linux.tgkill(
462 std.os.linux.getpid(),478 std.os.linux.getpid(),
463 @bitCast(cancel_tid),479 @bitCast(cancel_tid),
464 std.posix.SIG.IO,480 posix.SIG.IO,
465 ),481 ),
466 else => {},482 else => {},
467 },483 },
...@@ -635,7 +651,7 @@ fn closeFile(userdata: ?*anyopaque, file: Io.File) void {...@@ -635,7 +651,7 @@ fn closeFile(userdata: ?*anyopaque, file: Io.File) void {
635 return fs_file.close();651 return fs_file.close();
636}652}
637653
638fn pread(userdata: ?*anyopaque, file: Io.File, buffer: []u8, offset: std.posix.off_t) Io.File.PReadError!usize {654fn pread(userdata: ?*anyopaque, file: Io.File, buffer: []u8, offset: posix.off_t) Io.File.PReadError!usize {
639 const pool: *Pool = @alignCast(@ptrCast(userdata));655 const pool: *Pool = @alignCast(@ptrCast(userdata));
640 try pool.checkCancel();656 try pool.checkCancel();
641 const fs_file: std.fs.File = .{ .handle = file.handle };657 const fs_file: std.fs.File = .{ .handle = file.handle };
...@@ -645,7 +661,7 @@ fn pread(userdata: ?*anyopaque, file: Io.File, buffer: []u8, offset: std.posix.o...@@ -645,7 +661,7 @@ fn pread(userdata: ?*anyopaque, file: Io.File, buffer: []u8, offset: std.posix.o
645 };661 };
646}662}
647663
648fn pwrite(userdata: ?*anyopaque, file: Io.File, buffer: []const u8, offset: std.posix.off_t) Io.File.PWriteError!usize {664fn pwrite(userdata: ?*anyopaque, file: Io.File, buffer: []const u8, offset: posix.off_t) Io.File.PWriteError!usize {
649 const pool: *Pool = @alignCast(@ptrCast(userdata));665 const pool: *Pool = @alignCast(@ptrCast(userdata));
650 try pool.checkCancel();666 try pool.checkCancel();
651 const fs_file: std.fs.File = .{ .handle = file.handle };667 const fs_file: std.fs.File = .{ .handle = file.handle };
...@@ -655,20 +671,20 @@ fn pwrite(userdata: ?*anyopaque, file: Io.File, buffer: []const u8, offset: std....@@ -655,20 +671,20 @@ fn pwrite(userdata: ?*anyopaque, file: Io.File, buffer: []const u8, offset: std.
655 };671 };
656}672}
657673
658fn now(userdata: ?*anyopaque, clockid: std.posix.clockid_t) Io.ClockGetTimeError!Io.Timestamp {674fn now(userdata: ?*anyopaque, clockid: posix.clockid_t) Io.ClockGetTimeError!Io.Timestamp {
659 const pool: *Pool = @alignCast(@ptrCast(userdata));675 const pool: *Pool = @alignCast(@ptrCast(userdata));
660 try pool.checkCancel();676 try pool.checkCancel();
661 const timespec = try std.posix.clock_gettime(clockid);677 const timespec = try posix.clock_gettime(clockid);
662 return @enumFromInt(@as(i128, timespec.sec) * std.time.ns_per_s + timespec.nsec);678 return @enumFromInt(@as(i128, timespec.sec) * std.time.ns_per_s + timespec.nsec);
663}679}
664680
665fn sleep(userdata: ?*anyopaque, clockid: std.posix.clockid_t, deadline: Io.Deadline) Io.SleepError!void {681fn sleep(userdata: ?*anyopaque, clockid: posix.clockid_t, deadline: Io.Deadline) Io.SleepError!void {
666 const pool: *Pool = @alignCast(@ptrCast(userdata));682 const pool: *Pool = @alignCast(@ptrCast(userdata));
667 const deadline_nanoseconds: i96 = switch (deadline) {683 const deadline_nanoseconds: i96 = switch (deadline) {
668 .duration => |duration| duration.nanoseconds,684 .duration => |duration| duration.nanoseconds,
669 .timestamp => |timestamp| @intFromEnum(timestamp),685 .timestamp => |timestamp| @intFromEnum(timestamp),
670 };686 };
671 var timespec: std.posix.timespec = .{687 var timespec: posix.timespec = .{
672 .sec = @intCast(@divFloor(deadline_nanoseconds, std.time.ns_per_s)),688 .sec = @intCast(@divFloor(deadline_nanoseconds, std.time.ns_per_s)),
673 .nsec = @intCast(@mod(deadline_nanoseconds, std.time.ns_per_s)),689 .nsec = @intCast(@mod(deadline_nanoseconds, std.time.ns_per_s)),
674 };690 };
...@@ -682,7 +698,7 @@ fn sleep(userdata: ?*anyopaque, clockid: std.posix.clockid_t, deadline: Io.Deadl...@@ -682,7 +698,7 @@ fn sleep(userdata: ?*anyopaque, clockid: std.posix.clockid_t, deadline: Io.Deadl
682 .FAULT => unreachable,698 .FAULT => unreachable,
683 .INTR => {},699 .INTR => {},
684 .INVAL => return error.UnsupportedClock,700 .INVAL => return error.UnsupportedClock,
685 else => |err| return std.posix.unexpectedErrno(err),701 else => |err| return posix.unexpectedErrno(err),
686 }702 }
687 }703 }
688}704}
...@@ -718,3 +734,206 @@ fn select(userdata: ?*anyopaque, futures: []const *Io.AnyFuture) usize {...@@ -718,3 +734,206 @@ fn select(userdata: ?*anyopaque, futures: []const *Io.AnyFuture) usize {
718 }734 }
719 return result.?;735 return result.?;
720}736}
737
738fn listen(userdata: ?*anyopaque, address: Io.net.IpAddress, options: Io.net.ListenOptions) Io.net.ListenError!Io.net.Server {
739 const pool: *Pool = @alignCast(@ptrCast(userdata));
740 try pool.checkCancel();
741
742 const nonblock: u32 = if (options.force_nonblocking) posix.SOCK.NONBLOCK else 0;
743 const sock_flags = posix.SOCK.STREAM | posix.SOCK.CLOEXEC | nonblock;
744 const proto: u32 = posix.IPPROTO.TCP;
745 const family = posixAddressFamily(address);
746 const sockfd = try posix.socket(family, sock_flags, proto);
747 const stream: std.net.Stream = .{ .handle = sockfd };
748 errdefer stream.close();
749
750 if (options.reuse_address) {
751 try posix.setsockopt(
752 sockfd,
753 posix.SOL.SOCKET,
754 posix.SO.REUSEADDR,
755 &std.mem.toBytes(@as(c_int, 1)),
756 );
757 if (@hasDecl(posix.SO, "REUSEPORT") and family != posix.AF.UNIX) {
758 try posix.setsockopt(
759 sockfd,
760 posix.SOL.SOCKET,
761 posix.SO.REUSEPORT,
762 &std.mem.toBytes(@as(c_int, 1)),
763 );
764 }
765 }
766
767 var storage: PosixAddress = undefined;
768 var socklen = addressToPosix(address, &storage);
769 try posix.bind(sockfd, &storage.any, socklen);
770 try posix.listen(sockfd, options.kernel_backlog);
771 try posix.getsockname(sockfd, &storage.any, &socklen);
772 return .{
773 .listen_address = addressFromPosix(&storage),
774 .stream = .{ .handle = stream.handle },
775 };
776}
777
778fn accept(userdata: ?*anyopaque, server: *Io.net.Server) Io.net.Server.AcceptError!Io.net.Server.Connection {
779 const pool: *Pool = @alignCast(@ptrCast(userdata));
780 try pool.checkCancel();
781
782 var storage: PosixAddress = undefined;
783 var addr_len: posix.socklen_t = @sizeOf(PosixAddress);
784 const fd = try posix.accept(server.stream.handle, &storage.any, &addr_len, posix.SOCK.CLOEXEC);
785 return .{
786 .stream = .{ .handle = fd },
787 .address = addressFromPosix(&storage),
788 };
789}
790
791fn netReadPosix(
792 userdata: ?*anyopaque,
793 stream: Io.net.Stream,
794 w: *Io.Writer,
795 limit: Io.Limit,
796) Io.net.Stream.Reader.Error!usize {
797 const pool: *Pool = @alignCast(@ptrCast(userdata));
798 try pool.checkCancel();
799
800 var iovecs_buffer: [max_iovecs_len]posix.iovec = undefined;
801 const dest = try w.writableVectorPosix(&iovecs_buffer, limit);
802 assert(dest[0].len > 0);
803 const n = try posix.readv(stream.handle, dest);
804 if (n == 0) return error.EndOfStream;
805 return n;
806}
807
808fn netWritePosix(
809 userdata: ?*anyopaque,
810 stream: Io.net.Stream,
811 header: []const u8,
812 data: []const []const u8,
813 splat: usize,
814) Io.net.Stream.Writer.Error!usize {
815 const pool: *Pool = @alignCast(@ptrCast(userdata));
816 try pool.checkCancel();
817
818 var iovecs: [max_iovecs_len]posix.iovec_const = undefined;
819 var msg: posix.msghdr_const = .{
820 .name = null,
821 .namelen = 0,
822 .iov = &iovecs,
823 .iovlen = 0,
824 .control = null,
825 .controllen = 0,
826 .flags = 0,
827 };
828 addBuf(&iovecs, &msg.iovlen, header);
829 for (data[0 .. data.len - 1]) |bytes| addBuf(&iovecs, &msg.iovlen, bytes);
830 const pattern = data[data.len - 1];
831 if (iovecs.len - msg.iovlen != 0) switch (splat) {
832 0 => {},
833 1 => addBuf(&iovecs, &msg.iovlen, pattern),
834 else => switch (pattern.len) {
835 0 => {},
836 1 => {
837 var backup_buffer: [splat_buffer_size]u8 = undefined;
838 const splat_buffer = &backup_buffer;
839 const memset_len = @min(splat_buffer.len, splat);
840 const buf = splat_buffer[0..memset_len];
841 @memset(buf, pattern[0]);
842 addBuf(&iovecs, &msg.iovlen, buf);
843 var remaining_splat = splat - buf.len;
844 while (remaining_splat > splat_buffer.len and iovecs.len - msg.iovlen != 0) {
845 assert(buf.len == splat_buffer.len);
846 addBuf(&iovecs, &msg.iovlen, splat_buffer);
847 remaining_splat -= splat_buffer.len;
848 }
849 addBuf(&iovecs, &msg.iovlen, splat_buffer[0..remaining_splat]);
850 },
851 else => for (0..@min(splat, iovecs.len - msg.iovlen)) |_| {
852 addBuf(&iovecs, &msg.iovlen, pattern);
853 },
854 },
855 };
856 const flags = posix.MSG.NOSIGNAL;
857 return posix.sendmsg(stream.handle, &msg, flags);
858}
859
860fn addBuf(v: []posix.iovec_const, i: *@FieldType(posix.msghdr_const, "iovlen"), bytes: []const u8) void {
861 // OS checks ptr addr before length so zero length vectors must be omitted.
862 if (bytes.len == 0) return;
863 if (v.len - i.* == 0) return;
864 v[i.*] = .{ .base = bytes.ptr, .len = bytes.len };
865 i.* += 1;
866}
867
868fn netClose(userdata: ?*anyopaque, stream: Io.net.Stream) void {
869 const pool: *Pool = @alignCast(@ptrCast(userdata));
870 _ = pool;
871 const net_stream: std.net.Stream = .{ .handle = stream.handle };
872 return net_stream.close();
873}
874
875const PosixAddress = extern union {
876 any: posix.sockaddr,
877 in: posix.sockaddr.in,
878 in6: posix.sockaddr.in6,
879};
880
881fn posixAddressFamily(a: Io.net.IpAddress) posix.sa_family_t {
882 return switch (a) {
883 .ip4 => posix.AF.INET,
884 .ip6 => posix.AF.INET6,
885 };
886}
887
888fn addressFromPosix(posix_address: *PosixAddress) Io.net.IpAddress {
889 return switch (posix_address.any.family) {
890 posix.AF.INET => .{ .ip4 = address4FromPosix(&posix_address.in) },
891 posix.AF.INET6 => .{ .ip6 = address6FromPosix(&posix_address.in6) },
892 else => unreachable,
893 };
894}
895
896fn addressToPosix(a: Io.net.IpAddress, storage: *PosixAddress) posix.socklen_t {
897 return switch (a) {
898 .ip4 => |ip4| {
899 storage.in = address4ToPosix(ip4);
900 return @sizeOf(posix.sockaddr.in);
901 },
902 .ip6 => |ip6| {
903 storage.in6 = address6ToPosix(ip6);
904 return @sizeOf(posix.sockaddr.in6);
905 },
906 };
907}
908
909fn address4FromPosix(in: *posix.sockaddr.in) Io.net.Ip4Address {
910 return .{
911 .port = std.mem.bigToNative(u16, in.port),
912 .bytes = @bitCast(in.addr),
913 };
914}
915
916fn address6FromPosix(in6: *posix.sockaddr.in6) Io.net.Ip6Address {
917 return .{
918 .port = std.mem.bigToNative(u16, in6.port),
919 .bytes = in6.addr,
920 .flowinfo = in6.flowinfo,
921 .scope_id = in6.scope_id,
922 };
923}
924
925fn address4ToPosix(a: Io.net.Ip4Address) posix.sockaddr.in {
926 return .{
927 .port = std.mem.nativeToBig(u16, a.port),
928 .addr = @bitCast(a.bytes),
929 };
930}
931
932fn address6ToPosix(a: Io.net.Ip6Address) posix.sockaddr.in6 {
933 return .{
934 .port = std.mem.nativeToBig(u16, a.port),
935 .flowinfo = a.flowinfo,
936 .addr = a.bytes,
937 .scope_id = a.scope_id,
938 };
939}
lib/std/Io/net.zig created+450
...@@ -0,0 +1,450 @@
1const builtin = @import("builtin");
2const native_os = builtin.os.tag;
3const std = @import("../std.zig");
4const Io = std.Io;
5
6pub const ListenError = std.net.Address.ListenError || Io.Cancelable;
7
8pub const ListenOptions = struct {
9 /// How many connections the kernel will accept on the application's behalf.
10 /// If more than this many connections pool in the kernel, clients will start
11 /// seeing "Connection refused".
12 kernel_backlog: u31 = 128,
13 /// Sets SO_REUSEADDR and SO_REUSEPORT on POSIX.
14 /// Sets SO_REUSEADDR on Windows, which is roughly equivalent.
15 reuse_address: bool = false,
16 force_nonblocking: bool = false,
17};
18
19pub const IpAddress = union(enum) {
20 ip4: Ip4Address,
21 ip6: Ip6Address,
22
23 /// Parse the given IP address string into an `IpAddress` value.
24 pub fn parse(name: []const u8, port: u16) !IpAddress {
25 if (parseIp4(name, port)) |ip4| return ip4 else |err| switch (err) {
26 error.Overflow,
27 error.InvalidEnd,
28 error.InvalidCharacter,
29 error.Incomplete,
30 error.NonCanonical,
31 => {},
32 }
33
34 if (parseIp6(name, port)) |ip6| return ip6 else |err| switch (err) {
35 error.Overflow,
36 error.InvalidEnd,
37 error.InvalidCharacter,
38 error.Incomplete,
39 error.InvalidIpv4Mapping,
40 => {},
41 }
42
43 return error.InvalidIpAddressFormat;
44 }
45
46 pub fn parseIp6(buffer: []const u8, port: u16) Ip6Address.ParseError!IpAddress {
47 return .{ .ip6 = try Ip6Address.parse(buffer, port) };
48 }
49
50 pub fn parseIp4(buffer: []const u8, port: u16) Ip4Address.ParseError!IpAddress {
51 return .{ .ip4 = try Ip4Address.parse(buffer, port) };
52 }
53
54 /// Returns the port in native endian.
55 pub fn getPort(a: IpAddress) u16 {
56 return switch (a) {
57 inline .ip4, .ip6 => |x| x.port,
58 };
59 }
60
61 /// `port` is native-endian.
62 pub fn setPort(a: *IpAddress, port: u16) void {
63 switch (a) {
64 inline .ip4, .ip6 => |*x| x.port = port,
65 }
66 }
67
68 pub fn format(a: IpAddress, w: *std.io.Writer) std.io.Writer.Error!void {
69 switch (a) {
70 .ip4, .ip6 => |x| return x.format(w),
71 }
72 }
73
74 pub fn eql(a: IpAddress, b: IpAddress) bool {
75 return switch (a) {
76 .ip4 => |a_ip4| switch (b) {
77 .ip4 => |b_ip4| a_ip4.eql(b_ip4),
78 else => false,
79 },
80 .ip6 => |a_ip6| switch (b) {
81 .ip6 => |b_ip6| a_ip6.eql(b_ip6),
82 else => false,
83 },
84 };
85 }
86
87 /// The returned `Server` has an open `stream`.
88 pub fn listen(address: IpAddress, io: Io, options: ListenOptions) ListenError!Server {
89 return io.vtable.listen(io.userdata, address, options);
90 }
91};
92
93pub const Ip4Address = struct {
94 bytes: [4]u8,
95 port: u16,
96
97 pub const ParseError = error{
98 Overflow,
99 InvalidEnd,
100 InvalidCharacter,
101 Incomplete,
102 NonCanonical,
103 };
104
105 pub fn parse(buffer: []const u8, port: u16) ParseError!Ip4Address {
106 var bytes: [4]u8 = @splat(0);
107 var index: u8 = 0;
108 var saw_any_digits = false;
109 var has_zero_prefix = false;
110 for (buffer) |c| switch (c) {
111 '.' => {
112 if (!saw_any_digits) return error.InvalidCharacter;
113 if (index == 3) return error.InvalidEnd;
114 index += 1;
115 saw_any_digits = false;
116 has_zero_prefix = false;
117 },
118 '0'...'9' => {
119 if (c == '0' and !saw_any_digits) {
120 has_zero_prefix = true;
121 } else if (has_zero_prefix) {
122 return error.NonCanonical;
123 }
124 saw_any_digits = true;
125 bytes[index] = try std.math.mul(u8, bytes[index], 10);
126 bytes[index] = try std.math.add(u8, bytes[index], c - '0');
127 },
128 else => return error.InvalidCharacter,
129 };
130 if (index == 3 and saw_any_digits) return .{
131 .bytes = bytes,
132 .port = port,
133 };
134 return error.Incomplete;
135 }
136
137 pub fn format(a: Ip4Address, w: *std.io.Writer) std.io.Writer.Error!void {
138 const bytes = &a.bytes;
139 try w.print("{d}.{d}.{d}.{d}:{d}", .{ bytes[0], bytes[1], bytes[2], bytes[3], a.port });
140 }
141
142 pub fn eql(a: Ip4Address, b: Ip4Address) bool {
143 const a_int: u32 = @bitCast(a.bytes);
144 const b_int: u32 = @bitCast(b.bytes);
145 return a.port == b.port and a_int == b_int;
146 }
147};
148
149pub const Ip6Address = struct {
150 /// Native endian
151 port: u16,
152 /// Big endian
153 bytes: [16]u8,
154 flowinfo: u32 = 0,
155 scope_id: u32 = 0,
156
157 pub const ParseError = error{
158 Overflow,
159 InvalidCharacter,
160 InvalidEnd,
161 InvalidIpv4Mapping,
162 Incomplete,
163 };
164
165 pub fn parse(buffer: []const u8, port: u16) ParseError!Ip6Address {
166 var result: Ip6Address = .{
167 .port = port,
168 .bytes = undefined,
169 };
170 var ip_slice: *[16]u8 = &result.bytes;
171
172 var tail: [16]u8 = undefined;
173
174 var x: u16 = 0;
175 var saw_any_digits = false;
176 var index: u8 = 0;
177 var scope_id = false;
178 var abbrv = false;
179 for (buffer, 0..) |c, i| {
180 if (scope_id) {
181 if (c >= '0' and c <= '9') {
182 const digit = c - '0';
183 {
184 const ov = @mulWithOverflow(result.scope_id, 10);
185 if (ov[1] != 0) return error.Overflow;
186 result.scope_id = ov[0];
187 }
188 {
189 const ov = @addWithOverflow(result.scope_id, digit);
190 if (ov[1] != 0) return error.Overflow;
191 result.scope_id = ov[0];
192 }
193 } else {
194 return error.InvalidCharacter;
195 }
196 } else if (c == ':') {
197 if (!saw_any_digits) {
198 if (abbrv) return error.InvalidCharacter; // ':::'
199 if (i != 0) abbrv = true;
200 @memset(ip_slice[index..], 0);
201 ip_slice = tail[0..];
202 index = 0;
203 continue;
204 }
205 if (index == 14) {
206 return error.InvalidEnd;
207 }
208 ip_slice[index] = @as(u8, @truncate(x >> 8));
209 index += 1;
210 ip_slice[index] = @as(u8, @truncate(x));
211 index += 1;
212
213 x = 0;
214 saw_any_digits = false;
215 } else if (c == '%') {
216 if (!saw_any_digits) {
217 return error.InvalidCharacter;
218 }
219 scope_id = true;
220 saw_any_digits = false;
221 } else if (c == '.') {
222 if (!abbrv or ip_slice[0] != 0xff or ip_slice[1] != 0xff) {
223 // must start with '::ffff:'
224 return error.InvalidIpv4Mapping;
225 }
226 const start_index = std.mem.lastIndexOfScalar(u8, buffer[0..i], ':').? + 1;
227 const addr = (Ip4Address.parse(buffer[start_index..], 0) catch {
228 return error.InvalidIpv4Mapping;
229 }).bytes;
230 ip_slice = result.bytes[0..];
231 ip_slice[10] = 0xff;
232 ip_slice[11] = 0xff;
233
234 ip_slice[12] = addr[0];
235 ip_slice[13] = addr[1];
236 ip_slice[14] = addr[2];
237 ip_slice[15] = addr[3];
238 return result;
239 } else {
240 const digit = try std.fmt.charToDigit(c, 16);
241 {
242 const ov = @mulWithOverflow(x, 16);
243 if (ov[1] != 0) return error.Overflow;
244 x = ov[0];
245 }
246 {
247 const ov = @addWithOverflow(x, digit);
248 if (ov[1] != 0) return error.Overflow;
249 x = ov[0];
250 }
251 saw_any_digits = true;
252 }
253 }
254
255 if (!saw_any_digits and !abbrv) {
256 return error.Incomplete;
257 }
258 if (!abbrv and index < 14) {
259 return error.Incomplete;
260 }
261
262 if (index == 14) {
263 ip_slice[14] = @as(u8, @truncate(x >> 8));
264 ip_slice[15] = @as(u8, @truncate(x));
265 return result;
266 } else {
267 ip_slice[index] = @as(u8, @truncate(x >> 8));
268 index += 1;
269 ip_slice[index] = @as(u8, @truncate(x));
270 index += 1;
271 @memcpy(result.bytes[16 - index ..][0..index], ip_slice[0..index]);
272 return result;
273 }
274 }
275
276 pub fn format(a: Ip6Address, w: *std.io.Writer) std.io.Writer.Error!void {
277 const bytes = &a.bytes;
278 if (std.mem.eql(u8, bytes[0..12], &[_]u8{ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0xff, 0xff })) {
279 try w.print("[::ffff:{d}.{d}.{d}.{d}]:{d}", .{
280 bytes[12], bytes[13], bytes[14], bytes[15], a.port,
281 });
282 return;
283 }
284 const parts: [8]u16 = .{
285 std.mem.readInt(u16, bytes[0..2], .big),
286 std.mem.readInt(u16, bytes[2..4], .big),
287 std.mem.readInt(u16, bytes[4..6], .big),
288 std.mem.readInt(u16, bytes[6..8], .big),
289 std.mem.readInt(u16, bytes[8..10], .big),
290 std.mem.readInt(u16, bytes[10..12], .big),
291 std.mem.readInt(u16, bytes[12..14], .big),
292 std.mem.readInt(u16, bytes[14..16], .big),
293 };
294
295 // Find the longest zero run
296 var longest_start: usize = 8;
297 var longest_len: usize = 0;
298 var current_start: usize = 0;
299 var current_len: usize = 0;
300
301 for (parts, 0..) |part, i| {
302 if (part == 0) {
303 if (current_len == 0) {
304 current_start = i;
305 }
306 current_len += 1;
307 if (current_len > longest_len) {
308 longest_start = current_start;
309 longest_len = current_len;
310 }
311 } else {
312 current_len = 0;
313 }
314 }
315
316 // Only compress if the longest zero run is 2 or more
317 if (longest_len < 2) {
318 longest_start = 8;
319 longest_len = 0;
320 }
321
322 try w.writeAll("[");
323 var i: usize = 0;
324 var abbrv = false;
325 while (i < parts.len) : (i += 1) {
326 if (i == longest_start) {
327 // Emit "::" for the longest zero run
328 if (!abbrv) {
329 try w.writeAll(if (i == 0) "::" else ":");
330 abbrv = true;
331 }
332 i += longest_len - 1; // Skip the compressed range
333 continue;
334 }
335 if (abbrv) {
336 abbrv = false;
337 }
338 try w.print("{x}", .{parts[i]});
339 if (i != parts.len - 1) {
340 try w.writeAll(":");
341 }
342 }
343 try w.print("]:{d}", .{a.port});
344 }
345
346 pub fn eql(a: Ip6Address, b: Ip6Address) bool {
347 return a.port == b.port and std.mem.eql(u8, &a.bytes, &b.bytes);
348 }
349};
350
351pub const Stream = struct {
352 /// Underlying platform-defined type which may or may not be
353 /// interchangeable with a file system file descriptor.
354 handle: Handle,
355
356 pub const Handle = switch (native_os) {
357 .windows => std.windows.ws2_32.SOCKET,
358 else => std.posix.fd_t,
359 };
360
361 pub fn close(s: Stream, io: Io) void {
362 return io.vtable.close(io.userdata, s);
363 }
364
365 pub const Reader = struct {
366 io: Io,
367 interface: Io.Reader,
368 stream: Stream,
369 err: ?Error,
370
371 pub const Error = std.net.Stream.ReadError || Io.Cancelable || Io.Writer.Error || error{EndOfStream};
372
373 pub fn init(stream: Stream, buffer: []u8) Reader {
374 return .{
375 .interface = .{
376 .vtable = &.{ .stream = streamImpl },
377 .buffer = buffer,
378 .seek = 0,
379 .end = 0,
380 },
381 .stream = stream,
382 .err = null,
383 };
384 }
385
386 fn streamImpl(io_r: *Io.Reader, io_w: *Io.Writer, limit: Io.Limit) Io.Reader.StreamError!usize {
387 const r: *Reader = @alignCast(@fieldParentPtr("interface", io_r));
388 const io = r.io;
389 return io.vtable.netRead(io.vtable.userdata, r.stream, io_w, limit);
390 }
391 };
392
393 pub const Writer = struct {
394 io: Io,
395 interface: Io.Writer,
396 stream: Stream,
397 err: ?Error = null,
398
399 pub const Error = std.net.Stream.WriteError || Io.Cancelable;
400
401 pub fn init(stream: Stream, buffer: []u8) Writer {
402 return .{
403 .stream = stream,
404 .interface = .{
405 .vtable = &.{ .drain = drain },
406 .buffer = buffer,
407 },
408 };
409 }
410
411 fn drain(io_w: *Io.Writer, data: []const []const u8, splat: usize) Io.Writer.Error!usize {
412 const w: *Writer = @alignCast(@fieldParentPtr("interface", io_w));
413 const io = w.io;
414 const buffered = io_w.buffered();
415 const n = try io.vtable.netWrite(io.vtable.userdata, w.stream, buffered, data, splat);
416 return io_w.consume(n);
417 }
418 };
419
420 pub fn reader(stream: Stream, buffer: []u8) Reader {
421 return .init(stream, buffer);
422 }
423
424 pub fn writer(stream: Stream, buffer: []u8) Writer {
425 return .init(stream, buffer);
426 }
427};
428
429pub const Server = struct {
430 listen_address: IpAddress,
431 stream: Stream,
432
433 pub const Connection = struct {
434 stream: Stream,
435 address: IpAddress,
436 };
437
438 pub fn deinit(s: *Server, io: Io) void {
439 s.stream.close(io);
440 s.* = undefined;
441 }
442
443 pub const AcceptError = std.posix.AcceptError || Io.Cancelable;
444
445 /// Blocks until a client connects to the server. The returned `Connection` has
446 /// an open stream.
447 pub fn accept(s: *Server, io: Io) AcceptError!Connection {
448 return io.vtable.accept(io, s);
449 }
450};