diff --git a/src/concurrent.zig b/src/concurrent.zig index beed0e8..8173d64 100644 --- a/src/concurrent.zig +++ b/src/concurrent.zig @@ -155,3 +155,160 @@ test "incrBy on a fresh key starts from zero and returns the running total" { try std.testing.expectEqual(@as(i64, 103), try guarded.incrBy(allocator, "ctr", 100)); } +const IncrWorker = struct { + guarded: *GuardedDb, + key: []const u8, + iterations: usize, + + fn run(self: IncrWorker) void { + const allocator = std.heap.page_allocator; + var i: usize = 0; + while (i < self.iterations) : (i += 1) { + _ = self.guarded.incrBy(allocator, self.key, 1) catch |err| { + std.debug.panic("incrBy failed under real thread contention: {s}", .{@errorName(err)}); + }; + } + } +}; + +test "incrBy survives real concurrent OS threads with zero lost updates" { + 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); + + const threadCount = 6; + const perThread = 300; + var threads: [threadCount]std.Thread = undefined; + for (&threads) |*t| { + t.* = try std.Thread.spawn(.{}, IncrWorker.run, .{IncrWorker{ + .guarded = &guarded, + .key = "shared-counter", + .iterations = perThread, + }}); + } + for (threads) |t| t.join(); + + const raw = try guarded.get(allocator, "shared-counter"); + defer allocator.free(raw.?); + const total = std.mem.readInt(i64, raw.?[0..8], .little); + // A bare unsynchronized `db.get` + `db.put` read-modify-write would + // routinely lose updates here (two threads read the same stale value, + // both write back the same next value, one increment vanishes). The + // mutex inside GuardedDb makes every incrBy fully serialized, so the + // total below must land exactly on threadCount*perThread every single + // run - not "usually", not "close to" - or the guard has a real bug. + try std.testing.expectEqual(@as(i64, threadCount * perThread), total); +} + +const TransferWorker = struct { + guarded: *GuardedDb, + rounds: usize, + + fn run(self: TransferWorker) void { + const allocator = std.heap.page_allocator; + var i: usize = 0; + while (i < self.rounds) : (i += 1) { + const g = self.guarded.acquire(); + defer g.release(); + const aRaw = g.db.get(allocator, "balanceA") catch @panic("get a failed"); + defer allocator.free(aRaw.?); + const bRaw = g.db.get(allocator, "balanceB") catch @panic("get b failed"); + defer allocator.free(bRaw.?); + const a = std.mem.readInt(i64, aRaw.?[0..8], .little); + const b = std.mem.readInt(i64, bRaw.?[0..8], .little); + if (a < 1) continue; + var aBuf: [8]u8 = undefined; + var bBuf: [8]u8 = undefined; + std.mem.writeInt(i64, &aBuf, a - 1, .little); + std.mem.writeInt(i64, &bBuf, b + 1, .little); + g.db.put("balanceA", &aBuf) catch @panic("put a failed"); + g.db.put("balanceB", &bBuf) catch @panic("put b failed"); + } + } +}; + +const InvariantReader = struct { + guarded: *GuardedDb, + checks: usize, + violations: *std.atomic.Value(usize), + + fn run(self: InvariantReader) void { + const allocator = std.heap.page_allocator; + var i: usize = 0; + while (i < self.checks) : (i += 1) { + const g = self.guarded.acquire(); + const aRaw = g.db.get(allocator, "balanceA") catch @panic("get a failed"); + const bRaw = g.db.get(allocator, "balanceB") catch @panic("get b failed"); + const a = std.mem.readInt(i64, aRaw.?[0..8], .little); + const b = std.mem.readInt(i64, bRaw.?[0..8], .little); + g.release(); + allocator.free(aRaw.?); + allocator.free(bRaw.?); + if (a + b != 1000) _ = self.violations.fetchAdd(1, .monotonic); + } + } +}; + +test "acquire lets a writer move value between two keys atomically under a concurrent reader" { + 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); + + var initA: [8]u8 = undefined; + var initB: [8]u8 = undefined; + std.mem.writeInt(i64, &initA, 1000, .little); + std.mem.writeInt(i64, &initB, 0, .little); + try guarded.put("balanceA", &initA); + try guarded.put("balanceB", &initB); + + var violations = std.atomic.Value(usize).init(0); + const writerThread = try std.Thread.spawn(.{}, TransferWorker.run, .{TransferWorker{ + .guarded = &guarded, + .rounds = 1000, + }}); + const readerThread = try std.Thread.spawn(.{}, InvariantReader.run, .{InvariantReader{ + .guarded = &guarded, + .checks = 1000, + .violations = &violations, + }}); + writerThread.join(); + readerThread.join(); + + // Every single reader observation is required to see the invariant + // balanceA + balanceB == 1000 hold, because each transfer's two puts + // happen inside one `acquire()`/`release()` pair the reader's own + // acquire() must wait out entirely - a real race would show up here + // as a nonzero violation count, not as a crash, which is why this + // asserts the counter is exactly zero instead of merely "did not hang". + try std.testing.expectEqual(@as(usize, 0), violations.load(.monotonic)); + + const finalARaw = try guarded.get(allocator, "balanceA"); + defer allocator.free(finalARaw.?); + const finalBRaw = try guarded.get(allocator, "balanceB"); + defer allocator.free(finalBRaw.?); + const finalA = std.mem.readInt(i64, finalARaw.?[0..8], .little); + const finalB = std.mem.readInt(i64, finalBRaw.?[0..8], .little); + try std.testing.expectEqual(@as(i64, 1000), finalA + finalB); + try std.testing.expectEqual(@as(i64, 0), finalA); + try std.testing.expectEqual(@as(i64, 1000), finalB); +} + +test "apply is usable concurrently: disjoint batches from different threads never clobber each other" { + 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); + + const ApplyWorker = struct { + guarded: *GuardedDb, + prefix: u8, + count: usize, +