mirrors/zits
1
0
Fork 0
mirror of https://git.robbyzambito.me/zits synced 2026-02-04 03:34:48 +00:00
Code Issues Packages Projects Releases Wiki Activity

starting zero alloc parsing

Browse Source
This commit is contained in:
Robby Zambito
2026-01-07 17:26:10 -05:00
parent e2a60c9427
commit 96a3705069
6 changed files with 1001 additions and 914 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
src/Server.zig
View File

@@ -200,32 +200,30 @@ fn handleConnection(
var client_task = try io.concurrent(Client.start, .{ &client, io }); var client_task = try io.concurrent(Client.start, .{ &client, io });
defer client_task.cancel(io) catch {}; defer client_task.cancel(io) catch {};
// Messages are owned by the server after they are received from the client while (client.next(server_allocator)) |ctrl| {
while (client.next(server_allocator)) |msg| { switch (ctrl) {
switch (msg) {
.PING => { .PING => {
// Respond to ping with pong. // Respond to ping with pong.
try client.send(io, .PONG); try client.recv_queue_write_lock.lock(io);
defer client.recv_queue_write_lock.unlock(io);
try client.send(io, "PONG\r\n");
}, },
.PUB => |pb| { .PUB => {
@branchHint(.likely); @branchHint(.likely);
defer pb.deinit(server_allocator);
try server.publishMessage(io, server_allocator, &client, msg); try server.publishMessage(io, server_allocator, &client, msg);
}, },
.HPUB => |hp| { .HPUB => {
@branchHint(.likely); @branchHint(.likely);
defer hp.deinit(server_allocator);
try server.publishMessage(io, server_allocator, &client, msg); try server.publishMessage(io, server_allocator, &client, msg);
}, },
.SUB => |sub| { .SUB => {
defer sub.deinit(server_allocator);
try server.subscribe(io, server_allocator, client, id, sub); try server.subscribe(io, server_allocator, client, id, sub);
}, },
.UNSUB => |unsub| { .UNSUB => {
defer unsub.deinit(server_allocator); defer unsub.deinit(server_allocator);
try server.unsubscribe(io, server_allocator, id, unsub); try server.unsubscribe(io, server_allocator, id, unsub);
}, },
.CONNECT => |connect| { .CONNECT => {
if (client.connect) |*current| { if (client.connect) |*current| {
current.deinit(server_allocator); current.deinit(server_allocator);
} }

192
src/Server/Client.zig
View File

@@ -11,29 +11,23 @@ pub const Msgs = union(enum) {
}; };
connect: ?Message.Connect, connect: ?Message.Connect,
// Used to own messages that we receive in our queues. // Byte queue for this client to receive.
alloc: std.mem.Allocator, recv_queue: *Queue(u8),
// Only necessary to hold this lock for writing to the queue (to avoid interleaving message writes).
// Messages for this client to receive. recv_queue_write_lock: std.Io.Mutex = .init,
recv_queue: *Queue(Message),
msg_queue: *Queue(Msgs),
from_client: *std.Io.Reader, from_client: *std.Io.Reader,
to_client: *std.Io.Writer, to_client: *std.Io.Writer,
pub fn init( pub fn init(
connect: ?Message.Connect, connect: ?Message.Connect,
alloc: std.mem.Allocator, recv_queue: *Queue(u8),
recv_queue: *Queue(Message),
msg_queue: *Queue(Msgs),
in: *std.Io.Reader, in: *std.Io.Reader,
out: *std.Io.Writer, out: *std.Io.Writer,
) Client { ) Client {
return .{ return .{
.connect = connect, .connect = connect,
.alloc = alloc,
.recv_queue = recv_queue, .recv_queue = recv_queue,
.msg_queue = msg_queue,
.from_client = in, .from_client = in,
.to_client = out, .to_client = out,
}; };
@@ -47,97 +41,91 @@ pub fn deinit(self: *Client, alloc: std.mem.Allocator) void {
} }
pub fn start(self: *Client, io: std.Io) !void { pub fn start(self: *Client, io: std.Io) !void {
var msgs_buf: [1024]Msgs = undefined; std.debug.assert(self.to_client.buffer.len > 0);
std.debug.assert(self.to_client.end == 0);
var recv_msgs_task = io.concurrent(Queue(Msgs).get, .{ self.msg_queue, io, &msgs_buf, 1 }) catch @panic("Concurrency unavailable");
errdefer _ = recv_msgs_task.cancel(io) catch {};
var recv_proto_task = io.concurrent(Queue(Message).getOne, .{ self.recv_queue, io }) catch unreachable;
errdefer _ = recv_proto_task.cancel(io) catch {};
while (true) { while (true) {
switch (try io.select(.{ .msgs = &recv_msgs_task, .proto = &recv_proto_task })) { self.to_client.end = try self.recv_queue.get(io, self.to_client.buffer, 1);
.msgs => |len_err| {
@branchHint(.likely);
const msgs = msgs_buf[0..try len_err];
for (0..msgs.len) |i| {
const msg = msgs[i];
defer switch (msg) {
.MSG => |m| m.deinit(self.alloc),
.HMSG => |h| h.deinit(self.alloc),
};
errdefer for (msgs[i + 1 ..]) |mg| switch (mg) {
.MSG => |m| {
m.deinit(self.alloc);
},
.HMSG => |h| {
h.deinit(self.alloc);
},
};
switch (msg) {
.MSG => |m| {
try self.to_client.print(
"MSG {s} {s} {s} {d}\r\n",
.{
m.subject,
m.sid,
m.reply_to orelse "",
m.payload.len,
},
);
try m.payload.write(self.to_client);
try self.to_client.print("\r\n", .{});
},
.HMSG => |hmsg| {
try self.to_client.print("HMSG {s} {s} {s} {d} {d}\r\n", .{
hmsg.msg.subject,
hmsg.msg.sid,
hmsg.msg.reply_to orelse "",
hmsg.header_bytes,
hmsg.msg.payload.len,
});
try hmsg.msg.payload.write(self.to_client);
try self.to_client.print("\r\n", .{});
},
}
}
recv_msgs_task = io.concurrent(Queue(Msgs).get, .{ self.msg_queue, io, &msgs_buf, 1 }) catch unreachable;
},
.proto => |msg_err| {
@branchHint(.unlikely);
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});
},
}
recv_proto_task = io.concurrent(Queue(Message).getOne, .{ self.recv_queue, io }) catch unreachable;
},
}
try self.to_client.flush(); try self.to_client.flush();
} }
// while (true) {
// switch (try io.select(.{ .msgs = &recv_msgs_task, .proto = &recv_proto_task })) {
// .msgs => |len_err| {
// @branchHint(.likely);
// const msgs = msgs_buf[0..try len_err];
// for (0..msgs.len) |i| {
// const msg = msgs[i];
// defer switch (msg) {
// .MSG => |m| m.deinit(self.alloc),
// .HMSG => |h| h.deinit(self.alloc),
// };
// errdefer for (msgs[i + 1 ..]) |mg| switch (mg) {
// .MSG => |m| {
// m.deinit(self.alloc);
// },
// .HMSG => |h| {
// h.deinit(self.alloc);
// },
// };
// switch (msg) {
// .MSG => |m| {
// try self.to_client.print(
// "MSG {s} {s} {s} {d}\r\n",
// .{
// m.subject,
// m.sid,
// m.reply_to orelse "",
// m.payload.len,
// },
// );
// try m.payload.write(self.to_client);
// try self.to_client.print("\r\n", .{});
// },
// .HMSG => |hmsg| {
// try self.to_client.print("HMSG {s} {s} {s} {d} {d}\r\n", .{
// hmsg.msg.subject,
// hmsg.msg.sid,
// hmsg.msg.reply_to orelse "",
// hmsg.header_bytes,
// hmsg.msg.payload.len,
// });
// try hmsg.msg.payload.write(self.to_client);
// try self.to_client.print("\r\n", .{});
// },
// }
// }
// recv_msgs_task = io.concurrent(Queue(Msgs).get, .{ self.msg_queue, io, &msgs_buf, 1 }) catch unreachable;
// },
// .proto => |msg_err| {
// @branchHint(.unlikely);
// 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});
// },
// }
// recv_proto_task = io.concurrent(Queue(Message).getOne, .{ self.recv_queue, io }) catch unreachable;
// },
// }
// try self.to_client.flush();
// }
} }
pub fn send(self: *Client, io: std.Io, msg: Message) !void { pub fn send(self: *Client, io: std.Io, msg: []const u8) !void {
switch (msg) { try self.recv_queue.putAll(io, msg);
.MSG => |m| try self.msg_queue.putOne(io, .{ .MSG = m }),
.HMSG => |m| try self.msg_queue.putOne(io, .{ .HMSG = m }),
else => try self.recv_queue.putOne(io, msg),
}
} }
test send { test send {
@@ -148,19 +136,15 @@ test send {
var buf: [1024]u8 = undefined; var buf: [1024]u8 = undefined;
break :blk &buf; break :blk &buf;
}); });
var recv_queue: Queue(Message) = .init(&.{}); var recv_queue: Queue(u8) = .init(&.{});
var msgs_queue: Queue(Msgs) = .init(blk: { var client: Client = .init(null, &recv_queue, undefined, &to_client);
var buf: [1]Msgs = undefined;
break :blk &buf;
});
var client: Client = .init(null, gpa, &recv_queue, &msgs_queue, undefined, &to_client);
defer client.deinit(gpa); defer client.deinit(gpa);
var c_task = try io.concurrent(Client.start, .{ &client, io }); var c_task = try io.concurrent(Client.start, .{ &client, io });
defer c_task.cancel(io) catch {}; defer c_task.cancel(io) catch {};
{ {
try client.send(io, .PONG); try client.send(io, "PONG\r\n");
// Wait for the concurrent client task to write to the writer // Wait for the concurrent client task to write to the writer
try io.sleep(.fromMilliseconds(1), .awake); try io.sleep(.fromMilliseconds(1), .awake);
try std.testing.expectEqualSlices(u8, "PONG\r\n", to_client.buffered()); try std.testing.expectEqualSlices(u8, "PONG\r\n", to_client.buffered());

0
src/Server/parse/Payload.zig → src/Server/Payload.zig
View File

30
src/Server/parse/message.zig → src/Server/message.zig
View File

@@ -106,13 +106,7 @@ pub const Message = union(enum) {
/// The reply subject that subscribers can use to send a response back to the publisher/requestor. /// The reply subject that subscribers can use to send a response back to the publisher/requestor.
reply_to: ?[]const u8 = null, reply_to: ?[]const u8 = null,
/// The message payload data. /// The message payload data.
payload: Payload, payload: []const u8,
pub fn deinit(self: Pub, alloc: Allocator) void {
alloc.free(self.subject);
self.payload.deinit(alloc);
if (self.reply_to) |r| alloc.free(r);
}
pub fn toMsg(self: Pub, alloc: Allocator, sid: []const u8) !Msg { pub fn toMsg(self: Pub, alloc: Allocator, sid: []const u8) !Msg {
const res: Msg = .{ const res: Msg = .{
@@ -183,26 +177,6 @@ pub const Message = union(enum) {
subject: []const u8, subject: []const u8,
sid: []const u8, sid: []const u8,
reply_to: ?[]const u8, reply_to: ?[]const u8,
payload: Payload, payload: []const u8,
pub fn deinit(self: Msg, alloc: Allocator) void {
alloc.free(self.subject);
alloc.free(self.sid);
if (self.reply_to) |r| alloc.free(r);
self.payload.deinit(alloc);
}
pub fn dupe(self: Msg, alloc: Allocator) !Msg {
var res: Msg = undefined;
res.subject = try alloc.dupe(u8, self.subject);
errdefer alloc.free(res.subject);
res.sid = try alloc.dupe(u8, self.sid);
errdefer alloc.free(res.sid);
res.reply_to = if (self.reply_to) |r| try alloc.dupe(u8, r) else null;
errdefer if (res.reply_to) |r| alloc.free(r);
res.payload = try self.payload.dupe(alloc);
errdefer alloc.free(res.payload);
return res;
}
}; };
}; };

1665
src/Server/parse.zig
View File

File diff suppressed because it is too large Load Diff

6
src/Server/serialize.zig Normal file
View File

@@ -0,0 +1,6 @@
const std = @import("std");
const message = @import("./message.zig");
const Message = message.Message;
pub fn message(msg: Message)