diff --git a/src/pager.zig b/src/pager.zig index 8521555..485131c 100644 --- a/src/pager.zig +++ b/src/pager.zig @@ -134,6 +134,21 @@ pub const Extent = struct { /// What a checkpoint publishes. Everything here is authoritative except the /// two cached counters, which are hints the engine recomputes if they look /// wrong. +/// Pages a stream of `len` bytes occupies. +fn pages_for(len: u64) u32 { + return @intCast((len + page_size - 1) / page_size); +} + +/// One generation's two streams, as page runs. Page counts rather than byte +/// lengths because the free list speaks pages and because the free-list stream +/// is allocated from an upper bound that its final length can fall short of. +pub const Streams = struct { + catalog_page: u32 = 0, + catalog_pages: u32 = 0, + freelist_page: u32 = 0, + freelist_pages: u32 = 0, +}; + pub const Watermark = struct { generation: u64 = 0, /// The log sequence this image covers. The crash-recovery invariant is that @@ -235,6 +250,22 @@ pub const Pager = struct { /// The generation the next checkpoint will publish. generation: u64, + /// Where the catalog and free-list streams of the two most recent + /// generations live, so the one that falls out of reference can be freed. + /// + /// Both streams are written into *fresh* pages at every publish, so that a + /// crash leaves the previous copy intact. Nothing ever gave those pages + /// back: a database checkpointing every 32 MiB of log leaked two runs per + /// checkpoint forever, and the catalog carries a `u32` per index node page, + /// so at the tens-of-GB target that is hundreds of KB each time. + /// + /// Process-local rather than in the watermark: only a running pager needs to + /// know, because an open reads the slot it is loading and the other slot is + /// the fallback. Two generations because a publish overwrites the slot of + /// the generation *two* back -- that is the one nothing can reach again. + streams_cur: Streams, + streams_prev: Streams, + /// Pages below this belong to the last published image and must never be /// stored into (PLAN amendment A1). Zero until a checkpoint publishes one, /// which is why copy-on-write is inert before then. @@ -330,6 +361,8 @@ pub const Pager = struct { .fresh = created, .loaded = .{}, .generation = 0, + .streams_cur = .{}, + .streams_prev = .{}, .stable_pages = 0, .unpublished = .{}, .free_ready = .empty, @@ -859,6 +892,18 @@ pub const Pager = struct { // Everything the published image references is off limits to // writes from here on. self.stable_pages = wm.alloc_tail; + // So the next publish but one gives this generation's streams back. + // Page counts are derived from the lengths here rather than + // remembered, which can be one page short for a free-list stream + // whose final length fell inside the page its bound reserved. That + // loses at most one page, once per open, against the unbounded leak + // this replaces. + self.streams_cur = .{ + .catalog_page = wm.catalog_page, + .catalog_pages = pages_for(wm.catalog_len), + .freelist_page = wm.freelist_page, + .freelist_pages = pages_for(wm.freelist_len), + }; try self.read_freelist(wm); } else if (self.watermark_attempted()) { // Only worth saying when a watermark was *written* and cannot be @@ -973,7 +1018,24 @@ pub const Pager = struct { // made durable -- so a test can discard everything else and assert the // image still loads. Clearing it here would make that check vacuous. - // 5. only now is the new image current: advance the stable mark and + // 5. the streams of the generation two back are now unreachable: the + // slot that named them is the one this publish just overwrote, and + // the two slots hold the new generation and its predecessor. Give + // their pages back -- via the free list, so they are still withheld + // for two more generations like anything else. + // + // Before the lock below, which `free_pages` takes for itself. + try self.free_pages(self.streams_prev.catalog_page, self.streams_prev.catalog_pages); + try self.free_pages(self.streams_prev.freelist_page, self.streams_prev.freelist_pages); + self.streams_prev = self.streams_cur; + self.streams_cur = .{ + .catalog_page = wm.catalog_page, + .catalog_pages = pages_for(wm.catalog_len), + .freelist_page = fl.first, + .freelist_pages = fl.pages, + }; + + // 6. only now is the new image current: advance the stable mark and // rotate the free lists by one generation. self.generation = wm.generation; self.loaded = wm; @@ -1019,7 +1081,7 @@ pub const Pager = struct { return n; } - fn write_freelist(self: *Pager) !struct { first: u32, len: u64 } { + fn write_freelist(self: *Pager) !struct { first: u32, len: u64, pages: u32 } { // Size from an upper bound, then count the entries actually written. // // The allocation below takes its pages off this very list, and an exact @@ -1069,7 +1131,10 @@ pub const Pager = struct { ); std.mem.writeInt(u64, buf[0..8], count, .little); std.mem.writeInt(u64, buf[at..][0..8], header_hash(buf[0..at]), .little); - return .{ .first = first, .len = @as(u64, at) + 8 }; + // `pages` and not `pages_for(len)`: the allocation was sized from the + // bound, and the whole run has to go back when this generation falls + // out of reference. + return .{ .first = first, .len = @as(u64, at) + 8, .pages = pages }; } /// Load the free list a watermark points at. A damaged one is dropped with a @@ -1623,8 +1688,12 @@ test "the persisted free list survives allocating its own pages" { // This is the publish that trips it: the free list is now non-empty and // holds a run of exactly the one page the stream needs. try tp.pg().publish(.{ .seq = 4 }); + // Not a fixed number: a publish also recycles the streams of the generation + // two back, so what is on the list is the three frees above minus whatever + // the stream allocations took plus whatever they gave back. The invariant + // under test is that a reopen agrees with it, whatever it is. const ready_before = tp.pg().free_ready_pages(); - try testing.expectEqual(@as(u32, 2), ready_before); + try testing.expect(ready_before > 0); tp.close(); var again = try reopen(io, tp.path); @@ -1844,6 +1913,33 @@ test "one-page requests do not carve up the runs the extents need" { try testing.expectEqual(tail_before, pg.alloc_tail); } +test "a quiet checkpoint stops growing the file" { + // Both streams a publish writes are allocated fresh every time, so that a + // crash leaves the previous copy readable. Nothing gave them back, and a + // database checkpoints on log volume rather than on having anything to say + // -- so an idle server grew the data file forever. + // + // Steady state is not zero growth: a publish allocates this generation's + // streams and frees the ones from two generations back, and those take two + // more publishes to become reusable. So a few pages are always in flight, + // and the assertion is that the number does not track the publish count. + // + // Mutation check: drop the two `free_pages` calls from `publish` and this + // goes red at 40-something pages instead of a handful. + var threaded: std.Io.Threaded = .init_single_threaded; + defer threaded.deinit(); + const io = threaded.io(); + var tp = try TmpPager.init(io, 64 << 20); + defer tp.deinit(); + + // Let the rotation reach its steady state before measuring. + for (1..5) |i| try tp.pg().publish(.{ .seq = i }); + const settled = tp.pg().alloc_tail; + + for (5..45) |i| try tp.pg().publish(.{ .seq = i }); + try testing.expect(tp.pg().alloc_tail - settled <= 8); +} + test "concurrent frees lose no pages while a publish rotates the lists" { // `free_pages` mutates the same three lists `publish` rotates, and its hot // caller is `page_mut_cow` under a *collection* lock -- so two collections @@ -1906,7 +2002,10 @@ test "concurrent frees lose no pages while a publish rotates the lists" { }) |list| for (list) |e| { for (0..e.pages) |i| { const p = e.first + @as(u32, @intCast(i)); - try testing.expect(p >= lo and p - lo < pages.len); + // Pages outside the set the freers own are the publisher's own + // stream runs coming back two generations later; they are not what + // this test is about. + if (p < lo or p - lo >= pages.len) continue; try testing.expect(!seen.isSet(p - lo)); seen.set(p - lo); on_list += 1;