authorgravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2026-04-01 17:45:29+02:00
committergravatar for andrew@ziglang.orgAndrew Kelley <andrew@ziglang.org> 2026-04-01 17:45:29+02:00
logb22eb176b0d6e2275b89b715f7e3f0dd0a5fc574
tree81c0f81fef1d3a9a7f0161ccd95482726dc5a22c
parent752f991c09901b07a7084555e140384d162d72c9
parent617979ef26bf11b0dc18ecfc2eddb4c6e2c3ed18

Merge pull request 'audit remaining uses of thread spawning' (#31725) from audit-thread-spawning into master

Reviewed-on: https://codeberg.org/ziglang/zig/pulls/31725

2 files changed, 29 insertions(+), 22 deletions(-)

lib/std/Build/WebServer.zig+21-16
...@@ -6,7 +6,7 @@ root_prog_node: std.Progress.Node,...@@ -6,7 +6,7 @@ root_prog_node: std.Progress.Node,
6watch: bool,6watch: bool,
77
8tcp_server: ?net.Server,8tcp_server: ?net.Server,
9serve_thread: ?std.Thread,9serve_task: ?Io.Future(Io.Cancelable!void),
1010
11/// Uses `Io.Clock.awake`.11/// Uses `Io.Clock.awake`.
12base_timestamp: Io.Timestamp,12base_timestamp: Io.Timestamp,
...@@ -103,7 +103,7 @@ pub fn init(opts: Options) WebServer {...@@ -103,7 +103,7 @@ pub fn init(opts: Options) WebServer {
103 .watch = opts.watch,103 .watch = opts.watch,
104104
105 .tcp_server = null,105 .tcp_server = null,
106 .serve_thread = null,106 .serve_task = null,
107107
108 .base_timestamp = opts.base_timestamp.raw,108 .base_timestamp = opts.base_timestamp.raw,
109 .step_names_trailing = step_names_trailing,109 .step_names_trailing = step_names_trailing,
...@@ -136,9 +136,9 @@ pub fn deinit(ws: *WebServer) void {...@@ -136,9 +136,9 @@ pub fn deinit(ws: *WebServer) void {
136 gpa.free(ws.time_report_msgs);136 gpa.free(ws.time_report_msgs);
137 gpa.free(ws.time_report_update_times);137 gpa.free(ws.time_report_update_times);
138138
139 if (ws.serve_thread) |t| {139 if (ws.serve_task) |t| {
140 if (ws.tcp_server) |*s| s.stream.close(io);140 if (ws.tcp_server) |*s| s.stream.close(io);
141 t.join();141 t.await();
142 }142 }
143 if (ws.tcp_server) |*s| s.deinit();143 if (ws.tcp_server) |*s| s.deinit();
144144
...@@ -146,15 +146,15 @@ pub fn deinit(ws: *WebServer) void {...@@ -146,15 +146,15 @@ pub fn deinit(ws: *WebServer) void {
146}146}
147pub fn start(ws: *WebServer) error{AlreadyReported}!void {147pub fn start(ws: *WebServer) error{AlreadyReported}!void {
148 assert(ws.tcp_server == null);148 assert(ws.tcp_server == null);
149 assert(ws.serve_thread == null);149 assert(ws.serve_task == null);
150 const io = ws.graph.io;150 const io = ws.graph.io;
151151
152 ws.tcp_server = ws.listen_address.listen(io, .{ .reuse_address = true }) catch |err| {152 ws.tcp_server = ws.listen_address.listen(io, .{ .reuse_address = true }) catch |err| {
153 log.err("failed to listen to port {d}: {s}", .{ ws.listen_address.getPort(), @errorName(err) });153 log.err("failed to listen to port {d}: {t}", .{ ws.listen_address.getPort(), err });
154 return error.AlreadyReported;154 return error.AlreadyReported;
155 };155 };
156 ws.serve_thread = std.Thread.spawn(.{}, serve, .{ws}) catch |err| {156 ws.serve_task = io.concurrent(serve, .{ws}) catch |err| {
157 log.err("unable to spawn web server thread: {s}", .{@errorName(err)});157 log.err("unable to spawn web server thread: {t}", .{err});
158 ws.tcp_server.?.deinit(io);158 ws.tcp_server.?.deinit(io);
159 ws.tcp_server = null;159 ws.tcp_server = null;
160 return error.AlreadyReported;160 return error.AlreadyReported;
...@@ -165,15 +165,20 @@ pub fn start(ws: *WebServer) error{AlreadyReported}!void {...@@ -165,15 +165,20 @@ pub fn start(ws: *WebServer) error{AlreadyReported}!void {
165 log.info("hint: pass '--webui={f}' to use the same port next time", .{ws.tcp_server.?.socket.address});165 log.info("hint: pass '--webui={f}' to use the same port next time", .{ws.tcp_server.?.socket.address});
166 }166 }
167}167}
168fn serve(ws: *WebServer) void {168fn serve(ws: *WebServer) Io.Cancelable!void {
169 const io = ws.graph.io;169 const io = ws.graph.io;
170 var group: Io.Group = .init;
171 defer group.cancel(io);
170 while (true) {172 while (true) {
171 var stream = ws.tcp_server.?.accept(io) catch |err| {173 var stream = ws.tcp_server.?.accept(io) catch |err| switch (err) {
172 log.err("failed to accept connection: {s}", .{@errorName(err)});174 error.Canceled => |e| return e,
173 return;175 else => |e| {
176 log.err("failed to accept connection: {t}", .{e});
177 return;
178 },
174 };179 };
175 _ = std.Thread.spawn(.{}, accept, .{ ws, stream }) catch |err| {180 group.concurrent(io, accept, .{ ws, stream }) catch |err| {
176 log.err("unable to spawn connection thread: {s}", .{@errorName(err)});181 log.err("unable to spawn connection thread: {t}", .{err});
177 stream.close(io);182 stream.close(io);
178 continue;183 continue;
179 };184 };
...@@ -303,8 +308,8 @@ fn serveWebSocket(ws: *WebServer, sock: *http.Server.WebSocket) !noreturn {...@@ -303,8 +308,8 @@ fn serveWebSocket(ws: *WebServer, sock: *http.Server.WebSocket) !noreturn {
303 copy.* = @atomicLoad(u8, shared, .monotonic);308 copy.* = @atomicLoad(u8, shared, .monotonic);
304 }309 }
305310
306 const recv_thread = try std.Thread.spawn(.{}, recvWebSocketMessages, .{ ws, sock });311 var recv_thread = try io.concurrent(recvWebSocketMessages, .{ ws, sock });
307 defer recv_thread.join();312 defer recv_thread.cancel(io);
308313
309 {314 {
310 const hello_header: abi.Hello = .{315 const hello_header: abi.Hello = .{
lib/std/fs/test.zig+8-6
...@@ -1758,9 +1758,7 @@ test "open file with exclusive and shared nonblocking lock" {...@@ -1758,9 +1758,7 @@ test "open file with exclusive and shared nonblocking lock" {
1758}1758}
17591759
1760test "open file with exclusive lock twice, make sure second lock waits" {1760test "open file with exclusive lock twice, make sure second lock waits" {
1761 if (builtin.single_threaded) return error.SkipZigTest;1761 testWithAllSupportedPathTypes(struct {
1762
1763 try testWithAllSupportedPathTypes(struct {
1764 fn impl(ctx: *TestContext) !void {1762 fn impl(ctx: *TestContext) !void {
1765 const io = ctx.io;1763 const io = ctx.io;
1766 const filename = try ctx.transformPath("file_lock_test.txt");1764 const filename = try ctx.transformPath("file_lock_test.txt");
...@@ -1781,8 +1779,8 @@ test "open file with exclusive lock twice, make sure second lock waits" {...@@ -1781,8 +1779,8 @@ test "open file with exclusive lock twice, make sure second lock waits" {
1781 var started: Io.Event = .unset;1779 var started: Io.Event = .unset;
1782 var locked: Io.Event = .unset;1780 var locked: Io.Event = .unset;
17831781
1784 const t = try std.Thread.spawn(.{}, S.checkFn, .{ ctx, filename, &started, &locked });1782 var t = try io.concurrent(S.checkFn, .{ ctx, filename, &started, &locked });
1785 defer t.join();1783 defer t.cancel(io) catch {};
17861784
1787 // Wait for the spawned thread to start trying to acquire the exclusive file lock.1785 // Wait for the spawned thread to start trying to acquire the exclusive file lock.
1788 // Then wait a bit to make sure that can't acquire it since we currently hold the file lock.1786 // Then wait a bit to make sure that can't acquire it since we currently hold the file lock.
...@@ -1795,8 +1793,12 @@ test "open file with exclusive lock twice, make sure second lock waits" {...@@ -1795,8 +1793,12 @@ test "open file with exclusive lock twice, make sure second lock waits" {
1795 // Release the file lock which should unlock the thread to lock it and set the locked event.1793 // Release the file lock which should unlock the thread to lock it and set the locked event.
1796 file.close(io);1794 file.close(io);
1797 try locked.wait(io);1795 try locked.wait(io);
1796 try t.await(io);
1798 }1797 }
1799 }.impl);1798 }.impl) catch |err| switch (err) {
1799 error.ConcurrencyUnavailable => return error.SkipZigTest,
1800 else => |e| return e,
1801 };
1800}1802}
18011803
1802test "open file with exclusive nonblocking lock twice (absolute paths)" {1804test "open file with exclusive nonblocking lock twice (absolute paths)" {