add cancel support to EvLoop.Fd

Author: zfl9

build.zig

@@ -701,7 +701,7 @@ fn setup_libexeobj_step(step: *LibExeObjStep) void {
 
     // compile
     if (step.kind == .obj)
-        step.use_stage1 = true; // required by async/await (.zig)
+        step.use_stage1 = true; // required by coroutine (.zig)
 
     step.single_threaded = true;
 

src/CacheMsg.zig

@@ -5,15 +5,15 @@ const cc = @import("cc.zig");
 const dns = @import("dns.zig");
 const log = @import("log.zig");
 const Rc = @import("Rc.zig");
-const ListNode = @import("ListNode.zig");
+const Node = @import("Node.zig");
 const Bytes = cc.Bytes;
 
 // =======================================================
 
 const CacheMsg = @This();
 
 next: ?*CacheMsg = null, // for hashmap
-list_node: ListNode = undefined,
+list_node: Node = undefined,
 update_time: c.time_t,
 hashv: c_uint,
 ttl: i32,
@@ -73,7 +73,7 @@ pub fn free(self: *CacheMsg) void {
         g.allocator.free(self.mem());
 }
 
-pub fn from_list_node(node: *ListNode) *CacheMsg {
+pub fn from_list_node(node: *Node) *CacheMsg {
     return @fieldParentPtr(CacheMsg, "list_node", node);
 }
 

src/EvLoop.zig

@@ -96,6 +96,7 @@ pub const Fd = struct {
     write_frame: ?anyframe = null, // waiting for writable event
     fd: c_int,
     rc: Rc = .{},
+    canceled: bool = false,
 
     /// ownership of `fd` is transferred to `fdobj`
     pub fn new(fd: c_int) *Fd {
@@ -139,6 +140,31 @@ pub const Fd = struct {
 
         return events;
     }
+
+    pub fn cancel(self: *Fd) void {
+        if (self.canceled)
+            return;
+
+        self.canceled = true;
+        // cc.set_errno(c.ECANCELED);
+
+        if (self.read_frame) |frame|
+            co.do_resume(frame);
+        if (self.write_frame) |frame|
+            co.do_resume(frame);
+
+        assert(self.read_frame == null);
+        assert(self.write_frame == null);
+    }
+
+    /// return `true` if canceled and set errno to `ECANCELED`
+    pub fn is_canceled(self: *const Fd) bool {
+        if (self.canceled) {
+            cc.set_errno(c.ECANCELED);
+            return true;
+        }
+        return false;
+    }
 };
 
 // =============================================================
@@ -256,6 +282,7 @@ fn del(self: *EvLoop, fdobj: *const Fd) bool {
 // ======================================================================
 
 fn set_frame(fdobj: *Fd, comptime field_name: []const u8, frame: anyframe) void {
+    assert(!fdobj.canceled);
     assert(@field(fdobj, field_name) == null);
     @field(fdobj, field_name) = frame;
 }
@@ -401,26 +428,34 @@ pub fn run(self: *EvLoop) void {
 
 // ========================================================================
 
-// socket API (non-blocking + async)
+// socket API (non-blocking + coroutine)
 
 comptime {
     assert(c.EAGAIN == c.EWOULDBLOCK);
 }
 
-/// used for external modules, not for this module:
-/// because the async-call chains consume at least 24 bytes per level (x86_64)
-pub fn wait_readable(self: *EvLoop, fdobj: *Fd) void {
+/// return `null` if fdobj is canceled. \
+/// used for external modules, not for this module: \
+/// because the coroutine chains consume at least 24 bytes per level (x86_64).
+pub fn wait_readable(self: *EvLoop, fdobj: *Fd) ?void {
     self.add_readable(fdobj, @frame());
     suspend {}
     self.del_readable(fdobj, @frame());
+
+    if (fdobj.is_canceled())
+        return null;
 }
 
-/// used for external modules, not for this module:
-/// because the async-call chains consume at least 24 bytes per level (x86_64)
-pub fn wait_writable(self: *EvLoop, fdobj: *Fd) void {
+/// return `null` if fdobj is canceled. \
+/// used for external modules, not for this module: \
+/// because the coroutine chains consume at least 24 bytes per level (x86_64).
+pub fn wait_writable(self: *EvLoop, fdobj: *Fd) ?void {
     self.add_writable(fdobj, @frame());
     suspend {}
     self.del_writable(fdobj, @frame());
+
+    if (fdobj.is_canceled())
+        return null;
 }
 
 pub fn connect(self: *EvLoop, fdobj: *Fd, addr: *const cc.SockAddr) ?void {
@@ -432,6 +467,9 @@ pub fn connect(self: *EvLoop, fdobj: *Fd, addr: *const cc.SockAddr) ?void {
         suspend {}
         self.del_writable(fdobj, @frame());
 
+        if (fdobj.is_canceled())
+            return null;
+
         if (net.getsockopt_int(fdobj.fd, c.SOL_SOCKET, c.SO_ERROR, "SO_ERROR")) |err| {
             if (err == 0) return;
             cc.set_errno(err);
@@ -444,7 +482,7 @@ pub fn connect(self: *EvLoop, fdobj: *Fd, addr: *const cc.SockAddr) ?void {
 }
 
 pub fn accept(self: *EvLoop, fdobj: *Fd, src_addr: ?*cc.SockAddr) ?c_int {
-    while (true) {
+    while (!fdobj.is_canceled()) {
         return cc.accept4(fdobj.fd, src_addr, c.SOCK_NONBLOCK | c.SOCK_CLOEXEC) orelse {
             if (cc.errno() != c.EAGAIN)
                 return null;
@@ -455,11 +493,11 @@ pub fn accept(self: *EvLoop, fdobj: *Fd, src_addr: ?*cc.SockAddr) ?c_int {
 
             continue;
         };
-    }
+    } else return null;
 }
 
 pub fn read(self: *EvLoop, fdobj: *Fd, buf: []u8) ?usize {
-    while (true) {
+    while (!fdobj.is_canceled()) {
         return cc.read(fdobj.fd, buf) orelse {
             if (cc.errno() != c.EAGAIN)
                 return null;
@@ -470,11 +508,11 @@ pub fn read(self: *EvLoop, fdobj: *Fd, buf: []u8) ?usize {
 
             continue;
         };
-    }
+    } else return null;
 }
 
 pub fn recvfrom(self: *EvLoop, fdobj: *Fd, buf: []u8, flags: c_int, src_addr: *cc.SockAddr) ?usize {
-    while (true) {
+    while (!fdobj.is_canceled()) {
         return cc.recvfrom(fdobj.fd, buf, flags, src_addr) orelse {
             if (cc.errno() != c.EAGAIN)
                 return null;
@@ -485,11 +523,11 @@ pub fn recvfrom(self: *EvLoop, fdobj: *Fd, buf: []u8, flags: c_int, src_addr: *c
 
             continue;
         };
-    }
+    } else return null;
 }
 
 pub fn recv(self: *EvLoop, fdobj: *Fd, buf: []u8, flags: c_int) ?usize {
-    while (true) {
+    while (!fdobj.is_canceled()) {
         return cc.recv(fdobj.fd, buf, flags) orelse {
             if (cc.errno() != c.EAGAIN)
                 return null;
@@ -500,16 +538,18 @@ pub fn recv(self: *EvLoop, fdobj: *Fd, buf: []u8, flags: c_int) ?usize {
 
             continue;
         };
-    }
+    } else return null;
 }
 
-const ReadErr = error{ eof, other };
+const ReadErr = error{ eof, errno };
 
-pub fn recv_exactly(self: *EvLoop, fdobj: *Fd, buf: []u8, flags: c_int) ReadErr!void {
+pub fn recv_full(self: *EvLoop, fdobj: *Fd, buf: []u8, flags: c_int) ReadErr!void {
     var nread: usize = 0;
     while (nread < buf.len) {
+        if (fdobj.is_canceled())
+            return ReadErr.errno; // ECANCELED
         const n = self.recv(fdobj, buf[nread..], flags) orelse
-            return ReadErr.other;
+            return ReadErr.errno;
         if (n == 0)
             return ReadErr.eof;
         nread += n;
@@ -525,6 +565,8 @@ pub fn recv_exactly(self: *EvLoop, fdobj: *Fd, buf: []u8, flags: c_int) ReadErr!
 pub fn send(self: *EvLoop, fdobj: *Fd, data: []const u8, flags: c_int) ?void {
     var nsend: usize = 0;
     while (nsend < data.len) {
+        if (fdobj.is_canceled())
+            return null;
         const n = cc.send(fdobj.fd, data[nsend..], flags) orelse b: {
             if (cc.errno() != c.EAGAIN)
                 return null;
@@ -544,6 +586,8 @@ pub fn send(self: *EvLoop, fdobj: *Fd, data: []const u8, flags: c_int) ?void {
 pub fn sendmsg(self: *EvLoop, fdobj: *Fd, msg: *const cc.msghdr_t, flags: c_int) ?void {
     var remain_len: usize = msg.calc_len();
     while (remain_len > 0) {
+        if (fdobj.is_canceled())
+            return null;
         const n = cc.sendmsg(fdobj.fd, msg, flags) orelse b: {
             if (cc.errno() != c.EAGAIN)
                 return null;

src/ListNode.zig src/Node.zig

@@ -5,55 +5,55 @@ const assert = std.debug.assert;
 
 // =====================================================
 
-const ListNode = @This();
+const Node = @This();
 
-prev: *ListNode,
-next: *ListNode,
+prev: *Node,
+next: *Node,
 
 // =================== `list_head(sentinel)` ===================
 
 /// empty list (sentinel node)
-pub fn init(list: *ListNode) void {
+pub fn init(list: *Node) void {
     list.prev = list;
     list.next = list;
 }
 
 /// first node
-pub inline fn head(list: *const ListNode) *ListNode {
+pub inline fn head(list: *const Node) *Node {
     return list.next;
 }
 
 /// last node
-pub inline fn tail(list: *const ListNode) *ListNode {
+pub inline fn tail(list: *const Node) *Node {
     return list.prev;
 }
 
 /// is sentinel node
-pub inline fn is_empty(list: *const ListNode) bool {
+pub inline fn is_empty(list: *const Node) bool {
     return list.head() == list;
 }
 
 /// `unlink(node)` and/or `free(node)` is safe
-pub fn iterator(list: *const ListNode) Iterator {
+pub fn iterator(list: *const Node) Iterator {
     return .{
         .sentinel = list,
         .node = list.head(),
     };
 }
 
 /// `unlink(node)` and/or `free(node)` is safe
-pub fn reverse_iterator(list: *const ListNode) ReverseIterator {
+pub fn reverse_iterator(list: *const Node) ReverseIterator {
     return .{
         .sentinel = list,
         .node = list.tail(),
     };
 }
 
 pub const Iterator = struct {
-    sentinel: *const ListNode,
-    node: *ListNode,
+    sentinel: *const Node,
+    node: *Node,
 
-    pub fn next(it: *Iterator) ?*ListNode {
+    pub fn next(it: *Iterator) ?*Node {
         const node = it.node;
         if (node != it.sentinel) {
             it.node = node.next;
@@ -64,10 +64,10 @@ pub const Iterator = struct {
 };
 
 pub const ReverseIterator = struct {
-    sentinel: *const ListNode,
-    node: *ListNode,
+    sentinel: *const Node,
+    node: *Node,
 
-    pub fn next(it: *ReverseIterator) ?*ListNode {
+    pub fn next(it: *ReverseIterator) ?*Node {
         const node = it.node;
         if (node != it.sentinel) {
             it.node = node.prev;
@@ -79,31 +79,31 @@ pub const ReverseIterator = struct {
 
 // =================== `node` ===================
 
-pub fn link_to_head(list: *ListNode, node: *ListNode) void {
+pub fn link_to_head(list: *Node, node: *Node) void {
     return node.link(list, list.head());
 }
 
-pub fn link_to_tail(list: *ListNode, node: *ListNode) void {
+pub fn link_to_tail(list: *Node, node: *Node) void {
     return node.link(list.tail(), list);
 }
 
 /// assume that the `node` is linked to the `list`
-pub fn move_to_head(list: *ListNode, node: *ListNode) void {
+pub fn move_to_head(list: *Node, node: *Node) void {
     if (node != list.head()) {
         node.unlink();
         list.link_to_head(node);
     }
 }
 
 /// assume that the `node` is linked to the `list`
-pub fn move_to_tail(list: *ListNode, node: *ListNode) void {
+pub fn move_to_tail(list: *Node, node: *Node) void {
     if (node != list.tail()) {
         node.unlink();
         list.link_to_tail(node);
     }
 }
 
-fn link(node: *ListNode, prev: *ListNode, next: *ListNode) void {
+fn link(node: *Node, prev: *Node, next: *Node) void {
     prev.next = node;
     node.prev = prev;
     node.next = next;
@@ -112,7 +112,7 @@ fn link(node: *ListNode, prev: *ListNode, next: *ListNode) void {
 
 /// `node.prev` and `node.next` are unmodified, use `node.init()` if needed.
 /// `list_head.unlink()` is not allowed unless `list_head` is an empty list.
-pub fn unlink(node: *const ListNode) void {
+pub fn unlink(node: *const Node) void {
     node.prev.next = node.next;
     node.next.prev = node.prev;
 }
@@ -121,15 +121,15 @@ pub fn unlink(node: *const ListNode) void {
 
 const Object = struct {
     id: u32,
-    node: ListNode,
+    node: Node,
 
-    pub fn from_node(node: *ListNode) *Object {
+    pub fn from_node(node: *Node) *Object {
         return @fieldParentPtr(Object, "node", node);
     }
 };
 
 pub fn @"test: linked list"() !void {
-    var list: ListNode = undefined;
+    var list: Node = undefined;
     list.init();
 
     defer {
@@ -195,7 +195,7 @@ pub fn @"test: linked list"() !void {
     }
 
     // link_to_head
-    var l: ListNode = undefined;
+    var l: Node = undefined;
     l.init();
 
     defer {

src/Upstream.zig

@@ -13,6 +13,7 @@ const Tag = @import("tag.zig").Tag;
 const DynStr = @import("DynStr.zig");
 const EvLoop = @import("EvLoop.zig");
 const RcMsg = @import("RcMsg.zig");
+const Node = @import("Node.zig");
 const flags_op = @import("flags_op.zig");
 const assert = std.debug.assert;
 
@@ -29,7 +30,8 @@ comptime {
 const Upstream = @This();
 
 // runtime info
-ctx: ?*anyopaque = null,
+data: ?*anyopaque = null,
+data2: ?*anyopaque = null, // ssl session for resumption
 
 // config info
 host: ?cc.ConstStr, // DoT SNI
@@ -70,7 +72,7 @@ fn init(tag: Tag, proto: Proto, addr: *const cc.SockAddr, host: []const u8, ip:
 }
 
 fn deinit(self: *const Upstream) void {
-    assert(self.ctx == null);
+    assert(self.data == null and self.data2 == null);
 
     if (self.host) |host|
         g.allocator.free(cc.strslice_c(host));
@@ -100,20 +102,38 @@ fn send(self: *Upstream, qmsg: *RcMsg) void {
 
 // ======================================================
 
+const TypedNode = struct {
+    type: enum { udp, tcp }, // `struct UDP` or `struct TCP`
+    node: Node,
+
+    pub fn from_node(node: *Node) *TypedNode {
+        return @fieldParentPtr(TypedNode, "node", node);
+    }
+};
+
+const UDP = struct {
+    typed_node: TypedNode,
+    fdobj: *EvLoop.Fd,
+    create_time: c.time_t,
+    query_rtime: u16, // relative to create_time
+    query_count: u16,
+    reply_count: u16,
+};
+
 fn udp_get_fdobj(self: *const Upstream) ?*EvLoop.Fd {
     assert(self.proto == .udpi or self.proto == .udp);
-    return cc.ptrcast(?*EvLoop.Fd, self.ctx);
+    return cc.ptrcast(?*EvLoop.Fd, self.data);
 }
 
 fn udp_set_fdobj(self: *Upstream, fdobj: ?*EvLoop.Fd) void {
     assert(self.proto == .udpi or self.proto == .udp);
-    self.ctx = fdobj;
+    self.data = fdobj;
 }
 
 fn udp_send(self: *Upstream, qmsg: *RcMsg) void {
     const fd = if (self.udp_get_fdobj()) |fdobj| fdobj.fd else b: {
         const fd = net.new_sock(self.addr.family(), .udp) orelse return;
-        co.create(udp_recv, .{ self, fd });
+        co.start(udp_recv, .{ self, fd });
         assert(self.udp_get_fdobj() != null);
         break :b fd;
     };
@@ -179,7 +199,7 @@ fn udp_recv(self: *Upstream, fd: c_int) void {
                     log.warn(@src(), "recv(%s) failed: (%d) %m", .{ self.url, cc.errno() });
                     return;
                 }
-                g.evloop.wait_readable(fdobj);
+                g.evloop.wait_readable(fdobj) orelse return;
                 continue;
             };
         } else return;
@@ -203,7 +223,8 @@ fn udp_on_eol(self: *Upstream) void {
     if (fdobj.read_frame) |frame| {
         co.do_resume(frame);
     } else {
-        // this coroutine may be sending a response to the tcp client (suspended)
+        // this coroutine may be sending a response to the tcp client (suspending)
+        // TODO: change to nosuspend send_reply
     }
 }
 
@@ -279,26 +300,26 @@ pub const TLS = struct {
 
 const TCP = struct {
     upstream: *const Upstream,
-    fdobj: ?*EvLoop.Fd = null,
+    fdobj: ?*EvLoop.Fd = null, // tcp connection
+    tls: TLS_ = .{}, // tls connection (DoT)
     send_list: MsgQueue = .{}, // qmsg to be sent
     ack_list: std.AutoHashMapUnmanaged(u16, *RcMsg) = .{}, // qmsg to be ack
     pending_n: u16 = 0, // outstanding queries: send_list + ack_list
     healthy: bool = false, // current connection processed at least one query ?
-    tls: TLS_ = .{}, // for DoT upstream
 
     const TLS_ = if (has_tls) TLS else struct {};
 
     /// must <= u16_max
     const PENDING_MAX = std.math.maxInt(u16);
 
     const MsgQueue = struct {
-        head: ?*Node = null,
-        tail: ?*Node = null,
+        head: ?*MsgNode = null,
+        tail: ?*MsgNode = null,
         waiter: ?anyframe = null,
 
-        const Node = struct {
+        const MsgNode = struct {
             msg: *RcMsg,
-            next: *Node,
+            next: *MsgNode,
         };
 
         /// `null`: cancel wait
@@ -312,7 +333,7 @@ const TCP = struct {
                 return;
             }
 
-            const node = g.allocator.create(Node) catch unreachable;
+            const node = g.allocator.create(MsgNode) catch unreachable;
             node.* = .{
                 .msg = msg,
                 .next = undefined,
@@ -342,7 +363,7 @@ const TCP = struct {
         }
 
         /// `null`: cancel wait
-        pub fn pop(self: *MsgQueue, blocking: bool) ?*RcMsg {
+        pub fn pop(self: *MsgQueue, suspending: bool) ?*RcMsg {
             if (self.head) |node| {
                 defer g.allocator.destroy(node);
                 if (node == self.tail) {
@@ -354,7 +375,7 @@ const TCP = struct {
                 }
                 return node.msg;
             } else {
-                if (!blocking)
+                if (!suspending)
                     return null;
                 self.waiter = @frame();
                 suspend {}
@@ -404,8 +425,7 @@ const TCP = struct {
             self.start();
     }
 
-    /// [async] used to send qmsg to upstream
-    /// pop from send_list && add to ack_list
+    /// [suspending] pop from send_list && add to ack_list
     fn pop_qmsg(self: *TCP, fdobj: *EvLoop.Fd) ?*RcMsg {
         if (!self.fdobj_ok(fdobj)) return null;
 
@@ -477,7 +497,7 @@ const TCP = struct {
         assert(self.ack_list.count() == 0);
 
         self.healthy = false;
-        co.create(TCP.send, .{self});
+        co.start(TCP.send, .{self});
     }
 
     fn send(self: *TCP) void {
@@ -494,7 +514,7 @@ const TCP = struct {
 
         self.do_connect(fdobj) orelse return;
 
-        co.create(recv, .{self});
+        co.start(recv, .{self});
 
         while (self.pop_qmsg(fdobj)) |qmsg| {
             self.do_send(fdobj, qmsg) orelse return;
@@ -574,11 +594,11 @@ const TCP = struct {
                     var err: c_int = undefined;
                     cc.SSL_connect(self.ssl(), &err) orelse switch (err) {
                         c.WOLFSSL_ERROR_WANT_READ => {
-                            g.evloop.wait_readable(fdobj);
+                            g.evloop.wait_readable(fdobj) orelse return null;
                             continue;
                         },
                         c.WOLFSSL_ERROR_WANT_WRITE => {
-                            g.evloop.wait_writable(fdobj);
+                            g.evloop.wait_writable(fdobj) orelse return null;
                             continue;
                         },
                         else => {
@@ -624,7 +644,7 @@ const TCP = struct {
                     .msg_iov = &iov,
                     .msg_iovlen = iov.len,
                 };
-                g.evloop.sendmsg(fdobj, &msg, 0) orelse break :e null; // async
+                g.evloop.sendmsg(fdobj, &msg, 0) orelse break :e null;
             } else if (has_tls) {
                 // merge into one ssl record
                 var buf: [2 + c.DNS_QMSG_MAXSIZE]u8 align(2) = undefined;
@@ -638,7 +658,7 @@ const TCP = struct {
                     var err: c_int = undefined;
                     cc.SSL_write(self.ssl(), data, &err) orelse switch (err) {
                         c.WOLFSSL_ERROR_WANT_WRITE => {
-                            g.evloop.wait_writable(fdobj); // async
+                            g.evloop.wait_writable(fdobj) orelse return null;
                             continue;
                         },
                         else => {
@@ -661,9 +681,9 @@ const TCP = struct {
             if (!self.fdobj_ok(fdobj)) return null;
 
             if (self.upstream.proto != .tls) {
-                g.evloop.recv_exactly(fdobj, buf, flags) catch |err| switch (err) {
+                g.evloop.recv_full(fdobj, buf, flags) catch |err| switch (err) {
                     error.eof => return null,
-                    error.other => break :e null,
+                    error.errno => break :e null,
                 };
             } else if (has_tls) {
                 var nread: usize = 0;
@@ -676,7 +696,7 @@ const TCP = struct {
                             return null;
                         },
                         c.WOLFSSL_ERROR_WANT_READ => {
-                            g.evloop.wait_readable(fdobj); // async
+                            g.evloop.wait_readable(fdobj) orelse return null;
                             continue;
                         },
                         else => {
@@ -696,9 +716,9 @@ const TCP = struct {
 
 fn tcp_ctx(self: *Upstream) *TCP {
     assert(self.proto == .tcpi or self.proto == .tcp or self.proto == .tls);
-    if (self.ctx == null)
-        self.ctx = TCP.new(self);
-    return cc.ptrcast(*TCP, self.ctx.?);
+    if (self.data == null)
+        self.data = TCP.new(self);
+    return cc.ptrcast(*TCP, self.data.?);
 }
 
 fn tcp_send(self: *Upstream, qmsg: *RcMsg) void {

src/cache.zig

@@ -3,15 +3,15 @@ const g = @import("g.zig");
 const c = @import("c.zig");
 const cc = @import("cc.zig");
 const dns = @import("dns.zig");
-const ListNode = @import("ListNode.zig");
+const Node = @import("Node.zig");
 const CacheMsg = @import("CacheMsg.zig");
 const cache_ignore = @import("cache_ignore.zig");
 const log = @import("log.zig");
 const assert = std.debug.assert;
 const Bytes = cc.Bytes;
 
 /// LRU
-var _list: ListNode = undefined;
+var _list: Node = undefined;
 
 pub fn module_init() void {
     _list.init();

src/co.zig

@@ -4,14 +4,13 @@ const g = @import("g.zig");
 const assert = std.debug.assert;
 
 /// create and start a new coroutine
-pub fn create(comptime func: anytype, args: anytype) void {
+pub fn start(comptime func: anytype, args: anytype) void {
     const buf = g.allocator.alignedAlloc(u8, std.Target.stack_align, @frameSize(func)) catch unreachable;
     _ = @asyncCall(buf, {}, func, args);
-    // @call(.{ .modifier = .async_kw, .stack = buf }, func, args);
     check_terminated();
 }
 
-/// if the coroutine is at the last pause point, its memory will be freed after resume
+/// if the coroutine is at the last suspend point, its memory will be freed after resume
 pub fn do_resume(frame: anyframe) void {
     resume frame;
     check_terminated();

src/dns.c

@@ -481,8 +481,8 @@ static bool get_ttl(struct dns_record *noalias record, int rnamelen, void *ud, b
     if (ntohs(record->rtype) != DNS_TYPE_OPT) {
         /* it is hereby specified that a TTL value is an unsigned number,
             with a minimum value of 0, and a maximum value of 2147483647. */
-        s32 ttl = ntohl(record->rttl);
-        s32 *final_ttl = ud;
+        i32 ttl = ntohl(record->rttl);
+        i32 *final_ttl = ud;
         if (ttl < *final_ttl)
             *final_ttl = ttl;
     }
@@ -497,21 +497,21 @@ static bool update_ttl(struct dns_record *noalias record, int rnamelen, void *ud
     if (ntohs(record->rtype) != DNS_TYPE_OPT) {
         /* it is hereby specified that a TTL value is an unsigned number,
             with a minimum value of 0, and a maximum value of 2147483647. */
-        s32 ttl = (s32)ntohl(record->rttl) + (intptr_t)ud;
+        i32 ttl = (i32)ntohl(record->rttl) + (intptr_t)ud;
         record->rttl = htonl(max(ttl, 1));
     }
 
     return true;
 }
 
-s32 dns_get_ttl(const void *noalias msg, ssize_t len, int qnamelen, s32 nodata_ttl) {
+i32 dns_get_ttl(const void *noalias msg, ssize_t len, int qnamelen, i32 nodata_ttl) {
     if (!is_normal_msg(msg))
         return -1;
 
     int count = get_records_count(msg);
     move_to_records(msg, len, qnamelen);
 
-    s32 ttl = INT32_MAX;
+    i32 ttl = INT32_MAX;
 
     unlikely_if (!foreach_record((void **)&msg, &len, count, get_ttl, &ttl))
         ttl = -1;
@@ -522,7 +522,7 @@ s32 dns_get_ttl(const void *noalias msg, ssize_t len, int qnamelen, s32 nodata_t
     return ttl;
 }
 
-void dns_update_ttl(void *noalias msg, ssize_t len, int qnamelen, s32 ttl_change) {
+void dns_update_ttl(void *noalias msg, ssize_t len, int qnamelen, i32 ttl_change) {
     int count = get_records_count(msg);
     move_to_records(msg, len, qnamelen);
 

src/dns.h

@@ -89,10 +89,10 @@ int dns_test_ip(const void *noalias msg, ssize_t len, int qnamelen, const struct
 void dns_add_ip(const void *noalias msg, ssize_t len, int qnamelen, struct ipset_addctx *noalias ctx);
 
 /* return -1 if failed */
-s32 dns_get_ttl(const void *noalias msg, ssize_t len, int qnamelen, s32 nodata_ttl);
+i32 dns_get_ttl(const void *noalias msg, ssize_t len, int qnamelen, i32 nodata_ttl);
 
 /* it should not fail because it has been checked by `get_ttl` */
-void dns_update_ttl(void *noalias msg, ssize_t len, int qnamelen, s32 ttl_change);
+void dns_update_ttl(void *noalias msg, ssize_t len, int qnamelen, i32 ttl_change);
 
 /*
 * `levels`: the level of the domain to get (8 bools)

src/main.zig

@@ -135,8 +135,8 @@ pub fn main() u8 {
     // ============================================================================
 
     // used only for business-independent initialization, such as global variables
-    init_all_module();
-    defer if (_debug) deinit_all_module();
+    call_module_fn("module_init", .{});
+    defer if (_debug) call_module_fn("module_deinit", .{});
 
     // ============================================================================
 
@@ -214,33 +214,23 @@ pub fn main() u8 {
     server.start();
 
     if (_debug)
-        co.create(memleak_checker, .{});
+        co.start(memleak_checker, .{});
 
     g.evloop.run();
 
     return 0;
 }
 
-fn init_all_module() void {
+fn call_module_fn(comptime fn_name: [:0]const u8, args: anytype) void {
     comptime var i = 0;
     inline while (i < modules.module_list.len) : (i += 1) {
         const module = modules.module_list[i];
         const module_name: cc.ConstStr = modules.name_list[i];
-        if (@hasDecl(module, "module_init")) {
-            if (false) log.debug(@src(), "%s.module_init()", .{module_name});
-            module.module_init();
-        }
-    }
-}
-
-fn deinit_all_module() void {
-    comptime var i = 0;
-    inline while (i < modules.module_list.len) : (i += 1) {
-        const module = modules.module_list[i];
-        const module_name: cc.ConstStr = modules.name_list[i];
-        if (@hasDecl(module, "module_deinit")) {
-            if (false) log.debug(@src(), "%s.module_deinit()", .{module_name});
-            module.module_deinit();
+        if (@hasDecl(module, fn_name)) {
+            if (false) log.debug(@src(), "%s.%s()", .{ module_name, fn_name.ptr });
+            const options: std.builtin.CallOptions = .{};
+            const func = @field(module, fn_name);
+            @call(options, func, args);
         }
     }
 }

src/misc.h

@@ -34,22 +34,22 @@ typedef uint16_t u16;
 typedef uint32_t u32;
 typedef uint64_t u64;
 
-typedef int8_t  s8;
-typedef int16_t s16;
-typedef int32_t s32;
-typedef int64_t s64;
+typedef int8_t  i8;
+typedef int16_t i16;
+typedef int32_t i32;
+typedef int64_t i64;
 
 #define U8C  UINT8_C
 #define U16C UINT16_C
 #define U32C UINT32_C
 #define U64C UINT64_C
 
-#define S8C  INT8_C
-#define S16C INT16_C
-#define S32C INT32_C
-#define S64C INT64_C
+#define I8C  INT8_C
+#define I16C INT16_C
+#define I32C INT32_C
+#define I64C INT64_C
 
-// typedef signed char byte; /* >= 8 bits */
+typedef signed char ibyte; /* >= 8 bits */
 typedef unsigned char ubyte; /* >= 8 bits */
 typedef unsigned short ushort; /* >= 16 bits */
 typedef unsigned int uint; /* >= 16 bits */
@@ -118,7 +118,7 @@ typedef u8 bitvec_t;
 
 /* align to `n` (struct, struct member) */
 #define struct_alignto(n) \
-    __attribute__((packed,aligned(n)))
+    __attribute__((packed, aligned(n)))
 
 /* ======================================================== */
 

src/modules.zig

@@ -1,11 +1,11 @@
-pub const name_list = .{ "CacheMsg", "DynStr", "EvLoop", "ListNode", "NoAAAA", "Rc", "RcMsg", "StrList", "Upstream", "c", "cache", "cache_ignore", "cc", "co", "dnl", "dns", "flags_op", "fmtchk", "g", "groups", "ipset", "local_rr", "log", "main", "modules", "net", "opt", "sentinel_vector", "server", "str2int", "tag", "tests", "verdict_cache" };
-pub const module_list = .{ CacheMsg, DynStr, EvLoop, ListNode, NoAAAA, Rc, RcMsg, StrList, Upstream, c, cache, cache_ignore, cc, co, dnl, dns, flags_op, fmtchk, g, groups, ipset, local_rr, log, main, modules, net, opt, sentinel_vector, server, str2int, tag, tests, verdict_cache };
+pub const name_list = .{ "CacheMsg", "DynStr", "EvLoop", "NoAAAA", "Node", "Rc", "RcMsg", "StrList", "Upstream", "c", "cache", "cache_ignore", "cc", "co", "dnl", "dns", "flags_op", "fmtchk", "g", "groups", "ipset", "local_rr", "log", "main", "modules", "net", "opt", "sentinel_vector", "server", "str2int", "tag", "tests", "verdict_cache" };
+pub const module_list = .{ CacheMsg, DynStr, EvLoop, NoAAAA, Node, Rc, RcMsg, StrList, Upstream, c, cache, cache_ignore, cc, co, dnl, dns, flags_op, fmtchk, g, groups, ipset, local_rr, log, main, modules, net, opt, sentinel_vector, server, str2int, tag, tests, verdict_cache };
 
 const CacheMsg = @import("CacheMsg.zig");
 const DynStr = @import("DynStr.zig");
 const EvLoop = @import("EvLoop.zig");
-const ListNode = @import("ListNode.zig");
 const NoAAAA = @import("NoAAAA.zig");
+const Node = @import("Node.zig");
 const Rc = @import("Rc.zig");
 const RcMsg = @import("RcMsg.zig");
 const StrList = @import("StrList.zig");

src/server.zig

@@ -14,7 +14,7 @@ const Upstream = @import("Upstream.zig");
 const NoAAAA = @import("NoAAAA.zig");
 const EvLoop = @import("EvLoop.zig");
 const RcMsg = @import("RcMsg.zig");
-const ListNode = @import("ListNode.zig");
+const Node = @import("Node.zig");
 const flags_op = @import("flags_op.zig");
 const verdict_cache = @import("verdict_cache.zig");
 const local_rr = @import("local_rr.zig");
@@ -28,7 +28,7 @@ comptime {
 
 const QueryCtx = struct {
     // linked list
-    list_node: ListNode = undefined,
+    list_node: Node = undefined,
 
     // alignment: 8/4
     fdobj: *EvLoop.Fd, // requester's fdobj
@@ -87,13 +87,13 @@ const QueryCtx = struct {
         g.allocator.destroy(self);
     }
 
-    pub fn from_list_node(node: *ListNode) *QueryCtx {
+    pub fn from_list_node(node: *Node) *QueryCtx {
         return @fieldParentPtr(QueryCtx, "list_node", node);
     }
 
     pub const List = struct {
         map: std.AutoHashMapUnmanaged(u16, *QueryCtx),
-        list: ListNode,
+        list: Node,
 
         var _last_qid: u16 = 0;
 
@@ -186,7 +186,7 @@ fn listen_tcp(fd: c_int, ip: cc.ConstStr, port: u16) void {
             continue;
         };
         net.setup_tcp_conn_sock(conn_fd);
-        co.create(service_tcp, .{ conn_fd, &src_addr });
+        co.start(service_tcp, .{ conn_fd, &src_addr });
     }
 }
 
@@ -215,9 +215,9 @@ fn service_tcp(fd: c_int, p_src_addr: *const cc.SockAddr) void {
         while (true) {
             // read len (be16)
             var len: u16 = undefined;
-            g.evloop.recv_exactly(fdobj, std.mem.asBytes(&len), 0) catch |err| switch (err) {
+            g.evloop.recv_full(fdobj, std.mem.asBytes(&len), 0) catch |err| switch (err) {
                 error.eof => return,
-                error.other => break :e .{ .op = "read_len" },
+                error.errno => break :e .{ .op = "read_len" },
             };
 
             len = cc.ntohs(len);
@@ -238,9 +238,9 @@ fn service_tcp(fd: c_int, p_src_addr: *const cc.SockAddr) void {
 
             // read msg
             qmsg.len = len;
-            g.evloop.recv_exactly(fdobj, qmsg.msg(), 0) catch |err| switch (err) {
+            g.evloop.recv_full(fdobj, qmsg.msg(), 0) catch |err| switch (err) {
                 error.eof => break :e .{ .op = "read_msg", .msg = "connection closed" },
-                error.other => break :e .{ .op = "read_msg" },
+                error.errno => break :e .{ .op = "read_msg" },
             };
 
             on_query(qmsg, fdobj, &src_addr, .from_tcp);
@@ -477,8 +477,10 @@ fn on_query(qmsg: *RcMsg, fdobj: *EvLoop.Fd, src_addr: *const cc.SockAddr, in_qf
 
         dns.make_reply(rmsg, msg, qnamelen, answer, answer_n);
 
-        // [async func]
-        if (g.verbose()) qlog.local_rr(answer_n, answer.len);
+        if (g.verbose())
+            qlog.local_rr(answer_n, answer.len);
+
+        // suspending
         return send_reply(rmsg, fdobj, src_addr, bufsz, id, qflags);
     }
 
@@ -493,7 +495,7 @@ fn on_query(qmsg: *RcMsg, fdobj: *EvLoop.Fd, src_addr: *const cc.SockAddr, in_qf
     var ttl_r: i32 = undefined;
     var add_ip: bool = undefined;
     if (cache.get(msg, qnamelen, &ttl, &ttl_r, &add_ip)) |cache_msg| {
-        // because send_reply is async func
+        // because send_reply is a suspending func
         cache.ref(cache_msg);
         defer cache.unref(cache_msg);
 
@@ -507,6 +509,7 @@ fn on_query(qmsg: *RcMsg, fdobj: *EvLoop.Fd, src_addr: *const cc.SockAddr, in_qf
             }
         }
 
+        // suspending
         send_reply(cache_msg, fdobj, src_addr, bufsz, id, qflags);
 
         if (ttl > ttl_r)
@@ -768,7 +771,8 @@ pub fn on_reply(rmsg: *RcMsg, upstream: *const Upstream) void {
             dns.add_ip(msg, qnamelen, addctx);
         };
 
-    // [async] send reply to client
+    // [suspending] send reply to client
+    // TODO: change to nosuspend send_reply
     if (!qctx.flags.has(.from_local))
         send_reply(msg, qctx.fdobj, &qctx.src_addr, qctx.bufsz, qctx.id, qctx.flags);
 
@@ -782,7 +786,7 @@ pub fn on_reply(rmsg: *RcMsg, upstream: *const Upstream) void {
     qctx.free();
 }
 
-/// [async]
+/// [suspending]
 fn send_reply(msg: []const u8, fdobj: *EvLoop.Fd, src_addr: *const cc.SockAddr, bufsz: u16, id: c.be16, qflags: QueryCtx.Flags) void {
     var iov = [_]cc.iovec_t{
         undefined, // for tcp
@@ -839,7 +843,7 @@ fn send_reply(msg: []const u8, fdobj: *EvLoop.Fd, src_addr: *const cc.SockAddr,
     );
 }
 
-/// [async] for bad query msg
+/// [suspending] for bad query msg
 fn send_reply_xxx(msg: []u8, fdobj: *EvLoop.Fd, src_addr: *const cc.SockAddr, qflags: QueryCtx.Flags) void {
     if (msg.len >= dns.header_len())
         _ = dns.empty_reply(msg, 0);
@@ -922,10 +926,10 @@ noinline fn do_start(ip: cc.ConstStr, port: u16, socktype: net.SockType) void {
         switch (socktype) {
             .tcp => {
                 cc.listen(fd, 1024) orelse break :e "listen";
-                co.create(listen_tcp, .{ fd, ip, port });
+                co.start(listen_tcp, .{ fd, ip, port });
             },
             .udp => {
-                co.create(listen_udp, .{ fd, ip, port });
+                co.start(listen_udp, .{ fd, ip, port });
             },
         }
         return;

src/str2int.zig

@@ -2,7 +2,7 @@ const std = @import("std");
 
 pub fn parse(comptime T: type, str: []const u8, radix: u8) ?T {
     if (@bitSizeOf(T) < 1 or @bitSizeOf(T) > 64)
-        @compileError("expected i1..i64 or s1..s64, found " ++ @typeName(T));
+        @compileError("expected i1..i64 or u1..u64, found " ++ @typeName(T));
 
     return @intCast(T, parse_internal(
         if (comptime std.meta.trait.isSignedInt(T)) i64 else u64,

src/wolfssl_opt.h

@@ -4,3 +4,5 @@
 #define WOLFSSL_NO_ATOMICS
 #define WOLFSSL_AEAD_ONLY
 #define LARGE_STATIC_BUFFERS
+#define WOLFSSL_JNI /* ssl->data (void *) */
+#define HAVE_EXT_CACHE /* new session callback */

tool/dns_cache_mgr.c

@@ -9,10 +9,10 @@
 #include "../src/misc.h"
 
 struct header {
-    s64 update_time;
+    i64 update_time;
     u32 hashv;
-    s32 ttl;
-    s32 ttl_r;
+    i32 ttl;
+    i32 ttl_r;
     u16 msg_len;
     u8 qnamelen;
     // msg: [msg_len]u8, // {header, question, answer, authority, additional}
@@ -55,13 +55,14 @@ static void list(FILE *file) {
     char buf[sizeof(*h)] alignto(__alignof__(*h));
     h = (void *)buf;
 
-    char name[DNS_NAME_MAXLEN + 1];
-
     void *msg = malloc(DNS_MSG_MAXSIZE);
+    char name[DNS_NAME_MAXLEN + 1];
 
-    s64 now = time(NULL);
+    i64 now = time(NULL);
     while (next(file, h, msg, name))
-        printf("%-60s qtype:%-5u ttl:%-10d size:%u\n", name, dns_get_qtype(msg, h->qnamelen), h->ttl - (s32)(now - h->update_time), h->msg_len);
+        printf("%-60s qtype:%-5u ttl:%-10d size:%u\n",
+            name, dns_get_qtype(msg, h->qnamelen),
+            h->ttl - (i32)(now - h->update_time), h->msg_len);
 
     free(msg);
 }