mirror of
https://git.robbyzambito.me/zaprus/
synced 2026-05-06 14:50:37 +00:00
Remove any possible exit paths from main loop.
This commit is contained in:
@@ -100,7 +100,7 @@ pub fn init() error{
|
|||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn setTimeout(self: *RawSocket, sec: isize, usec: i64) !void {
|
pub fn setTimeout(self: *RawSocket, sec: isize, usec: i64) error{SetTimeoutError}!void {
|
||||||
const timeout: std.os.linux.timeval = .{ .sec = sec, .usec = usec };
|
const timeout: std.os.linux.timeval = .{ .sec = sec, .usec = usec };
|
||||||
const timeout_ret = std.os.linux.setsockopt(self.fd, std.os.linux.SOL.SOCKET, std.os.linux.SO.RCVTIMEO, @ptrCast(&timeout), @sizeOf(@TypeOf(timeout)));
|
const timeout_ret = std.os.linux.setsockopt(self.fd, std.os.linux.SOL.SOCKET, std.os.linux.SO.RCVTIMEO, @ptrCast(&timeout), @sizeOf(@TypeOf(timeout)));
|
||||||
if (timeout_ret != 0) return error.SetTimeoutError;
|
if (timeout_ret != 0) return error.SetTimeoutError;
|
||||||
|
|||||||
65
src/main.zig
65
src/main.zig
@@ -148,7 +148,7 @@ pub fn main(init: std.process.Init) !void {
|
|||||||
reconnect: while (true) {
|
reconnect: while (true) {
|
||||||
client = SaprusClient.init() catch |err| switch (err) {
|
client = SaprusClient.init() catch |err| switch (err) {
|
||||||
error.NoInterfaceFound => {
|
error.NoInterfaceFound => {
|
||||||
try init.io.sleep(.fromMilliseconds(100), .boot);
|
init.io.sleep(.fromMilliseconds(100), .boot) catch unreachable;
|
||||||
continue :reconnect;
|
continue :reconnect;
|
||||||
},
|
},
|
||||||
else => |e| return e,
|
else => |e| return e,
|
||||||
@@ -156,24 +156,44 @@ pub fn main(init: std.process.Init) !void {
|
|||||||
defer client.deinit();
|
defer client.deinit();
|
||||||
log.debug("Starting connection", .{});
|
log.debug("Starting connection", .{});
|
||||||
|
|
||||||
try client.socket.setTimeout(if (is_debug) 3 else 25, 0);
|
client.socket.setTimeout(if (is_debug) 3 else 25, 0) catch {
|
||||||
|
log.err("Unable to set timeout", .{});
|
||||||
|
init.io.sleep(.fromMilliseconds(100), .boot) catch unreachable;
|
||||||
|
continue :reconnect;
|
||||||
|
};
|
||||||
var connection = client.connect(init.io, w.buffered()) catch {
|
var connection = client.connect(init.io, w.buffered()) catch {
|
||||||
log.debug("Connection timed out", .{});
|
log.debug("Connection timed out", .{});
|
||||||
continue;
|
continue :reconnect;
|
||||||
};
|
};
|
||||||
|
|
||||||
log.debug("Connection started", .{});
|
log.debug("Connection started", .{});
|
||||||
|
|
||||||
next_message: while (true) {
|
next_message: while (true) {
|
||||||
var res_buf: [2048]u8 = undefined;
|
var res_buf: [2048]u8 = undefined;
|
||||||
try client.socket.setTimeout(if (is_debug) 60 else 600, 0);
|
client.socket.setTimeout(if (is_debug) 60 else 600, 0) catch {
|
||||||
|
log.err("Unable to set timeout", .{});
|
||||||
|
init.io.sleep(.fromMilliseconds(100), .boot) catch unreachable;
|
||||||
|
continue :reconnect;
|
||||||
|
};
|
||||||
const next = connection.next(init.io, &res_buf) catch {
|
const next = connection.next(init.io, &res_buf) catch {
|
||||||
continue :reconnect;
|
continue :reconnect;
|
||||||
};
|
};
|
||||||
|
|
||||||
const b64d = std.base64.standard.Decoder;
|
const b64d = std.base64.standard.Decoder;
|
||||||
var connection_payload_buf: [2048]u8 = undefined;
|
var connection_payload_buf: [2048]u8 = undefined;
|
||||||
const connection_payload = connection_payload_buf[0..try b64d.calcSizeForSlice(next)];
|
const connection_payload = blk: {
|
||||||
|
const size = b64d.calcSizeForSlice(next) catch |err| switch (err) {
|
||||||
|
error.InvalidCharacter, error.InvalidPadding => {
|
||||||
|
log.warn("Invalid base64 message received, ignoring: '{s}'", .{next});
|
||||||
|
continue :next_message;
|
||||||
|
},
|
||||||
|
error.NoSpaceLeft => {
|
||||||
|
log.warn("No space left when decoding base64 string, ignoring.", .{});
|
||||||
|
continue :next_message;
|
||||||
|
},
|
||||||
|
};
|
||||||
|
break :blk connection_payload_buf[0..size];
|
||||||
|
};
|
||||||
b64d.decode(connection_payload, next) catch {
|
b64d.decode(connection_payload, next) catch {
|
||||||
log.debug("Failed to decode message, skipping: '{s}'", .{connection_payload});
|
log.debug("Failed to decode message, skipping: '{s}'", .{connection_payload});
|
||||||
continue;
|
continue;
|
||||||
@@ -221,7 +241,7 @@ pub fn main(init: std.process.Init) !void {
|
|||||||
|
|
||||||
var is_killed: std.atomic.Value(bool) = .init(false);
|
var is_killed: std.atomic.Value(bool) = .init(false);
|
||||||
|
|
||||||
var kill_task = try init.io.concurrent(killProcessAfter, .{ init.io, &child, .fromSeconds(3), &is_killed });
|
var kill_task = init.io.concurrent(killProcessAfter, .{ init.io, &child, .fromSeconds(3), &is_killed }) catch unreachable;
|
||||||
defer _ = kill_task.cancel(init.io) catch {};
|
defer _ = kill_task.cancel(init.io) catch {};
|
||||||
|
|
||||||
var cmd_output_buf: [SaprusClient.max_payload_len * 2]u8 = undefined;
|
var cmd_output_buf: [SaprusClient.max_payload_len * 2]u8 = undefined;
|
||||||
@@ -244,12 +264,41 @@ pub fn main(init: std.process.Init) !void {
|
|||||||
break;
|
break;
|
||||||
},
|
},
|
||||||
};
|
};
|
||||||
cmd_output.print("{b64}", .{try child_output_reader.interface.takeArray(child_output_buf.len)}) catch unreachable;
|
const child_output_chunk = child_output_reader.interface.takeArray(child_output_buf.len) catch |err| switch (err) {
|
||||||
|
error.EndOfStream => {
|
||||||
|
log.warn("Reached end of stream when reading from the child process. Maybe this should be handled more gracefull, but ignoring for now.", .{});
|
||||||
|
continue :next_message;
|
||||||
|
},
|
||||||
|
error.ReadFailed => if (child_output_reader.err) |co_err| switch (co_err) {
|
||||||
|
error.AccessDenied,
|
||||||
|
error.ConnectionResetByPeer,
|
||||||
|
error.InputOutput,
|
||||||
|
error.IsDir,
|
||||||
|
error.LockViolation,
|
||||||
|
error.NotOpenForReading,
|
||||||
|
error.SocketUnconnected,
|
||||||
|
error.SystemResources,
|
||||||
|
error.WouldBlock,
|
||||||
|
=> |e| {
|
||||||
|
log.err("Unending error reading output from child process: {t}", .{e});
|
||||||
|
continue :next_message;
|
||||||
|
},
|
||||||
|
error.Canceled => |e| return e,
|
||||||
|
error.Unexpected => {
|
||||||
|
log.err("Unexpected error reading output from child process.", .{});
|
||||||
|
continue :next_message;
|
||||||
|
},
|
||||||
|
} else {
|
||||||
|
log.err("Shouldn't get here :(", .{});
|
||||||
|
continue :next_message;
|
||||||
|
},
|
||||||
|
};
|
||||||
|
cmd_output.print("{b64}", .{child_output_chunk}) catch unreachable;
|
||||||
connection.send(init.io, .{}, cmd_output.buffered()) catch |err| {
|
connection.send(init.io, .{}, cmd_output.buffered()) catch |err| {
|
||||||
log.debug("Failed to send connection chunk: {t}", .{err});
|
log.debug("Failed to send connection chunk: {t}", .{err});
|
||||||
continue :next_message;
|
continue :next_message;
|
||||||
};
|
};
|
||||||
try init.io.sleep(.fromMilliseconds(40), .boot);
|
init.io.sleep(.fromMilliseconds(40), .boot) catch unreachable;
|
||||||
} else {
|
} else {
|
||||||
kill_task.cancel(init.io) catch {};
|
kill_task.cancel(init.io) catch {};
|
||||||
killProcessAfter(init.io, &child, .zero, &is_killed) catch |err| {
|
killProcessAfter(init.io, &child, .zero, &is_killed) catch |err| {
|
||||||
|
|||||||
Reference in New Issue
Block a user