From 14dfdcf57ad6dab33a091cc404b426c6816cd06d Mon Sep 17 00:00:00 2001 From: DualFroz Date: Wed, 2 Sep 2026 18:18:03 +0200 Subject: [PATCH] feat: add GuardedDb and Guard --- src/concurrent.zig | 157 +++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 157 insertions(+) create mode 100644 src/concurrent.zig diff --git a/src/concurrent.zig b/src/concurrent.zig new file mode 100644 index 0000000..beed0e8 --- /dev/null +++ b/src/concurrent.zig @@ -0,0 +1,157 @@ +const std = @import("std"); +const dbmod = @import("db.zig"); +const batchmod = @import("batch.zig"); +const txnmod = @import("txn.zig"); +const txntable = @import("txn_table.zig"); +const querymod = @import("query.zig"); + +const Allocator = std.mem.Allocator; +const Db = dbmod.Db; +const Transaction = txnmod.Transaction; +const TableTxn = txntable.TableTxn; +const WriteBatch = batchmod.WriteBatch; + +/// A thread-safe facade over `*Db`. `Db` itself keeps no internal locking - +/// a fact documented (and worked around ad hoc) at every call site that +/// already needs concurrent access from more than one OS thread: +/// server.zig takes its own `mutex` before touching `db` on every request, +/// and every session note about txn_table.zig/query.zig/collections.zig +/// repeats the same known gap for the same reason. `GuardedDb` is that +/// exact pattern promoted to one reusable, independently tested primitive +/// instead of being reimplemented by hand wherever it is next needed. +/// +/// This does not make `Db` itself thread-safe: it makes "every access to +/// this particular `Db` goes through this particular lock" enforceable, +/// which is the same guarantee `Server` already relies on. Two independent +/// `GuardedDb` wrapping the same underlying `*Db` would NOT be safe - the +/// mutex only serializes callers that agree to go through it. +pub const GuardedDb = struct { + db: *Db, + mutex: std.Thread.Mutex = .{}, + + pub fn init(db: *Db) GuardedDb { + return .{ .db = db }; + } + + /// Exclusive handle returned by `acquire()`. Holds the lock until + /// `release()` runs, so a caller that needs several `Db` calls to + /// happen as one atomic unit (a read-modify-write, or several puts + /// that must never be observed half-applied by another guarded + /// reader) can acquire once and issue every call through `.db` + /// without releasing in between - the same shape as + /// `Server.dispatch` locking `self.mutex` once per request no matter + /// how many `Db` calls that request needs. + pub const Guard = struct { + db: *Db, + mutex: *std.Thread.Mutex, + + pub fn release(self: Guard) void { + self.mutex.unlock(); + } + }; + + pub fn acquire(self: *GuardedDb) Guard { + self.mutex.lock(); + return .{ .db = self.db, .mutex = &self.mutex }; + } + + pub fn put(self: *GuardedDb, key: []const u8, value: []const u8) !void { + const g = self.acquire(); + defer g.release(); + try g.db.put(key, value); + } + + pub fn delete(self: *GuardedDb, key: []const u8) !void { + const g = self.acquire(); + defer g.release(); + try g.db.delete(key); + } + + pub fn get(self: *GuardedDb, allocator: Allocator, key: []const u8) !?[]u8 { + const g = self.acquire(); + defer g.release(); + return g.db.get(allocator, key); + } + + pub fn apply(self: *GuardedDb, batch: *WriteBatch) !void { + const g = self.acquire(); + defer g.release(); + try g.db.apply(batch); + } + + pub fn flush(self: *GuardedDb) !void { + const g = self.acquire(); + defer g.release(); + try g.db.flush(); + } + + pub fn compact(self: *GuardedDb) !void { + const g = self.acquire(); + defer g.release(); + try g.db.compact(); + } + + /// Atomic read-modify-write counter: decodes the current value at + /// `key` as a little-endian i64 (0 if absent or the wrong width), + /// adds `delta`, writes the new value back - the whole read, decide, + /// write sequence runs under one lock acquisition, so no other + /// guarded caller can observe or clobber the value partway through. + /// A bare `db.get` followed later by a separate `db.put` cannot make + /// that guarantee, which is exactly the lost-update hazard this + /// exists to close (see the test below that proves it with real + /// concurrent OS threads, not a simulated interleaving). + pub fn incrBy(self: *GuardedDb, allocator: Allocator, key: []const u8, delta: i64) !i64 { + const g = self.acquire(); + defer g.release(); + var current: i64 = 0; + if (try g.db.get(allocator, key)) |raw| { + defer allocator.free(raw); + if (raw.len == 8) current = std.mem.readInt(i64, raw[0..8], .little); + } + const next = current + delta; + var buf: [8]u8 = undefined; + std.mem.writeInt(i64, &buf, next, .little); + try g.db.put(key, &buf); + return next; + } +}; + +fn openTemp(tmp: *std.testing.TmpDir, allocator: Allocator) !Db { + const dbPath = try tmp.dir.realpathAlloc(allocator, "."); + defer allocator.free(dbPath); + return Db.open(allocator, dbPath, .{}); +} + +test "put/get/delete round-trip through the guard like a plain Db call" { + const allocator = std.testing.allocator; + var tmp = std.testing.tmpDir(.{}); + defer tmp.cleanup(); + var db = try openTemp(&tmp, allocator); + defer db.close(); + var guarded = GuardedDb.init(&db); + + try guarded.put("k", "v1"); + { + const got = try guarded.get(allocator, "k"); + defer if (got) |g| allocator.free(g); + try std.testing.expectEqualStrings("v1", got.?); + } + + try guarded.delete("k"); + const afterDelete = try guarded.get(allocator, "k"); + try std.testing.expectEqual(@as(?[]u8, null), afterDelete); +} + +test "incrBy on a fresh key starts from zero and returns the running total" { + const allocator = std.testing.allocator; + var tmp = std.testing.tmpDir(.{}); + defer tmp.cleanup(); + var db = try openTemp(&tmp, allocator); + defer db.close(); + var guarded = GuardedDb.init(&db); + + try std.testing.expectEqual(@as(i64, 5), try guarded.incrBy(allocator, "ctr", 5)); + try std.testing.expectEqual(@as(i64, 3), try guarded.incrBy(allocator, "ctr", -2)); + try std.testing.expectEqual(@as(i64, 103), try guarded.incrBy(allocator, "ctr", 100)); +} +