244 lines
11 KiB
Zig
244 lines
11 KiB
Zig
/// Parent-side endpoint actor.
|
|
///
|
|
/// Wraps a subprocess, accumulates framed CBOR from its stdout, and dispatches
|
|
/// decoded messages to local actors or proxies.
|
|
///
|
|
/// Outbound messages (local → endpoint):
|
|
/// {"send", from_id: u64, to_id: u64, payload} — via proxy, by remote ID
|
|
/// {"send", from_id: u64, to_name: text, payload} — direct by name
|
|
/// {"proxy_exit", remote_id: u64, reason: text} — proxy notifying of exit
|
|
const std = @import("std");
|
|
const tp = @import("thespian");
|
|
const cbor = @import("cbor");
|
|
const framing = @import("framing");
|
|
const protocol = @import("protocol");
|
|
const proxy = @import("proxy");
|
|
|
|
const subprocess = tp.subprocess;
|
|
|
|
const proc_tag = "endpoint_proc";
|
|
|
|
pub const Args = struct {
|
|
allocator: std.mem.Allocator,
|
|
/// CBOR-encoded argv array for the child binary.
|
|
/// Must be heap-allocated; endpoint.init will free it.
|
|
argv: []const u8,
|
|
};
|
|
|
|
const Endpoint = struct {
|
|
allocator: std.mem.Allocator,
|
|
proc: subprocess,
|
|
accumulator: framing.Accumulator,
|
|
/// Inbound proxy table: remote actor ID → local proxy pid.
|
|
proxies: std.AutoHashMap(u64, tp.pid),
|
|
/// Outbound table: local actor handle pointer → assigned wire ID.
|
|
outbound: std.AutoHashMap(usize, u64),
|
|
/// Local actor table: wire ID → local actor pid (reverse of outbound).
|
|
local_actors: std.AutoHashMap(u64, tp.pid),
|
|
next_id: u64,
|
|
receiver: tp.Receiver(*@This()),
|
|
|
|
fn start(args: Args) tp.result {
|
|
return init(args) catch |e| return tp.exit_error(e, @errorReturnTrace());
|
|
}
|
|
|
|
fn init(args: Args) !void {
|
|
defer args.allocator.free(args.argv);
|
|
const proc = try subprocess.init(args.allocator, tp.message{ .buf = args.argv }, proc_tag, .Pipe);
|
|
const self = try args.allocator.create(@This());
|
|
self.* = .{
|
|
.allocator = args.allocator,
|
|
.proc = proc,
|
|
.accumulator = .{},
|
|
.proxies = std.AutoHashMap(u64, tp.pid).init(args.allocator),
|
|
.outbound = std.AutoHashMap(usize, u64).init(args.allocator),
|
|
.local_actors = std.AutoHashMap(u64, tp.pid).init(args.allocator),
|
|
.next_id = 1,
|
|
.receiver = .init(receive_fn, deinit, self),
|
|
};
|
|
errdefer self.deinit();
|
|
tp.receive(&self.receiver);
|
|
}
|
|
|
|
fn deinit(self: *@This()) void {
|
|
// Shut down all live proxies. Use "endpoint_exit" rather than "exit"
|
|
// so the proxy can distinguish this deinit-time message (which has no
|
|
// live sender context, i.e. from.instance_id()==0) from a Thespian
|
|
// trapped-exit sent by a linked local actor.
|
|
var proxy_it = self.proxies.valueIterator();
|
|
while (proxy_it.next()) |p| {
|
|
p.send(.{ "endpoint_exit", "transport_closed" }) catch {};
|
|
p.deinit();
|
|
}
|
|
self.proxies.deinit();
|
|
var actor_it = self.local_actors.valueIterator();
|
|
while (actor_it.next()) |p| p.deinit();
|
|
self.local_actors.deinit();
|
|
self.outbound.deinit();
|
|
self.proc.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(), _: tp.pid_ref, m: tp.message) !void {
|
|
var bytes: []const u8 = "";
|
|
var from_id: u64 = 0;
|
|
var to_id: u64 = 0;
|
|
var to_name: []const u8 = "";
|
|
var payload: []const u8 = "";
|
|
var remote_id: u64 = 0;
|
|
var reason: []const u8 = "";
|
|
|
|
if (try m.match(.{ proc_tag, "stdout", tp.extract(&bytes) })) {
|
|
if (self.accumulator.feed(bytes)) |frame| try self.dispatch_inbound(frame);
|
|
} else if (try m.match(.{ proc_tag, "stderr", tp.any })) {
|
|
} else if (try m.match(.{ "send", tp.extract(&from_id), tp.extract(&to_id), cbor.extract_cbor(&payload) })) {
|
|
// Outbound send by remote ID — from a local proxy.
|
|
// from_id carries the sender's handle pointer; resolve to wire ID.
|
|
const wire_from_id = try self.get_or_assign_outbound_id(from_id);
|
|
try self.send_wire_by_id(wire_from_id, to_id, payload);
|
|
} else if (try m.match(.{ "send", tp.extract(&from_id), tp.extract(&to_name), cbor.extract_cbor(&payload) })) {
|
|
// Outbound send_named — from a local actor addressing a well-known remote actor by name.
|
|
try self.send_wire_named(from_id, to_name, payload);
|
|
} else if (try m.match(.{ "link_wire", tp.extract(&from_id), tp.extract(&to_id) })) {
|
|
// A local proxy is establishing a remote link: from_id=local handle pointer,
|
|
// to_id=remote_id of the target on the remote system.
|
|
const wire_from_id = try self.get_or_assign_outbound_id(from_id);
|
|
try self.send_wire_link(wire_from_id, to_id);
|
|
} else if (try m.match(.{ "local_link_exit", tp.extract(&from_id), tp.extract(&reason) })) {
|
|
// A local actor that a proxy was Thespian-linked to has exited.
|
|
// from_id is the actor's handle pointer; look up its wire ID and
|
|
// forward an exit wire message so the remote side can propagate it.
|
|
const key: usize = @intCast(from_id);
|
|
if (self.outbound.get(key)) |wire_id|
|
|
try self.send_wire_exit(wire_id, reason);
|
|
// If not in the outbound table the actor never sent outbound; ignore.
|
|
} else if (try m.match(.{ "proxy_exit", tp.extract(&remote_id), tp.any })) {
|
|
// A local proxy has exited; remove it from the proxy table.
|
|
if (self.proxies.fetchRemove(remote_id)) |entry| entry.value.deinit();
|
|
} else if (try m.match(.{ proc_tag, "term", tp.any, tp.any })) {
|
|
return tp.exit("transport_closed");
|
|
} else {
|
|
return tp.unexpected(m);
|
|
}
|
|
}
|
|
|
|
/// Look up or assign a stable wire ID for a local actor identified by its
|
|
/// handle pointer. Also records the pid so inbound sends can be routed back.
|
|
fn get_or_assign_outbound_id(self: *@This(), handle_int: u64) !u64 {
|
|
const key: usize = @intCast(handle_int);
|
|
if (self.outbound.get(key)) |id| return id;
|
|
const id = self.next_id;
|
|
self.next_id += 1;
|
|
// Reconstruct a pid_ref from the handle pointer and clone it for storage.
|
|
const pid_ref: tp.pid_ref = .{ .h = @ptrFromInt(key) };
|
|
try self.local_actors.put(id, pid_ref.clone());
|
|
try self.outbound.put(key, id);
|
|
return id;
|
|
}
|
|
|
|
/// Get the local proxy for remote_id, creating one if it doesn't exist yet.
|
|
fn get_or_create_proxy(self: *@This(), remote_id: u64) !tp.pid_ref {
|
|
if (self.proxies.getPtr(remote_id)) |p| return p.ref();
|
|
const p = try tp.spawn(self.allocator, proxy.Args{
|
|
.allocator = self.allocator,
|
|
.endpoint = tp.self_pid().clone(),
|
|
.remote_id = remote_id,
|
|
}, proxy.start, "proxy");
|
|
try self.proxies.put(remote_id, p);
|
|
return self.proxies.getPtr(remote_id).?.ref();
|
|
}
|
|
|
|
fn dispatch_inbound(self: *@This(), frame: []const u8) !void {
|
|
const msg = try protocol.decode(frame);
|
|
switch (msg) {
|
|
.send => |s| {
|
|
// Ensure a local proxy exists for the sender.
|
|
const prx = try self.get_or_create_proxy(s.from_id);
|
|
// Route to the local actor that owns this wire ID, delivering
|
|
// FROM the proxy so the recipient can reply naturally.
|
|
if (self.local_actors.getPtr(s.to_id)) |actor_ptr|
|
|
try prx.send(.{ "deliver_pid", @as(u64, @intFromPtr(actor_ptr.h)), protocol.RawCbor{ .bytes = s.payload } })
|
|
else
|
|
return tp.exit_error(error.UnknownLocalActor, null);
|
|
},
|
|
.send_named => |s| {
|
|
if (s.from_id != 0) {
|
|
// Deliver FROM the proxy so the recipient sees the proxy as
|
|
// `from` and can reply by sending back through it.
|
|
const prx = try self.get_or_create_proxy(s.from_id);
|
|
try prx.send(.{ "deliver_named", s.to_name, protocol.RawCbor{ .bytes = s.payload } });
|
|
} else {
|
|
// No sender identity; deliver directly without from substitution.
|
|
const actor = tp.env.get().proc(s.to_name);
|
|
try actor.send_raw(tp.message{ .buf = s.payload });
|
|
}
|
|
},
|
|
.exit => |e| {
|
|
// Remote actor has exited; notify its local proxy.
|
|
if (self.proxies.get(e.id)) |p|
|
|
p.send(.{ "exit", e.reason }) catch {};
|
|
},
|
|
.link => |lnk| {
|
|
// Remote side is establishing a link: lnk.local_id is the remote
|
|
// actor's wire ID (= our proxy key), lnk.remote_id is the local
|
|
// actor's wire ID in our local_actors table.
|
|
const prx = try self.get_or_create_proxy(lnk.local_id);
|
|
if (self.local_actors.getPtr(lnk.remote_id)) |actor_ptr|
|
|
try prx.send(.{ "set_notify", @as(u64, @intFromPtr(actor_ptr.h)) })
|
|
else
|
|
return tp.exit_error(error.UnknownLocalActor, null);
|
|
},
|
|
.transport_error => |te| return tp.exit(te.reason),
|
|
}
|
|
}
|
|
|
|
fn send_wire_by_id(self: *@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 self.proc.send(frame_stream.buffered());
|
|
}
|
|
|
|
fn send_wire_link(self: *@This(), local_id: u64, remote_id: u64) !void {
|
|
var msg_buf: [framing.max_frame_size]u8 = undefined;
|
|
var msg_stream: std.Io.Writer = .fixed(&msg_buf);
|
|
try protocol.encode_link(&msg_stream, local_id, remote_id);
|
|
|
|
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 self.proc.send(frame_stream.buffered());
|
|
}
|
|
|
|
fn send_wire_exit(self: *@This(), id: u64, reason: []const u8) !void {
|
|
var msg_buf: [framing.max_frame_size]u8 = undefined;
|
|
var msg_stream: std.Io.Writer = .fixed(&msg_buf);
|
|
try protocol.encode_exit(&msg_stream, id, reason);
|
|
|
|
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 self.proc.send(frame_stream.buffered());
|
|
}
|
|
|
|
fn send_wire_named(self: *@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 self.proc.send(frame_stream.buffered());
|
|
}
|
|
};
|
|
|
|
pub const start = Endpoint.start;
|