diff --git a/src/db.zig b/src/db.zig index c595778..3af904f 100644 --- a/src/db.zig +++ b/src/db.zig @@ -121,9 +121,16 @@ pub const Collection = struct { /// extent it falls in. slab_tail: u64, slab_end: u64, - /// Document bytes written into this collection's slab since the last - /// rebuild. `slab_tail` cannot answer that -- it is an absolute file offset, - /// so it jumps forward whenever a fresh extent is taken. + /// Slab this collection has consumed and not yet given back. `slab_tail` + /// cannot answer that -- it is an absolute file offset, so it jumps forward + /// whenever a fresh extent is taken. + /// + /// It used to mean "bytes ever appended since the last rebuild", which was + /// the same thing while a rebuild was the only way to get slab back. Window + /// reclamation subtracts from it, and that is what keeps + /// `slab_used - live_bytes` equal to the garbage the collection still has + /// -- with no new persistent field, since both halves are already in the + /// catalog. slab_used: u64, /// This collection's outstanding page promise, for the document slab. Per /// collection because concurrent writers must not release each other's -- @@ -148,6 +155,15 @@ pub const Collection = struct { /// update. What it costs is only that garbage from before a restart is not /// reclaimed window-wise; it still arms compaction like any other. dead_unlocated: u64, + /// Windows whose counter has reached `map_align`, i.e. how much there is + /// for the next checkpoint to give back. + /// + /// It exists so that a checkpoint costs nothing on a collection with + /// nothing to reclaim. Scanning would otherwise be O(slab) per checkpoint + /// whatever the workload -- and the workload this design is *known* not to + /// help, small documents on large system pages, is exactly the one that + /// would pay that for no return. + full_windows: u32, /// Slab handed back to the pager by window reclamation, cumulative for the /// life of the process. Purely an observation: it is what distinguishes /// "the ratio improved because reclamation worked" from "the ratio improved @@ -203,6 +219,7 @@ pub const Collection = struct { .slab_used = 0, .live_bytes = 0, .dead_unlocated = 0, + .full_windows = 0, .reclaimed_bytes = 0, .hold = .{}, .indexes = .empty, @@ -316,7 +333,9 @@ pub const Collection = struct { // this means the same range was marked twice -- a double eviction, // or a recycled offset marked against the previous owner's map. assert_msg(r.dead[w] + n <= pgr.map_align, "a slab window holds more dead bytes than it has"); + const was_full = r.dead[w] == pgr.map_align; r.dead[w] += @intCast(n); + if (!was_full and r.dead[w] == pgr.map_align) self.full_windows += 1; pos += n; } if (pos < stop) self.dead_unlocated += stop - pos; @@ -338,6 +357,122 @@ pub const Collection = struct { fn free_runs(self: *Collection, gpa: std.mem.Allocator) void { for (self.slab_runs.items) |r| gpa.free(r.dead); self.slab_runs.clearRetainingCapacity(); + self.full_windows = 0; + } + + /// One piece of a run that survives reclamation, with a window map of its + /// own copied out of the original. + /// + /// Every kept piece gets a fresh array, including a run nothing was taken + /// from. Moving the original array instead would save a copy and make the + /// failure path have to know which arrays it still owns -- the version that + /// tried it had a double free in the out-of-memory case, which is the one + /// case nothing exercises. + fn keep_piece( + out: *std.ArrayListUnmanaged(SlabRun), + gpa: std.mem.Allocator, + r: SlabRun, + p0: u32, + p1: u32, + ) !void { + const wf = std.mem.alignForward(u64, @as(u64, p0) << pgr.page_shift, pgr.map_align); + const we = std.mem.alignBackward(u64, @as(u64, p1) << pgr.page_shift, pgr.map_align); + const count: usize = if (we > wf) @intCast((we - wf) / pgr.map_align) else 0; + const dead = try gpa.alloc(WindowDead, count); + errdefer gpa.free(dead); + // A piece boundary is either the run's own start/end or a window + // boundary, so the piece's windows line up with a contiguous stretch of + // the original's and the counters can be copied rather than rebuilt. + const base: usize = @intCast((wf - r.window_first) / pgr.map_align); + @memcpy(dead, r.dead[base..][0..count]); + try out.append(gpa, .{ .first = p0, .pages = p1 - p0, .window_first = wf, .dead = dead }); + } + + /// Give back every window with nothing live left in it, splitting the runs + /// around what is kept. Returns the bytes handed to the pager. + /// + /// Counting is the whole test: a window reaches `map_align` dead only once + /// every document with a byte in it has been through `evict_doc`, which + /// removes its index entries before marking it. So "no live bytes" and "no + /// reference to these bytes" are the same statement, and nothing has to be + /// scanned to establish it. + /// + /// Fallible, and arranged so a failure changes nothing: the replacement + /// list is built whole before the old one is touched. The garbage simply + /// stays and the next checkpoint tries again. + fn reclaim_windows(self: *Collection, gpa: std.mem.Allocator) !u64 { + assert_msg( + self.slab_used >= self.live_bytes, + "a collection cannot hold more live bytes than it ever appended", + ); + // The identity, checked where every window is being walked anyway. + assert_msg( + self.dead_located() + self.dead_unlocated == self.slab_used - self.live_bytes, + "the collection's placed and unplaced garbage must add up to its garbage", + ); + var out: std.ArrayListUnmanaged(SlabRun) = .empty; + errdefer { + for (out.items) |p| gpa.free(p.dead); + out.deinit(gpa); + } + var give: std.ArrayListUnmanaged(pgr.Extent) = .empty; + defer give.deinit(gpa); + + var freed: u64 = 0; + for (self.slab_runs.items) |r| { + var keep_from = r.first; + var i: usize = 0; + while (i < r.dead.len) { + if (r.dead[i] != pgr.map_align) { + i += 1; + continue; + } + var j = i + 1; + while (j < r.dead.len and r.dead[j] == pgr.map_align) j += 1; + const from = r.window_first + i * pgr.map_align; + const to = r.window_first + j * pgr.map_align; + // The appender's own extent is off limits, and not by + // filtering: bytes above the cursor have never been written, so + // no window covering them can have reached `map_align` dead. + // Tripping this means a range was marked dead twice. + assert_msg( + to <= self.slab_tail or from >= self.slab_end, + "reclaiming a slab window the append cursor is still walking", + ); + const p_from: u32 = @intCast(from >> pgr.page_shift); + const p_to: u32 = @intCast(to >> pgr.page_shift); + if (p_from > keep_from) try keep_piece(&out, gpa, r, keep_from, p_from); + try give.append(gpa, .{ .first = p_from, .pages = p_to - p_from }); + freed += to - from; + keep_from = p_to; + i = j; + } + if (keep_from < r.first + r.pages) { + try keep_piece(&out, gpa, r, keep_from, r.first + r.pages); + } + } + if (freed == 0) { + for (out.items) |p| gpa.free(p.dead); + out.deinit(gpa); + // Every full window was given back or there were none, so nothing + // is left for the next checkpoint to find. + self.full_windows = 0; + return 0; + } + + // Past the last fallible step: swap the list in, then hand the pages + // over. A `free_pages` that fails here leaks the run -- it is no longer + // the collection's and not yet the pager's -- which costs space and + // nothing else. The other order would leave the same pages owned twice. + for (self.slab_runs.items) |r| gpa.free(r.dead); + self.slab_runs.deinit(gpa); + self.slab_runs = out; + self.full_windows = 0; + for (give.items) |e| self.pager.free_pages(e.first, e.pages) catch {}; + assert_msg(self.slab_used >= self.live_bytes + freed, "reclaiming more slab than the collection has"); + self.slab_used -= freed; + self.reclaimed_bytes += freed; + return freed; } /// Append `bytes` to the slab, returning its flat offset. The last @@ -2314,6 +2449,62 @@ pub const Engine = struct { self.committed_seq = 0; } + /// Hand back every slab window with nothing live left in it, across every + /// collection. The first phase of a checkpoint. + /// + /// Inside the checkpoint rather than a hook after it, and that placement is + /// the whole safety argument. Reclamation changes two things: it splits + /// `slab_runs`, which the catalog describes, and it calls + /// `pager.free_pages`, which the free list describes. The checkpoint's + /// single `publish` makes both durable together, so a crash before it + /// leaves the old catalog and the old free list -- no reclamation happened + /// -- and a crash after leaves both describing the new ownership. There is + /// no order in between to get wrong, and no new record type or replay path. + /// + /// Batched with the checkpoint for a second reason: the write path pays + /// only for a counter update, and the cadence of actually returning pages + /// is the checkpoint threshold rather than per-delete. + /// + /// Catalog shared, then each collection exclusive, one at a time -- the + /// order `compact` and `write_catalog` both use. + fn reclaim_slabs(self: *Engine) void { + self.catalog_lock.lockSharedUncancelable(self.io); + defer self.catalog_lock.unlockShared(self.io); + var db_it = self.dbs.iterator(); + while (db_it.next()) |db_entry| { + var coll_it = db_entry.value_ptr.collections.iterator(); + while (coll_it.next()) |ce| self.reclaim_collection(ce.value_ptr.*); + } + } + + fn reclaim_collection(self: *Engine, coll: *Collection) void { + coll.lock.lockUncancelable(self.io); + defer coll.lock.unlock(self.io); + // The common case, and the reason this is affordable at every + // checkpoint: a collection with no full window is not scanned at all. + if (coll.full_windows == 0) return; + // Out of memory here means the garbage stays where it is. Nothing is + // lost and the next checkpoint tries again. + const freed = coll.reclaim_windows(self.gpa) catch return; + if (freed == 0) return; + self.counter_lock.lockUncancelable(self.io); + assert_msg(self.dead_bytes >= freed, "reclaiming more slab than the engine counts as dead"); + self.dead_bytes -= freed; + self.counter_lock.unlock(self.io); + // A cursor holding slab offsets is now holding some that name pages + // this collection no longer owns -- and reading them would succeed, + // since the pages are only on the free list, so the answer would be + // plausible garbage rather than an error. `cursor_still_valid` compares + // this epoch and kills such a cursor with QueryPlanKilled, which is + // what a rebuild already does to it. + // + // Only when something was actually given back: a collection that + // reclaimed nothing must not have its cursors killed on the cadence of + // the checkpoint. + self.layout_epoch_seq += 1; + coll.layout_epoch = self.layout_epoch_seq; + } + /// Publish the current state as a checkpoint. /// /// The watermark equals the sequence the log has already made durable, never @@ -2323,6 +2514,7 @@ pub const Engine = struct { /// D6) reduced to an ordering. pub fn checkpoint(self: *Engine) !void { try self.commit(); + self.reclaim_slabs(); var buf: std.ArrayListUnmanaged(u8) = .empty; defer buf.deinit(self.gpa); @@ -3350,6 +3542,134 @@ test "a restart forgets where the garbage is, not that there is any" { while (it.next()) |entry| try testing.expect(coll.run_of(entry.off) != null); } +/// Whether `page` falls in one of the pager's free-list generations. +fn in_extents(list: []const pgr.Extent, page: u32) bool { + for (list) |e| if (page >= e.first and page < e.first + e.pages) return true; + return false; +} + +test "a slab window with one live document in it is never given back" { + // The load-bearing test of window reclamation. A window goes back when its + // counter reaches `map_align`, which is a statement about *bytes*, not + // about documents -- so the thing that must never happen is a window handed + // to the pager while a document still sits in it. The document would still + // read, because a freed page is only on a list, so the failure would be + // silent until the pages were handed out again and overwritten. + // + // Mutation check: relax the fullness test in `reclaim_windows` to + // `r.dead[i] + 2048 < pgr.map_align`, so a window 2 KiB short of empty + // qualifies. On its own that aborts the whole suite on the append-cursor + // assertion instead -- the appender's own window is the first thing a + // loosened test reaches, which is worth knowing. Drop that assertion too + // and this is the test that goes red, alone. + var threaded: std.Io.Threaded = .init_single_threaded; + defer threaded.deinit(); + var env = test_env(&threaded); + const io = env.io; + const gpa = testing.allocator; + + var tmp = try TmpLog.init(gpa); + defer tmp.deinit(gpa); + var engine = try Engine.open(gpa, io, tmp.path); + defer engine.deinit(); + engine.compact_threshold = std.math.maxInt(u64); // no rebuild may intervene + try engine.lock(); + defer engine.unlock(); + + var i: i32 = 0; + while (i < 200) : (i += 1) { + var d = try make_padded(gpa, i, 2000); + defer d.deinit(); + try engine.insert("app", "c", &d, &env.gen); + } + try engine.commit(); + const coll = engine.get_collection("app", "c").?; + + // Everything dies but one document, roughly in the middle of the slab. + i = 0; + while (i < 200) : (i += 1) { + if (i == 100) continue; + try testing.expect(try engine.remove_by_id("app", "c", .{ .int32 = i })); + } + try engine.commit(); + const survivor_enc = try id_key_for(gpa, bson.Value{ .int32 = 100 }); + defer gpa.free(survivor_enc); + const survivor = coll.id_index.lookup_exact(survivor_enc).?; + + try engine.checkpoint(); + // Most of the slab went back... + try testing.expect(coll.reclaimed_bytes > 100 * 2000); + // ...but not the window the survivor is in, and it still reads. + try testing.expect(coll.run_of(survivor) != null); + try testing.expect(std.mem.indexOf(u8, coll.doc_bytes(survivor), "xxxx") != null); + // The accounting followed the pages: what is left is what is left. + try testing.expectEqual(coll.slab_used - coll.live_bytes, engine.dead_bytes); + try testing.expectEqual( + coll.slab_used - coll.live_bytes, + coll.dead_located() + coll.dead_unlocated, + ); + + // And once the survivor is gone, its window goes too. + try testing.expect(try engine.remove_by_id("app", "c", .{ .int32 = 100 })); + try engine.commit(); + try engine.checkpoint(); + try testing.expect(coll.run_of(survivor) == null); +} + +test "a reclaimed slab window is not reusable until two publishes later" { + // Reclamation hands pages to `free_pages`, which withholds them for two + // generations -- and it has to, because the image one generation back is + // still the fallback a crash would open, and its catalog still claims them. + // Handing them straight out would let a write land on pages the recovery + // path is about to read as documents. + // + // Asserted on the pager's own lists rather than on `free_ready_pages()`, + // whose total also moves for copy-on-write victims and the catalog stream. + var threaded: std.Io.Threaded = .init_single_threaded; + defer threaded.deinit(); + var env = test_env(&threaded); + const io = env.io; + const gpa = testing.allocator; + + var tmp = try TmpLog.init(gpa); + defer tmp.deinit(gpa); + var engine = try Engine.open(gpa, io, tmp.path); + defer engine.deinit(); + engine.compact_threshold = std.math.maxInt(u64); + try engine.lock(); + defer engine.unlock(); + + var i: i32 = 0; + while (i < 200) : (i += 1) { + var d = try make_padded(gpa, i, 2000); + defer d.deinit(); + try engine.insert("app", "c", &d, &env.gen); + } + try engine.commit(); + const coll = engine.get_collection("app", "c").?; + const owned_before = coll.slab_runs.items[0]; + + i = 0; + while (i < 200) : (i += 1) _ = try engine.remove_by_id("app", "c", .{ .int32 = i }); + try engine.commit(); + try engine.checkpoint(); + try testing.expect(coll.reclaimed_bytes > 0); + + // A page from the first window given back: the run's first window is all + // dead now, so its first page is no longer the collection's. + const gone = @as(u32, @intCast(owned_before.window_first >> pgr.page_shift)); + try testing.expect(coll.run_of(@as(u64, gone) << pgr.page_shift) == null); + try testing.expect(gone >= owned_before.first); + + // One publish has happened, so it is held, not ready. + try testing.expect(!in_extents(engine.pager.free_ready.items, gone)); + try testing.expect(in_extents(engine.pager.free_hold.items, gone)); + + // The second publish is what makes it allocatable. + try engine.checkpoint(); + try testing.expect(in_extents(engine.pager.free_ready.items, gone)); +} + test "a replace that changes nothing is not a write" { // Mutation check: delete the byte comparison in `upsert`'s `.replace` arm. // Red on all three: the log grows, the document is superseded so the engine @@ -3656,9 +3976,19 @@ test "the slab counts what the appender skips" { try testing.expectEqual(gap + abandoned, engine.dead_bytes); try testing.expectEqual(coll.slab_used - coll.live_bytes, engine.dead_bytes); - // And it survives the round trip, because it is in `slab_used`. + // And the next checkpoint gives most of it straight back. An abandoned + // extent tail is whole windows with nothing live in them, which is exactly + // what window reclamation is for -- so counting it was not bookkeeping for + // its own sake, it is what made this reclaimable at all. + // + // What stays is the edges: the round-up gap, which shares its window with + // the live documents below it, and the bytes of the run outside any whole + // window. try engine.checkpoint(); - try testing.expectEqual(gap + abandoned, engine.dead_bytes); + try testing.expect(coll.reclaimed_bytes > 4 * 1024 * 1024); + try testing.expectEqual(gap + abandoned - coll.reclaimed_bytes, engine.dead_bytes); + try testing.expectEqual(coll.slab_used - coll.live_bytes, engine.dead_bytes); + try testing.expect(engine.dead_bytes < gap + 2 * pgr.map_align); } test "a rebuild leaves behind what its own copying skipped" { @@ -3706,11 +4036,16 @@ test "a rebuild leaves behind what its own copying skipped" { const coll = engine.get_collection("app", "c").?; try testing.expectEqual(@as(u64, 2), coll.doc_count); - // The deleted document is gone from the slab, but the gap the copy left - // between the two survivors is not -- and the engine says so. + // The deleted document is gone from the slab, and the gap the copy left + // between the two survivors is now mostly gone too -- `compact` ends in a + // checkpoint, and a checkpoint reclaims whole windows. What survives is the + // edges of that gap, which is a smaller number than this test used to + // assert but the same statement: the counter is *not* zeroed, and it equals + // what the collection actually has. try testing.expectEqual(coll.slab_used - coll.live_bytes, engine.dead_bytes); - try testing.expect(engine.dead_bytes > 2 * 1024 * 1024); - try testing.expect(engine.dead_bytes < 4 * 1024 * 1024); + try testing.expect(coll.reclaimed_bytes > 2 * 1024 * 1024); + try testing.expect(engine.dead_bytes > 0); + try testing.expect(engine.dead_bytes < 4 * pgr.map_align); } test "dropping a collection does not arm compaction" { @@ -5119,8 +5454,35 @@ test "the epochs that invalidate a cursor move exactly when they must" { try testing.expect(after_recreate != after_rebuild); try testing.expect(after_recreate != before); - // And the index-level token, which guards the position hint. + // Reclamation does not move a live document, so a cursor's *live* offsets + // stay good -- but the pages it gives back can be handed out again, and a + // cursor's saved offset list may name one of them. Same remedy, and the + // same token. + // + // Both halves matter. A checkpoint that reclaims nothing must leave the + // epoch alone, or every open cursor on a busy collection dies on the + // checkpoint cadence for nothing. try engine.lock(); + var j: i32 = 0; + while (j < 200) : (j += 1) { + var d = try make_padded(gpa, 1000 + j, 2000); + defer d.deinit(); + try engine.insert("app", "c", &d, &env.gen); + } + try engine.commit(); + const quiet_before = engine.get_collection("app", "c").?.layout_epoch; + try engine.checkpoint(); + try testing.expectEqual(quiet_before, engine.get_collection("app", "c").?.layout_epoch); + + j = 0; + while (j < 200) : (j += 1) _ = try engine.remove_by_id("app", "c", .{ .int32 = 1000 + j }); + try engine.commit(); + try engine.checkpoint(); + const after_reclaim = engine.get_collection("app", "c").?.layout_epoch; + try testing.expect(engine.get_collection("app", "c").?.reclaimed_bytes > 0); + try testing.expect(after_reclaim != quiet_before); + + // And the index-level token, which guards the position hint. const coll = engine.get_collection("app", "c").?; const index_before = coll.id_index.epoch; try coll.id_index.reset_tree(gpa);