349 lines
13 KiB
Zig
349 lines
13 KiB
Zig
/// Child binary for the endpoint test.
|
|
///
|
|
/// Runs a full Thespian context with actors registered in a StdioEndpoint:
|
|
///
|
|
/// "echo" (wire ID 2): receives {from_id, payload}, replies via send_named
|
|
/// to "test_receiver" on the parent side.
|
|
///
|
|
/// "echo_id" (wire ID 3): receives {from_id, payload}.
|
|
/// - If from_id == 0: trigger mode — sends send_named to "test_receiver"
|
|
/// so the parent can establish a proxy and reply by ID.
|
|
/// - If from_id != 0: reply mode — sends wire send to from_id with ["done"].
|
|
///
|
|
/// StdioEndpoint dispatches inbound wire messages to actors, passing
|
|
/// {from_id, payload} so actors can reply by ID when needed.
|
|
const std = @import("std");
|
|
const tp = @import("thespian");
|
|
const cbor = @import("cbor");
|
|
const framing = @import("framing");
|
|
const protocol = @import("protocol");
|
|
|
|
var trace_file: ?std.fs.File = null;
|
|
var trace_buf: [4096]u8 = undefined;
|
|
var trace_file_writer: std.fs.File.Writer = undefined;
|
|
|
|
fn trace_handler(buf: tp.message.c_buffer_type) callconv(.c) void {
|
|
if (trace_file == null) return;
|
|
cbor.toJsonWriter(buf.base[0..buf.len], &trace_file_writer.interface, .{}) catch return;
|
|
trace_file_writer.interface.writeByte('\n') catch return;
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// EchoActor (wire ID 2)
|
|
// ---------------------------------------------------------------------------
|
|
|
|
const EchoActor = struct {
|
|
allocator: std.mem.Allocator,
|
|
receiver: tp.Receiver(*@This()),
|
|
|
|
const Args = struct {
|
|
allocator: std.mem.Allocator,
|
|
/// Owned pid of the StdioEndpoint to notify when ready.
|
|
parent: tp.pid,
|
|
};
|
|
|
|
fn start(args: Args) tp.result {
|
|
return init(args) catch |e| return tp.exit_error(e, @errorReturnTrace());
|
|
}
|
|
|
|
fn init(args: Args) !void {
|
|
defer args.parent.deinit();
|
|
const self = try args.allocator.create(@This());
|
|
self.* = .{
|
|
.allocator = args.allocator,
|
|
.receiver = .init(receive_fn, deinit, self),
|
|
};
|
|
errdefer self.deinit();
|
|
try args.parent.send(.{"echo_ready"});
|
|
tp.receive(&self.receiver);
|
|
}
|
|
|
|
fn deinit(self: *@This()) void {
|
|
self.allocator.destroy(self);
|
|
}
|
|
|
|
fn receive_fn(self: *@This(), from: tp.pid_ref, m: tp.message) tp.result {
|
|
return self.receive(from, m) catch |e| return tp.exit_error(e, @errorReturnTrace());
|
|
}
|
|
|
|
fn receive(self: *@This(), from: tp.pid_ref, m: tp.message) !void {
|
|
var from_id: u64 = 0;
|
|
var payload: []const u8 = "";
|
|
if (try m.match(.{ tp.extract(&from_id), cbor.extract_cbor(&payload) })) {
|
|
// Reply via send_named to "test_receiver" on the parent side.
|
|
try from.send(.{ "send", @as(u64, 2), "test_receiver", protocol.RawCbor{ .bytes = payload } });
|
|
} else {
|
|
return tp.unexpected(m);
|
|
}
|
|
_ = self;
|
|
}
|
|
};
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// EchoIdActor (wire ID 3)
|
|
// ---------------------------------------------------------------------------
|
|
|
|
const EchoIdActor = struct {
|
|
allocator: std.mem.Allocator,
|
|
receiver: tp.Receiver(*@This()),
|
|
|
|
const Args = struct {
|
|
allocator: std.mem.Allocator,
|
|
/// Owned pid of the StdioEndpoint to notify when ready.
|
|
parent: tp.pid,
|
|
};
|
|
|
|
fn start(args: Args) tp.result {
|
|
return init(args) catch |e| return tp.exit_error(e, @errorReturnTrace());
|
|
}
|
|
|
|
fn init(args: Args) !void {
|
|
defer args.parent.deinit();
|
|
const self = try args.allocator.create(@This());
|
|
self.* = .{
|
|
.allocator = args.allocator,
|
|
.receiver = .init(receive_fn, deinit, self),
|
|
};
|
|
errdefer self.deinit();
|
|
try args.parent.send(.{"echo_id_ready"});
|
|
tp.receive(&self.receiver);
|
|
}
|
|
|
|
fn deinit(self: *@This()) void {
|
|
self.allocator.destroy(self);
|
|
}
|
|
|
|
fn receive_fn(self: *@This(), from: tp.pid_ref, m: tp.message) tp.result {
|
|
return self.receive(from, m) catch |e| return tp.exit_error(e, @errorReturnTrace());
|
|
}
|
|
|
|
fn receive(self: *@This(), from: tp.pid_ref, m: tp.message) !void {
|
|
var from_id: u64 = 0;
|
|
var payload: []const u8 = "";
|
|
if (try m.match(.{ tp.extract(&from_id), cbor.extract_cbor(&payload) })) {
|
|
if (from_id == 0) {
|
|
// Trigger mode: send send_named so the parent can establish
|
|
// a proxy and will reply by ID.
|
|
try from.send(.{ "send", @as(u64, 3), "test_receiver", protocol.RawCbor{ .bytes = payload } });
|
|
} else {
|
|
// Reply mode: send ["done"] back to the parent actor by wire ID.
|
|
var done_buf: [8]u8 = undefined;
|
|
const done_msg = try tp.message.fmtbuf(&done_buf, .{"done"});
|
|
try from.send(.{ "send", @as(u64, 3), from_id, protocol.RawCbor{ .bytes = done_msg.buf } });
|
|
}
|
|
} else {
|
|
return tp.unexpected(m);
|
|
}
|
|
_ = self;
|
|
}
|
|
};
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// StdioEndpoint
|
|
// ---------------------------------------------------------------------------
|
|
|
|
const StdioEndpoint = struct {
|
|
allocator: std.mem.Allocator,
|
|
fd_stdin: tp.file_descriptor,
|
|
accumulator: framing.Accumulator,
|
|
read_buf: [4096]u8,
|
|
/// Maps wire ID → local actor pid for routing inbound send-by-ID.
|
|
wire_ids: std.AutoHashMap(u64, tp.pid),
|
|
/// Number of child actors that have reported ready; arm stdin when all ready.
|
|
ready_count: u8,
|
|
receiver: tp.Receiver(*@This()),
|
|
|
|
const Args = struct { allocator: std.mem.Allocator };
|
|
|
|
fn start(args: Args) tp.result {
|
|
return init(args) catch |e| return tp.exit_error(e, @errorReturnTrace());
|
|
}
|
|
|
|
fn init(args: Args) !void {
|
|
const fd_stdin = try tp.file_descriptor.init("stdin", 0);
|
|
const self = try args.allocator.create(@This());
|
|
self.* = .{
|
|
.allocator = args.allocator,
|
|
.fd_stdin = fd_stdin,
|
|
.accumulator = .{},
|
|
.read_buf = undefined,
|
|
.wire_ids = std.AutoHashMap(u64, tp.pid).init(args.allocator),
|
|
.ready_count = 0,
|
|
.receiver = .init(receive_fn, deinit, self),
|
|
};
|
|
errdefer self.deinit();
|
|
|
|
_ = try tp.spawn_link(args.allocator, EchoActor.Args{
|
|
.allocator = args.allocator,
|
|
.parent = tp.self_pid().clone(),
|
|
}, EchoActor.start, "echo");
|
|
|
|
_ = try tp.spawn_link(args.allocator, EchoIdActor.Args{
|
|
.allocator = args.allocator,
|
|
.parent = tp.self_pid().clone(),
|
|
}, EchoIdActor.start, "echo_id");
|
|
|
|
tp.receive(&self.receiver);
|
|
}
|
|
|
|
fn deinit(self: *@This()) void {
|
|
var it = self.wire_ids.valueIterator();
|
|
while (it.next()) |p| p.deinit();
|
|
self.wire_ids.deinit();
|
|
self.fd_stdin.deinit();
|
|
self.allocator.destroy(self);
|
|
}
|
|
|
|
fn receive_fn(self: *@This(), from: tp.pid_ref, m: tp.message) tp.result {
|
|
return self.receive(from, m) catch |e| return tp.exit_error(e, @errorReturnTrace());
|
|
}
|
|
|
|
fn receive(self: *@This(), from: tp.pid_ref, m: tp.message) !void {
|
|
var from_id: u64 = 0;
|
|
var to_id_wire: u64 = 0;
|
|
var to_name: []const u8 = "";
|
|
var payload: []const u8 = "";
|
|
|
|
if (try m.match(.{"echo_ready"})) {
|
|
tp.env.get().proc_set("echo", from);
|
|
try self.wire_ids.put(2, from.clone());
|
|
self.ready_count += 1;
|
|
if (self.ready_count == 2) try self.fd_stdin.wait_read();
|
|
} else if (try m.match(.{"echo_id_ready"})) {
|
|
tp.env.get().proc_set("echo_id", from);
|
|
try self.wire_ids.put(3, from.clone());
|
|
self.ready_count += 1;
|
|
if (self.ready_count == 2) try self.fd_stdin.wait_read();
|
|
} else if (try m.match(.{ "fd", "stdin", "read_ready" })) {
|
|
try self.dispatch_stdin();
|
|
try self.fd_stdin.wait_read();
|
|
} else if (try m.match(.{ "fd", "stdin", "read_error", tp.any, tp.any })) {
|
|
self.send_transport_error("stdin_closed");
|
|
return tp.exit("stdin_closed");
|
|
} else if (try m.match(.{ "send", tp.extract(&from_id), tp.extract(&to_id_wire), cbor.extract_cbor(&payload) })) {
|
|
try self.send_wire_by_id(from_id, to_id_wire, payload);
|
|
} else if (try m.match(.{ "send", tp.extract(&from_id), tp.extract(&to_name), cbor.extract_cbor(&payload) })) {
|
|
try self.send_wire(from_id, to_name, payload);
|
|
} else {
|
|
return tp.unexpected(m);
|
|
}
|
|
}
|
|
|
|
fn dispatch_stdin(self: *@This()) !void {
|
|
const n = std.fs.File.stdin().read(&self.read_buf) catch |e| switch (e) {
|
|
error.WouldBlock => return,
|
|
else => return tp.exit_error(e, @errorReturnTrace()),
|
|
};
|
|
if (n == 0) {
|
|
self.send_transport_error("stdin_closed");
|
|
return tp.exit("stdin_closed");
|
|
}
|
|
if (self.accumulator.feed(self.read_buf[0..n])) |frame| {
|
|
try self.dispatch_frame(frame);
|
|
}
|
|
}
|
|
|
|
fn dispatch_frame(self: *@This(), frame: []const u8) !void {
|
|
const msg = try protocol.decode(frame);
|
|
switch (msg) {
|
|
.send_named => |s| {
|
|
const actor = tp.env.get().proc(s.to_name);
|
|
// Include from_id so the actor can reply by ID if needed.
|
|
try actor.send(.{ s.from_id, protocol.RawCbor{ .bytes = s.payload } });
|
|
},
|
|
.send => |s| {
|
|
// Route by wire ID.
|
|
if (self.wire_ids.get(s.to_id)) |actor|
|
|
try actor.send(.{ s.from_id, protocol.RawCbor{ .bytes = s.payload } })
|
|
else
|
|
return tp.exit_error(error.UnknownWireId, null);
|
|
},
|
|
.transport_error => |te| return tp.exit(te.reason),
|
|
.link, .exit => return tp.exit_error(error.UnexpectedMessage, null),
|
|
}
|
|
}
|
|
|
|
fn send_wire_by_id(_: *@This(), from_id: u64, to_id: u64, payload: []const u8) !void {
|
|
var msg_buf: [framing.max_frame_size]u8 = undefined;
|
|
var msg_stream: std.Io.Writer = .fixed(&msg_buf);
|
|
try protocol.encode_send(&msg_stream, from_id, to_id, payload);
|
|
|
|
var frame_buf: [framing.max_frame_size + 4]u8 = undefined;
|
|
var frame_stream: std.Io.Writer = .fixed(&frame_buf);
|
|
try framing.write_frame(&frame_stream, msg_stream.buffered());
|
|
try std.fs.File.stdout().writeAll(frame_stream.buffered());
|
|
}
|
|
|
|
fn send_wire(_: *@This(), from_id: u64, to_name: []const u8, payload: []const u8) !void {
|
|
var msg_buf: [framing.max_frame_size]u8 = undefined;
|
|
var msg_stream: std.Io.Writer = .fixed(&msg_buf);
|
|
try protocol.encode_send_named(&msg_stream, from_id, to_name, payload);
|
|
|
|
var frame_buf: [framing.max_frame_size + 4]u8 = undefined;
|
|
var frame_stream: std.Io.Writer = .fixed(&frame_buf);
|
|
try framing.write_frame(&frame_stream, msg_stream.buffered());
|
|
try std.fs.File.stdout().writeAll(frame_stream.buffered());
|
|
}
|
|
|
|
fn send_transport_error(_: *@This(), reason: []const u8) void {
|
|
var msg_buf: [framing.max_frame_size]u8 = undefined;
|
|
var msg_stream: std.Io.Writer = .fixed(&msg_buf);
|
|
protocol.encode_transport_error(&msg_stream, reason) catch return;
|
|
var frame_buf: [framing.max_frame_size + 4]u8 = undefined;
|
|
var frame_stream: std.Io.Writer = .fixed(&frame_buf);
|
|
framing.write_frame(&frame_stream, msg_stream.buffered()) catch return;
|
|
std.fs.File.stdout().writeAll(frame_stream.buffered()) catch {};
|
|
}
|
|
};
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// main
|
|
// ---------------------------------------------------------------------------
|
|
|
|
pub fn main() !void {
|
|
var gpa: std.heap.GeneralPurposeAllocator(.{}) = .{};
|
|
defer _ = gpa.deinit();
|
|
const allocator = gpa.allocator();
|
|
|
|
var initial_env: ?tp.env = null;
|
|
if (std.posix.getenv("TRACE") != null) {
|
|
const f = try std.fs.cwd().createFile("remote_child_endpoint_trace.json", .{});
|
|
trace_file = f;
|
|
trace_file_writer = f.writer(&trace_buf);
|
|
var e = tp.env.init();
|
|
e.on_trace(&trace_handler);
|
|
e.enable_all_channels();
|
|
initial_env = e;
|
|
}
|
|
|
|
var ctx = try tp.context.init(allocator);
|
|
defer ctx.deinit();
|
|
|
|
var exit_ok = true;
|
|
var exit_handler = tp.make_exit_handler(&exit_ok, struct {
|
|
fn handle(ok: *bool, status: []const u8) void {
|
|
if (!std.mem.eql(u8, status, "stdin_closed") and
|
|
!std.mem.eql(u8, status, "normal"))
|
|
ok.* = false;
|
|
}
|
|
}.handle);
|
|
|
|
_ = try ctx.spawn_link(
|
|
StdioEndpoint.Args{ .allocator = allocator },
|
|
StdioEndpoint.start,
|
|
"stdio_endpoint",
|
|
&exit_handler,
|
|
if (initial_env) |*e| e else null,
|
|
);
|
|
|
|
ctx.run();
|
|
|
|
if (initial_env) |e| {
|
|
trace_file_writer.interface.flush() catch {};
|
|
trace_file.?.close();
|
|
trace_file = null;
|
|
e.deinit();
|
|
}
|
|
|
|
std.process.exit(if (exit_ok) 0 else 1);
|
|
}
|