Using separate queue for high throughput messages

1 files changed, 92 insertions, 58 deletions

diff --git a/src/server/Client.zig b/src/server/Client.zig
index 1e93360..8ff92e8 100644
--- a/src/server/Client.zig
+++ b/src/server/Client.zig
@@ -3,10 +3,16 @@ const std = @import("std");
 
 const Client = @This();
 
+pub const Msgs = union(enum) {
+    MSG: Message.Msg,
+    HMSG: Message.HMsg,
+};
+
 connect: ?Message.Connect,
 
 // Messages for this client to receive.
 recv_queue: *std.Io.Queue(Message),
+msg_queue: *std.Io.Queue(Msgs),
 
 from_client: *std.Io.Reader,
 to_client: *std.Io.Writer,
@@ -14,12 +20,14 @@ to_client: *std.Io.Writer,
 pub fn init(
     connect: ?Message.Connect,
     recv_queue: *std.Io.Queue(Message),
+    msg_queue: *std.Io.Queue(Msgs),
     in: *std.Io.Reader,
     out: *std.Io.Writer,
 ) Client {
     return .{
         .connect = connect,
         .recv_queue = recv_queue,
+        .msg_queue = msg_queue,
         .from_client = in,
         .to_client = out,
     };
@@ -33,72 +41,98 @@ pub fn deinit(self: *Client, alloc: std.mem.Allocator) void {
 }
 
 pub fn start(self: *Client, io: std.Io, alloc: std.mem.Allocator) !void {
-    var msgs: [256]Message = undefined;
+    var msgs_buf: [1024]Msgs = undefined;
+
+    var recv_msgs_task = io.concurrent(recvMsgsMsg, .{ self, io, &msgs_buf }) catch @panic("Concurrency unavailable");
+    errdefer _ = recv_msgs_task.cancel(io) catch {};
+
+    var recv_proto_task = io.concurrent(recvProtoMsg, .{ self, io }) catch unreachable;
+    errdefer _ = recv_proto_task.cancel(io) catch {};
 
     while (true) {
-        const len = try self.recv_queue.get(io, &msgs, 1);
-        std.debug.assert(len <= msgs.len);
-        for (0..len) |i| {
-            const msg = msgs[i];
-            defer switch (msg) {
-                .MSG => |m| m.deinit(alloc),
-                .HMSG => |h| h.deinit(alloc),
-                else => {},
-            };
-            errdefer for (msgs[i + 1 .. len]) |mg| switch (mg) {
-                .MSG => |m| {
-                    m.deinit(alloc);
-                },
-                else => {},
-            };
-            switch (msg) {
-                .@"+OK" => {
-                    _ = try self.to_client.write("+OK\r\n");
-                },
-                .PONG => {
-                    _ = try self.to_client.write("PONG\r\n");
-                },
-                .INFO => |info| {
-                    _ = try self.to_client.write("INFO ");
-                    try std.json.Stringify.value(info, .{}, self.to_client);
-                    _ = try self.to_client.write("\r\n");
-                },
-                .MSG => |m| {
-                    @branchHint(.likely);
-                    try self.to_client.print(
-                        "MSG {s} {s} {s} {d}\r\n{s}\r\n",
-                        .{
-                            m.subject,
-                            m.sid,
-                            m.reply_to orelse "",
-                            m.payload.len,
-                            m.payload,
+        switch (try io.select(.{ .msgs = &recv_msgs_task, .proto = &recv_proto_task })) {
+            .msgs => |msgs_err| {
+                @branchHint(.likely);
+                defer recv_msgs_task = io.concurrent(recvMsgsMsg, .{ self, io, &msgs_buf }) catch unreachable;
+                const msgs = try msgs_err;
+                for (0..msgs.len) |i| {
+                    const msg = msgs[i];
+                    defer switch (msg) {
+                        .MSG => |m| m.deinit(alloc),
+                        .HMSG => |h| h.deinit(alloc),
+                    };
+                    errdefer for (msgs[i + 1 ..]) |mg| switch (mg) {
+                        .MSG => |m| {
+                            m.deinit(alloc);
                         },
-                    );
-                },
-                .HMSG => |hmsg| {
-                    @branchHint(.likely);
-                    try self.to_client.print("HMSG {s} {s} {s} {d} {d}\r\n{s}\r\n", .{
-                        hmsg.msg.subject,
-                        hmsg.msg.sid,
-                        hmsg.msg.reply_to orelse "",
-                        hmsg.header_bytes,
-                        hmsg.msg.payload.len,
-                        hmsg.msg.payload,
-                    });
-                },
-                .@"-ERR" => |s| {
-                    _ = try self.to_client.print("-ERR '{s}'\r\n", .{s});
-                },
-                else => |m| {
-                    std.debug.panic("unimplemented write: {any}\n", .{m});
-                },
-            }
+                        .HMSG => |h| {
+                            h.deinit(alloc);
+                        },
+                    };
+                    switch (msg) {
+                        .MSG => |m| {
+                            try self.to_client.print(
+                                "MSG {s} {s} {s} {d}\r\n{s}\r\n",
+                                .{
+                                    m.subject,
+                                    m.sid,
+                                    m.reply_to orelse "",
+                                    m.payload.len,
+                                    m.payload,
+                                },
+                            );
+                        },
+                        .HMSG => |hmsg| {
+                            try self.to_client.print("HMSG {s} {s} {s} {d} {d}\r\n{s}\r\n", .{
+                                hmsg.msg.subject,
+                                hmsg.msg.sid,
+                                hmsg.msg.reply_to orelse "",
+                                hmsg.header_bytes,
+                                hmsg.msg.payload.len,
+                                hmsg.msg.payload,
+                            });
+                        },
+                    }
+                }
+            },
+            .proto => |msg_err| {
+                @branchHint(.unlikely);
+                defer recv_proto_task = io.concurrent(recvProtoMsg, .{ self, io }) catch unreachable;
+                const msg = try msg_err;
+                switch (msg) {
+                    .@"+OK" => {
+                        _ = try self.to_client.write("+OK\r\n");
+                    },
+                    .PONG => {
+                        _ = try self.to_client.write("PONG\r\n");
+                    },
+                    .INFO => |info| {
+                        _ = try self.to_client.write("INFO ");
+                        try std.json.Stringify.value(info, .{}, self.to_client);
+                        _ = try self.to_client.write("\r\n");
+                    },
+                    .@"-ERR" => |s| {
+                        _ = try self.to_client.print("-ERR '{s}'\r\n", .{s});
+                    },
+                    else => |m| {
+                        std.debug.panic("unimplemented write: {any}\n", .{m});
+                    },
+                }
+            },
         }
         try self.to_client.flush();
     }
 }
 
+fn recvProtoMsg(self: *Client, io: std.Io) !Message {
+    return self.recv_queue.getOne(io);
+}
+
+fn recvMsgsMsg(self: *Client, io: std.Io, buf: []Msgs) ![]Msgs {
+    const len = try self.msg_queue.get(io, buf, 1);
+    return buf[0..len];
+}
+
 pub fn send(self: *Client, io: std.Io, msg: Message) !void {
     try self.recv_queue.putOne(io, msg);
 }