diff --git a/src/concurrent.zig b/src/concurrent.zig index 8173d64..7e25402 100644 --- a/src/concurrent.zig +++ b/src/concurrent.zig @@ -312,3 +312,126 @@ test "apply is usable concurrently: disjoint batches from different threads neve prefix: u8, count: usize, + fn run(self: @This()) void { + var batch = WriteBatch.init(std.heap.page_allocator); + defer batch.deinit(); + var i: usize = 0; + while (i < self.count) : (i += 1) { + const key = [_]u8{ self.prefix, @intCast(i) }; + batch.put(&key, "v") catch @panic("batch put failed"); + } + self.guarded.apply(&batch) catch |err| { + std.debug.panic("apply failed under contention: {s}", .{@errorName(err)}); + }; + } + }; + + var threads: [4]std.Thread = undefined; + for (&threads, 0..) |*t, idx| { + t.* = try std.Thread.spawn(.{}, ApplyWorker.run, .{ApplyWorker{ + .guarded = &guarded, + .prefix = @intCast('a' + idx), + .count = 50, + }}); + } + for (threads) |t| t.join(); + + inline for (.{ 'a', 'b', 'c', 'd' }) |prefix| { + var iter = try guarded.db.scanPrefix(allocator, &[_]u8{prefix}); + defer iter.deinit(); + var n: usize = 0; + while (try iter.next()) |_| n += 1; + try std.testing.expectEqual(@as(usize, 50), n); + } +} + +const TableTransferWorker = struct { + guarded: *GuardedDb, + rounds: usize, + + /// Runs a full `Transaction` + `TableTxn` transfer lifecycle (begin, + /// read both rows, stage both writes, commit) under a single + /// `acquire()`/`release()` pair, closing the specific gap every prior + /// session on this project flagged and deferred: "brak testu + /// wspolbieznosci wielowatkowej (dwa watki komituja TableTxn na tym + /// samym Db jednoczesnie)". `Db.get`/`Transaction.commit` touch shared + /// mutable state (the memtable, the OCC read-set bookkeeping) with no + /// locking of their own, so without the guard two real OS threads + /// racing through `TableTxn.open`/`get`/`put`/`commit` at once would + /// be exactly the kind of unguarded concurrent access every other + /// module in this project takes its own mutex to avoid. + fn run(self: TableTransferWorker) 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(); + + var txn = Transaction.begin(g.db); + defer txn.deinit(); + var t = TableTxn.open(&txn, allocator, "accounts") catch @panic("TableTxn.open failed"); + defer t.deinit(); + + var from = (t.get(.{ .int = 1 }) catch @panic("get 1 failed")).?; + defer from.deinit(); + var to = (t.get(.{ .int = 2 }) catch @panic("get 2 failed")).?; + defer to.deinit(); + if (from.get("balance").?.int < 1) continue; + + t.put(.{ .int = 1 }, &.{.{ .name = "balance", .value = .{ .int = from.get("balance").?.int - 1 } }}) catch @panic("put 1 failed"); + t.put(.{ .int = 2 }, &.{.{ .name = "balance", .value = .{ .int = to.get("balance").?.int + 1 } }}) catch @panic("put 2 failed"); + + // The whole sequence runs under one lock, so no external writer + // can ever appear mid-transfer here: commit() must never see a + // real error.Conflict in this test, unlike the deliberately + // racy scenario in txn_table.zig's own "all-or-nothing" test. + txn.commit() catch |err| std.debug.panic("unexpected commit failure under serialized access: {s}", .{@errorName(err)}); + } + } +}; + +test "TableTxn commits from many real OS threads never lose or duplicate a unit of balance" { + const allocator = std.testing.allocator; + var tmp = std.testing.tmpDir(.{}); + defer tmp.cleanup(); + var db = try openTemp(&tmp, allocator); + defer db.close(); + + const startBalance: i64 = 2000; + { + var setup = try querymod.Table.open(&db, allocator, "accounts"); + defer setup.deinit(); + try setup.put(.{ .int = 1 }, &.{.{ .name = "balance", .value = .{ .int = startBalance } }}); + try setup.put(.{ .int = 2 }, &.{.{ .name = "balance", .value = .{ .int = 0 } }}); + } + + var guarded = GuardedDb.init(&db); + const threadCount = 4; + const roundsPerThread = 200; + var threads: [threadCount]std.Thread = undefined; + for (&threads) |*t| { + t.* = try std.Thread.spawn(.{}, TableTransferWorker.run, .{TableTransferWorker{ + .guarded = &guarded, + .rounds = roundsPerThread, + }}); + } + for (threads) |t| t.join(); + + var verify = try querymod.Table.open(&db, allocator, "accounts"); + defer verify.deinit(); + var row1 = (try verify.get(.{ .int = 1 })).?; + defer row1.deinit(); + var row2 = (try verify.get(.{ .int = 2 })).?; + defer row2.deinit(); + const balance1 = row1.get("balance").?.int; + const balance2 = row2.get("balance").?.int; + + // Every transfer moves exactly 1 unit and every transfer's two writes + // commit together or not at all, so the total must be exactly + // preserved and account 2 must have gained exactly what account 1 + // lost - no interleaving of `get`+`put`+`commit` across threads can + // produce any other pair of numbers. + try std.testing.expectEqual(startBalance, balance1 + balance2); + try std.testing.expectEqual(@as(i64, threadCount * roundsPerThread), balance2); + try std.testing.expectEqual(startBalance - threadCount * roundsPerThread, balance1); +}