this is a little bit faster, but not as correct

hashing.zig

@@ -1,7 +1,18 @@
 const std = @import("std");
 
 pub fn hashKey(k: []const u8) u32 {
-    return djb2(k);
+    return fnv1a(k);
+}
+
+pub fn fnv1a(key: []const u8) u32 {
+    var hash: u32 = 2166136261;
+
+    for (key) |c| {
+        hash ^= c;
+        hash *%= 16777619;
+    }
+
+    return hash;
 }
 
 pub fn xoramasrosas(k: []const u8) u32 {

index.zig

@@ -3,7 +3,6 @@ const storage = @import("storage.zig");
 
 var tree_arena = std.heap.ArenaAllocator.init(std.heap.page_allocator);
 const tree_allocator = tree_arena.allocator();
-
 const temp_allocator = std.heap.c_allocator;
 
 const RadixNode = struct {

main.zig

@@ -6,11 +6,27 @@ const fmt = std.fmt;
 
 const socket = @import("socket.zig");
 const command = @import("command.zig");
+const storage = @import("storage.zig");
 const persistence = @import("persistence.zig");
 
 const PORT = 8085;
+var should_exit = std.atomic.Value(bool).init(false);
+fn handleSignal(sig: c_int) callconv(.c) void {
+    _ = sig;
+    should_exit.store(true, .seq_cst);
+}
 
 pub fn main() !void {
+    const empty_mask = std.mem.zeroes(posix.sigset_t);
+    const act = posix.Sigaction{
+        .handler = .{ .handler = handleSignal },
+        .mask = empty_mask,
+        .flags = 0,
+    };
+
+    _ = posix.sigaction(posix.SIG.TERM, &act, null);
+    _ = posix.sigaction(posix.SIG.INT, &act, null);
+
     const listener = try socket.init(PORT);
     defer posix.close(listener);
 
@@ -18,30 +34,66 @@ pub fn main() !void {
     std.debug.print("Commands:\n\nread key\nwrite key|value\ndelete key\nkeys\nreads prefix\nstatus\n", .{});
     std.debug.print("---------\n", .{});
 
+    storage.init();
     try persistence.init();
 
-    while (true) {
+    while (!should_exit.load(.seq_cst)) {
+        var poll_fds = [_]posix.pollfd{
+            .{
+                .fd = listener,
+                .events = posix.POLL.IN,
+                .revents = 0,
+            },
+        };
+
+        const ready = posix.poll(&poll_fds, 1000) catch |err| {
+            if (should_exit.load(.seq_cst)) break;
+            std.debug.print("poll error: {any}\n", .{err});
+            continue;
+        };
+
+        if (ready == 0) {
+            continue;
+        }
+
+        if (should_exit.load(.seq_cst)) break;
+
         var client_address: net.Address = undefined;
         var client_address_len: posix.socklen_t = @sizeOf(net.Address);
 
         const conn = posix.accept(listener, &client_address.any, &client_address_len, 0) catch |err| {
-            std.debug.print("error accept: {any}", .{err});
+            if (should_exit.load(.seq_cst)) break;
+            std.debug.print("error accept: {any}\n", .{err});
             continue;
         };
 
+        if (should_exit.load(.seq_cst)) {
+            posix.close(conn);
+            break;
+        }
+
+        posix.setsockopt(conn, posix.IPPROTO.TCP, posix.TCP.NODELAY, &std.mem.toBytes(@as(c_int, 1))) catch {};
+
         const thread = try std.Thread.spawn(.{}, handleConnection, .{conn});
         thread.detach();
     }
+
+    std.debug.print("\nShutdown signal received...\n", .{});
+    persistence.flush() catch |err| {
+        std.debug.print("Failed to flush persistence: {any}\n", .{err});
+    };
 }
 
 pub fn handleConnection(conn: posix.socket_t) !void {
     defer posix.close(conn);
 
     var requestBuffer: [1024 * 1024]u8 = undefined;
-    var responseBuffer: [1024 * 1024 * 4]u8 = undefined;
 
     while (true) {
-        const n = try socket.readUntilCR(conn, &requestBuffer);
+        const n = socket.readUntilCR(conn, &requestBuffer) catch |err| {
+            if (err == error.ConnectionClosed) break;
+            return err;
+        };
         if (n == 0) {
             break;
         }
@@ -53,9 +105,12 @@ pub fn handleConnection(conn: posix.socket_t) !void {
             continue;
         };
 
-        @memcpy(responseBuffer[0..cmdResponse.len], cmdResponse);
-        responseBuffer[cmdResponse.len] = '\r';
-        socket.write(conn, responseBuffer[0 .. cmdResponse.len + 1]) catch |err| {
+        const terminator = "\r";
+        const iovecs = [_]posix.iovec_const{
+            .{ .base = cmdResponse.ptr, .len = cmdResponse.len },
+            .{ .base = terminator.ptr, .len = 1 },
+        };
+        socket.writev(conn, &iovecs) catch |err| {
             std.debug.print("error writing: {any}", .{err});
         };
     }

persistence.zig

@@ -2,22 +2,23 @@ const std = @import("std");
 const storage = @import("storage.zig");
 
 const MAX_PERSISTENCE_SIZE = 10_000_000 * 100;
+const BUFFER_SIZE = 1024 * 1024 * 8;
+const FLUSH_THRESHOLD = (BUFFER_SIZE * 3) / 4;
+
 var storage_file: ?std.fs.File = null;
-var thread_pool: std.Thread.Pool = undefined;
-var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator);
-const allocator = arena.allocator();
 const c_allocator = std.heap.c_allocator;
 
 var mutex: std.Thread.Mutex = .{};
 
+var write_buffer: [BUFFER_SIZE]u8 = undefined;
+var buffer_position: usize = 0;
+
 const OPCode = enum {
     W,
     D,
 };
 
 pub fn init() !void {
-    try std.Thread.Pool.init(&thread_pool, .{ .allocator = allocator, .n_jobs = 4 });
-
     const cwd = std.fs.cwd();
     storage_file = cwd.openFile(".db", .{ .mode = .read_write }) catch |err| {
         if (err == std.fs.File.OpenError.FileNotFound) {
@@ -78,21 +79,48 @@ pub fn persist(opcode: u8, key: []const u8, value: []const u8) void {
         std.debug.print("Failed to format record for persistence\n", .{});
         return;
     };
+    defer c_allocator.free(record);
+
+    mutex.lock();
+    defer mutex.unlock();
 
-    thread_pool.spawn(persistRecord, .{record}) catch |err| {
-        std.debug.print("Failed to spawn persistence job: {any}\n", .{err});
+    if (buffer_position + record.len > FLUSH_THRESHOLD) {
+        flushBuffer() catch |err| {
+            std.debug.print("Failed to flush buffer: {any}\n", .{err});
+            return;
+        };
+    }
+
+    if (record.len > BUFFER_SIZE) {
+        _ = storage_file.?.write(record) catch |err| {
+            std.debug.print("Failed to write large record to storage file: {any}\n", .{err});
+            return;
+        };
         return;
-    };
-}
+    }
 
-fn persistRecord(record: []const u8) void {
-    defer c_allocator.free(record);
+    if (buffer_position + record.len > BUFFER_SIZE) {
+        flushBuffer() catch |err| {
+            std.debug.print("Failed to flush buffer: {any}\n", .{err});
+            return;
+        };
+    }
+
+    @memcpy(write_buffer[buffer_position .. buffer_position + record.len], record);
+    buffer_position += record.len;
+}
 
+pub fn flush() !void {
     mutex.lock();
     defer mutex.unlock();
+    try flushBuffer();
+}
 
-    _ = storage_file.?.write(record) catch |err| {
-        std.debug.print("Failed to write record to storage file: {any}\n", .{err});
+fn flushBuffer() !void {
+    if (buffer_position == 0) {
         return;
-    };
+    }
+
+    _ = try storage_file.?.write(write_buffer[0..buffer_position]);
+    buffer_position = 0;
 }

socket.zig

@@ -11,8 +11,10 @@ pub fn init(port: u16) !posix.socket_t {
     const listener = try posix.socket(address.any.family, tpe, protocol);
 
     try posix.setsockopt(listener, posix.SOL.SOCKET, posix.SO.REUSEADDR, &std.mem.toBytes(@as(c_int, 1)));
+    try posix.setsockopt(listener, posix.SOL.SOCKET, posix.SO.RCVBUF, &std.mem.toBytes(@as(c_int, 1048576))); // 1MB receive buffer
+    try posix.setsockopt(listener, posix.SOL.SOCKET, posix.SO.SNDBUF, &std.mem.toBytes(@as(c_int, 1048576))); // 1MB send buffer
     try posix.bind(listener, &address.any, address.getOsSockLen());
-    try posix.listen(listener, 128);
+    try posix.listen(listener, 1024); // Increased backlog
 
     return listener;
 }
@@ -54,3 +56,15 @@ pub fn write(conn: posix.socket_t, msg: []const u8) !void {
         return error.PartialWrite;
     }
 }
+
+pub fn writev(conn: posix.socket_t, iovecs: []const posix.iovec_const) !void {
+    var total: usize = 0;
+    for (iovecs) |iov| {
+        total += iov.len;
+    }
+
+    const written = try posix.writev(conn, iovecs);
+    if (written != total) {
+        return error.PartialWrite;
+    }
+}

storage.zig

@@ -4,27 +4,67 @@ const index = @import("index.zig");
 const hashing = @import("hashing.zig");
 const persistence = @import("persistence.zig");
 
-const MAX_RECORDS = 10_000_000;
+const INITIAL_BUCKETS = 1_048_576;
 const Entry = struct {
     key: []const u8,
     value: []const u8,
+    hash: u32,
     next: ?*Entry,
 };
-var buf: [MAX_RECORDS]?*Entry = undefined;
+
+var buckets: []?*Entry = undefined;
+var buckets_initialized: bool = false;
 
 var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator);
 const allocator = arena.allocator();
 
-const EMPTY = "";
-var mutex: std.Thread.Mutex = .{};
+var rwlock: std.Thread.RwLock = .{};
+var init_mutex: std.Thread.Mutex = .{};
+
+pub fn init() void {
+    if (buckets_initialized) return;
+
+    init_mutex.lock();
+    defer init_mutex.unlock();
+
+    if (!buckets_initialized) {
+        buckets = allocator.alloc(?*Entry, INITIAL_BUCKETS) catch unreachable;
+        @memset(buckets, null);
+        buckets_initialized = true;
+    }
+}
+
+pub fn restore(key: []const u8, value: []const u8) bool {
+    rwlock.lock();
+    defer rwlock.unlock();
+
+    const entry = writeVolatile(key, value);
+    if (entry != null) {
+        index.insert(entry.?.key);
+        return true;
+    }
+    return false;
+}
+
+pub fn restoreDelete(key: []const u8) bool {
+    rwlock.lock();
+    defer rwlock.unlock();
+
+    const deleted = deleteVolatile(key);
+    if (deleted) {
+        index.delete(key);
+        return true;
+    }
+    return false;
+}
 
 pub fn writeVolatile(key: []const u8, value: []const u8) ?*Entry {
     const hash = hashing.hashKey(key);
-    const bufIdx = hash % buf.len;
+    const bucketIdx = hash % buckets.len;
 
-    var current = buf[bufIdx];
+    var current = buckets[bucketIdx];
     while (current) |entry| {
-        if (std.mem.eql(u8, entry.key, key)) {
+        if (entry.hash == hash and std.mem.eql(u8, entry.key, key)) {
             allocator.free(entry.value);
             entry.value = allocator.dupe(u8, value) catch return null;
             return entry;
@@ -38,35 +78,38 @@ pub fn writeVolatile(key: []const u8, value: []const u8) ?*Entry {
     newEntry.* = Entry{
         .key = allocator.dupe(u8, key) catch return null,
         .value = allocator.dupe(u8, value) catch return null,
-        .next = buf[bufIdx],
+        .hash = hash, // Cache hash value
+        .next = buckets[bucketIdx],
     };
 
-    buf[bufIdx] = newEntry;
-    index.insert(key);
+    buckets[bucketIdx] = newEntry;
     return newEntry;
 }
 
 pub fn write(key: []const u8, value: []const u8) bool {
-    mutex.lock();
-    defer mutex.unlock();
+    rwlock.lock();
+    defer rwlock.unlock();
 
     const entry = writeVolatile(key, value);
     if (entry == null) {
         return false;
     }
 
+    index.insert(entry.?.key);
     persistence.persist('W', entry.?.key, entry.?.value);
     return true;
 }
 
 pub fn read(key: []const u8) ?[]const u8 {
-    mutex.lock();
-    defer mutex.unlock();
+    rwlock.lockShared();
+    defer rwlock.unlockShared();
+
+    if (!buckets_initialized) return null;
 
     const hash = hashing.hashKey(key);
-    var current = buf[hash % buf.len];
+    var current = buckets[hash % buckets.len];
     while (current) |entry| {
-        if (std.mem.eql(u8, entry.key, key)) {
+        if (entry.hash == hash and std.mem.eql(u8, entry.key, key)) {
             return entry.value;
         }
         current = entry.next;
@@ -76,21 +119,21 @@ pub fn read(key: []const u8) ?[]const u8 {
 }
 
 pub fn deleteVolatile(key: []const u8) bool {
+    if (!buckets_initialized) return false;
     const hash = hashing.hashKey(key);
-    const bufIdx = hash % buf.len;
+    const bucketIdx = hash % buckets.len;
 
-    var current = buf[bufIdx];
+    var current = buckets[bucketIdx];
     var prev: ?*Entry = null;
 
     while (current) |entry| {
-        if (std.mem.eql(u8, entry.key, key)) {
+        if (entry.hash == hash and std.mem.eql(u8, entry.key, key)) {
             if (prev) |p| {
                 p.next = entry.next;
             } else {
-                buf[bufIdx] = entry.next;
+                buckets[bucketIdx] = entry.next;
             }
 
-            index.delete(key);
             allocator.free(entry.key);
             allocator.free(entry.value);
             allocator.destroy(entry);
@@ -105,11 +148,12 @@ pub fn deleteVolatile(key: []const u8) bool {
 }
 
 pub fn delete(key: []const u8) bool {
-    mutex.lock();
-    defer mutex.unlock();
+    rwlock.lock();
+    defer rwlock.unlock();
 
     const deleted = deleteVolatile(key);
     if (deleted) {
+        index.delete(key);
         persistence.persist('D', key, "");
     }