authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2018-09-30 10:37:58-04:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2018-09-30 10:37:58-04:00
log57c4d38c556238530af9b643b4dbc481de417782
treee04e9cba5369315c6a4a5701a8bbb4aaad9306db
parenta42a213e35842e18750dd8cfdfc0d024e23da50b
signature Commit is signed but in an unrecognized format.

rename std.event.tcp to std.event.net

4 files changed, 235 insertions(+), 235 deletions(-)

CMakeLists.txt+1-1
...@@ -492,9 +492,9 @@ set(ZIG_STD_FILES...@@ -492,9 +492,9 @@ set(ZIG_STD_FILES
492 "event/lock.zig"492 "event/lock.zig"
493 "event/locked.zig"493 "event/locked.zig"
494 "event/loop.zig"494 "event/loop.zig"
495 "event/net.zig"
495 "event/rwlock.zig"496 "event/rwlock.zig"
496 "event/rwlocked.zig"497 "event/rwlocked.zig"
497 "event/tcp.zig"
498 "fmt/errol/enum3.zig"498 "fmt/errol/enum3.zig"
499 "fmt/errol/index.zig"499 "fmt/errol/index.zig"
500 "fmt/errol/lookup.zig"500 "fmt/errol/lookup.zig"
std/event.zig+2-2
...@@ -7,7 +7,7 @@ pub const RwLock = @import("event/rwlock.zig").RwLock;...@@ -7,7 +7,7 @@ pub const RwLock = @import("event/rwlock.zig").RwLock;
7pub const RwLocked = @import("event/rwlocked.zig").RwLocked;7pub const RwLocked = @import("event/rwlocked.zig").RwLocked;
8pub const Loop = @import("event/loop.zig").Loop;8pub const Loop = @import("event/loop.zig").Loop;
9pub const fs = @import("event/fs.zig");9pub const fs = @import("event/fs.zig");
10pub const tcp = @import("event/tcp.zig");10pub const net = @import("event/net.zig");
1111
12test "import event tests" {12test "import event tests" {
13 _ = @import("event/channel.zig");13 _ = @import("event/channel.zig");
...@@ -19,5 +19,5 @@ test "import event tests" {...@@ -19,5 +19,5 @@ test "import event tests" {
19 _ = @import("event/rwlock.zig");19 _ = @import("event/rwlock.zig");
20 _ = @import("event/rwlocked.zig");20 _ = @import("event/rwlocked.zig");
21 _ = @import("event/loop.zig");21 _ = @import("event/loop.zig");
22 _ = @import("event/tcp.zig");22 _ = @import("event/net.zig");
23}23}
std/event/net.zig created+232
...@@ -0,0 +1,232 @@
1const std = @import("../index.zig");
2const builtin = @import("builtin");
3const assert = std.debug.assert;
4const event = std.event;
5const mem = std.mem;
6const posix = std.os.posix;
7const windows = std.os.windows;
8const Loop = std.event.Loop;
9
10pub const Server = struct {
11 handleRequestFn: async<*mem.Allocator> fn (*Server, *const std.net.Address, *const std.os.File) void,
12
13 loop: *Loop,
14 sockfd: ?i32,
15 accept_coro: ?promise,
16 listen_address: std.net.Address,
17
18 waiting_for_emfile_node: PromiseNode,
19 listen_resume_node: event.Loop.ResumeNode,
20
21 const PromiseNode = std.LinkedList(promise).Node;
22
23 pub fn init(loop: *Loop) Server {
24 // TODO can't initialize handler coroutine here because we need well defined copy elision
25 return Server{
26 .loop = loop,
27 .sockfd = null,
28 .accept_coro = null,
29 .handleRequestFn = undefined,
30 .waiting_for_emfile_node = undefined,
31 .listen_address = undefined,
32 .listen_resume_node = event.Loop.ResumeNode{
33 .id = event.Loop.ResumeNode.Id.Basic,
34 .handle = undefined,
35 .overlapped = event.Loop.ResumeNode.overlapped_init,
36 },
37 };
38 }
39
40 pub fn listen(
41 self: *Server,
42 address: *const std.net.Address,
43 handleRequestFn: async<*mem.Allocator> fn (*Server, *const std.net.Address, *const std.os.File) void,
44 ) !void {
45 self.handleRequestFn = handleRequestFn;
46
47 const sockfd = try std.os.posixSocket(posix.AF_INET, posix.SOCK_STREAM | posix.SOCK_CLOEXEC | posix.SOCK_NONBLOCK, posix.PROTO_tcp);
48 errdefer std.os.close(sockfd);
49 self.sockfd = sockfd;
50
51 try std.os.posixBind(sockfd, &address.os_addr);
52 try std.os.posixListen(sockfd, posix.SOMAXCONN);
53 self.listen_address = std.net.Address.initPosix(try std.os.posixGetSockName(sockfd));
54
55 self.accept_coro = try async<self.loop.allocator> Server.handler(self);
56 errdefer cancel self.accept_coro.?;
57
58 self.listen_resume_node.handle = self.accept_coro.?;
59 try self.loop.linuxAddFd(sockfd, &self.listen_resume_node, posix.EPOLLIN | posix.EPOLLOUT | posix.EPOLLET);
60 errdefer self.loop.removeFd(sockfd);
61 }
62
63 /// Stop listening
64 pub fn close(self: *Server) void {
65 self.loop.linuxRemoveFd(self.sockfd.?);
66 std.os.close(self.sockfd.?);
67 }
68
69 pub fn deinit(self: *Server) void {
70 if (self.accept_coro) |accept_coro| cancel accept_coro;
71 if (self.sockfd) |sockfd| std.os.close(sockfd);
72 }
73
74 pub async fn handler(self: *Server) void {
75 while (true) {
76 var accepted_addr: std.net.Address = undefined;
77 if (std.os.posixAccept(self.sockfd.?, &accepted_addr.os_addr, posix.SOCK_NONBLOCK | posix.SOCK_CLOEXEC)) |accepted_fd| {
78 var socket = std.os.File.openHandle(accepted_fd);
79 _ = async<self.loop.allocator> self.handleRequestFn(self, accepted_addr, socket) catch |err| switch (err) {
80 error.OutOfMemory => {
81 socket.close();
82 continue;
83 },
84 };
85 } else |err| switch (err) {
86 error.WouldBlock => {
87 suspend; // we will get resumed by epoll_wait in the event loop
88 continue;
89 },
90 error.ProcessFdQuotaExceeded => {
91 errdefer std.os.emfile_promise_queue.remove(&self.waiting_for_emfile_node);
92 suspend {
93 self.waiting_for_emfile_node = PromiseNode.init(@handle());
94 std.os.emfile_promise_queue.append(&self.waiting_for_emfile_node);
95 }
96 continue;
97 },
98 error.ConnectionAborted, error.FileDescriptorClosed => continue,
99
100 error.PageFault => unreachable,
101 error.InvalidSyscall => unreachable,
102 error.FileDescriptorNotASocket => unreachable,
103 error.OperationNotSupported => unreachable,
104
105 error.SystemFdQuotaExceeded, error.SystemResources, error.ProtocolFailure, error.BlockedByFirewall, error.Unexpected => {
106 @panic("TODO handle this error");
107 },
108 }
109 }
110 }
111};
112
113pub async fn connectUnixSocket(loop: *Loop, path: []const u8) !i32 {
114 const sockfd = try std.os.posixSocket(
115 posix.AF_UNIX,
116 posix.SOCK_STREAM | posix.SOCK_CLOEXEC | posix.SOCK_NONBLOCK,
117 0,
118 );
119 errdefer std.os.close(sockfd);
120
121 var sock_addr = posix.sockaddr{
122 .un = posix.sockaddr_un{
123 .family = posix.AF_UNIX,
124 .path = undefined,
125 },
126 };
127
128 if (path.len > @typeOf(sock_addr.un.path).len) return error.NameTooLong;
129 mem.copy(u8, sock_addr.un.path[0..], path);
130 const size = @intCast(u32, @sizeOf(posix.sa_family_t) + path.len);
131 try std.os.posixConnectAsync(sockfd, &sock_addr, size);
132 try await try async loop.linuxWaitFd(sockfd, posix.EPOLLIN | posix.EPOLLOUT | posix.EPOLLET);
133 try std.os.posixGetSockOptConnectError(sockfd);
134
135 return sockfd;
136}
137
138pub async fn socketRead(loop: *std.event.Loop, fd: i32, buffer: []u8) !void {
139 while (true) {
140 return std.os.posixRead(fd, buffer) catch |err| switch (err) {
141 error.WouldBlock => {
142 try await try async loop.linuxWaitFd(fd, std.os.posix.EPOLLET | std.os.posix.EPOLLIN);
143 continue;
144 },
145 else => return err,
146 };
147 }
148}
149pub async fn socketWrite(loop: *std.event.Loop, fd: i32, buffer: []const u8) !void {
150 while (true) {
151 return std.os.posixWrite(fd, buffer) catch |err| switch (err) {
152 error.WouldBlock => {
153 try await try async loop.linuxWaitFd(fd, std.os.posix.EPOLLET | std.os.posix.EPOLLOUT);
154 continue;
155 },
156 else => return err,
157 };
158 }
159}
160
161pub async fn connect(loop: *Loop, _address: *const std.net.Address) !std.os.File {
162 var address = _address.*; // TODO https://github.com/ziglang/zig/issues/733
163
164 const sockfd = try std.os.posixSocket(posix.AF_INET, posix.SOCK_STREAM | posix.SOCK_CLOEXEC | posix.SOCK_NONBLOCK, posix.PROTO_tcp);
165 errdefer std.os.close(sockfd);
166
167 try std.os.posixConnectAsync(sockfd, &address.os_addr, @sizeOf(posix.sockaddr_in));
168 try await try async loop.linuxWaitFd(sockfd, posix.EPOLLIN | posix.EPOLLOUT | posix.EPOLLET);
169 try std.os.posixGetSockOptConnectError(sockfd);
170
171 return std.os.File.openHandle(sockfd);
172}
173
174test "listen on a port, send bytes, receive bytes" {
175 if (builtin.os != builtin.Os.linux) {
176 // TODO build abstractions for other operating systems
177 return error.SkipZigTest;
178 }
179
180 const MyServer = struct {
181 tcp_server: Server,
182
183 const Self = @This();
184 async<*mem.Allocator> fn handler(tcp_server: *Server, _addr: *const std.net.Address, _socket: *const std.os.File) void {
185 const self = @fieldParentPtr(Self, "tcp_server", tcp_server);
186 var socket = _socket.*; // TODO https://github.com/ziglang/zig/issues/733
187 defer socket.close();
188 // TODO guarantee elision of this allocation
189 const next_handler = async errorableHandler(self, _addr, socket) catch unreachable;
190 (await next_handler) catch |err| {
191 std.debug.panic("unable to handle connection: {}\n", err);
192 };
193 suspend {
194 cancel @handle();
195 }
196 }
197 async fn errorableHandler(self: *Self, _addr: *const std.net.Address, _socket: std.os.File) !void {
198 const addr = _addr.*; // TODO https://github.com/ziglang/zig/issues/733
199 var socket = _socket; // TODO https://github.com/ziglang/zig/issues/733
200
201 var adapter = std.io.FileOutStream.init(socket);
202 var stream = &adapter.stream;
203 try stream.print("hello from server\n");
204 }
205 };
206
207 const ip4addr = std.net.parseIp4("127.0.0.1") catch unreachable;
208 const addr = std.net.Address.initIp4(ip4addr, 0);
209
210 var loop: Loop = undefined;
211 try loop.initSingleThreaded(std.debug.global_allocator);
212 var server = MyServer{ .tcp_server = Server.init(&loop) };
213 defer server.tcp_server.deinit();
214 try server.tcp_server.listen(addr, MyServer.handler);
215
216 const p = try async<std.debug.global_allocator> doAsyncTest(&loop, server.tcp_server.listen_address, &server.tcp_server);
217 defer cancel p;
218 loop.run();
219}
220
221async fn doAsyncTest(loop: *Loop, address: *const std.net.Address, server: *Server) void {
222 errdefer @panic("test failure");
223
224 var socket_file = try await try async connect(loop, address);
225 defer socket_file.close();
226
227 var buf: [512]u8 = undefined;
228 const amt_read = try socket_file.read(buf[0..]);
229 const msg = buf[0..amt_read];
230 assert(mem.eql(u8, msg, "hello from server\n"));
231 server.close();
232}
std/event/tcp.zig deleted-232
...@@ -1,232 +0,0 @@
1const std = @import("../index.zig");
2const builtin = @import("builtin");
3const assert = std.debug.assert;
4const event = std.event;
5const mem = std.mem;
6const posix = std.os.posix;
7const windows = std.os.windows;
8const Loop = std.event.Loop;
9
10pub const Server = struct {
11 handleRequestFn: async<*mem.Allocator> fn (*Server, *const std.net.Address, *const std.os.File) void,
12
13 loop: *Loop,
14 sockfd: ?i32,
15 accept_coro: ?promise,
16 listen_address: std.net.Address,
17
18 waiting_for_emfile_node: PromiseNode,
19 listen_resume_node: event.Loop.ResumeNode,
20
21 const PromiseNode = std.LinkedList(promise).Node;
22
23 pub fn init(loop: *Loop) Server {
24 // TODO can't initialize handler coroutine here because we need well defined copy elision
25 return Server{
26 .loop = loop,
27 .sockfd = null,
28 .accept_coro = null,
29 .handleRequestFn = undefined,
30 .waiting_for_emfile_node = undefined,
31 .listen_address = undefined,
32 .listen_resume_node = event.Loop.ResumeNode{
33 .id = event.Loop.ResumeNode.Id.Basic,
34 .handle = undefined,
35 .overlapped = event.Loop.ResumeNode.overlapped_init,
36 },
37 };
38 }
39
40 pub fn listen(
41 self: *Server,
42 address: *const std.net.Address,
43 handleRequestFn: async<*mem.Allocator> fn (*Server, *const std.net.Address, *const std.os.File) void,
44 ) !void {
45 self.handleRequestFn = handleRequestFn;
46
47 const sockfd = try std.os.posixSocket(posix.AF_INET, posix.SOCK_STREAM | posix.SOCK_CLOEXEC | posix.SOCK_NONBLOCK, posix.PROTO_tcp);
48 errdefer std.os.close(sockfd);
49 self.sockfd = sockfd;
50
51 try std.os.posixBind(sockfd, &address.os_addr);
52 try std.os.posixListen(sockfd, posix.SOMAXCONN);
53 self.listen_address = std.net.Address.initPosix(try std.os.posixGetSockName(sockfd));
54
55 self.accept_coro = try async<self.loop.allocator> Server.handler(self);
56 errdefer cancel self.accept_coro.?;
57
58 self.listen_resume_node.handle = self.accept_coro.?;
59 try self.loop.linuxAddFd(sockfd, &self.listen_resume_node, posix.EPOLLIN | posix.EPOLLOUT | posix.EPOLLET);
60 errdefer self.loop.removeFd(sockfd);
61 }
62
63 /// Stop listening
64 pub fn close(self: *Server) void {
65 self.loop.linuxRemoveFd(self.sockfd.?);
66 std.os.close(self.sockfd.?);
67 }
68
69 pub fn deinit(self: *Server) void {
70 if (self.accept_coro) |accept_coro| cancel accept_coro;
71 if (self.sockfd) |sockfd| std.os.close(sockfd);
72 }
73
74 pub async fn handler(self: *Server) void {
75 while (true) {
76 var accepted_addr: std.net.Address = undefined;
77 if (std.os.posixAccept(self.sockfd.?, &accepted_addr.os_addr, posix.SOCK_NONBLOCK | posix.SOCK_CLOEXEC)) |accepted_fd| {
78 var socket = std.os.File.openHandle(accepted_fd);
79 _ = async<self.loop.allocator> self.handleRequestFn(self, accepted_addr, socket) catch |err| switch (err) {
80 error.OutOfMemory => {
81 socket.close();
82 continue;
83 },
84 };
85 } else |err| switch (err) {
86 error.WouldBlock => {
87 suspend; // we will get resumed by epoll_wait in the event loop
88 continue;
89 },
90 error.ProcessFdQuotaExceeded => {
91 errdefer std.os.emfile_promise_queue.remove(&self.waiting_for_emfile_node);
92 suspend {
93 self.waiting_for_emfile_node = PromiseNode.init(@handle());
94 std.os.emfile_promise_queue.append(&self.waiting_for_emfile_node);
95 }
96 continue;
97 },
98 error.ConnectionAborted, error.FileDescriptorClosed => continue,
99
100 error.PageFault => unreachable,
101 error.InvalidSyscall => unreachable,
102 error.FileDescriptorNotASocket => unreachable,
103 error.OperationNotSupported => unreachable,
104
105 error.SystemFdQuotaExceeded, error.SystemResources, error.ProtocolFailure, error.BlockedByFirewall, error.Unexpected => {
106 @panic("TODO handle this error");
107 },
108 }
109 }
110 }
111};
112
113pub async fn connectUnixSocket(loop: *Loop, path: []const u8) !i32 {
114 const sockfd = try std.os.posixSocket(
115 posix.AF_UNIX,
116 posix.SOCK_STREAM | posix.SOCK_CLOEXEC | posix.SOCK_NONBLOCK,
117 0,
118 );
119 errdefer std.os.close(sockfd);
120
121 var sock_addr = posix.sockaddr{
122 .un = posix.sockaddr_un{
123 .family = posix.AF_UNIX,
124 .path = undefined,
125 },
126 };
127
128 if (path.len > @typeOf(sock_addr.un.path).len) return error.NameTooLong;
129 mem.copy(u8, sock_addr.un.path[0..], path);
130 const size = @intCast(u32, @sizeOf(posix.sa_family_t) + path.len);
131 try std.os.posixConnectAsync(sockfd, &sock_addr, size);
132 try await try async loop.linuxWaitFd(sockfd, posix.EPOLLIN | posix.EPOLLOUT | posix.EPOLLET);
133 try std.os.posixGetSockOptConnectError(sockfd);
134
135 return sockfd;
136}
137
138pub async fn socketRead(loop: *std.event.Loop, fd: i32, buffer: []u8) !void {
139 while (true) {
140 return std.os.posixRead(fd, buffer) catch |err| switch (err) {
141 error.WouldBlock => {
142 try await try async loop.linuxWaitFd(fd, std.os.posix.EPOLLET | std.os.posix.EPOLLIN);
143 continue;
144 },
145 else => return err,
146 };
147 }
148}
149pub async fn socketWrite(loop: *std.event.Loop, fd: i32, buffer: []const u8) !void {
150 while (true) {
151 return std.os.posixWrite(fd, buffer) catch |err| switch (err) {
152 error.WouldBlock => {
153 try await try async loop.linuxWaitFd(fd, std.os.posix.EPOLLET | std.os.posix.EPOLLOUT);
154 continue;
155 },
156 else => return err,
157 };
158 }
159}
160
161pub async fn connect(loop: *Loop, _address: *const std.net.Address) !std.os.File {
162 var address = _address.*; // TODO https://github.com/ziglang/zig/issues/733
163
164 const sockfd = try std.os.posixSocket(posix.AF_INET, posix.SOCK_STREAM | posix.SOCK_CLOEXEC | posix.SOCK_NONBLOCK, posix.PROTO_tcp);
165 errdefer std.os.close(sockfd);
166
167 try std.os.posixConnectAsync(sockfd, &address.os_addr, @sizeOf(posix.sockaddr_in));
168 try await try async loop.linuxWaitFd(sockfd, posix.EPOLLIN | posix.EPOLLOUT | posix.EPOLLET);
169 try std.os.posixGetSockOptConnectError(sockfd);
170
171 return std.os.File.openHandle(sockfd);
172}
173
174test "listen on a port, send bytes, receive bytes" {
175 if (builtin.os != builtin.Os.linux) {
176 // TODO build abstractions for other operating systems
177 return error.SkipZigTest;
178 }
179
180 const MyServer = struct {
181 tcp_server: Server,
182
183 const Self = @This();
184 async<*mem.Allocator> fn handler(tcp_server: *Server, _addr: *const std.net.Address, _socket: *const std.os.File) void {
185 const self = @fieldParentPtr(Self, "tcp_server", tcp_server);
186 var socket = _socket.*; // TODO https://github.com/ziglang/zig/issues/733
187 defer socket.close();
188 // TODO guarantee elision of this allocation
189 const next_handler = async errorableHandler(self, _addr, socket) catch unreachable;
190 (await next_handler) catch |err| {
191 std.debug.panic("unable to handle connection: {}\n", err);
192 };
193 suspend {
194 cancel @handle();
195 }
196 }
197 async fn errorableHandler(self: *Self, _addr: *const std.net.Address, _socket: std.os.File) !void {
198 const addr = _addr.*; // TODO https://github.com/ziglang/zig/issues/733
199 var socket = _socket; // TODO https://github.com/ziglang/zig/issues/733
200
201 var adapter = std.io.FileOutStream.init(socket);
202 var stream = &adapter.stream;
203 try stream.print("hello from server\n");
204 }
205 };
206
207 const ip4addr = std.net.parseIp4("127.0.0.1") catch unreachable;
208 const addr = std.net.Address.initIp4(ip4addr, 0);
209
210 var loop: Loop = undefined;
211 try loop.initSingleThreaded(std.debug.global_allocator);
212 var server = MyServer{ .tcp_server = Server.init(&loop) };
213 defer server.tcp_server.deinit();
214 try server.tcp_server.listen(addr, MyServer.handler);
215
216 const p = try async<std.debug.global_allocator> doAsyncTest(&loop, server.tcp_server.listen_address, &server.tcp_server);
217 defer cancel p;
218 loop.run();
219}
220
221async fn doAsyncTest(loop: *Loop, address: *const std.net.Address, server: *Server) void {
222 errdefer @panic("test failure");
223
224 var socket_file = try await try async connect(loop, address);
225 defer socket_file.close();
226
227 var buf: [512]u8 = undefined;
228 const amt_read = try socket_file.read(buf[0..]);
229 const msg = buf[0..amt_read];
230 assert(mem.eql(u8, msg, "hello from server\n"));
231 server.close();
232}