diff --git a/src/db.zig b/src/db.zig index 84b1cce..3d7929d 100644 --- a/src/db.zig +++ b/src/db.zig @@ -50,6 +50,10 @@ pub const Collection = struct { /// rebuild. `slab_tail` cannot answer that -- it is an absolute file offset, /// so it jumps forward whenever a fresh extent is taken. slab_used: u64, + /// This collection's outstanding page promise, for the document slab. Per + /// collection because concurrent writers must not release each other's -- + /// see `pager.Reservation`. + hold: pgr.Reservation, /// Of those bytes, the ones still reachable. `slab_used - live_bytes` is /// this collection's slab garbage, which only a rebuild reclaims. Kept per /// collection so dropping one can move the right amount from the engine's @@ -91,6 +95,7 @@ pub const Collection = struct { .slab_end = 0, .slab_used = 0, .live_bytes = 0, + .hold = .{}, .indexes = .empty, .id_index = undefined, }; @@ -140,8 +145,8 @@ pub const Collection = struct { slab_extent_pages, (len + pgr.page_size - 1) / pgr.page_size, )); - try self.pager.reserve_pages(want_pages); - const first = self.pager.alloc_pages_assume_reserved(want_pages); + try self.pager.reserve_pages(&self.hold, want_pages); + const first = self.pager.alloc_pages_assume_reserved(&self.hold, want_pages); try self.slab_extents.append(gpa, .{ .first = first, .pages = want_pages }); self.slab_tail = @as(u64, first) << pgr.page_shift; self.slab_end = self.slab_tail + (@as(u64, want_pages) << pgr.page_shift); @@ -261,6 +266,10 @@ pub const Engine = struct { /// 16 KiB documents and one of 40 B documents look identical. live_bytes: u64 = 0, dead_bytes: u64 = 0, + /// The checkpoint's own page promise, for the catalog and free-list pages it + /// writes. Separate from any collection's for the same reason those are + /// separate from each other. + hold: pgr.Reservation = .{}, /// Set when the log has grown enough since the last checkpoint to be worth /// reclaiming. Read by the write epilogue and the TTL monitor, both of which /// run without holding a collection lock. @@ -420,6 +429,15 @@ pub const Engine = struct { /// own total (a rebuild appends through it too, and its copies are live by /// definition); the engine's total only moves when a document actually /// becomes live, which a rebuild's copies do not. + /// Drop the unclaimed part of every promise a write to this collection took: + /// the slab's and one per index. Called after the write is published, under + /// the same collection lock the reservations were taken under. + fn release_write_reservations(self: *Engine, coll: *Collection) void { + self.pager.release_reservation(&coll.hold); + self.pager.release_reservation(&coll.id_index.hold); + for (coll.indexes.items) |ix| self.pager.release_reservation(&ix.hold); + } + fn publish_doc_bytes(self: *Engine, coll: *Collection, bytes: []const u8) u64 { const off = coll.slab_append(bytes); self.live_bytes += bytes.len; @@ -794,8 +812,10 @@ pub const Engine = struct { b.ix.insert_entries(&b.built, off); } // The write is published; anything the reservations above did not claim - // is dead. Leaving it promised would grow the file on every write. - self.pager.release_reservation(); + // is dead. Leaving it promised would grow the file on every write. Every + // consumer this upsert reserved through, and only those: another + // collection may be mid-write on another thread. + self.release_write_reservations(coll); self.note_compact(); self.note_checkpoint(); } @@ -919,7 +939,7 @@ pub const Engine = struct { try ix.append_doc_entries(self.gpa, coll.doc_bytes(entry.off), entry.off); } _ = try ix.finish_bulk(self.gpa, true); - self.pager.release_reservation(); + self.pager.release_reservation(&ix.hold); // Reserve the collection slot, then persist and publish. try coll.indexes.ensureUnusedCapacity(self.gpa, 1); @@ -1284,7 +1304,7 @@ pub const Engine = struct { const bytes = doc_bytes_in(self.pager, entry.off); try coll.slab_reserve(self.gpa, bytes.len); const new_off = coll.slab_append(bytes); - self.pager.release_reservation(); + self.pager.release_reservation(&coll.hold); try moved.append(self.gpa, .{ .off = new_off }); } // Republish the offsets. @@ -1313,7 +1333,7 @@ pub const Engine = struct { // Duplicates are tolerated here for the same reason they are on open: // refusing would make a maintenance task able to take the database down. _ = ix.finish_bulk(self.gpa, false) catch |err| return err; - self.pager.release_reservation(); + self.pager.release_reservation(&ix.hold); } /// Re-emit one collection's index specs and documents into the compacted @@ -1367,7 +1387,7 @@ pub const Engine = struct { }; } // Tolerated, not enforced: the database must always open. - defer self.pager.release_reservation(); + defer self.pager.release_reservation(&ix.hold); if (try ix.finish_bulk(self.gpa, false)) { std.debug.print( "multiforadb: WARNING: unique index '{s}' has duplicate keys in existing " ++ @@ -1710,7 +1730,7 @@ pub const Engine = struct { self.log_lock.unlock(self.io); return err; }; - self.pager.release_reservation(); + self.pager.release_reservation(&self.hold); // The watermark is durable, so every record it covers is now // redundant. Strictly after the publish: the other order loses data // if a crash lands between them. @@ -1844,7 +1864,7 @@ fn apply_record(ctx: *anyopaque, record: storage.Record, doc: *bson.Document) an } else { self.index_one(&coll.id_index, doc_bytes, off); } - self.pager.release_reservation(); + self.release_write_reservations(coll); // The _id_ entry is added after replay, in build_all_indexes, // together with the secondary indexes. // diff --git a/src/index.zig b/src/index.zig index 4de40bf..c143cf2 100644 --- a/src/index.zig +++ b/src/index.zig @@ -217,6 +217,10 @@ pub const Index = struct { /// `Slot.off` means two different things depending on `spill` is unchanged, /// only the second meaning moved. ovf_extents: std.ArrayListUnmanaged(pgr.Extent), + /// This index's outstanding page promise. Per index rather than per pager + /// because two collections are written concurrently and each one's promise + /// has to survive the other's release -- see `pager.Reservation`. + hold: pgr.Reservation, ovf_tail: u64, ovf_end: u64, /// Bulk-build staging: entries appended unsorted by append_doc_entries, @@ -255,6 +259,7 @@ pub const Index = struct { .pager = pager, .node_pages = .empty, .ovf_extents = .empty, + .hold = .{}, .ovf_tail = 0, .ovf_end = 0, .staging = .empty, @@ -281,9 +286,9 @@ pub const Index = struct { // Slot 0 is a dummy (0 is the null node id); the root is one empty // leaf, so a fresh index is always a valid tree. try self.node_pages.ensureUnusedCapacity(gpa, 2); - try pager.reserve_pages(2); - self.node_pages.appendAssumeCapacity(pager.alloc_pages_assume_reserved(1)); - self.node_pages.appendAssumeCapacity(pager.alloc_pages_assume_reserved(1)); + try pager.reserve_pages(&self.hold, 2); + self.node_pages.appendAssumeCapacity(pager.alloc_pages_assume_reserved(&self.hold, 1)); + self.node_pages.appendAssumeCapacity(pager.alloc_pages_assume_reserved(&self.hold, 1)); self.page_mut(0).* = empty_node(0); self.page_mut(1).* = empty_node(1); self.root = 1; @@ -435,7 +440,7 @@ pub const Index = struct { // a slot in the id->page table, and the insertion after the log append // must not be able to fail on either. try self.node_pages.ensureUnusedCapacity(gpa, @intCast(extra_nodes)); - try self.pager.reserve_pages(@intCast(extra_nodes)); + try self.pager.reserve_pages(&self.hold, @intCast(extra_nodes)); try self.reserve_overflow(gpa, entries); } @@ -467,8 +472,8 @@ pub const Index = struct { ovf_extent_pages, (overflow_bytes + pgr.page_size - 1) / pgr.page_size, )); - try self.pager.reserve_pages(want_pages); - const first = self.pager.alloc_pages_assume_reserved(want_pages); + try self.pager.reserve_pages(&self.hold, want_pages); + const first = self.pager.alloc_pages_assume_reserved(&self.hold, want_pages); try self.ovf_extents.append(gpa, .{ .first = first, .pages = want_pages }); self.ovf_tail = @as(u64, first) << pgr.page_shift; self.ovf_end = self.ovf_tail + (@as(u64, want_pages) << pgr.page_shift); @@ -593,9 +598,9 @@ pub const Index = struct { self.ovf_tail = 0; self.ovf_end = 0; try self.node_pages.ensureUnusedCapacity(gpa, 2); - try self.pager.reserve_pages(2); - self.node_pages.appendAssumeCapacity(self.pager.alloc_pages_assume_reserved(1)); - self.node_pages.appendAssumeCapacity(self.pager.alloc_pages_assume_reserved(1)); + try self.pager.reserve_pages(&self.hold, 2); + self.node_pages.appendAssumeCapacity(self.pager.alloc_pages_assume_reserved(&self.hold, 1)); + self.node_pages.appendAssumeCapacity(self.pager.alloc_pages_assume_reserved(&self.hold, 1)); self.page_mut(0).* = empty_node(0); self.page_mut(1).* = empty_node(1); self.root = 1; @@ -952,7 +957,7 @@ pub const Index = struct { /// honest response. fn alloc_node(self: *Index) u32 { assert_msg(self.node_pages.items.len < self.node_pages.capacity, "node allocation overran reserve_for's bound"); - const p = self.pager.alloc_pages_assume_reserved(1); + const p = self.pager.alloc_pages_assume_reserved(&self.hold, 1); self.node_pages.appendAssumeCapacity(p); const id: u32 = @intCast(self.node_pages.items.len - 1); self.page_mut(id).* = empty_node(0); diff --git a/src/pager.zig b/src/pager.zig index 0d9309a..9c35d3d 100644 --- a/src/pager.zig +++ b/src/pager.zig @@ -169,6 +169,26 @@ pub const OpenOptions = struct { reserve_bytes: usize = default_reserve_bytes, }; +/// One consumer's outstanding promise, in pages. +/// +/// The promise used to be a single counter on the pager, and that is not +/// something concurrent writers can share. Two upserts on different +/// collections run at the same time (they hold different collection locks), +/// and each one ends by dropping "whatever is still promised" -- so the first +/// to finish zeroed the second's promise, and the second's supposedly +/// infallible allocation then tripped its own tripwire: +/// +/// assertion failed: page allocation overran reserve_pages' promise +/// src/index.zig:955 in alloc_node +/// +/// Reliably, at four concurrent clients, on the first benchmark run after the +/// data file landed (PLAN risk 3). Each consumer -- every index, every +/// collection's slab, the checkpoint -- now holds its own, and only ever +/// releases its own. The pager keeps the sum, which is all `grow_to` needs. +pub const Reservation = struct { + pages: u32 = 0, +}; + pub const Pager = struct { gpa: std.mem.Allocator, io: std.Io, @@ -196,6 +216,15 @@ pub const Pager = struct { /// first one's allocation asserts. Which is exactly what happened, on a /// 512 MB load, with the tripwire in alloc_node catching it. reserved_pages: u32, + /// Serialises the allocator's bookkeeping: the tail, the reservation total, + /// the free lists, the unpublished set and file growth. Writers on different + /// collections hold different collection locks and allocate from this one + /// pager, so none of that can be a plain field (PLAN risk 3). + /// + /// Taken uncancelable: the critical section is bookkeeping that leaves the + /// allocator inconsistent if abandoned half way, and it is never held across + /// the log append -- that is what per-consumer reservations buy. + alloc_lock: std.Io.Mutex, /// True when this file was created by this open (no checkpoint to load). fresh: bool, @@ -297,6 +326,7 @@ pub const Pager = struct { .file_pages = @intCast(existing_len / page_size), .alloc_tail = page_first_data, .reserved_pages = 0, + .alloc_lock = .init, .fresh = created, .loaded = .{}, .generation = 0, @@ -444,14 +474,17 @@ pub const Pager = struct { // -- allocation --------------------------------------------------------- - /// Hand out `n` contiguous pages, growing the file if needed. + /// Hand out `n` contiguous pages, growing the file if needed. For callers + /// with nothing to protect against failure -- copy-on-write, the catalog -- + /// where reserving and claiming are one step. pub fn alloc_pages(self: *Pager, n: u32) !u32 { assert(n > 0); - try self.reserve_pages(n); - return self.alloc_pages_assume_reserved(n); + var hold: Reservation = .{}; + try self.reserve_pages(&hold, n); + return self.alloc_pages_assume_reserved(&hold, n); } - /// Guarantee that the next `n` pages can be handed out without failing. + /// Guarantee that `hold` can have `n` more pages handed out without failing. /// Fallible, and meant to run *before* the log append on a write path, so /// that publishing afterwards cannot fail — the invariant that keeps a /// document from ever being live but unindexed. @@ -459,25 +492,35 @@ pub const Pager = struct { /// Nothing is dirtied here, so nothing is allocated on disk: `setLength` /// leaves a sparse file, `ls -l` grows and `du` does not. That is what makes /// a generous reservation cheap. - pub fn reserve_pages(self: *Pager, n: u32) !void { - // Additive: room for what is already promised *plus* this. Two - // consumers reserving before the same log append must both be able to + pub fn reserve_pages(self: *Pager, hold: *Reservation, n: u32) !void { + self.alloc_lock.lockUncancelable(self.io); + defer self.alloc_lock.unlock(self.io); + // Additive: room for every promise outstanding anywhere *plus* this one. + // Two consumers reserving before the same log append must both be able to // rely on their promise. try self.grow_to(self.alloc_tail + self.reserved_pages + n); self.reserved_pages += n; + hold.pages += n; } - /// Drop whatever is still promised but unclaimed. + /// Drop whatever `hold` still promises but has not claimed. /// /// A reservation is scoped to one write: it is taken before the log append /// so the publish afterwards cannot fail, and once the publish is done /// anything unclaimed is dead. Without this the promise accumulates -- a /// tree reservation covers the worst case of several splits and a typical - /// insert causes none, so `reserved_pages` grew by a handful per write and - /// dragged the file up with it. It showed as a 1.89 GB data file for 512 MB - /// of documents. - pub fn release_reservation(self: *Pager) void { - self.reserved_pages = 0; + /// insert causes none, so the total grew by a handful per write and dragged + /// the file up with it. It showed as a 1.89 GB data file for 512 MB of + /// documents. + pub fn release_reservation(self: *Pager, hold: *Reservation) void { + self.alloc_lock.lockUncancelable(self.io); + defer self.alloc_lock.unlock(self.io); + assert_msg( + self.reserved_pages >= hold.pages, + "a consumer released more pages than the pager had promised", + ); + self.reserved_pages -= hold.pages; + hold.pages = 0; } /// Hand out `n` pages against a previous `reserve_pages`. Infallible. @@ -540,17 +583,24 @@ pub const Pager = struct { self.free_ready.items.len = w + 1; } - pub fn alloc_pages_assume_reserved(self: *Pager, n: u32) u32 { + pub fn alloc_pages_assume_reserved(self: *Pager, hold: *Reservation, n: u32) u32 { assert(n > 0); + self.alloc_lock.lockUncancelable(self.io); + defer self.alloc_lock.unlock(self.io); + assert_msg( + n <= hold.pages, + "page allocation overran reserve_pages' promise", + ); assert_msg( n <= self.reserved_pages, - "page allocation overran reserve_pages' promise", + "page allocation overran the pager's total promise", ); assert_msg( self.alloc_tail + n <= self.mapped_pages, "page allocation past the mapped end of the data file", ); self.reserved_pages -= n; + hold.pages -= n; // Reuse before growing. Without this the free list is decorative and the // file grows without bound under churn, because copy-on-write abandons // every page it touches in every generation (PLAN amendment A2). @@ -841,6 +891,10 @@ pub const Pager = struct { // rotate the free lists by one generation. self.generation = wm.generation; self.loaded = wm; + // The free lists and the unpublished set are allocator state, so the + // rotation below takes the same lock every allocation does. + self.alloc_lock.lockUncancelable(self.io); + defer self.alloc_lock.unlock(self.io); self.stable_pages = self.alloc_tail; self.protect_image(); try self.free_ready.appendSlice(self.gpa, self.free_hold.items); @@ -1104,15 +1158,50 @@ test "two consumers reserving before one commit both keep their promise" { // Consumer A reserves a few pages, then consumer B reserves a large extent // and takes it -- exactly the order upsert uses. - try pg.reserve_pages(8); - try pg.reserve_pages(2048); - const b_first = pg.alloc_pages_assume_reserved(2048); + var a: Reservation = .{}; + var b: Reservation = .{}; + try pg.reserve_pages(&a, 8); + try pg.reserve_pages(&b, 2048); + const b_first = pg.alloc_pages_assume_reserved(&b, 2048); // A's promise must have survived B's reservation *and* B's allocation. - const a_first = pg.alloc_pages_assume_reserved(8); + const a_first = pg.alloc_pages_assume_reserved(&a, 8); try testing.expect(a_first >= b_first + 2048); try testing.expectEqual(@as(u32, 0), pg.reserved_pages); } +test "one consumer's release leaves another's promise intact" { + // Mutation check: make `release_reservation` zero `self.reserved_pages` + // (what it did when the promise was a single counter on the pager) and the + // allocation below goes red on `overran the pager's total promise`. + // + // Not hypothetical: two upserts on different collections hold different + // collection locks and run at the same time, so the first to publish + // released the second's promise out from under it. It aborted the server + // reliably at four concurrent clients (PLAN risk 3). + var threaded: std.Io.Threaded = .init_single_threaded; + defer threaded.deinit(); + const io = threaded.io(); + var tp = try TmpPager.init(io, 256 << 20); + defer tp.deinit(); + const pg = tp.pg(); + + var writer_a: Reservation = .{}; + var writer_b: Reservation = .{}; + try pg.reserve_pages(&writer_a, 16); + try pg.reserve_pages(&writer_b, 16); + + // A finishes its write and drops what it did not use. + _ = pg.alloc_pages_assume_reserved(&writer_a, 4); + pg.release_reservation(&writer_a); + try testing.expectEqual(@as(u32, 0), writer_a.pages); + + // B's promise is untouched, and still claimable in full. + try testing.expectEqual(@as(u32, 16), writer_b.pages); + try testing.expectEqual(@as(u32, 16), pg.reserved_pages); + for (0..16) |_| _ = pg.alloc_pages_assume_reserved(&writer_b, 1); + try testing.expectEqual(@as(u32, 0), pg.reserved_pages); +} + test "a reservation makes the following allocation infallible" { var threaded: std.Io.Threaded = .init_single_threaded; defer threaded.deinit(); @@ -1121,10 +1210,11 @@ test "a reservation makes the following allocation infallible" { defer tp.deinit(); const pg = tp.pg(); - try pg.reserve_pages(64); + var hold: Reservation = .{}; + try pg.reserve_pages(&hold, 64); const before = pg.alloc_tail; // Exactly the promised amount, one page at a time. - for (0..64) |_| _ = pg.alloc_pages_assume_reserved(1); + for (0..64) |_| _ = pg.alloc_pages_assume_reserved(&hold, 1); try testing.expectEqual(before + 64, pg.alloc_tail); }