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

reorganize but crashing

Browse Source 
not sure why, seems like i'm using the right allocators everywhere?
need to take another pass at this later.
This commit is contained in:
Robby Zambito
2026-01-02 03:00:49 +00:00
parent a71f08e1f6
commit 4bf064056c
3 changed files with 70 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
src/server/client.zig
View File

@@ -23,9 +23,20 @@ pub fn init(
}; };
} }
pub fn start(self: *Client, io: std.Io, alloc: std.mem.Allocator, queue: *std.Io.Queue(Message)) !void { pub fn deinit(self: *Client, alloc: std.mem.Allocator) void {
if (self.connect) |c| {
c.deinit(alloc);
}
self.* = undefined;
}
pub fn start(self: *Client, io: std.Io, alloc: std.mem.Allocator, queue: *std.Io.Queue(Message), server_info: Message.ServerInfo) !void {
self.recv_queue = queue; self.recv_queue = queue;
var msgs: [8]Message = undefined; var msgs: [8]Message = undefined;
// Do initial handshake with client
try queue.putOne(io, .{ .info = server_info });
while (true) { while (true) {
const len = try queue.get(io, &msgs, 1); const len = try queue.get(io, &msgs, 1);
std.debug.assert(len <= msgs.len); std.debug.assert(len <= msgs.len);

33
src/server/main.zig
View File

@@ -87,7 +87,6 @@ pub fn start(server: *Server, io: std.Io, gpa: std.mem.Allocator) !void {
defer client_group.cancel(io); defer client_group.cancel(io);
var id: usize = 0; var id: usize = 0;
// Run until SIGINT is handled, then exit gracefully
while (true) : (id +%= 1) { while (true) : (id +%= 1) {
std.debug.print("in server loop\n", .{}); std.debug.print("in server loop\n", .{});
if (server.clients.contains(id)) continue; if (server.clients.contains(id)) continue;
@@ -140,10 +139,10 @@ fn handleConnection(
) !void { ) !void {
defer stream.close(io); defer stream.close(io);
var client_allocator: std.heap.DebugAllocator(.{}) = .init; //var client_allocator: std.heap.DebugAllocator(.{}) = .init;
client_allocator.backing_allocator = server_allocator; //client_allocator.backing_allocator = server_allocator;
defer _ = client_allocator.deinit(); //defer _ = client_allocator.deinit();
const allocator = if (builtin.mode == .Debug or builtin.mode == .ReleaseSafe) client_allocator.allocator() else server_allocator; //const allocator = if (builtin.mode == .Debug or builtin.mode == .ReleaseSafe) client_allocator.allocator() else server_allocator;
// Set up client writer // Set up client writer
var w_buffer: [1024]u8 = undefined; var w_buffer: [1024]u8 = undefined;
@@ -157,6 +156,8 @@ fn handleConnection(
// Create client // Create client
var client: Client = .init(null, in, out); var client: Client = .init(null, in, out);
defer client.deinit(server_allocator);
try server.addClient(server_allocator, id, &client); try server.addClient(server_allocator, id, &client);
defer server.removeClient(io, server_allocator, id); defer server.removeClient(io, server_allocator, id);
@@ -172,17 +173,9 @@ fn handleConnection(
} else |_| {} } else |_| {}
} }
var client_task = try io.concurrent(Client.start, .{ &client, io, server_allocator, &queue }); var client_task = try io.concurrent(Client.start, .{ &client, io, server_allocator, &queue, server.info });
defer client_task.cancel(io) catch {}; defer client_task.cancel(io) catch {};
try io.sleep(std.Io.Duration.fromMilliseconds(5), .real);
// Do initial handshake with client
try client.send(io, .{ .info = server.info });
var connect_arena: std.heap.ArenaAllocator = .init(allocator);
defer connect_arena.deinit();
client.connect = (Message.next(connect_arena.allocator(), in) catch return).connect;
// Messages are owned by the server after they are received from the client // Messages are owned by the server after they are received from the client
while (client.next(server_allocator)) |msg| { while (client.next(server_allocator)) |msg| {
switch (msg) { switch (msg) {
@@ -191,6 +184,7 @@ fn handleConnection(
try client.send(io, .pong); try client.send(io, .pong);
}, },
.@"pub" => |pb| { .@"pub" => |pb| {
defer pb.deinit(server_allocator);
try server.publishMessage(io, server_allocator, &client, pb); try server.publishMessage(io, server_allocator, &client, pb);
}, },
.sub => |sub| { .sub => |sub| {
@@ -199,6 +193,12 @@ fn handleConnection(
.unsub => |unsub| { .unsub => |unsub| {
try server.unsubscribe(io, server_allocator, id, unsub); try server.unsubscribe(io, server_allocator, id, unsub);
}, },
.connect => |connect| {
if (client.connect) |*current| {
current.deinit(server_allocator);
}
client.connect = connect;
},
else => |e| { else => |e| {
std.debug.panic("Unimplemented message: {any}\n", .{e}); std.debug.panic("Unimplemented message: {any}\n", .{e});
}, },
@@ -213,8 +213,8 @@ fn handleConnection(
} }
} }
fn subjectMatches(expected: []const u8, actual: []const u8) bool { fn subjectMatches(sub_subject: []const u8, pub_subject: []const u8) bool {
return std.mem.eql(u8, expected, actual); return std.mem.eql(u8, sub_subject, pub_subject);
} }
fn publishMessage(server: *Server, io: std.Io, alloc: std.mem.Allocator, source_client: *Client, msg: Message.Pub) !void { fn publishMessage(server: *Server, io: std.Io, alloc: std.mem.Allocator, source_client: *Client, msg: Message.Pub) !void {
@@ -225,7 +225,6 @@ fn publishMessage(server: *Server, io: std.Io, alloc: std.mem.Allocator, source_
} }
} }
} }
defer msg.deinit(alloc);
try server.subs_lock.lock(io); try server.subs_lock.lock(io);
defer server.subs_lock.unlock(io); defer server.subs_lock.unlock(io);
for (server.subscriptions.items) |subscription| { for (server.subscriptions.items) |subscription| {

70
src/server/message_parser.zig
View File

@@ -102,24 +102,22 @@ pub const Message = union(MessageType) {
} }
pub fn dupe(self: Connect, alloc: std.mem.Allocator) !Connect { pub fn dupe(self: Connect, alloc: std.mem.Allocator) !Connect {
return .{ var res = self;
.verbose = self.verbose, res.auth_token = if (self.auth_token) |a| try alloc.dupe(u8, a) else null;
.pedantic = self.pedantic, errdefer if (res.auth_token) |a| alloc.free(a);
.tls_required = self.tls_required, res.user = if (self.user) |u| try alloc.dupe(u8, u) else null;
.auth_token = if (self.auth_token) |a| try alloc.dupe(u8, a) else null, errdefer if (res.user) |u| alloc.free(u);
.user = if (self.user) |u| try alloc.dupe(u8, u) else null, res.pass = if (self.pass) |p| try alloc.dupe(u8, p) else null;
.pass = if (self.pass) |p| try alloc.dupe(u8, p) else null, errdefer if (res.pass) |p| alloc.free(p);
.name = if (self.name) |n| try alloc.dupe(u8, n) else null, res.name = if (self.name) |n| try alloc.dupe(u8, n) else null;
.lang = self.lang, errdefer if (res.name) |n| alloc.free(n);
.version = self.version, res.sig = if (self.sig) |s| try alloc.dupe(u8, s) else null;
.protocol = self.protocol, errdefer if (res.sig) |s| alloc.free(s);
.echo = self.echo, res.jwt = if (self.jwt) |j| try alloc.dupe(u8, j) else null;
.sig = if (self.sig) |s| try alloc.dupe(u8, s) else null, errdefer if (res.jwt) |j| alloc.free(j);
.jwt = if (self.jwt) |j| try alloc.dupe(u8, j) else null, res.nkey = if (self.nkey) |n| try alloc.dupe(u8, n) else null;
.no_responders = self.no_responders, errdefer if (res.nkey) |n| alloc.free(n);
.headers = self.headers, return res;
.nkey = if (self.nkey) |n| try alloc.dupe(u8, n) else null,
};
} }
}; };
pub const Pub = struct { pub const Pub = struct {
@@ -165,17 +163,21 @@ pub const Message = union(MessageType) {
pub fn deinit(self: Msg, alloc: std.mem.Allocator) void { pub fn deinit(self: Msg, alloc: std.mem.Allocator) void {
alloc.free(self.subject); alloc.free(self.subject);
alloc.free(self.sid); alloc.free(self.sid);
alloc.free(self.payload);
if (self.reply_to) |r| alloc.free(r); if (self.reply_to) |r| alloc.free(r);
alloc.free(self.payload);
} }
pub fn dupe(self: Msg, alloc: std.mem.Allocator) !Msg { pub fn dupe(self: Msg, alloc: std.mem.Allocator) !Msg {
return .{ var res: Msg = undefined;
.subject = try alloc.dupe(u8, self.subject), res.subject = try alloc.dupe(u8, self.subject);
.sid = try alloc.dupe(u8, self.sid), errdefer alloc.free(res.subject);
.reply_to = if (self.reply_to) |r| try alloc.dupe(u8, r) else null, res.sid = try alloc.dupe(u8, self.sid);
.payload = try alloc.dupe(u8, self.payload), errdefer alloc.free(res.sid);
}; res.reply_to = if (self.reply_to) |r| alloc.dupe(u8, r) else null;
errdefer if (res.reply_to) |r| alloc.free(r);
res.payload = try alloc.dupe(u8, self.payload);
errdefer alloc.free(res.payload);
return res;
} }
}; };
@@ -221,11 +223,16 @@ pub const Message = union(MessageType) {
switch (operation) { switch (operation) {
.connect => { .connect => {
// for storing the json string
var connect_string_writer_allocating: std.Io.Writer.Allocating = try .initCapacity(alloc, 1024);
defer connect_string_writer_allocating.deinit();
var connect_string_writer = connect_string_writer_allocating.writer;
// for parsing the json string
var connect_arena_allocator: std.heap.ArenaAllocator = .init(alloc); var connect_arena_allocator: std.heap.ArenaAllocator = .init(alloc);
defer connect_arena_allocator.deinit(); defer connect_arena_allocator.deinit();
const connect_allocator = connect_arena_allocator.allocator(); const connect_allocator = connect_arena_allocator.allocator();
const connect_string_writer_allocating: std.Io.Writer.Allocating = try .initCapacity(connect_allocator, 1024);
var connect_string_writer = connect_string_writer_allocating.writer;
try in.discardAll(1); // throw away space try in.discardAll(1); // throw away space
// Should read the next JSON object to the fixed buffer writer. // Should read the next JSON object to the fixed buffer writer.
@@ -234,7 +241,12 @@ pub const Message = union(MessageType) {
try expectStreamBytes(in, "}\r\n"); // discard '}\r\n' try expectStreamBytes(in, "}\r\n"); // discard '}\r\n'
// TODO: should be CONNECTION allocator // TODO: should be CONNECTION allocator
const res = try std.json.parseFromSliceLeaky(Connect, connect_allocator, connect_string_writer.buffered(), .{ .allocate = .alloc_always }); const res = try std.json.parseFromSliceLeaky(
Connect,
connect_allocator,
connect_string_writer.buffered(),
.{ .allocate = .alloc_always },
);
return .{ .connect = try res.dupe(alloc) }; return .{ .connect = try res.dupe(alloc) };
}, },
@@ -243,6 +255,7 @@ pub const Message = union(MessageType) {
// Parse subject // Parse subject
const subject: []const u8 = try readSubject(alloc, in); const subject: []const u8 = try readSubject(alloc, in);
errdefer alloc.free(subject);
// Parse byte count // Parse byte count
const byte_count = blk: { const byte_count = blk: {
@@ -267,6 +280,7 @@ pub const Message = union(MessageType) {
const payload = blk: { const payload = blk: {
const bytes = try alloc.alloc(u8, byte_count); const bytes = try alloc.alloc(u8, byte_count);
errdefer alloc.free(bytes);
try in.readSliceAll(bytes); try in.readSliceAll(bytes);
try expectStreamBytes(in, "\r\n"); try expectStreamBytes(in, "\r\n");
break :blk bytes; break :blk bytes;