Significant speed improvement

2 files changed, 13 insertions, 20 deletions

diff --git a/src/Server.zig b/src/Server.zig
index f0ee1fe..7dc3783 100644
--- a/src/Server.zig
+++ b/src/Server.zig
@@ -39,15 +39,14 @@ const Subscription = struct {
         if (self.queue_group) |g| alloc.free(g);
     }
 
-    fn send(self: *Subscription, io: Io, bytes: []const []const u8) !void {
-        try self.queue_lock.lock(io);
-        defer self.queue_lock.unlock(io);
+    fn send(self: *Subscription, io: Io, buf: []u8, bytes: []const []const u8) !void {
+        var w: std.Io.Writer = .fixed(buf);
         for (bytes) |chunk| {
-            // Uncancelable because canceling the sender in the middle of this loop
-            // would put an invalid set series of bytes in the receivers queue.
-            _ = try self.queue.putUncancelable(io, chunk, chunk.len);
+            w.writeAll(chunk) catch unreachable;
         }
-        try io.checkCancel();
+        try self.queue_lock.lock(io);
+        defer self.queue_lock.unlock(io);
+        try self.queue.putAll(io, w.buffered());
     }
 };
 
@@ -181,6 +180,9 @@ fn handleConnection(
     var recv_queue: Queue(u8) = .init(qbuf);
     defer recv_queue.close(io);
 
+    const msg_write_buf: []u8 = try alloc.alignedAlloc(u8, .fromByteUnits(std.atomic.cache_line), 1 * 1024 * 1024);
+    defer alloc.free(msg_write_buf);
+
     // Create client
     var client: Client = .init(null, &recv_queue, in, out);
     defer client.deinit(server_allocator);
@@ -207,7 +209,7 @@ fn handleConnection(
             .PUB => {
                 @branchHint(.likely);
                 // log.debug("received a pub msg", .{});
-                server.publishMessage(io, rand, server_allocator, &client, .@"pub") catch |err| switch (err) {
+                server.publishMessage(io, rand, server_allocator, msg_write_buf, &client, .@"pub") catch |err| switch (err) {
                     error.ReadFailed => return reader.err.?,
                     error.EndOfStream => return error.ClientDisconnected,
                     else => |e| return e,
@@ -215,7 +217,7 @@ fn handleConnection(
             },
             .HPUB => {
                 @branchHint(.likely);
-                server.publishMessage(io, rand, server_allocator, &client, .hpub) catch |err| switch (err) {
+                server.publishMessage(io, rand, server_allocator, msg_write_buf, &client, .hpub) catch |err| switch (err) {
                     error.ReadFailed => return reader.err.?,
                     error.EndOfStream => return error.ClientDisconnected,
                     else => |e| return e,
@@ -298,6 +300,7 @@ fn publishMessage(
     io: Io,
     rand: std.Random,
     alloc: Allocator,
+    msg_write_buf: []u8,
     source_client: *Client,
     comptime pub_or_hpub: enum { @"pub", hpub },
 ) !void {
@@ -384,7 +387,7 @@ fn publishMessage(
             );
             msg_chunks.appendAssumeCapacity(msg.payload);
 
-            subscription.send(io, msg_chunks.items[0..chunk_count]) catch |err| switch (err) {
+            subscription.send(io, msg_write_buf, msg_chunks.items[0..chunk_count]) catch |err| switch (err) {
                 error.Closed => {},
                 error.Canceled => |e| return e,
             };
diff --git a/src/Server/Client.zig b/src/Server/Client.zig
index 6ad1804..77034fd 100644
--- a/src/Server/Client.zig
+++ b/src/Server/Client.zig
@@ -40,16 +40,6 @@ pub fn start(self: *Client, io: std.Io) !void {
     std.debug.assert(self.to_client.end == 0);
     while (true) {
         self.to_client.end = try self.recv_queue.get(io, self.to_client.buffer, 1);
-        // Wait 1 nanosecond to see if more data is in the queue.
-        // If there is, add it to the write buffer before sending it.
-        // The reason for this is because if we send the first chunk as soon as we get it,
-        // we will likely be sending a partial message, which will end up being way slower.
-        try io.sleep(.fromNanoseconds(1), .awake);
-        self.to_client.end += try self.recv_queue.get(
-            io,
-            self.to_client.buffer[self.to_client.end..],
-            0,
-        );
         try self.to_client.flush();
     }
 }