neurocyte/thespian
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: return all subprocess errors as term messages

Browse source
This commit is contained in:
CJ van den Berg 2024-12-16 19:05:21 +01:00
parent e44e6ed306
commit 9df7beb192
Signed by: neurocyte
GPG key ID: 8EB1E1BB660E3FB9
2 changed files with 48 additions and 42 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

55
src/subprocess.zig
View file

@ -130,18 +130,15 @@ const Proc = struct {
fn start(self: *Proc) tp.result { fn start(self: *Proc) tp.result {
errdefer self.deinit(); errdefer self.deinit();
self.child.spawn() catch |e| { self.child.spawn() catch |e| return self.handle_error(e);
try self.parent.send(.{ self.tag, "term", e, 1 });
return tp.exit_normal();
};
_ = self.args.reset(.free_all); _ = self.args.reset(.free_all);
if (self.child.stdin_behavior == .Pipe) if (self.child.stdin_behavior == .Pipe)
self.fd_stdin = tp.file_descriptor.init("stdin", self.child.stdin.?.handle) catch |e| return tp.exit_error(e, @errorReturnTrace()); self.fd_stdin = tp.file_descriptor.init("stdin", self.child.stdin.?.handle) catch |e| return self.handle_error(e);
self.fd_stdout = tp.file_descriptor.init("stdout", self.child.stdout.?.handle) catch |e| return tp.exit_error(e, @errorReturnTrace()); self.fd_stdout = tp.file_descriptor.init("stdout", self.child.stdout.?.handle) catch |e| return self.handle_error(e);
self.fd_stderr = tp.file_descriptor.init("stderr", self.child.stderr.?.handle) catch |e| return tp.exit_error(e, @errorReturnTrace()); self.fd_stderr = tp.file_descriptor.init("stderr", self.child.stderr.?.handle) catch |e| return self.handle_error(e);
if (self.fd_stdout) |fd_stdout| fd_stdout.wait_read() catch |e| return tp.exit_error(e, @errorReturnTrace()); if (self.fd_stdout) |fd_stdout| fd_stdout.wait_read() catch |e| return self.handle_error(e);
if (self.fd_stderr) |fd_stderr| fd_stderr.wait_read() catch |e| return tp.exit_error(e, @errorReturnTrace()); if (self.fd_stderr) |fd_stderr| fd_stderr.wait_read() catch |e| return self.handle_error(e);
tp.receive(&self.receiver); tp.receive(&self.receiver);
} }
@ -153,22 +150,22 @@ const Proc = struct {
var err_msg: []u8 = ""; var err_msg: []u8 = "";
if (try m.match(.{ "fd", "stdout", "read_ready" })) { if (try m.match(.{ "fd", "stdout", "read_ready" })) {
try self.dispatch_stdout(); try self.dispatch_stdout();
if (self.fd_stdout) |fd_stdout| fd_stdout.wait_read() catch |e| return tp.exit_error(e, @errorReturnTrace()); if (self.fd_stdout) |fd_stdout| fd_stdout.wait_read() catch |e| return self.handle_error(e);
} else if (try m.match(.{ "fd", "stderr", "read_ready" })) { } else if (try m.match(.{ "fd", "stderr", "read_ready" })) {
try self.dispatch_stderr(); try self.dispatch_stderr();
if (self.fd_stderr) |fd_stderr| fd_stderr.wait_read() catch |e| return tp.exit_error(e, @errorReturnTrace()); if (self.fd_stderr) |fd_stderr| fd_stderr.wait_read() catch |e| return self.handle_error(e);
} else if (try m.match(.{ "fd", "stdin", "write_ready" })) { } else if (try m.match(.{ "fd", "stdin", "write_ready" })) {
if (self.stdin_buffer.items.len > 0) { if (self.stdin_buffer.items.len > 0) {
if (self.child.stdin) |stdin| { if (self.child.stdin) |stdin| {
const written = stdin.write(self.stdin_buffer.items) catch |e| switch (e) { const written = stdin.write(self.stdin_buffer.items) catch |e| switch (e) {
error.WouldBlock => { error.WouldBlock => {
if (self.fd_stdin) |fd_stdin| { if (self.fd_stdin) |fd_stdin| {
fd_stdin.wait_write() catch |e_| return tp.exit_error(e_, @errorReturnTrace()); fd_stdin.wait_write() catch |e_| return self.handle_error(e_);
self.write_pending = true; self.write_pending = true;
return; return;
} else return tp.exit_error(error.WouldBlock, @errorReturnTrace()); } else return self.handle_error(error.WouldBlock);
}, },
else => return tp.exit_error(e, @errorReturnTrace()), else => return self.handle_error(e),
}; };
self.write_pending = false; self.write_pending = false;
defer { defer {
@ -181,7 +178,7 @@ const Proc = struct {
std.mem.copyForwards(u8, self.stdin_buffer.items, self.stdin_buffer.items[written..]); std.mem.copyForwards(u8, self.stdin_buffer.items, self.stdin_buffer.items[written..]);
self.stdin_buffer.items.len = self.stdin_buffer.items.len - written; self.stdin_buffer.items.len = self.stdin_buffer.items.len - written;
if (self.fd_stdin) |fd_stdin| { if (self.fd_stdin) |fd_stdin| {
fd_stdin.wait_write() catch |e| return tp.exit_error(e, @errorReturnTrace()); fd_stdin.wait_write() catch |e| return self.handle_error(e);
self.write_pending = true; self.write_pending = true;
} }
} }
@ -189,8 +186,8 @@ const Proc = struct {
} }
} else if (try m.match(.{ "stdin", tp.extract(&bytes) })) { } else if (try m.match(.{ "stdin", tp.extract(&bytes) })) {
if (self.fd_stdin) |fd_stdin| { if (self.fd_stdin) |fd_stdin| {
self.stdin_buffer.appendSlice(bytes) catch |e| return tp.exit_error(e, @errorReturnTrace()); self.stdin_buffer.appendSlice(bytes) catch |e| return self.handle_error(e);
fd_stdin.wait_write() catch |e| return tp.exit_error(e, @errorReturnTrace()); fd_stdin.wait_write() catch |e| return self.handle_error(e);
self.write_pending = true; self.write_pending = true;
} }
} else if (try m.match(.{"stdin_close"})) { } else if (try m.match(.{"stdin_close"})) {
@ -210,10 +207,11 @@ const Proc = struct {
self.child.stderr = null; self.child.stderr = null;
} }
} else if (try m.match(.{"term"})) { } else if (try m.match(.{"term"})) {
const term_ = self.child.kill() catch |e| return tp.exit_error(e, @errorReturnTrace()); const term_ = self.child.kill() catch |e| return self.handle_error(e);
return self.handle_term(term_); return self.handle_term(term_);
} else if (try m.match(.{ "fd", tp.any, "read_error", tp.extract(&err), tp.extract(&err_msg) })) { } else if (try m.match(.{ "fd", tp.any, "read_error", tp.extract(&err), tp.extract(&err_msg) })) {
return tp.exit(err_msg); try self.parent.send(.{ self.tag, "term", err_msg, 1 });
return tp.exit_normal();
} }
} }
@ -227,10 +225,10 @@ const Proc = struct {
fn dispatch_stdout(self: *Proc) tp.result { fn dispatch_stdout(self: *Proc) tp.result {
var buffer: [max_chunk_size]u8 = undefined; var buffer: [max_chunk_size]u8 = undefined;
const stdout = self.child.stdout orelse return tp.exit("cannot read closed stdout"); const stdout = self.child.stdout orelse return self.handle_error(error.ReadNoStdout);
const bytes = stdout.read(&buffer) catch |e| switch (e) { const bytes = stdout.read(&buffer) catch |e| switch (e) {
error.WouldBlock => return, error.WouldBlock => return,
else => return tp.exit_error(e, @errorReturnTrace()), else => return self.handle_error(e),
}; };
if (bytes == 0) if (bytes == 0)
return self.handle_terminate(); return self.handle_terminate();
@ -239,21 +237,21 @@ const Proc = struct {
fn dispatch_stderr(self: *Proc) tp.result { fn dispatch_stderr(self: *Proc) tp.result {
var buffer: [max_chunk_size]u8 = undefined; var buffer: [max_chunk_size]u8 = undefined;
const stderr = self.child.stderr orelse return tp.exit("cannot read closed stderr"); const stderr = self.child.stderr orelse return self.handle_error(error.ReadNoStderr);
const bytes = stderr.read(&buffer) catch |e| switch (e) { const bytes = stderr.read(&buffer) catch |e| switch (e) {
error.WouldBlock => return, error.WouldBlock => return,
else => return tp.exit_error(e, @errorReturnTrace()), else => return self.handle_error(e),
}; };
if (bytes == 0) if (bytes == 0)
return; return;
try self.parent.send(.{ self.tag, "stderr", buffer[0..bytes] }); try self.parent.send(.{ self.tag, "stderr", buffer[0..bytes] });
} }
fn handle_terminate(self: *Proc) tp.result { fn handle_terminate(self: *Proc) error{Exit} {
return self.handle_term(self.child.wait() catch |e| return tp.exit_error(e, @errorReturnTrace())); return self.handle_term(self.child.wait() catch |e| return self.handle_error(e));
} }
fn handle_term(self: *Proc, term_: std.process.Child.Term) tp.result { fn handle_term(self: *Proc, term_: std.process.Child.Term) error{Exit} {
(switch (term_) { (switch (term_) {
.Exited => |val| self.parent.send(.{ self.tag, "term", "exited", val }), .Exited => |val| self.parent.send(.{ self.tag, "term", "exited", val }),
.Signal => |val| self.parent.send(.{ self.tag, "term", "signal", val }), .Signal => |val| self.parent.send(.{ self.tag, "term", "signal", val }),
@ -262,4 +260,9 @@ const Proc = struct {
}) catch {}; }) catch {};
return tp.exit_normal(); return tp.exit_normal();
} }
fn handle_error(self: *Proc, e: anyerror) error{Exit} {
try self.parent.send(.{ self.tag, "term", e, 1 });
return tp.exit_normal();
}
}; };

35
src/subprocess_windows.zig
View file

@ -123,16 +123,13 @@ const Proc = struct {
fn start(self: *Proc) tp.result { fn start(self: *Proc) tp.result {
errdefer self.deinit(); errdefer self.deinit();
self.child.spawn() catch |e| { self.child.spawn() catch |e| return self.handle_error(e);
try self.parent.send(.{ self.tag, "term", e, 1 });
return tp.exit_normal();
};
_ = self.args.reset(.free_all); _ = self.args.reset(.free_all);
self.stream_stdout = tp.file_stream.init("stdout", self.child.stdout.?.handle) catch |e| return tp.exit_error(e, @errorReturnTrace()); self.stream_stdout = tp.file_stream.init("stdout", self.child.stdout.?.handle) catch |e| return self.handle_error(e);
self.stream_stderr = tp.file_stream.init("stderr", self.child.stderr.?.handle) catch |e| return tp.exit_error(e, @errorReturnTrace()); self.stream_stderr = tp.file_stream.init("stderr", self.child.stderr.?.handle) catch |e| return self.handle_error(e);
if (self.stream_stdout) |stream| stream.start_read() catch |e| return tp.exit_error(e, @errorReturnTrace()); if (self.stream_stdout) |stream| stream.start_read() catch |e| return self.handle_error(e);
if (self.stream_stderr) |stream| stream.start_read() catch |e| return tp.exit_error(e, @errorReturnTrace()); if (self.stream_stderr) |stream| stream.start_read() catch |e| return self.handle_error(e);
tp.receive(&self.receiver); tp.receive(&self.receiver);
} }
@ -145,10 +142,10 @@ const Proc = struct {
var err_msg: []u8 = ""; var err_msg: []u8 = "";
if (try m.match(.{ "stream", "stdout", "read_complete", tp.extract(&bytes) })) { if (try m.match(.{ "stream", "stdout", "read_complete", tp.extract(&bytes) })) {
try self.dispatch_stdout(bytes); try self.dispatch_stdout(bytes);
if (self.stream_stdout) |stream| stream.start_read() catch |e| return tp.exit_error(e, @errorReturnTrace()); if (self.stream_stdout) |stream| stream.start_read() catch |e| return self.handle_error(e);
} else if (try m.match(.{ "stream", "stderr", "read_complete", tp.extract(&bytes) })) { } else if (try m.match(.{ "stream", "stderr", "read_complete", tp.extract(&bytes) })) {
try self.dispatch_stderr(bytes); try self.dispatch_stderr(bytes);
if (self.stream_stderr) |stream| stream.start_read() catch |e| return tp.exit_error(e, @errorReturnTrace()); if (self.stream_stderr) |stream| stream.start_read() catch |e| return self.handle_error(e);
} else if (try m.match(.{ "stdin", tp.extract(&bytes) })) { } else if (try m.match(.{ "stdin", tp.extract(&bytes) })) {
try self.start_write(bytes); try self.start_write(bytes);
} else if (try m.match(.{"stdin_close"})) { } else if (try m.match(.{"stdin_close"})) {
@ -164,7 +161,7 @@ const Proc = struct {
self.child.stderr = null; self.child.stderr = null;
} }
} else if (try m.match(.{"term"})) { } else if (try m.match(.{"term"})) {
const term_ = self.child.kill() catch |e| return tp.exit_error(e, @errorReturnTrace()); const term_ = self.child.kill() catch |e| return self.handle_error(e);
return self.handle_term(term_); return self.handle_term(term_);
} else if (try m.match(.{ "stream", "stdout", "read_error", 109, tp.extract(&err_msg) })) { } else if (try m.match(.{ "stream", "stdout", "read_error", 109, tp.extract(&err_msg) })) {
// stdout closed // stdout closed
@ -174,13 +171,14 @@ const Proc = struct {
// stderr closed // stderr closed
self.child.stderr = null; self.child.stderr = null;
} else if (try m.match(.{ "stream", tp.extract(&stream_name), "read_error", tp.extract(&err), tp.extract(&err_msg) })) { } else if (try m.match(.{ "stream", tp.extract(&stream_name), "read_error", tp.extract(&err), tp.extract(&err_msg) })) {
return tp.exit_fmt("{s} read_error: {s}", .{ stream_name, err_msg }); try self.parent.send(.{ self.tag, "term", err_msg, 1 });
return tp.exit_normal();
} }
} }
fn start_write(self: *Proc, bytes: []const u8) tp.result { fn start_write(self: *Proc, bytes: []const u8) tp.result {
if (self.child.stdin) |stdin| if (self.child.stdin) |stdin|
stdin.writeAll(bytes) catch |e| return tp.exit_error(e, @errorReturnTrace()); stdin.writeAll(bytes) catch |e| return self.handle_error(e);
} }
fn stdin_close(self: *Proc) void { fn stdin_close(self: *Proc) void {
@ -203,11 +201,11 @@ const Proc = struct {
try self.parent.send(.{ self.tag, "stderr", bytes }); try self.parent.send(.{ self.tag, "stderr", bytes });
} }
fn handle_terminate(self: *Proc) tp.result { fn handle_terminate(self: *Proc) error{Exit} {
return self.handle_term(self.child.wait() catch |e| return tp.exit_error(e, @errorReturnTrace())); return self.handle_term(self.child.wait() catch |e| return self.handle_error(e));
} }
fn handle_term(self: *Proc, term_: std.process.Child.Term) tp.result { fn handle_term(self: *Proc, term_: std.process.Child.Term) error{Exit} {
(switch (term_) { (switch (term_) {
.Exited => |val| self.parent.send(.{ self.tag, "term", "exited", val }), .Exited => |val| self.parent.send(.{ self.tag, "term", "exited", val }),
.Signal => |val| self.parent.send(.{ self.tag, "term", "signal", val }), .Signal => |val| self.parent.send(.{ self.tag, "term", "signal", val }),
@ -216,4 +214,9 @@ const Proc = struct {
}) catch {}; }) catch {};
return tp.exit_normal(); return tp.exit_normal();
} }
fn handle_error(self: *Proc, e: anyerror) error{Exit} {
try self.parent.send(.{ self.tag, "term", e, 1 });
return tp.exit_normal();
}
}; };