シングルスレッド非同期 TCP エコーサーバ

main.zig

//! Zig 作者のコードを参考にした
//! https://gist.github.com/andrewrk/34c21bdc1600b0884a3ab9fa9aa485b8
//! https://gist.github.com/karlseguin/53bb8ebf945b20aa0b7472d9d30de801
//! 非同期版 tcp echo server

const std = @import("std");
const net = std.net;
const os = std.os;
const mem = std.mem;
const log = std.log;
const testing = std.testing;

const Queue = std.atomic.Queue;

pub const io_mode = .evented;

pub fn main() !void {
    var general_purpose_allocator = std.heap.GeneralPurposeAllocator(.{}){};
    var gpa = &general_purpose_allocator.allocator();

    const host = "127.0.0.1";
    const port = 5555;

    const address = std.net.Address.parseIp(host, port) catch unreachable;
    var listener = std.net.StreamServer.init(.{ .reuse_address = true });
    listener.listen(address) catch |err| {
        log.err("{}", .{err});
        std.process.exit(0);
    };
    log.debug("start listener: {s}:{d}", .{ host, port });

    var cleanup = &Queue(*Conn).init();
    _ = async cleaner(cleanup, gpa);

    while (true) {
        const c = try listener.accept();
        var conn = try gpa.create(Conn);
        log.debug("created: {*}", .{conn});
        conn.* = Conn{
            .stream = c.stream,
            .address = c.address,
            .handle_frame = async conn.handle(cleanup),
        };
        log.debug("accepted: {}", .{c.address});
    }
}

const Conn = struct {
    stream: net.Stream,
    address: net.Address,
    handle_frame: @Frame(handle),

    fn handle(self: *Conn, cleanup: *Queue(*Conn)) !void {
        const stream = self.stream;
        defer {
            stream.close();
            log.debug("closed: {}", .{self.address});
            var node = Queue(*Conn).Node{ .data = self };
            cleanup.put(&node);
        }

        while (true) {
            var buf: [100]u8 = undefined;
            const n = try stream.read(&buf);
            if (n == 0) {
                return;
            }
            const data = buf[0..n];
            log.debug("{s}", .{std.fmt.fmtSliceHexUpper(data)});
            _ = try stream.write(data);
        }
    }
};

fn cleaner(cleanup: *Queue(*Conn), gpa: *mem.Allocator) !void {
    while (true) {
        while (cleanup.get()) |node| {
            log.debug("destroyed: {*}", .{node.data});
            gpa.destroy(&node.data);
        }
        // 5 seconds (in nano seconds)
        std.time.sleep(5000000000);
    }
}