authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2021-01-11 15:35:00-08:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2021-01-11 15:35:00-08:00
logec9158305d0131c8f218120f6bd75a95a37cbf18
treec073c699bc3a6f0be4f9e3ae5607480d83ea9613
parent29c9d5896c0f3349b37eff4cbce280fffa7e2924
parent34720da3d0a4a98383639cbedcbe1aa885217ed6
signature Signed by PGP key 4AEE18F83AFDEB23

Merge pull request #7124 from LemonBoy/netstuff1

std: Decouple network streams from fs.File

2 files changed, 109 insertions(+), 11 deletions(-)

lib/std/net.zig+62-8
...@@ -10,8 +10,13 @@ const net = @This();...@@ -10,8 +10,13 @@ const net = @This();
10const mem = std.mem;10const mem = std.mem;
11const os = std.os;11const os = std.os;
12const fs = std.fs;12const fs = std.fs;
13const io = std.io;
1314
14pub const has_unix_sockets = @hasDecl(os, "sockaddr_un");15// Windows 10 added support for unix sockets in build 17063, redstone 4 is the
16// first release to support them.
17pub const has_unix_sockets = @hasDecl(os, "sockaddr_un") and
18 (builtin.os.tag != .windows or
19 std.Target.current.os.version_range.windows.isAtLeast(.win10_rs4) orelse false);
1520
16pub const Address = extern union {21pub const Address = extern union {
17 any: os.sockaddr,22 any: os.sockaddr,
...@@ -596,7 +601,7 @@ pub const Ip6Address = extern struct {...@@ -596,7 +601,7 @@ pub const Ip6Address = extern struct {
596 }601 }
597};602};
598603
599pub fn connectUnixSocket(path: []const u8) !fs.File {604pub fn connectUnixSocket(path: []const u8) !Stream {
600 const opt_non_block = if (std.io.is_async) os.SOCK_NONBLOCK else 0;605 const opt_non_block = if (std.io.is_async) os.SOCK_NONBLOCK else 0;
601 const sockfd = try os.socket(606 const sockfd = try os.socket(
602 os.AF_UNIX,607 os.AF_UNIX,
...@@ -614,7 +619,7 @@ pub fn connectUnixSocket(path: []const u8) !fs.File {...@@ -614,7 +619,7 @@ pub fn connectUnixSocket(path: []const u8) !fs.File {
614 try os.connect(sockfd, &addr.any, addr.getOsSockLen());619 try os.connect(sockfd, &addr.any, addr.getOsSockLen());
615 }620 }
616621
617 return fs.File{622 return Stream{
618 .handle = sockfd,623 .handle = sockfd,
619 };624 };
620}625}
...@@ -648,7 +653,7 @@ pub const AddressList = struct {...@@ -648,7 +653,7 @@ pub const AddressList = struct {
648};653};
649654
650/// All memory allocated with `allocator` will be freed before this function returns.655/// All memory allocated with `allocator` will be freed before this function returns.
651pub fn tcpConnectToHost(allocator: *mem.Allocator, name: []const u8, port: u16) !fs.File {656pub fn tcpConnectToHost(allocator: *mem.Allocator, name: []const u8, port: u16) !Stream {
652 const list = try getAddressList(allocator, name, port);657 const list = try getAddressList(allocator, name, port);
653 defer list.deinit();658 defer list.deinit();
654659
...@@ -665,7 +670,7 @@ pub fn tcpConnectToHost(allocator: *mem.Allocator, name: []const u8, port: u16)...@@ -665,7 +670,7 @@ pub fn tcpConnectToHost(allocator: *mem.Allocator, name: []const u8, port: u16)
665 return std.os.ConnectError.ConnectionRefused;670 return std.os.ConnectError.ConnectionRefused;
666}671}
667672
668pub fn tcpConnectToAddress(address: Address) !fs.File {673pub fn tcpConnectToAddress(address: Address) !Stream {
669 const nonblock = if (std.io.is_async) os.SOCK_NONBLOCK else 0;674 const nonblock = if (std.io.is_async) os.SOCK_NONBLOCK else 0;
670 const sock_flags = os.SOCK_STREAM | nonblock |675 const sock_flags = os.SOCK_STREAM | nonblock |
671 (if (builtin.os.tag == .windows) 0 else os.SOCK_CLOEXEC);676 (if (builtin.os.tag == .windows) 0 else os.SOCK_CLOEXEC);
...@@ -679,7 +684,7 @@ pub fn tcpConnectToAddress(address: Address) !fs.File {...@@ -679,7 +684,7 @@ pub fn tcpConnectToAddress(address: Address) !fs.File {
679 try os.connect(sockfd, &address.any, address.getOsSockLen());684 try os.connect(sockfd, &address.any, address.getOsSockLen());
680 }685 }
681686
682 return fs.File{ .handle = sockfd };687 return Stream{ .handle = sockfd };
683}688}
684689
685/// Call `AddressList.deinit` on the result.690/// Call `AddressList.deinit` on the result.
...@@ -1580,6 +1585,55 @@ fn dnsParseCallback(ctx: dpc_ctx, rr: u8, data: []const u8, packet: []const u8)...@@ -1580,6 +1585,55 @@ fn dnsParseCallback(ctx: dpc_ctx, rr: u8, data: []const u8, packet: []const u8)
1580 }1585 }
1581}1586}
15821587
1588pub const Stream = struct {
1589 // Underlying socket descriptor.
1590 // Note that on some platforms this may not be interchangeable with a
1591 // regular files descriptor.
1592 handle: os.socket_t,
1593
1594 pub fn close(self: Stream) void {
1595 os.closeSocket(self.handle);
1596 }
1597
1598 pub const ReadError = os.ReadError;
1599 pub const WriteError = os.WriteError;
1600
1601 pub const Reader = io.Reader(Stream, ReadError, read);
1602 pub const Writer = io.Writer(Stream, WriteError, write);
1603
1604 pub fn reader(self: Stream) Reader {
1605 return .{ .context = self };
1606 }
1607
1608 pub fn writer(self: Stream) Writer {
1609 return .{ .context = self };
1610 }
1611
1612 pub fn read(self: Stream, buffer: []u8) ReadError!usize {
1613 if (std.Target.current.os.tag == .windows) {
1614 return os.windows.ReadFile(self.handle, buffer, null, io.default_mode);
1615 }
1616
1617 if (std.io.is_async) {
1618 return std.event.Loop.instance.?.read(self.handle, buffer, false);
1619 } else {
1620 return os.read(self.handle, buffer);
1621 }
1622 }
1623
1624 pub fn write(self: Stream, buffer: []const u8) WriteError!usize {
1625 if (std.Target.current.os.tag == .windows) {
1626 return os.windows.WriteFile(self.handle, buffer, null, io.default_mode);
1627 }
1628
1629 if (std.io.is_async) {
1630 return std.event.Loop.instance.?.write(self.handle, buffer, false);
1631 } else {
1632 return os.write(self.handle, buffer);
1633 }
1634 }
1635};
1636
1583pub const StreamServer = struct {1637pub const StreamServer = struct {
1584 /// Copied from `Options` on `init`.1638 /// Copied from `Options` on `init`.
1585 kernel_backlog: u31,1639 kernel_backlog: u31,
...@@ -1686,7 +1740,7 @@ pub const StreamServer = struct {...@@ -1686,7 +1740,7 @@ pub const StreamServer = struct {
1686 } || os.UnexpectedError;1740 } || os.UnexpectedError;
16871741
1688 pub const Connection = struct {1742 pub const Connection = struct {
1689 file: fs.File,1743 stream: Stream,
1690 address: Address,1744 address: Address,
1691 };1745 };
16921746
...@@ -1705,7 +1759,7 @@ pub const StreamServer = struct {...@@ -1705,7 +1759,7 @@ pub const StreamServer = struct {
17051759
1706 if (accept_result) |fd| {1760 if (accept_result) |fd| {
1707 return Connection{1761 return Connection{
1708 .file = fs.File{ .handle = fd },1762 .stream = Stream{ .handle = fd },
1709 .address = accepted_addr,1763 .address = accepted_addr,
1710 };1764 };
1711 } else |err| switch (err) {1765 } else |err| switch (err) {
lib/std/net/test.zig+47-3
...@@ -145,7 +145,7 @@ test "listen on a port, send bytes, receive bytes" {...@@ -145,7 +145,7 @@ test "listen on a port, send bytes, receive bytes" {
145145
146 // Try only the IPv4 variant as some CI builders have no IPv6 localhost146 // Try only the IPv4 variant as some CI builders have no IPv6 localhost
147 // configured.147 // configured.
148 const localhost = try net.Address.parseIp("127.0.0.1", 8080);148 const localhost = try net.Address.parseIp("127.0.0.1", 0);
149149
150 var server = net.StreamServer.init(.{});150 var server = net.StreamServer.init(.{});
151 defer server.deinit();151 defer server.deinit();
...@@ -165,8 +165,9 @@ test "listen on a port, send bytes, receive bytes" {...@@ -165,8 +165,9 @@ test "listen on a port, send bytes, receive bytes" {
165 defer t.wait();165 defer t.wait();
166166
167 var client = try server.accept();167 var client = try server.accept();
168 defer client.stream.close();
168 var buf: [16]u8 = undefined;169 var buf: [16]u8 = undefined;
169 const n = try client.file.reader().read(&buf);170 const n = try client.stream.reader().read(&buf);
170171
171 testing.expectEqual(@as(usize, 12), n);172 testing.expectEqual(@as(usize, 12), n);
172 testing.expectEqualSlices(u8, "Hello world!", buf[0..n]);173 testing.expectEqualSlices(u8, "Hello world!", buf[0..n]);
...@@ -249,6 +250,49 @@ fn testServer(server: *net.StreamServer) anyerror!void {...@@ -249,6 +250,49 @@ fn testServer(server: *net.StreamServer) anyerror!void {
249250
250 var client = try server.accept();251 var client = try server.accept();
251252
252 const stream = client.file.writer();253 const stream = client.stream.writer();
253 try stream.print("hello from server\n", .{});254 try stream.print("hello from server\n", .{});
254}255}
256
257test "listen on a unix socket, send bytes, receive bytes" {
258 if (builtin.single_threaded) return error.SkipZigTest;
259 if (!net.has_unix_sockets) return error.SkipZigTest;
260
261 if (std.builtin.os.tag == .windows) {
262 _ = try std.os.windows.WSAStartup(2, 2);
263 }
264 defer {
265 if (std.builtin.os.tag == .windows) {
266 std.os.windows.WSACleanup() catch unreachable;
267 }
268 }
269
270 var server = net.StreamServer.init(.{});
271 defer server.deinit();
272
273 const socket_path = "socket.unix";
274
275 var socket_addr = try net.Address.initUnix(socket_path);
276 defer std.fs.cwd().deleteFile(socket_path) catch {};
277 try server.listen(socket_addr);
278
279 const S = struct {
280 fn clientFn(_: void) !void {
281 const socket = try net.connectUnixSocket(socket_path);
282 defer socket.close();
283
284 _ = try socket.writer().writeAll("Hello world!");
285 }
286 };
287
288 const t = try std.Thread.spawn({}, S.clientFn);
289 defer t.wait();
290
291 var client = try server.accept();
292 defer client.stream.close();
293 var buf: [16]u8 = undefined;
294 const n = try client.stream.reader().read(&buf);
295
296 testing.expectEqual(@as(usize, 12), n);
297 testing.expectEqualSlices(u8, "Hello world!", buf[0..n]);
298}