2 tcp connections between parent and child process

client.zig

//! client started by server as subprocess
const std = @import("std");
const builtin = @import("builtin");

const net = std.x.net;
const os = std.x.os;
const testing = std.testing;

const tcp = net.tcp;
const ip = net.ip;
const native_os = builtin.os;

const IPv4 = os.IPv4;
const IPv6 = os.IPv6;
const Socket = os.Socket;
const Buffer = os.Buffer;

// 1. server opens socket for client to connect
// 2. server --spawn child_process--> client with transmit ip address to connect to
// 3. client connects to socket
// 4. test stuff

pub fn main() !void {
    const allocator = testing.allocator;
    std.debug.print("client started\n", .{});
    //TODO no os-independent getPid() in std.process:

    var it = try std.process.argsWithAllocator(allocator);
    defer it.deinit(); // no-op unless WASI or Windows
    _ = it.next(); // ignore binary name
    const s_ip_ctrl = it.next().?;
    const s_port_ctrl = it.next().?;
    const s_ip_data = it.next().?;
    const s_port_data = it.next().?;
    try std.testing.expect(!it.skip());

    std.debug.print("{s}\n", .{s_ip_ctrl});
    std.debug.print("{s}\n", .{s_port_ctrl});
    std.debug.print("{s}\n", .{s_ip_data});
    std.debug.print("{s}\n", .{s_port_data});

    const client_ctrl = try tcp.Client.init(.ip, .{ .close_on_exec = true });
    defer client_ctrl.deinit();
    const ipv4_ctrl = try IPv4.parse(s_ip_ctrl);
    const port_ctrl = try std.fmt.parseUnsigned(u16, s_port_ctrl, 10);
    const addr_ctrl = ip.Address.initIPv4(ipv4_ctrl, port_ctrl);
    try client_ctrl.connect(addr_ctrl);

    const client_data = try tcp.Client.init(.ip, .{ .close_on_exec = true });
    defer client_data.deinit();
    const ipv4_data = try IPv4.parse(s_ip_data);
    const port_data = try std.fmt.parseUnsigned(u16, s_port_data, 10);
    const addr_data = ip.Address.initIPv4(ipv4_data, port_data);
    try client_data.connect(addr_data);

    const message = "hello world";
    _ = try client_ctrl.writeMessage(Socket.Message.fromBuffers(&[_]Buffer{
        Buffer.from(message[0 .. message.len / 2]),
        Buffer.from(message[message.len / 2 ..]),
    }), 0);

    //try client.shutdown(std.os.ShutdownHow.both);
    std.debug.print("client exited\n", .{});
}

server.zig

//! server starts client as subprocess
const std = @import("std");
const builtin = @import("builtin");

const native_os = builtin.os;
const net = std.x.net;
const os = std.x.os;
const testing = std.testing;
const ChildProcess = std.ChildProcess;

const tcp = net.tcp;
const ip = net.ip;
const IPv4 = os.IPv4;
const IPv6 = os.IPv6;
const Socket = os.Socket;
const Buffer = os.Buffer;

const have_ifnamesize = @hasDecl(std.os.system, "IFNAMESIZE"); // interface namesize

// 1. server opens socket for client to connect
// 2. server --spawn child_process--> client + transmit ip address to connect to
// 3. client connects to socket
// 4. test stuff

const TcpIo = struct {
    listen_ctrl: tcp.Listener,
    listen_data: tcp.Listener,
    addr_ctrl: ip.Address,
    addr_data: ip.Address,
    conn_ctrl: tcp.Connection,
    conn_data: tcp.Connection,
    // perf optimization: use comptime-computed offset array + continuous buffer
    // blocked by stage2, see #10920
    // max ip nr IPv4: 255.255.255.255 => 12 number digits + 3 dots => 15 digits
    s_ip_ctrl: [15]u8,
    s_ip_data: [15]u8,
    len_ip_ctrl: u4,
    len_ip_data: u4,
    // max port nr IPv4: unsigned 16-bit integer, so 65535 => 5 (number) digits
    // we only use 4 digits for 9084, 9085
    s_port_ctrl: [4]u8,
    s_port_data: [4]u8,
};

fn setupListener() !TcpIo {
    var tcpio = TcpIo{
        .listen_ctrl = try tcp.Listener.init(.ip, .{ .close_on_exec = true }),
        .listen_data = try tcp.Listener.init(.ip, .{ .close_on_exec = true }),
        .addr_ctrl = undefined,
        .addr_data = undefined,
        .s_ip_ctrl = undefined,
        .s_ip_data = undefined,
        .len_ip_ctrl = undefined,
        .len_ip_data = undefined,
        .s_port_ctrl = undefined,
        .s_port_data = undefined,
        .conn_ctrl = undefined,
        .conn_data = undefined,
    };

    // Zig Test (ascii ZT: 90, 84) => port 9084, 9085
    try tcpio.listen_ctrl.bind(ip.Address.initIPv4(IPv4.unspecified, 9084));
    try tcpio.listen_data.bind(ip.Address.initIPv4(IPv4.unspecified, 9085));
    try tcpio.listen_ctrl.listen(1);
    try tcpio.listen_data.listen(1);
    tcpio.addr_ctrl = try tcpio.listen_ctrl.getLocalAddress();
    tcpio.addr_data = try tcpio.listen_data.getLocalAddress();
    switch (tcpio.addr_ctrl) {
        .ipv4 => |*ipv4| ipv4.host = IPv4.localhost,
        .ipv6 => unreachable,
    }
    switch (tcpio.addr_data) {
        .ipv4 => |*ipv4| ipv4.host = IPv4.localhost,
        .ipv6 => unreachable,
    }
    std.debug.print("ip_ctrl: {s}\n", .{tcpio.addr_ctrl.ipv4.host});
    std.debug.print("port_ctrl: {d}\n", .{tcpio.addr_ctrl.ipv4.port});
    std.debug.print("ip_data: {s}\n", .{tcpio.addr_data.ipv4.host});
    std.debug.print("port_data: {d}\n", .{tcpio.addr_data.ipv4.port});
    const ip_ctrl = try std.fmt.bufPrint(tcpio.s_ip_ctrl[0..], "{s}", .{tcpio.addr_ctrl.ipv4.host});
    const ip_data = try std.fmt.bufPrint(tcpio.s_ip_data[0..], "{s}", .{tcpio.addr_data.ipv4.host});
    _ = try std.fmt.bufPrint(tcpio.s_port_ctrl[0..], "{d}", .{tcpio.addr_ctrl.ipv4.port}); // no padding
    _ = try std.fmt.bufPrint(tcpio.s_port_data[0..], "{d}", .{tcpio.addr_data.ipv4.port});
    //_ = std.fmt.formatIntBuf(tcpio.s_port_ctrl[0..], tcpio.addr_ctrl.ipv4.port, 10, std.fmt.Case.lower, .{ .fill = '0', .width = tcpio.s_port_ctrl.len },);
    //_ = std.fmt.formatIntBuf(tcpio.s_port_data[0..], tcpio.addr_ctrl.ipv4.port, 10, std.fmt.Case.lower, .{ .fill = '0', .width = tcpio.s_port_data.len },);

    tcpio.len_ip_ctrl = @intCast(u4, ip_ctrl.len);
    tcpio.len_ip_data = @intCast(u4, ip_data.len);
    // remaining undefined: .conn_ctrl .conn_data
    return tcpio;
}

// running on raw sockets is not possible without additional permissions
// https://squidarth.com/networking/systems/rc/2018/05/28/using-raw-sockets.html
// only alternative: tcp/udp
pub fn main() !void {
    if (!have_ifnamesize) return error.FAILURE;
    const allocator = testing.allocator;
    std.debug.print("server started\n", .{});

    var tcpio = try setupListener();
    _ = tcpio;
    _ = allocator;

    std.debug.print("{s}\n", .{tcpio.s_ip_ctrl}); // TODO must store accurate length+slice it
    std.debug.print("{s}\n", .{tcpio.s_ip_data});
    std.debug.print("{d}\n", .{tcpio.s_port_ctrl});
    std.debug.print("{d}\n", .{tcpio.s_port_data});

    const ipv4_ctrl = try IPv4.parse(tcpio.s_ip_ctrl[0..tcpio.len_ip_ctrl]);
    const ipv4_data = try IPv4.parse(tcpio.s_ip_data[0..tcpio.len_ip_ctrl]);
    const port_ctrl = try std.fmt.parseUnsigned(u16, &tcpio.s_port_ctrl, 10);
    const port_data = try std.fmt.parseUnsigned(u16, &tcpio.s_port_data, 10);
    std.debug.print("{s}\n", .{ipv4_ctrl});
    std.debug.print("{s}\n", .{ipv4_data});
    std.debug.print("{d}\n", .{port_ctrl});
    std.debug.print("{d}\n", .{port_data});

    const cwd_path = try std.process.getCwdAlloc(allocator);
    std.debug.print("cwd_path: {s}\n", .{cwd_path});
    // TODO absolute paths
    const args = [_][]const u8{
        "./client",
        tcpio.s_ip_ctrl[0..tcpio.len_ip_ctrl],
        tcpio.s_port_ctrl[0..],
        tcpio.s_ip_data[0..tcpio.len_ip_data],
        tcpio.s_port_data[0..],
    };
    var child_proc = try ChildProcess.init(&args, allocator);
    defer child_proc.deinit();
    try child_proc.spawn();
    std.debug.print("child_proc spawned:\n", .{});
    std.debug.print("args {s}\n", .{args});

    // connection list
    // server, client
    // -> stdin: start at x, ipctrl portctrl ipdata portdata
    // <- stdout
    // <- stderr
    // <- ctrl
    // <- data

    // TODO problem: std.testing.expect should output to port etc instead of File(stdout/stderr)
    // => how to fix this?

    tcpio.conn_ctrl = try tcpio.listen_ctrl.accept(.{ .close_on_exec = true }); // accept is blocking
    tcpio.conn_data = try tcpio.listen_data.accept(.{ .close_on_exec = true }); // accept is blocking
    defer tcpio.conn_ctrl.deinit();
    defer tcpio.conn_data.deinit();
    std.debug.print("connections succesful\n", .{});

    const message = "hello world";

    var buf: [message.len + 1]u8 = undefined;
    var msg = Socket.Message.fromBuffers(&[_]Buffer{
        Buffer.from(buf[0 .. message.len / 2]),
        Buffer.from(buf[message.len / 2 ..]),
    });
    _ = try tcpio.conn_ctrl.client.readMessage(&msg, 0);
    try testing.expectEqualStrings(message, buf[0..message.len]);
    std.debug.print("comparison successful\n", .{});

    // server wants to initiate shutdown before waiting on client
    // reason: client process may not finish => we do not get stdout/stderr and shutdown
    //how to detect unresponsive children for try tcpio.conn_ctrl.shutdown(); + killing?
    const ret_val = child_proc.wait();
    try testing.expectEqual(ret_val, .{ .Exited = 0 });
    std.debug.print("server exited\n", .{});
}