also looking at this
Implement the web UI server with Zig I/O
#26
8 files
+513
-114
Review threads live on the whole diff, not on one commit, so none are shown here - a thread's line means something in the branch's final form, and painting it into an intermediate step would put it on code it is not about.
| @@ -56,6 +56,13 @@ pub fn build(b: *std.Build) void { | ||
| 56 | 56 | // Unit tests for the framework itself |
| 57 | 57 | const lib_unit_tests = b.addTest(.{ |
| 58 | 58 | .root_module = lib_module, |
| 59 | // The Zig server-mode test runner can deadlock with Threaded network | |
| 60 | // I/O on Windows. Run this networking-heavy test artifact through the | |
| 61 | // normal process protocol and use its exit status instead. | |
| 62 | .test_runner = if (target.result.os.tag == .windows) .{ | |
| 63 | .path = b.path("tests/simple_test_runner.zig"), | |
| 64 | .mode = .simple, | |
| 65 | } else null, | |
| 59 | 66 | }); |
| 60 | 67 | |
| 61 | 68 | const run_lib_unit_tests = b.addRunArtifact(lib_unit_tests); |
| @@ -78,27 +78,43 @@ pub const UIServer = struct { | ||
| 78 | 78 | /// Stop accepting connections and close every active event stream. |
| 79 | 79 | pub fn stop(self: *Self) void { |
| 80 | 80 | const was_running = self.running.swap(false, .acq_rel); |
| 81 | var wake_io: ?std.Io.Threaded = null; | |
| 82 | var wake_stream: ?std.Io.net.Stream = null; | |
| 83 | ||
| 81 | 84 | if (was_running) { |
| 82 | if (self.listener) |server| { | |
| 83 | const listening_stream = std.Io.net.Stream{ .socket = server.socket }; | |
| 84 | listening_stream.shutdown(self.threaded_io.io(), .both) catch {}; | |
| 85 | server.socket.close(self.threaded_io.io()); | |
| 85 | // A client may have connected without finishing its request | |
| 86 | // headers. Shutdown unblocks the parser while leaving ownership of | |
| 87 | // the socket with the accept thread. | |
| 88 | self.mutex.lock(); | |
| 89 | if (self.active_client) |client| client.shutdown(self.threaded_io.io(), .both) catch {}; | |
| 90 | self.mutex.unlock(); | |
| 91 | ||
| 92 | // Closing a listening socket from another thread makes the current | |
| 93 | // Zig Windows backend treat the canceled accept as unreachable. | |
| 94 | // Wake it with a loopback connection on an independent I/O runtime: | |
| 95 | // sharing the server runtime here can deadlock its blocking accept | |
| 96 | // on Linux and Windows. Keep the connection alive until accept has | |
| 97 | // returned so the peer cannot disappear from the listen queue. | |
| 98 | wake_io = .init(self.allocator, .{ .environ = .empty }); | |
| 99 | if (wake_io) |*threaded| { | |
| 100 | wake_stream = self.connectWake(threaded.io()); | |
| 86 | 101 | } |
| 87 | 102 | } |
| 88 | 103 | |
| 89 | // A client may have connected without finishing its request headers. | |
| 90 | // Shutting down that in-flight socket ensures shutdown cannot hang | |
| 91 | // waiting for the request parser; the accept thread remains its owner. | |
| 92 | self.mutex.lock(); | |
| 93 | if (self.active_client) |client| client.shutdown(self.threaded_io.io(), .both) catch {}; | |
| 94 | self.mutex.unlock(); | |
| 95 | ||
| 96 | 104 | if (self.thread) |thread| { |
| 97 | 105 | thread.join(); |
| 98 | 106 | self.thread = null; |
| 99 | 107 | } |
| 100 | 108 | |
| 101 | self.listener = null; | |
| 109 | if (wake_stream) |stream| { | |
| 110 | if (wake_io) |*threaded| stream.close(threaded.io()); | |
| 111 | } | |
| 112 | if (wake_io) |*threaded| threaded.deinit(); | |
| 113 | ||
| 114 | if (self.listener) |*server| { | |
| 115 | server.deinit(self.threaded_io.io()); | |
| 116 | self.listener = null; | |
| 117 | } | |
| 102 | 118 | |
| 103 | 119 | self.mutex.lock(); |
| 104 | 120 | defer self.mutex.unlock(); |
| @@ -106,6 +122,26 @@ pub const UIServer = struct { | ||
| 106 | 122 | self.clients.clearRetainingCapacity(); |
| 107 | 123 | } |
| 108 | 124 | |
| 125 | fn connectWake(self: *Self, io: std.Io) ?std.Io.net.Stream { | |
| 126 | const server = self.listener orelse return null; | |
| 127 | var address = server.socket.address; | |
| 128 | switch (address) { | |
| 129 | .ip4 => |ip4| { | |
| 130 | if (std.mem.eql(u8, &ip4.bytes, &.{ 0, 0, 0, 0 })) { | |
| 131 | address = .{ .ip4 = .loopback(ip4.port) }; | |
| 132 | } | |
| 133 | }, | |
| 134 | .ip6 => |ip6| { | |
| 135 | const unspecified: [16]u8 = @splat(0); | |
| 136 | if (std.mem.eql(u8, &ip6.bytes, &unspecified)) { | |
| 137 | address = .{ .ip6 = .loopback(ip6.port) }; | |
| 138 | } | |
| 139 | }, | |
| 140 | } | |
| 141 | ||
| 142 | return address.connect(io, .{ .mode = .stream }) catch null; | |
| 143 | } | |
| 144 | ||
| 109 | 145 | /// Return the actual bound port. This differs from the requested port when |
| 110 | 146 | /// port zero asks the operating system for an ephemeral port. |
| 111 | 147 | pub fn port(self: *const Self) u16 { |
| @@ -0,0 +1,74 @@ | ||
| 1 | //! Sequential test runner for test artifacts that exercise threaded network I/O. | |
| 2 | const builtin = ; | |
| 3 | const std = ; | |
| 4 | const testing = std.testing; | |
| 5 | ||
| 6 | pub const std_options: std.Options = .{ | |
| 7 | .logFn = log, | |
| 8 | }; | |
| 9 | ||
| 10 | var log_error_count: usize = 0; | |
| 11 | ||
| 12 | pub fn main(init: std.process.Init.Minimal) void { | |
| 13 | @disableInstrumentation(); | |
| 14 | ||
| 15 | var passed: usize = 0; | |
| 16 | var skipped: usize = 0; | |
| 17 | var failed: usize = 0; | |
| 18 | var leaked: usize = 0; | |
| 19 | ||
| 20 | for (builtin.test_functions, 0..) |test_fn, index| { | |
| 21 | testing.allocator_instance = .init(std.heap.page_allocator, .{ | |
| 22 | .canary = 0xc3a701ba, | |
| 23 | .check_write_after_free = true, | |
| 24 | }); | |
| 25 | testing.io_instance = .init(testing.allocator, .{ | |
| 26 | .argv0 = .init(init.args), | |
| 27 | .environ = init.environ, | |
| 28 | }); | |
| 29 | testing.log_level = .warn; | |
| 30 | testing.environ = init.environ; | |
| 31 | ||
| 32 | std.debug.print("{d}/{d} {s}...", .{ index + 1, builtin.test_functions.len, test_fn.name }); | |
| 33 | if (test_fn.func()) |_| { | |
| 34 | passed += 1; | |
| 35 | std.debug.print("OK\n", .{}); | |
| 36 | } else |err| switch (err) { | |
| 37 | error.SkipZigTest => { | |
| 38 | skipped += 1; | |
| 39 | std.debug.print("SKIP\n", .{}); | |
| 40 | }, | |
| 41 | else => { | |
| 42 | failed += 1; | |
| 43 | std.debug.print("FAIL ({t})\n", .{err}); | |
| 44 | if (@errorReturnTrace()) |trace| std.debug.dumpErrorReturnTrace(trace); | |
| 45 | }, | |
| 46 | } | |
| 47 | ||
| 48 | testing.io_instance.deinit(); | |
| 49 | leaked += @intFromBool(testing.allocator_instance.deinit() != 0); | |
| 50 | } | |
| 51 | ||
| 52 | std.debug.print("{d} passed; {d} skipped; {d} failed.\n", .{ passed, skipped, failed }); | |
| 53 | if (leaked != 0) std.debug.print("{d} tests leaked memory.\n", .{leaked}); | |
| 54 | if (log_error_count != 0) std.debug.print("{d} errors were logged.\n", .{log_error_count}); | |
| 55 | if (failed != 0 or leaked != 0 or log_error_count != 0) std.process.exit(1); | |
| 56 | } | |
| 57 | ||
| 58 | fn log( | |
| 59 | comptime message_level: std.log.Level, | |
| 60 | comptime scope: @EnumLiteral(), | |
| 61 | comptime format: []const u8, | |
| 62 | args: anytype, | |
| 63 | ) void { | |
| 64 | @disableInstrumentation(); | |
| 65 | if (@backingInt(message_level) <= @backingInt(std.log.Level.err)) { | |
| 66 | log_error_count +|= 1; | |
| 67 | } | |
| 68 | if (@backingInt(message_level) <= @backingInt(testing.log_level)) { | |
| 69 | std.debug.print( | |
| 70 | "[" ++ @tagName(scope) ++ "] (" ++ @tagName(message_level) ++ "): " ++ format ++ "\n", | |
| 71 | args, | |
| 72 | ); | |
| 73 | } | |
| 74 | } | |