diff --git a/src/db.zig b/src/db.zig index 74dfb8e..ffffd14 100644 --- a/src/db.zig +++ b/src/db.zig @@ -552,3 +552,71 @@ test "compaction rewrites log and keeps data" { } engine3.unlock(); } + +test "concurrent readers and writers on a threaded Io" { + // Real worker threads: writers hold the exclusive lock, readers the + // shared lock. Proves the RwLock split keeps committed writes visible + // to concurrent readers and never corrupts the maps. + const gpa = testing.allocator; + var threaded: std.Io.Threaded = .init(gpa, .{}); + defer threaded.deinit(); + const io = threaded.io(); + + var tmp = try TmpLog.init(gpa); + defer tmp.deinit(gpa); + var engine = try Engine.open(gpa, io, tmp.path); + defer engine.deinit(); + + const writers = 4; + const readers = 4; + const per_writer: i32 = 200; + const total: i32 = writers * per_writer; + var next_id = std.atomic.Value(i32).init(1); + var remaining = std.atomic.Value(usize).init(@intCast(total)); + + const Worker = struct { + fn writer(e: *Engine, id_counter: *std.atomic.Value(i32), pending: *std.atomic.Value(usize), alloc: std.mem.Allocator) error{Canceled}!void { + while (true) { + const id = id_counter.fetchAdd(1, .monotonic); + if (id > total) return; + var doc = make_doc(alloc, id, "user") catch return error.Canceled; + defer doc.deinit(); + e.lock() catch return error.Canceled; + defer e.unlock(); + e.insert("app", "users", &doc, undefined) catch return error.Canceled; + _ = pending.fetchSub(1, .monotonic); + } + } + + fn reader(e: *Engine, pending: *std.atomic.Value(usize)) error{Canceled}!void { + while (pending.load(.acquire) > 0) { + e.lock_read() catch return error.Canceled; + defer e.unlock_read(); + if (e.get_collection("app", "users")) |coll| { + var n: usize = 0; + var it = coll.docs.iterator(); + while (it.next()) |_| n += 1; + // A reader must never observe more docs than can exist. + if (n > @as(usize, @intCast(total))) return error.Canceled; + } + } + } + }; + + var group: std.Io.Group = .init; + defer group.cancel(io); + for (0..readers) |_| group.async(io, Worker.reader, .{ &engine, &remaining }); + for (0..writers) |_| group.async(io, Worker.writer, .{ &engine, &next_id, &remaining, gpa }); + try group.await(io); + + // Every committed write must be visible once all writers finish. + try engine.lock_read(); + defer engine.unlock_read(); + const coll = engine.get_collection("app", "users") orelse return error.TestUnexpectedResult; + try testing.expectEqual(@as(usize, @intCast(total)), coll.docs.count()); + for (1..total + 1) |i| { + const id_key = try bson.serialize_value(gpa, bson.Value{ .int32 = @intCast(i) }); + defer gpa.free(id_key); + try testing.expect(engine.get_doc("app", "users", id_key) != null); + } +}