//! The data file: a checkpoint of the engine's structures, mapped rather than //! parsed. The log (storage.zig) remains the WAL and the source of truth; this //! file is a snapshot at some sequence number, and on open the log replays //! whatever happened after it (PLAN D4). //! //! File layout — an array of 4 KiB pages: //! page 0 header. Written at create, never rewritten. //! [0..4) u32 magic "MFDB" (0x4D464442) //! [4] u8 format version (1) //! [5] u8 page_shift (12) //! [6..8) u16 reserved (0) //! [8..24) u128 database uuid — must match the log's, so a data file //! can never be paired with a foreign log //! [24..32) u64 xxhash3 over [0..24) //! [32..4096) zero //! pages 1,2 watermark slots A and B. Generation N writes slot (N & 1), //! so a torn write can never leave zero valid slots -- writing //! a new generation over the only copy of the old one could. //! The newer slot whose hash validates is authoritative. //! [0..8) u64 generation (monotonic, >= 1) //! [8..16) u64 seq -- the log sequence this image covers //! [16..24) u64 alloc_tail -- pages [0, alloc_tail) are allocated //! [24..32) u64 file_pages //! [32..40) u64 catalog_page -- first page of the catalog stream //! [40..48) u64 catalog_len -- bytes of it //! [48..56) u64 freelist_page //! [56..64) u64 freelist_len //! [64..72) u64 prev_generation -- kept intact for fallback //! [72..80) u64 live_docs -- cached hint //! [80..88) u64 dead_bytes -- cached hint; the engine recomputes it //! from the catalog on open //! [88..4088) reserved (zero) //! [4088..4096) u64 xxhash3 over [0..4088) //! pages 3.. data, handed out by a tail-bump extent allocator //! //! The catalog stream and the free list are written *wholesale into freshly //! allocated pages* at every checkpoint, never mutated in place. That is what //! makes them untearable: the previous copy stays intact and referenced by the //! previous watermark until the new watermark switches over. //! //! Every persisted reference is a page number (u32) or an absolute file byte //! offset (u64) — never a pointer — so where the file happens to be mapped is //! irrelevant. Node pages are raw host memory (D4: on-disk format == in-memory //! format, no serialization on page-in), which makes the data file //! little-endian-only; the log, framed field by field, stays portable. //! //! PLAN D6.1 called for a region table. It cannot be that: there is one node //! arena per index and one document slab per collection, so the region count is //! dynamic and unbounded, and N contiguous regions cannot all grow at the tail. //! With one page array and extent allocation there is exactly one growth path, //! therefore exactly one place where the write-then-extend discipline lives, //! and `ls`/`du` stay honest for the backup story (D6.6). //! //! Why this reaches for std.posix instead of std.Io, against the house style: //! `std.Io.File.MemoryMap` prefaults by default (`populate = true`), which is //! the opposite of "RSS = working set"; it exposes no NORESERVE, FIXED or //! address hint, so it cannot express an address-space reservation; and its //! `setLength` is mremap on Linux and unsupported on darwin, which this project //! develops on. Everything else here goes through std.Io.File. const std = @import("std"); const builtin = @import("builtin"); const assert = @import("assert.zig").assert; const assert_msg = @import("assert.zig").assert_msg; /// Test-only: record which pages have been written since the last sync, so a /// test can simulate the one failure a `kill -9` cannot produce -- writeback /// that never happened. /// /// This exists because the ordering it guards is the whole crash-recovery /// argument and is otherwise untestable. Killing a process does not lose /// page-cache writes, so removing the msync before the watermark leaves every /// test green while making a real power loss unrecoverable. Compiled out /// entirely outside tests, so `page_mut` keeps no branch in a real build. const track_dirty = builtin.is_test; /// Also make the published image *hardware* read-only, in builds that can /// afford it. /// /// The assert in `page_mut` catches a caller that asks for a writable page below /// the stable mark. It cannot catch one that holds a pointer obtained *before* /// the mark moved and writes through it afterwards -- and that is the realistic /// mistake, because tree code holds `*Node` across calls. mprotect catches it, /// as an immediate segfault at the offending store rather than as a corrupt /// database discovered after a power loss. /// /// Off in ReleaseFast so the hot path keeps no extra syscall, on everywhere /// else. One mprotect per checkpoint over one range is the whole cost. const protect_stable = builtin.mode != .ReleaseFast; pub const page_shift: u6 = 12; pub const page_size: usize = 1 << page_shift; /// Page numbers of the fixed-position pages. pub const page_header: u32 = 0; pub const page_watermark_a: u32 = 1; pub const page_watermark_b: u32 = 2; /// The first page the allocator may hand out. pub const page_first_data: u32 = 3; const magic: u32 = 0x4D464442; // "MFDB" const format_version: u8 = 1; const header_hashed_len: usize = 24; /// The alignment mmap requires, which is the *system* page size and not ours. /// 16 KiB on Apple Silicon against a 4 KiB logical page, so four logical pages /// share one system page there — which is why the checkpoint rounds its append /// cursors up to this rather than to page_size. pub const map_align = std.heap.page_size_min; /// Logical pages per system page. At least one: the comptime block below only /// requires one of the two sizes to divide the other. pub const pages_per_map_align: u32 = @max(1, map_align / page_size); /// Growth granularity. Large enough that growth is rare and each `setLength` /// covers many allocations, and a multiple of every supported system page size. const grow_chunk_pages: u32 = 2048; // 8 MiB /// Address space reserved by default. Reserved, not committed: a PROT_NONE /// anonymous NORESERVE mapping is one VMA and zero pages. Sized for the /// tens-of-GB target (D3) with room to spare. pub const default_reserve_bytes: usize = 64 << 30; comptime { assert(page_size == 4096); assert(page_first_data == 3); assert(grow_chunk_pages * page_size % map_align == 0); assert(map_align % page_size == 0 or page_size % map_align == 0); // Node pages are raw host memory in this file (see the header comment). assert(@import("builtin").cpu.arch.endian() == .little); } /// A run of contiguous pages. pub const Extent = struct { first: u32, pages: u32, }; /// 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 run of `len` bytes occupies, rounded up. The one place the /// partial-page rule is written. pub 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 /// this never exceeds the durable log tail (PLAN D6), so replay after open /// is exactly the records above it. seq: u64 = 0, alloc_tail: u32 = page_first_data, file_pages: u32 = page_first_data, catalog_page: u32 = 0, catalog_len: u64 = 0, freelist_page: u32 = 0, freelist_len: u64 = 0, prev_generation: u64 = 0, live_docs: u64 = 0, dead_bytes: u64 = 0, }; const wm_hashed_len: usize = page_size - 8; pub const Error = error{ /// Not a data file, or a version this build cannot read. InvalidDataFile, /// The file belongs to a different database than the log beside it. DatabaseMismatch, /// Growth would exceed the reserved address space. DatabaseTooLarge, }; pub const OpenOptions = struct { /// Must match the log's uuid. Zero means "adopt whatever the file has", /// which is only for tests that do not care. uuid: u128 = 0, 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, file: std.Io.File, path: []u8, uuid: u128, /// The address-space reservation. Never moves for the life of the process, /// so a pointer obtained from `page`/`page_mut` stays valid across growth — /// which the old ArrayList-backed arena could not promise, and which is the /// reason a whole class of dangling-pointer bugs disappears here. reserve: []align(map_align) u8, /// Pages actually mapped to the file. Always >= alloc_tail. mapped_pages: u32, /// Pages the file is long enough to hold. file_pages: u32, /// Pages [0, alloc_tail) have been handed out. alloc_tail: u32, /// Pages promised by `reserve_pages` and not yet handed out. /// /// A count rather than a tail mark, because there is more than one consumer: /// an upsert reserves tree pages for every index *and* slab room for the /// document, all before the log append. A tail mark cannot express that -- /// the second reserver overwrites the first one's promise, and then the /// 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, /// Separates a publish from the appends that are mid-flight. /// /// An appender asks `is_unpublished_at` whether its cursor is still /// writable, and copies bytes there afterwards. `publish` clears the whole /// unpublished set and mprotects the image between those two steps, so the /// answer was stale by the time it was used and the copy landed in the /// published image: SIGBUS where the protection is compiled in, a silent /// store into the durable image in ReleaseFast, where it is not. /// /// Shared by appenders so writers on different collections still run /// concurrently -- the decomposition ROADMAP item 5 measured is not given /// back. Exclusive only for the tail of a publish, which happens once per /// checkpoint. append_lock: std.Io.RwLock, /// True when this file was created by this open (no checkpoint to load). fresh: bool, /// The watermark this pager opened from, or a zero one when fresh. The /// engine reads `seq` to decide where replay starts. loaded: Watermark, /// 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. /// /// A *bound*, not the membership test: see `unpublished`. stable_pages: u32, /// Pages handed out since the last publish, and therefore not referenced by /// any durable image -- free to be written in place however low their page /// number is. /// /// `p >= stable_pages` was used for this and is not the same question. It is /// right for tail-bumped pages and wrong for recycled ones, which come off /// the free list *below* the mark and are nonetheless writable. The /// difference is not cosmetic: with the numeric test, every write to a /// recycled node page copied it and freed it again, and every recycled slab /// extent was abandoned after a single document -- so nothing was ever really /// reused and the file grew without bound under churn. The churn gate /// measured 7.2x live data over six rounds, still climbing linearly. /// /// One bit per page: 32 KiB per GiB of database, one load on the COW path /// against the 4 KiB copy it avoids. Cleared wholesale at each publish, /// which is exactly when every allocated page becomes part of the image. unpublished: std.DynamicBitSetUnmanaged, /// Free pages, in three stages. `ready` may be handed out now; `hold` was /// freed one generation ago; `pending` was freed during this generation. /// /// The two-generation delay is not caution, it is what makes generation N-1 /// remain a usable image: its pages are still allocated while N is current, /// so a torn watermark or a bad catalog can fall back to it instead of /// discarding the database. Releasing one generation early would silently /// take that away. free_ready: std.ArrayListUnmanaged(Extent), free_hold: std.ArrayListUnmanaged(Extent), free_pending: std.ArrayListUnmanaged(Extent), /// See `track_dirty`. Pages written since the last sync. dirty: if (track_dirty) std.AutoHashMapUnmanaged(u32, void) else void, /// Guards `dirty`, and only `dirty`. Writers to the file itself are kept /// apart by the locks their *owners* hold -- a collection's, an index's -- /// and the pages two of them touch never overlap. This set is the one thing /// they share: two appenders on different collections record into the same /// hash map at the same time, which is a torn map rather than a torn page. /// Test-only instrumentation, so it costs the server nothing. dirty_lock: if (track_dirty) std.Io.Mutex else void, /// Whether the last `protect_image` actually took effect. Checked by a test: /// an mprotect that silently fails would leave the belt looking present and /// doing nothing, which is worse than not having it. protect_ok: bool, pub fn open( gpa: std.mem.Allocator, io: std.Io, path: []const u8, opts: OpenOptions, ) !Pager { // Absolute, for the same reason storage.Log resolves its own path: the // rebuild renames this file and must not depend on the caller's working // directory. const owned_path = blk: { if (path.len > 0 and path[0] == '/') break :blk try gpa.dupe(u8, path); const cwd = try std.process.currentPathAlloc(io, gpa); defer gpa.free(cwd); break :blk try std.fmt.allocPrint(gpa, "{s}/{s}", .{ cwd, path }); }; errdefer gpa.free(owned_path); const dir = std.Io.Dir.cwd(); var created = false; var file = dir.openFile(io, owned_path, .{ .mode = .read_write }) catch |err| switch (err) { error.FileNotFound => blk: { created = true; break :blk try dir.createFile(io, owned_path, .{ .read = true }); }, else => return err, }; errdefer file.close(io); // A zero-length file is as good as absent: a create that died before // its header landed must not look like a corrupt database. const existing_len = try file.length(io); if (existing_len == 0) created = true; const reserve = try reserve_address_space(opts.reserve_bytes); errdefer std.posix.munmap(reserve); var self: Pager = .{ .gpa = gpa, .io = io, .file = file, .path = owned_path, .uuid = opts.uuid, .reserve = reserve, .mapped_pages = 0, .file_pages = @intCast(existing_len / page_size), .alloc_tail = page_first_data, .reserved_pages = 0, .alloc_lock = .init, .append_lock = .init, .fresh = created, .loaded = .{}, .generation = 0, .streams_cur = .{}, .streams_prev = .{}, .stable_pages = 0, .unpublished = .{}, .free_ready = .empty, .free_hold = .empty, .free_pending = .empty, .dirty = if (track_dirty) .empty else {}, .dirty_lock = if (track_dirty) .init else {}, .protect_ok = false, }; // The bit set is the one heap allocation `self` owns before `deinit` can // be reached: a rejected header returns from the middle of this function. errdefer self.unpublished.deinit(gpa); if (created) { try self.grow_to(page_first_data); try self.write_header(); } else { try self.grow_to(@max(page_first_data, self.file_pages)); try self.read_header(); // Everything already in the file is allocated until a watermark // narrows it down. self.alloc_tail = @max(page_first_data, self.file_pages); try self.load_watermark(); if (self.loaded.generation == 0) { // No usable checkpoint: either none was ever published, or both // watermark slots were unreadable. Then *nothing* in this file is // referenced -- the log is the whole truth and replay is about to // rebuild the slab, the trees and the overflow from it -- so the // file is free space, not allocated space. // // Leaving `alloc_tail` at the file end instead made every reopen // append a fresh copy above the last one, with no watermark and // therefore no free list to ever give the old copy back. Linear // growth per open, unbounded, and it does not need a crash: a // database small enough never to reach the checkpoint threshold // never publishes a watermark at all, so *every* clean reopen // took this path. Measured at 20 documents per cycle: +17 MB per // reopen, 269 MB after twelve, on course for DatabaseTooLarge. // // The file is not truncated here on purpose. The mapping is // already established over these pages, and `grow_to` extends the // file only when the mapping is too small -- so shortening the // file behind a mapping that still covers it would turn a later // write into SIGBUS. Reusing from the front bounds the file at // its high-water mark, which is what the unbounded growth needed. self.alloc_tail = page_first_data; } } return self; } pub fn deinit(self: *Pager) void { self.unpublished.deinit(self.gpa); self.free_ready.deinit(self.gpa); self.free_hold.deinit(self.gpa); self.free_pending.deinit(self.gpa); if (track_dirty) self.dirty.deinit(self.gpa); std.posix.munmap(self.reserve); self.file.close(self.io); self.gpa.free(self.path); } // -- page access -------------------------------------------------------- /// The page's bytes, for reading. pub inline fn page(self: *const Pager, p: u32) *align(page_size) const [page_size]u8 { assert_msg(p < self.mapped_pages, "read of a page past the mapped end of the data file"); return @ptrCast(@alignCast(self.reserve.ptr + (@as(usize, p) << page_shift))); } /// The page's bytes, for writing. Callers holding a page number that can be /// *updated* use `page_mut_cow`; append-only consumers pass a page they /// allocated themselves. /// /// There is deliberately no assert on the page number here. A page recycled /// off the free list is below the stable mark and is legitimately writable -- /// it was freed two generations ago and no live image references it any more /// -- so the page number alone cannot distinguish a violation from a reuse. /// `unpublished` can, but asserting on it here would only restate what /// `page_mut_cow` has already decided one frame up. /// /// The invariant is enforced the two ways PLAN amendment A1 describes /// instead: structurally, because every write to a tree node goes through /// `page_mut_cow`, and mechanically, because `protect_stable` builds make the /// image hardware read-only, which `unprotect` lifts for exactly the pages /// recycling hands back. pub inline fn page_mut(self: *Pager, p: u32) *align(page_size) [page_size]u8 { assert_msg(p < self.mapped_pages, "write to a page past the mapped end of the data file"); self.note_dirty(p, p); return @ptrCast(@alignCast(self.reserve.ptr + (@as(usize, p) << page_shift))); } /// A writable page for an owner that can be told where the page moved. /// /// `slot` must be the *single* owner of this page number. That is the /// property the whole scheme rests on: copy-on-write relocates the page, and /// if a second reference existed it would still point at the abandoned copy. /// For the B+tree that owner is the id->page table entry, which is precisely /// why node ids are not page numbers (PLAN amendment A1). /// /// The victim goes on the free list, which withholds it for two generations, /// so the image that still references it stays intact and usable. pub fn page_mut_cow(self: *Pager, slot: *u32) !*align(page_size) [page_size]u8 { if (self.is_unpublished(slot.*)) return self.page_mut(slot.*); const copy = try self.alloc_pages(1); @memcpy( @as(*[page_size]u8, @ptrCast(self.page_mut(copy))), @as(*const [page_size]u8, @ptrCast(self.page(slot.*))), ); try self.free_pages(slot.*, 1); slot.* = copy; return self.page_mut(copy); } /// Whether this page was handed out since the last publish, and so may be /// written in place. See the `unpublished` field. pub inline fn is_unpublished(self: *const Pager, p: u32) bool { return p < self.unpublished.bit_length and self.unpublished.isSet(p); } /// Reclaim the unwritten tail of an extent for appending again after a /// checkpoint, from `off` (which must already be clear of every byte the /// published image references) to the end of the extent. /// /// The append cursors need this because `publish` clears `unpublished` /// wholesale, which makes an extent the appender still owns read as part of /// the image. Without it the only safe move was to abandon the rest of the /// extent and take a fresh one -- ~8 MiB per collection at every checkpoint, /// never reclaimed when the workload produces no garbage for compaction to /// find. Measured: 40 collections of pure inserts put the data file at 11.8x /// the live data and rising by ~335 MB per checkpoint, on course to exhaust /// the address-space reservation after about 6 GB of real data. /// /// The caller's contract, which is what makes this sound: `off` is rounded up /// past the *live* append cursor to a system-page boundary, so no page in /// `[off, end)` holds a byte referenced by the image or by the live indexes. /// System pages rather than 4 KiB ones because writeback tears at the /// granularity the kernel manages -- a 4 KiB store dirties the whole 16 KiB /// page on Apple Silicon, and a torn writeback there would take out the image /// bytes sharing it. pub fn mark_appendable(self: *Pager, off: u64, end: u64) void { assert(off <= end); assert(off % map_align == 0); const first: u32 = @intCast(off >> page_shift); const last: u32 = @intCast(end >> page_shift); // exclusive if (last <= first) return; self.alloc_lock.lockUncancelable(self.io); defer self.alloc_lock.unlock(self.io); assert_msg( last <= self.mapped_pages, "marking pages appendable past the mapped end of the data file", ); self.unpublished.setRangeValue(.{ .start = first, .end = last }, true); // These pages may sit below the stable mark, where `protect_image` has // made them hardware read-only. self.unprotect(first, last - first); } /// The same question for the byte-offset consumers: may an append at `off` /// land in place, or does its page belong to the durable image? Used by the /// document slab and the overflow slab, which would otherwise have to guess /// from the offset (and, using `stable_bytes`, guessed wrong for every /// recycled extent). pub inline fn is_unpublished_at(self: *const Pager, off: u64) bool { return self.is_unpublished(@intCast(off >> page_shift)); } /// Hold off the next publish while an append decides where to put its bytes /// and puts them there. Uncancelable and infallible: the append runs after /// the log record is durable, where there is nowhere to report a failure. /// See `append_lock`. pub fn lock_append(self: *Pager) void { self.append_lock.lockSharedUncancelable(self.io); } pub fn unlock_append(self: *Pager) void { self.append_lock.unlockShared(self.io); } /// The one page a write may legitimately land on below the stable mark: a /// watermark slot. Overwriting the *inactive* slot is the whole mechanism -- /// alternating by generation parity is what makes it safe, where every other /// page below the mark is part of the image a crash must find unchanged. /// /// Deliberately not public and deliberately narrow: it takes the slot number /// and asserts it is one, so it cannot become a general escape hatch from /// the check in `page_mut`. inline fn page_mut_slot(self: *Pager, p: u32) *align(page_size) [page_size]u8 { assert(p == page_watermark_a or p == page_watermark_b); assert_msg(p < self.mapped_pages, "write to a watermark slot past the mapped end"); self.note_dirty(p, p); return @ptrCast(@alignCast(self.reserve.ptr + (@as(usize, p) << page_shift))); } /// Bytes above the stable mark, i.e. the first byte a write may touch. pub inline fn stable_bytes(self: *const Pager) u64 { return @as(u64, self.stable_pages) << page_shift; } /// Bytes at an absolute file offset, for the consumers whose references are /// byte offsets rather than page numbers: the document slab and the /// overflow slab. pub inline fn bytes(self: *const Pager, off: u64, len: usize) []const u8 { assert_msg(off + len <= @as(u64, self.mapped_pages) << page_shift, "read past the mapped end of the data file"); return self.reserve[@intCast(off)..][0..len]; } pub inline fn bytes_mut(self: *Pager, off: u64, len: usize) []u8 { assert_msg(off + len <= @as(u64, self.mapped_pages) << page_shift, "write past the mapped end of the data file"); if (track_dirty) { self.note_dirty(@intCast(off >> page_shift), @intCast((off + len - 1) >> page_shift)); } return self.reserve[@intCast(off)..][0..len]; } /// Record pages `first..=last` as written since the last sync. See /// `dirty_lock` for why this is the one shared structure on the write path. inline fn note_dirty(self: *Pager, first: u32, last: u32) void { if (!track_dirty) return; self.dirty_lock.lockUncancelable(self.io); defer self.dirty_lock.unlock(self.io); var p = first; while (p <= last) : (p += 1) self.dirty.put(self.gpa, p, {}) catch {}; } // -- allocation --------------------------------------------------------- /// 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); var hold: Reservation = .{}; try self.reserve_pages(&hold, n); return self.alloc_pages_assume_reserved(&hold, n); } /// 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. /// /// 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, hold: *Reservation, n: u32) !void { self.alloc_lock.lockUncancelable(self.io); defer self.alloc_lock.unlock(self.io); return self.reserve_pages_locked(hold, n); } /// For callers already holding `alloc_lock`. The lock is not reentrant, so /// the split is what lets `write_freelist` hold it across reading the lists /// *and* allocating the pages it writes them into. fn reserve_pages_locked(self: *Pager, hold: *Reservation, n: u32) !void { // 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 `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 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. /// Take a run from the free list if one fits, else bump the tail. A recycled /// page was inside a published image once, so its protection has to be /// lifted before it is handed out again. /// Best fit, not first fit. First fit lets single-page requests cannibalise /// the large runs: copy-on-write asks for one page thousands of times per /// generation, each carving a page off the front of whatever run comes first, /// and a 2048-page slab extent is shaved to nothing while never being usable /// as an extent. That is what the churn gate saw -- the free list draining to /// zero every generation with the file still growing by the full write volume. /// Smallest-sufficient keeps the big runs whole for the consumers that need /// them, and there is nothing else that wants the one-page holes. fn take_free(self: *Pager, n: u32) ?u32 { // The `pages == n` early exit is the exact-match shortcut only; removing // it must leave behaviour identical, and turning the loop into a plain // first-fit `break` is the mutation the extent test names. var best: ?usize = null; for (self.free_ready.items, 0..) |e, i| { if (e.pages < n) continue; if (best == null or e.pages < self.free_ready.items[best.?].pages) best = i; if (e.pages == n) break; // cannot do better than exact } const i = best orelse return null; const e = self.free_ready.items[i]; const first = e.first; if (e.pages == n) { _ = self.free_ready.swapRemove(i); } else { self.free_ready.items[i] = .{ .first = e.first + n, .pages = e.pages - n }; } return first; } /// Take a run off the free list for a document slab: at least `min_pages`, /// at most `max_pages`, starting and ending on a system-page boundary. /// Returns null when nothing on the list qualifies, and the caller bumps the /// tail instead. /// /// `take_free` cannot serve this, and that is the whole reason this exists. /// It is deliberately best fit -- smallest sufficient run -- so that the /// thousands of single-page copy-on-write requests per generation cannot /// dismantle the large runs. A slab extent asks for 2048 pages, and window /// reclamation gives back runs a few pages at a time, so with an exact-size /// rule the free list could fill up with reclaimed slab that no slab request /// would ever take: the pages come back, the file keeps growing, the ratio /// does not move. That is the failure this whole milestone is measured /// against. /// /// So this one takes a partial run when it cannot get a whole one, and is /// allowed to trim a larger one. There is no cannibalisation to fear here: /// the request is itself large (the caller's floor is 1 MiB), so what it /// leaves behind is still a usable run rather than a hole. The one-page /// requests still go through `take_free` unchanged, and its pinned mutation /// test is untouched. /// /// The alignment is not cosmetic. `map_align` is the granularity writeback /// works at and the granularity reclamation gives back at, so a run that /// starts mid-system-page both wastes its first window and shares a kernel /// page with whatever occupies the rest of it -- which for a page still in /// the published image is the tearing `mark_appendable` refuses to risk. pub fn alloc_slab_run(self: *Pager, hold: *Reservation, min_pages: u32, max_pages: u32) ?Extent { assert(min_pages > 0); assert(min_pages <= max_pages); self.alloc_lock.lockUncancelable(self.io); defer self.alloc_lock.unlock(self.io); // A split can leave a piece at each end, so one entry may become two. // Out of memory before anything is disturbed: the caller falls back to // bumping the tail, which is what it would have done anyway. self.free_ready.ensureUnusedCapacity(self.gpa, 1) catch return null; var best: ?usize = null; var best_first: u32 = 0; var best_take: u32 = 0; var best_src: u32 = 0; for (self.free_ready.items, 0..) |e, i| { const from = std.mem.alignForward(u32, e.first, pages_per_map_align); const to = std.mem.alignBackward(u32, e.first + e.pages, pages_per_map_align); if (to <= from) continue; const usable = to - from; if (usable < min_pages) continue; const take = @min(usable, max_pages); // The longest run available, so the collection switches extents as // rarely as possible -- every switch abandons what is left of the // one before it. Ties go to the smallest source run, which leaves // the big ones as whole as it can. `best_take` starts at zero and // every candidate takes at least `min_pages`, so "nothing yet" is // already encoded. if (take > best_take or (take == best_take and e.pages < best_src)) { best = i; best_first = from; best_take = take; best_src = e.pages; } } const i = best orelse return null; const e = self.free_ready.items[i]; const head = best_first - e.first; const tail_first = best_first + best_take; const tail = (e.first + e.pages) - tail_first; if (head > 0) { self.free_ready.items[i] = .{ .first = e.first, .pages = head }; if (tail > 0) self.free_ready.appendAssumeCapacity(.{ .first = tail_first, .pages = tail }); } else if (tail > 0) { self.free_ready.items[i] = .{ .first = tail_first, .pages = tail }; } else { _ = self.free_ready.swapRemove(i); } self.claim_locked(hold, best_first, best_take); return .{ .first = best_first, .pages = best_take }; } /// Merge runs that touch, so the holes single-page frees leave behind can add /// up to an extent again. Without it the free list only ever fragments: every /// generation returns thousands of one-page copy-on-write victims, and an /// 8 MiB slab extent request never finds a home however much space is free. /// Once per publish, over a list whose length is the generation's free count. fn coalesce_free_ready(self: *Pager) void { const items = self.free_ready.items; if (items.len < 2) return; std.mem.sort(Extent, items, {}, struct { fn less(_: void, a: Extent, b: Extent) bool { return a.first < b.first; } }.less); var w: usize = 0; for (items[1..]) |e| { const prev = &items[w]; if (prev.first + prev.pages == e.first) { prev.pages += e.pages; } else { w += 1; items[w] = e; } } self.free_ready.items.len = w + 1; } pub fn alloc_pages_assume_reserved(self: *Pager, hold: *Reservation, n: u32) u32 { self.alloc_lock.lockUncancelable(self.io); defer self.alloc_lock.unlock(self.io); return self.alloc_assume_reserved_locked(hold, n); } /// For callers already holding `alloc_lock`; see `reserve_pages_locked`. fn alloc_assume_reserved_locked(self: *Pager, hold: *Reservation, n: u32) u32 { assert(n > 0); assert_msg( self.alloc_tail + n <= self.mapped_pages, "page allocation past the mapped end of the data file", ); // 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). if (self.take_free(n)) |recycled| { self.claim_locked(hold, recycled, n); return recycled; } const first = self.alloc_tail; self.alloc_tail += n; self.claim_locked(hold, first, n); return first; } /// Charge `[first, first+pages)` against the reservation and make it /// writable. The one place a claim is booked, because there are two /// allocation policies above it and hand-copying this is how they drift -- /// the copy in `alloc_slab_run` had already lost one of the preconditions. fn claim_locked(self: *Pager, hold: *Reservation, first: u32, pages: u32) void { assert_msg(pages <= hold.pages, "page allocation overran reserve_pages' promise"); assert_msg(pages <= self.reserved_pages, "page allocation overran the pager's total promise"); self.reserved_pages -= pages; hold.pages -= pages; // A recycled page was inside a published image once, so its protection // has to be lifted before it is handed out again. self.unprotect(first, pages); self.mark_unpublished(first, pages); } fn mark_unpublished(self: *Pager, first: u32, n: u32) void { // `grow_to` sizes the set to the mapping, and `reserve_pages` has already // grown the mapping past this run, so the range is in bounds. assert_msg( @as(usize, first) + n <= self.unpublished.bit_length, "allocated a page outside the unpublished set", ); self.unpublished.setRangeValue(.{ .start = first, .end = first + n }, true); } /// Bytes currently allocated, i.e. the extent of the live image. pub fn allocated_bytes(self: *const Pager) u64 { return @as(u64, self.alloc_tail) << page_shift; } // -- durability --------------------------------------------------------- /// Flush every allocated page and the inode. Only the checkpoint calls /// this: between checkpoints dirty pages may sit in the page cache /// indefinitely, because recovery is `image + replay(seq > watermark)` and /// the image's pages are never written *differently*. That is what keeps /// the write path at exactly one fsync — the WAL's (PLAN amendment A1). pub fn sync(self: *Pager) !void { const len = std.mem.alignForward(usize, @intCast(self.allocated_bytes()), map_align); if (len > 0) { try std.posix.msync(self.reserve[0..len], std.posix.MSF.SYNC); } try self.file.sync(self.io); if (track_dirty) { self.dirty_lock.lockUncancelable(self.io); defer self.dirty_lock.unlock(self.io); self.dirty.clearRetainingCapacity(); } } /// Make the published image read-only at the hardware level. See /// `protect_stable`. The watermark slots stay writable: overwriting the /// inactive one is the publication mechanism, not a violation. fn protect_image(self: *Pager) void { if (!protect_stable) return; const first = std.mem.alignForward(usize, @as(usize, page_first_data) << page_shift, map_align); const end = std.mem.alignBackward(usize, @as(usize, self.stable_pages) << page_shift, map_align); if (end <= first) return; // Best effort as far as the database is concerned -- losing the check // leaves a weaker test build, not wrong data -- but recorded, so a test // can tell the difference between a belt that is working and one that is // silently doing nothing. std.posix has no mprotect wrapper in Zig 0.16, // hence the libc call. const rc = std.c.mprotect(@ptrCast(@alignCast(self.reserve.ptr + first)), end - first, .{ .READ = true }); self.protect_ok = rc == 0; } /// Lift the protection over a range that is about to leave the image -- /// which only copy-on-write does, when it recycles a freed page. fn unprotect(self: *Pager, first_page: u32, pages: u32) void { if (!protect_stable) return; const first = std.mem.alignBackward(usize, @as(usize, first_page) << page_shift, map_align); const end = std.mem.alignForward(usize, @as(usize, first_page + pages) << page_shift, map_align); _ = std.c.mprotect(@ptrCast(@alignCast(self.reserve.ptr + first)), end - first, .{ .READ = true, .WRITE = true }); } /// Test-only: overwrite, in the *file*, every page written since the last /// sync -- the state a power loss can leave behind and a `kill -9` cannot. /// Call it with the mapping already closed. fn simulate_lost_writeback(gpa: std.mem.Allocator, io: std.Io, path: []const u8, lost: []const u32) !void { var f = try std.Io.Dir.cwd().openFile(io, path, .{ .mode = .read_write }); defer f.close(io); const junk = try gpa.alloc(u8, page_size); defer gpa.free(junk); // Not zeroes: zeroes are a plausible value, and a test that passes // because the lost page happened to read as zero proves nothing. @memset(junk, 0xDD); for (lost) |p| try f.writePositionalAll(io, junk, @as(u64, p) << page_shift); } /// Test-only: the pages `simulate_lost_writeback` would discard. fn dirty_pages(self: *const Pager, gpa: std.mem.Allocator) ![]u32 { var out: std.ArrayListUnmanaged(u32) = .empty; errdefer out.deinit(gpa); var it = self.dirty.keyIterator(); while (it.next()) |k| try out.append(gpa, k.*); return out.toOwnedSlice(gpa); } // -- growth ------------------------------------------------------------- /// Make at least `want_pages` pages mapped and file-backed. /// /// The order is the whole point: extend the file, *then* map the new /// suffix. A store into a mapped page past end-of-file raises SIGBUS, which /// no Zig error path can catch, so the file must be long enough before any /// page in the range is reachable. The accessors assert against /// `mapped_pages` so a violation is a panic with a message rather than a /// signal. fn grow_to(self: *Pager, want_pages: u32) !void { if (want_pages <= self.mapped_pages) return; // Round up to a growth chunk, and to the system page size, so a // 16 KiB-page host never gets a partial mapping request. // // `alignForwardAnyAlign`, not `alignForward`: the chunk is a *proportion* // of the current size once the file passes 64 MiB, and `mapped_pages / 8` // is not a power of two. `alignForward` asserts that it is -- so this // panicked in safe builds and, worse, in ReleaseFast (where the assert is // compiled out) computed `(addr + align - 1) & ~(align - 1)` with a // non-power-of-two mask, which can round *down*. A mapping longer than the // file is the one thing this function exists to prevent: a store into a // mapped page past end-of-file raises SIGBUS, which no error path catches. // // Never noticed because no unit test grew a pager past 64 MiB, which is // where the chunk stops being `grow_chunk_pages`. const chunk = @max(grow_chunk_pages, self.mapped_pages / 8); var new_pages = std.mem.alignForwardAnyAlign(u32, want_pages, chunk); const sys_pages: u32 = pages_per_map_align; if (sys_pages > 1) new_pages = std.mem.alignForward(u32, new_pages, sys_pages); if (@as(u64, new_pages) << page_shift > self.reserve.len) { // One more try at exactly what was asked for: a reservation the // chunking would overshoot is still usable up to its end. new_pages = want_pages; if (sys_pages > 1) new_pages = std.mem.alignForward(u32, new_pages, sys_pages); if (@as(u64, new_pages) << page_shift > self.reserve.len) return Error.DatabaseTooLarge; } const new_len: u64 = @as(u64, new_pages) << page_shift; if (new_len > self.file_pages_bytes()) { try self.file.setLength(self.io, new_len); self.file_pages = new_pages; } // Before the mapping grows, so a failure here cannot leave pages // reachable that the set has no bit for. try self.unpublished.resize(self.gpa, new_pages, false); const off: usize = @as(usize, self.mapped_pages) << page_shift; const len: usize = @intCast(new_len - (@as(u64, self.mapped_pages) << page_shift)); _ = try std.posix.mmap( @alignCast(self.reserve.ptr + off), len, .{ .READ = true, .WRITE = true }, .{ .TYPE = .SHARED, .FIXED = true }, self.file.handle, off, ); self.mapped_pages = new_pages; } fn file_pages_bytes(self: *const Pager) u64 { return @as(u64, self.file_pages) << page_shift; } // -- watermark ---------------------------------------------------------- /// Adopt the newest valid watermark, or none. Never fails on a damaged /// watermark: a data file whose slots are both unreadable is treated as /// having no checkpoint, which means a full replay -- the database must /// always open (ground rule 4). The caller learns which happened from /// `loaded.generation == 0`. fn load_watermark(self: *Pager) !void { const a = self.read_slot(page_watermark_a); const b = self.read_slot(page_watermark_b); // Validate first, compare generations second. The other order picks a // torn slot whenever its garbage generation happens to be larger. const pick: ?Watermark = if (a != null and b != null) (if (a.?.generation >= b.?.generation) a else b) else if (a != null) a else b; if (pick) |wm| { self.loaded = wm; self.generation = wm.generation; self.alloc_tail = wm.alloc_tail; self.reserved_pages = 0; // 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 // read: a data file that simply never reached its first checkpoint // is the normal state of a young database, and warning about it // trains people to ignore the warning that matters. std.debug.print( "multiforadb: WARNING: data file '{s}' has a damaged watermark; " ++ "treating it as having no checkpoint and replaying the log in full\n", .{self.path}, ); } } /// Whether either slot holds anything at all. A never-checkpointed file has /// both slots as written by create: all zeroes. fn watermark_attempted(self: *const Pager) bool { for ([_]u32{ page_watermark_a, page_watermark_b }) |p| { if (p >= self.mapped_pages) continue; for (self.page(p)) |byte| if (byte != 0) return true; } return false; } fn read_slot(self: *const Pager, p: u32) ?Watermark { if (p >= self.mapped_pages) return null; const b = self.page(p); const want = std.mem.readInt(u64, b[wm_hashed_len..][0..8], .little); if (header_hash(b[0..wm_hashed_len]) != want) return null; const wm: Watermark = .{ .generation = std.mem.readInt(u64, b[0..8], .little), .seq = std.mem.readInt(u64, b[8..16], .little), .alloc_tail = @intCast(std.mem.readInt(u64, b[16..24], .little)), .file_pages = @intCast(std.mem.readInt(u64, b[24..32], .little)), .catalog_page = @intCast(std.mem.readInt(u64, b[32..40], .little)), .catalog_len = std.mem.readInt(u64, b[40..48], .little), .freelist_page = @intCast(std.mem.readInt(u64, b[48..56], .little)), .freelist_len = std.mem.readInt(u64, b[56..64], .little), .prev_generation = std.mem.readInt(u64, b[64..72], .little), .live_docs = std.mem.readInt(u64, b[72..80], .little), .dead_bytes = std.mem.readInt(u64, b[80..88], .little), }; // A slot whose hash matches but whose contents are impossible is worse // than a torn one, because it would be trusted. // // The generation check is belt-and-braces: an all-zero slot is already // rejected by the hash, so no test reddens when it is removed. Kept // because it costs nothing and it is the field every comparison below // depends on. The alloc_tail check is *not* redundant -- a slot can be // rehashed to validate with nonsense in it, which is what the // "hash-valid but impossible" test does. if (wm.generation == 0) return null; if (wm.alloc_tail < page_first_data) return null; return wm; } /// Publish `wm` as the new durable image. /// /// The ordering here is the entire crash-recovery argument, so it is worth /// stating plainly: every page the watermark describes is made durable /// *before* the watermark that describes them. Reverse the two and a crash /// in between leaves a watermark pointing at pages that were never written, /// with the log already truncated below it -- unrecoverable. /// /// The slot alternates by generation parity, so the previous watermark /// survives this write and a torn slot leaves the older one intact. pub fn publish(self: *Pager, wm_in: Watermark) !void { var wm = wm_in; wm.generation = self.generation + 1; wm.prev_generation = self.generation; // 1. every data page durable. try self.sync(); // 2. the free list, into fresh pages so the old one stays intact. const fl = try self.write_freelist(); wm.freelist_page = fl.first; wm.freelist_len = fl.len; try self.sync(); // 3. only now record the extent, *after* the free list has taken its // pages. Capturing alloc_tail before this left those pages outside // the published image, so the next open would hand them out again // while the watermark still pointed at them. wm.alloc_tail = self.alloc_tail; wm.file_pages = self.file_pages; // 4. the watermark itself, last. const slot = if (wm.generation & 1 == 1) page_watermark_a else page_watermark_b; const b = self.page_mut_slot(slot); @memset(b, 0); std.mem.writeInt(u64, b[0..8], wm.generation, .little); std.mem.writeInt(u64, b[8..16], wm.seq, .little); std.mem.writeInt(u64, b[16..24], wm.alloc_tail, .little); std.mem.writeInt(u64, b[24..32], wm.file_pages, .little); std.mem.writeInt(u64, b[32..40], wm.catalog_page, .little); std.mem.writeInt(u64, b[40..48], wm.catalog_len, .little); std.mem.writeInt(u64, b[48..56], wm.freelist_page, .little); std.mem.writeInt(u64, b[56..64], wm.freelist_len, .little); std.mem.writeInt(u64, b[64..72], wm.prev_generation, .little); std.mem.writeInt(u64, b[72..80], wm.live_docs, .little); std.mem.writeInt(u64, b[80..88], wm.dead_bytes, .little); std.mem.writeInt(u64, b[wm_hashed_len..][0..8], header_hash(b[0..wm_hashed_len]), .little); const off = @as(usize, slot) << page_shift; const end = std.mem.alignForward(usize, off + page_size, map_align); const start = std.mem.alignBackward(usize, off, map_align); try std.posix.msync(@alignCast(self.reserve[start..end]), std.posix.MSF.SYNC); try self.file.sync(self.io); // Deliberately does *not* clear the dirty record. After a correct // publish the only page left in it is this slot, which this msync just // made durable -- so a test can discard everything else and assert the // image still loads. Clearing it here would make that check vacuous. // 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. // // Exclusive against the appenders for this step alone. Freezing the // image while one of them is between "is my cursor still writable" // and the copy that relies on the answer is what put documents into // the durable image; holding them off here is what makes the answer // still true when it is used. Taken before `alloc_lock`, the order // `mark_appendable` uses on the appender's side. self.append_lock.lockUncancelable(self.io); defer self.append_lock.unlock(self.io); 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); self.free_hold.clearRetainingCapacity(); try self.free_hold.appendSlice(self.gpa, self.free_pending.items); self.free_pending.clearRetainingCapacity(); self.coalesce_free_ready(); // Every page handed out so far is now part of the published image, so // nothing may be written in place any more until it is allocated afresh. self.unpublished.setRangeValue(.{ .start = 0, .end = self.unpublished.bit_length }, false); } // -- free list ---------------------------------------------------------- /// Give back a run of pages. They become reusable two generations later -- /// see the field comment on `free_ready`. /// /// Under the allocation lock, like every other mutation of the free lists. /// It was not, and the hot caller is `page_mut_cow`, which runs under a /// *collection* lock: two collections doing copy-on-write concurrently /// appended to the same list, and `publish` rotated all three lists /// underneath them. No caller holds the lock already -- `page_mut_cow` takes /// it inside `alloc_pages` and has released it by here -- so this cannot /// recurse. pub fn free_pages(self: *Pager, first: u32, pages: u32) !void { if (pages == 0) return; self.alloc_lock.lockUncancelable(self.io); defer self.alloc_lock.unlock(self.io); try self.free_pending.append(self.gpa, .{ .first = first, .pages = pages }); } /// Whether any page of `[first, first+pages)` has been handed to the free /// list. A consumer that still claims one is claiming a page the pager is /// about to give to somebody else, and the symptom is a document quietly /// overwritten rather than anything failing -- so this is the detector the /// enlarged free list deserves, and it is why the free lists are readable /// from outside at all. /// /// Walks three lists, so it is for assertions in test and Debug builds. pub fn owns_freed(self: *Pager, first: u32, pages: u32) bool { self.alloc_lock.lockUncancelable(self.io); defer self.alloc_lock.unlock(self.io); for ([_][]const Extent{ self.free_pending.items, self.free_hold.items, self.free_ready.items, }) |list| { for (list) |e| { if (first < e.first + e.pages and e.first < first + pages) return true; } } return false; } /// Pages available for immediate reuse. pub fn free_ready_pages(self: *const Pager) u32 { var n: u32 = 0; for (self.free_ready.items) |e| n += e.pages; return n; } 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 // fit removes the entry it took (`take_free`). A count captured // beforehand therefore claims one entry more than the loop writes: the // hash lands eight bytes short of where `read_freelist` looks for it and // the whole list is dropped as corrupt on the next open. The stream is // one page and a one-page run is the commonest thing on the list, so // that is the ordinary case rather than a corner. // // `take_free` never *adds* an entry, so the bound holds and one // allocation is enough. // // Under `alloc_lock` for the whole of it, allocation included. Reading // the three lists is as much a use of them as appending is: a concurrent // `free_pages` -- copy-on-write in some collection, which a checkpoint // does not exclude -- grows `free_pending` while the loop below walks it, // and a growth that reallocates leaves the loop on freed memory. A free // that lands after this point simply waits for the next generation's // list; the page stays allocated one generation longer, which is the // safe direction. self.alloc_lock.lockUncancelable(self.io); defer self.alloc_lock.unlock(self.io); const bound = self.free_ready.items.len + self.free_hold.items.len + self.free_pending.items.len; const pages: u32 = pages_for(8 + bound * 8 + 8); var hold: Reservation = .{}; try self.reserve_pages_locked(&hold, pages); const first = self.alloc_assume_reserved_locked(&hold, pages); const buf = self.bytes_mut(@as(u64, first) << page_shift, @as(usize, pages) << page_shift); @memset(buf, 0); var at: usize = 8; for ([_][]const Extent{ self.free_ready.items, self.free_hold.items, self.free_pending.items, }) |list| { for (list) |e| { std.mem.writeInt(u32, buf[at..][0..4], e.first, .little); std.mem.writeInt(u32, buf[at + 4 ..][0..4], e.pages, .little); at += 8; } } const count = (at - 8) / 8; assert_msg( count == bound or count + 1 == bound, "the free list changed size while it was being written", ); std.mem.writeInt(u64, buf[0..8], count, .little); std.mem.writeInt(u64, buf[at..][0..8], header_hash(buf[0..at]), .little); // `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 /// warning rather than refused: losing it wastes space, refusing to open /// loses the database. fn read_freelist(self: *Pager, wm: Watermark) !void { if (wm.freelist_len == 0 or wm.freelist_page == 0) return; if (wm.freelist_page >= self.mapped_pages) return; const buf = self.bytes(@as(u64, wm.freelist_page) << page_shift, @intCast(wm.freelist_len)); const count = std.mem.readInt(u64, buf[0..8], .little); const want_len = 8 + count * 8 + 8; if (want_len != wm.freelist_len) return; const at: usize = @intCast(8 + count * 8); if (header_hash(buf[0..at]) != std.mem.readInt(u64, buf[at..][0..8], .little)) { std.debug.print("multiforadb: WARNING: data file free list is corrupt; its pages stay in use\n", .{}); return; } // Everything the previous generations freed is reusable now: this open // *is* a new generation, and nothing from before it is referenced. var i: usize = 0; while (i < count) : (i += 1) { const o = 8 + i * 8; try self.free_ready.append(self.gpa, .{ .first = std.mem.readInt(u32, buf[o..][0..4], .little), .pages = std.mem.readInt(u32, buf[o + 4 ..][0..4], .little), }); } } // -- header ------------------------------------------------------------- fn write_header(self: *Pager) !void { const h = self.page_mut(page_header); @memset(h, 0); std.mem.writeInt(u32, h[0..4], magic, .little); h[4] = format_version; h[5] = page_shift; std.mem.writeInt(u128, h[8..24], self.uuid, .little); std.mem.writeInt(u64, h[24..32], header_hash(h[0..header_hashed_len]), .little); } fn read_header(self: *Pager) !void { const h = self.page(page_header); if (std.mem.readInt(u32, h[0..4], .little) != magic) return Error.InvalidDataFile; if (h[4] != format_version) return Error.InvalidDataFile; if (h[5] != page_shift) return Error.InvalidDataFile; const want = std.mem.readInt(u64, h[24..32], .little); if (header_hash(h[0..header_hashed_len]) != want) return Error.InvalidDataFile; const file_uuid = std.mem.readInt(u128, h[8..24], .little); if (self.uuid == 0) { self.uuid = file_uuid; } else if (file_uuid != self.uuid) { return Error.DatabaseMismatch; } } }; fn header_hash(b: []const u8) u64 { return std.hash.XxHash3.hash(0, b); } /// Reserve address space without committing memory: PROT_NONE, anonymous and /// NORESERVE is one VMA and zero pages, and is not charged even under strict /// overcommit. Nothing is mapped to the file yet, so nothing here can fault. /// /// Halves on refusal rather than failing outright, so a container with a tight /// address-space limit still opens — with a smaller ceiling, which `grow_to` /// reports as DatabaseTooLarge if it is ever actually reached. fn reserve_address_space(want: usize) ![]align(map_align) u8 { var len = std.mem.alignForward(usize, want, map_align); while (true) { if (std.posix.mmap( null, len, .{}, // PROT_NONE .{ .TYPE = .PRIVATE, .ANONYMOUS = true, .NORESERVE = true }, -1, 0, )) |m| { return m; } else |err| { if (len <= 1 << 30) return err; len /= 2; } } } // --------------------------------------------------------------------------- // Tests // --------------------------------------------------------------------------- const testing = std.testing; /// A pager over a temp file, with a reservation small enough that the growth /// ceiling is reachable in a test. const TmpPager = struct { tmp: std.testing.TmpDir, path: []u8, /// Optional so a test can close it early -- several deliberately break the /// file and must not have to reopen one just to satisfy teardown. pager: ?Pager, fn init(io: std.Io, reserve_bytes: usize) !TmpPager { const tmp = std.testing.tmpDir(.{}); const path = try std.fmt.allocPrint( testing.allocator, ".zig-cache/tmp/{s}/data", .{tmp.sub_path}, ); return .{ .tmp = tmp, .path = path, .pager = try Pager.open(testing.allocator, io, path, .{ .uuid = 7, .reserve_bytes = reserve_bytes }), }; } fn deinit(self: *TmpPager) void { if (self.pager) |*p| p.deinit(); self.pager = null; self.tmp.cleanup(); testing.allocator.free(self.path); } fn close(self: *TmpPager) void { if (self.pager) |*p| p.deinit(); self.pager = null; } fn pg(self: *TmpPager) *Pager { return &self.pager.?; } }; test "pages survive growth and the mapping base never moves" { // The reason the fixed reservation exists: a pointer handed out before a // growth must still be valid after it. The ArrayList-backed arena this // replaces could not promise that, and the workarounds for it (copying // promoted keys into a scratch buffer) exist only because of it. // // Mutation check: make grow_to munmap and re-map the whole prefix at a // kernel-chosen address and the base assertion goes red. 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(); const base = @intFromPtr(pg.reserve.ptr); const first = try pg.alloc_pages(1); const first_ptr = pg.page_mut(first); @memset(first_ptr, 0xAB); // Grow well past several chunk boundaries. var written: std.ArrayListUnmanaged(u32) = .empty; defer written.deinit(testing.allocator); for (0..6000) |i| { const p = try pg.alloc_pages(1); const bytes = pg.page_mut(p); @memset(bytes, @truncate(i)); try written.append(testing.allocator, p); } try testing.expectEqual(base, @intFromPtr(pg.reserve.ptr)); // The pointer taken before all that growth still addresses its page. for (first_ptr) |b| try testing.expectEqual(@as(u8, 0xAB), b); for (written.items, 0..) |p, i| { const want: u8 = @truncate(i); try testing.expectEqual(want, pg.page(p)[0]); try testing.expectEqual(want, pg.page(p)[page_size - 1]); } } test "the file is extended before any page in the range is reachable" { // Write-then-extend, asserted as an ordering rather than by provoking the // failure: a store past end-of-file is SIGBUS, which cannot be caught // in-process, so the check is that the file is always long enough for // everything mapped, and everything allocated is always mapped. 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(); for (0..40) |_| { _ = try pg.alloc_pages(97); // not a chunk divisor, so growth is ragged try testing.expect(pg.alloc_tail <= pg.mapped_pages); try testing.expect(pg.mapped_pages <= pg.file_pages); const on_disk = try pg.file.length(io); try testing.expect(on_disk >= @as(u64, pg.mapped_pages) << page_shift); } } test "two consumers reserving before one commit both keep their promise" { // The reservation is a count, not a tail mark, and this is why. An upsert // reserves tree pages for every index *and* slab room for the document, // all before the log append, and both allocations happen after it. With a // tail mark the second reserver overwrote the first one's promise and the // first one's allocation then asserted -- which is how this was found, on a // 512 MB load, by the tripwire in alloc_node. // // Mutation check: make reserve_pages assign `alloc_tail + n` instead of // accumulating, and this goes red. 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(); // Consumer A reserves a few pages, then consumer B reserves a large extent // and takes it -- exactly the order upsert uses. 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(&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(); const io = threaded.io(); var tp = try TmpPager.init(io, 256 << 20); defer tp.deinit(); const pg = tp.pg(); 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(&hold, 1); try testing.expectEqual(before + 64, pg.alloc_tail); } test "growth past the reservation is an error, not a crash" { // A bounded reservation must fail cleanly at its ceiling: this is the path // a container with a tight address-space limit takes. var threaded: std.Io.Threaded = .init_single_threaded; defer threaded.deinit(); const io = threaded.io(); var tp = try TmpPager.init(io, 1 << 30); defer tp.deinit(); const pg = tp.pg(); const capacity: u32 = @intCast(pg.reserve.len / page_size); try testing.expectError(Error.DatabaseTooLarge, pg.alloc_pages(capacity + 1)); // And the pager is still usable afterwards. const p = try pg.alloc_pages(1); @memset(pg.page_mut(p), 1); } test "header round-trips and rejects a foreign database" { 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(); try testing.expect(tp.pg().fresh); const p = try tp.pg().alloc_pages(3); @memset(tp.pg().page_mut(p), 0x5A); try tp.pg().sync(); const path = try testing.allocator.dupe(u8, tp.path); defer testing.allocator.free(path); tp.close(); // Reopening finds the same header and is not fresh. var again = try Pager.open(testing.allocator, io, path, .{ .uuid = 7, .reserve_bytes = 64 << 20 }); try testing.expect(!again.fresh); try testing.expectEqual(@as(u128, 7), again.uuid); try testing.expectEqual(@as(u8, 0x5A), again.page(p)[0]); again.deinit(); // A different uuid is a different database; pairing them would silently // replay one database's log onto another's checkpoint. // Mutation check: drop the uuid comparison in read_header and this passes. try testing.expectError( Error.DatabaseMismatch, Pager.open(testing.allocator, io, path, .{ .uuid = 8, .reserve_bytes = 64 << 20 }), ); } test "a corrupt header is rejected rather than misread" { 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(); const path = try testing.allocator.dupe(u8, tp.path); defer testing.allocator.free(path); try tp.pg().sync(); tp.close(); // Flip a byte inside the hashed region. { var f = try std.Io.Dir.cwd().openFile(io, path, .{ .mode = .read_write }); defer f.close(io); var b: [1]u8 = undefined; _ = try f.readPositionalAll(io, &b, 6); b[0] ^= 0xFF; try f.writePositionalAll(io, &b, 6); } // Mutation check: drop the hash comparison in read_header and this passes, // which is worse than failing — a header believed on faith describes where // every structure in the file lives. try testing.expectError( Error.InvalidDataFile, Pager.open(testing.allocator, io, path, .{ .uuid = 7, .reserve_bytes = 64 << 20 }), ); } /// Reopen the same file, so a test can check what survived a publish. fn reopen(io: std.Io, path: []const u8) !Pager { return Pager.open(testing.allocator, io, path, .{ .uuid = 7, .reserve_bytes = 64 << 20 }); } test "a published watermark is what the next open loads" { 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(); const p = try tp.pg().alloc_pages(4); @memset(tp.pg().page_mut(p), 0xC3); try tp.pg().publish(.{ .seq = 4242, .catalog_page = p, .catalog_len = 99, .live_docs = 7 }); // After publish, not before: publish allocates the free list's own pages, // and the watermark has to cover them. const tail_at_publish = tp.pg().alloc_tail; tp.close(); var again = try reopen(io, tp.path); defer again.deinit(); try testing.expect(!again.fresh); try testing.expectEqual(@as(u64, 1), again.loaded.generation); try testing.expectEqual(@as(u64, 4242), again.loaded.seq); try testing.expectEqual(p, again.loaded.catalog_page); try testing.expectEqual(@as(u64, 99), again.loaded.catalog_len); try testing.expectEqual(@as(u64, 7), again.loaded.live_docs); try testing.expectEqual(tail_at_publish, again.alloc_tail); // Everything the image references is off limits to writes. try testing.expectEqual(tail_at_publish, again.stable_pages); try testing.expectEqual(@as(u8, 0xC3), again.page(p)[0]); } test "slots alternate by generation, so a torn write leaves the older one" { // Writing each new generation over the only copy of the previous one risks // having *neither* valid after a crash. Alternating is what makes a torn // slot survivable, so both halves are asserted: the parity, and that // corrupting the newer slot falls back rather than failing. // // Mutation checks: always write slot A, and the fallback finds nothing; // compare generations before validating the hash, and the torn slot wins. 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(); // Three generations: A, B, A. try tp.pg().publish(.{ .seq = 10 }); try tp.pg().publish(.{ .seq = 20 }); try tp.pg().publish(.{ .seq = 30 }); try testing.expectEqual(@as(u64, 3), tp.pg().generation); // Generation 3 is odd, so slot A. Generation 2 must still be in slot B. const a_gen = std.mem.readInt(u64, tp.pg().page(page_watermark_a)[0..8], .little); const b_gen = std.mem.readInt(u64, tp.pg().page(page_watermark_b)[0..8], .little); try testing.expectEqual(@as(u64, 3), a_gen); try testing.expectEqual(@as(u64, 2), b_gen); tp.close(); // Tear the newer slot: the open must fall back to generation 2, not fail // and not believe the garbage. { var f = try std.Io.Dir.cwd().openFile(io, tp.path, .{ .mode = .read_write }); defer f.close(io); var junk: [64]u8 = undefined; @memset(&junk, 0xFF); try f.writePositionalAll(io, &junk, page_size); } var again = try reopen(io, tp.path); defer again.deinit(); try testing.expectEqual(@as(u64, 2), again.loaded.generation); try testing.expectEqual(@as(u64, 20), again.loaded.seq); } test "both slots unreadable opens anyway, with no checkpoint" { // Ground rule 4: the database must always open. Once the log is truncated // this costs data, which is why it warns -- but refusing to start would // cost all of it. // // Mutation check: return an error from load_watermark instead and this // fails outright. 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(); try tp.pg().publish(.{ .seq = 5 }); try tp.pg().publish(.{ .seq = 6 }); tp.close(); { var f = try std.Io.Dir.cwd().openFile(io, tp.path, .{ .mode = .read_write }); defer f.close(io); var junk: [page_size * 2]u8 = undefined; @memset(&junk, 0xA5); try f.writePositionalAll(io, &junk, page_size); } var again = try reopen(io, tp.path); defer again.deinit(); try testing.expectEqual(@as(u64, 0), again.loaded.generation); try testing.expectEqual(@as(u64, 0), again.loaded.seq); } test "a hash-valid but impossible slot is refused" { // A slot that validates is *trusted*, so it has to be checked for sense as // well as integrity: a generation of 0 or an alloc_tail below the first data // page would be believed and would describe a file that does not exist. 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(); try tp.pg().publish(.{ .seq = 11 }); tp.close(); { var f = try std.Io.Dir.cwd().openFile(io, tp.path, .{ .mode = .read_write }); defer f.close(io); var b: [page_size]u8 = undefined; _ = try f.readPositionalAll(io, &b, page_size); // slot A holds gen 1 // alloc_tail = 1, below page_first_data, and re-hash so it validates. std.mem.writeInt(u64, b[16..24], 1, .little); std.mem.writeInt(u64, b[wm_hashed_len..][0..8], header_hash(b[0..wm_hashed_len]), .little); try f.writePositionalAll(io, &b, page_size); } var again = try reopen(io, tp.path); defer again.deinit(); // Refused, so no checkpoint rather than a nonsensical one. try testing.expectEqual(@as(u64, 0), again.loaded.generation); } test "freed pages are withheld for two generations and survive a reopen" { // The delay is what keeps generation N-1 usable as a fallback image: its // pages must still be allocated while N is current. // // Mutation check: move free_pending straight into free_ready at publish and // the withholding assertions go red. 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(); const victim = try tp.pg().alloc_pages(2); @memset(tp.pg().page_mut(victim), 0x11); try tp.pg().publish(.{ .seq = 1 }); try tp.pg().free_pages(victim, 2); try testing.expectEqual(@as(u32, 0), tp.pg().free_ready_pages()); try tp.pg().publish(.{ .seq = 2 }); // pending -> hold try testing.expectEqual(@as(u32, 0), tp.pg().free_ready_pages()); // Still intact, which is the point: generation 1 references it. try testing.expectEqual(@as(u8, 0x11), tp.pg().page(victim)[0]); try tp.pg().publish(.{ .seq = 3 }); // hold -> ready try testing.expectEqual(@as(u32, 2), tp.pg().free_ready_pages()); tp.close(); // And the list itself is durable. var again = try reopen(io, tp.path); defer again.deinit(); try testing.expectEqual(@as(u32, 2), again.free_ready_pages()); } test "the persisted free list survives allocating its own pages" { // `write_freelist` allocates the pages it is about to write into, and that // allocation goes through `take_free` like any other. On an exact fit the // entry is removed, so a count captured beforehand describes one entry more // than the loop writes, the hash lands short of where the reader looks, and // the whole list is dropped as corrupt on the next open. // // The stream is one page and a one-page run is the commonest thing on the // list, so this is the normal case, not a corner. The two-generation test // above misses it because its free run is two pages and the stream asks for // one: shrinking an entry keeps the count right, only removing it does not. // // Mutation check: compute `count` before `alloc_pages` again and the reopen // assertion goes red with "data file free list is corrupt". 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(); // Five pages, of which three go back one at a time with a gap between each // -- adjacent runs would be coalesced back into one and the list would be // too short for a lost entry to show. const base = try tp.pg().alloc_pages(5); try tp.pg().publish(.{ .seq = 1 }); try tp.pg().free_pages(base, 1); try tp.pg().free_pages(base + 2, 1); try tp.pg().free_pages(base + 4, 1); try tp.pg().publish(.{ .seq = 2 }); // pending -> hold try tp.pg().publish(.{ .seq = 3 }); // hold -> ready try testing.expectEqual(@as(u32, 3), tp.pg().free_ready_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.expect(ready_before > 0); tp.close(); var again = try reopen(io, tp.path); defer again.deinit(); try testing.expectEqual(ready_before, again.free_ready_pages()); } test "the watermark is never published before the pages it describes" { // The load-bearing ordering of the whole design: every page a watermark // describes is durable before the watermark that describes it. Reverse them // and a crash in between leaves a watermark pointing at pages that were // never written, with the log already truncated below it. // // A kill -9 cannot produce that state -- the page cache survives the // process -- so this simulates the writeback that never happened, using the // pager's own record of which pages were dirtied since the last sync. That // is the only honest way to make the ordering mutation-checkable, and // without it the mutation "delete the sync before the watermark" leaves // every other test in this file green. // // Mutation checks, stated precisely because the obvious one is not a // violation: publish syncs twice before the watermark (data pages, then the // free list), so removing *either* call alone is harmless and leaves this // green -- correctly. What it catches is removing both, or moving them after // the watermark write. The invariant is "at least one full sync between the // last data write and the watermark", not any particular call. const gpa = testing.allocator; 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(); // A page the checkpoint will reference, written and then published. const cat = try tp.pg().alloc_pages(2); @memset(tp.pg().page_mut(cat), 0x77); @memset(tp.pg().page_mut(cat + 1), 0x77); try tp.pg().publish(.{ .seq = 900, .catalog_page = cat, .catalog_len = 2 * page_size }); // After a correct publish the only page still pending is the watermark slot // itself, which its own msync made durable. Everything else the watermark // references was flushed first, so discarding whatever is still pending -- // minus that slot -- must leave a loadable image. const pending = try tp.pg().dirty_pages(gpa); defer gpa.free(pending); const slot: u32 = if (tp.pg().generation & 1 == 1) page_watermark_a else page_watermark_b; var lost: std.ArrayListUnmanaged(u32) = .empty; defer lost.deinit(gpa); for (pending) |pg_no| if (pg_no != slot) try lost.append(gpa, pg_no); tp.close(); try Pager.simulate_lost_writeback(gpa, io, tp.path, lost.items); var again = try reopen(io, tp.path); defer again.deinit(); try testing.expectEqual(@as(u64, 1), again.loaded.generation); try testing.expectEqual(@as(u64, 900), again.loaded.seq); try testing.expectEqual(cat, again.loaded.catalog_page); // The referenced page survived, because publish synced it first. for (again.page(cat)) |byte| try testing.expectEqual(@as(u8, 0x77), byte); for (again.page(cat + 1)) |byte| try testing.expectEqual(@as(u8, 0x77), byte); } test "copy-on-write leaves the published image byte-identical" { // The single most important property in this milestone. Recovery is // `image + replay(seq > watermark)`, and that is only correct because no // page the image references is ever written *differently* after it is // published. Everything else about the crash story follows from this. // // Mutation checks: make page_mut_cow return the page without copying, or // without updating the slot, and the snapshot comparison goes red. Remove // the stable-mark assert in page_mut and this still passes -- that assert is // the mechanical belt for callers that bypass COW entirely, and its own test // is "a direct write inside the image is refused" below. const gpa = testing.allocator; 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(); // Build an image of distinguishable pages, each owned by a slot, and // publish it. var slots: [64]u32 = undefined; for (&slots, 0..) |*slot, i| { slot.* = try pg.alloc_pages(1); @memset(pg.page_mut(slot.*), @truncate(i + 1)); } try pg.publish(.{ .seq = 1 }); // Snapshot every byte the image covers. const snapshot = try gpa.alloc(u8, @intCast(pg.stable_bytes())); defer gpa.free(snapshot); @memcpy(snapshot, pg.reserve[0..snapshot.len]); // Now rewrite every one of those pages through COW. for (&slots, 0..) |*slot, i| { const before = slot.*; const p = try pg.page_mut_cow(slot); @memset(p, @truncate(0x80 + i)); // The page moved out of the image, and the slot followed it. try testing.expect(slot.* != before); try testing.expect(slot.* >= pg.stable_pages); } // The published image is untouched, byte for byte. try testing.expectEqualSlices(u8, snapshot, pg.reserve[0..snapshot.len]); // And the new contents are readable through the updated slots. for (slots, 0..) |slot, i| { const want: u8 = @truncate(0x80 + i); try testing.expectEqual(want, pg.page(slot)[0]); } } test "a page already above the mark is written in place, not copied" { // COW must not copy what it does not have to: a page allocated since the // last publish is not part of any image, so writing it is free. Getting this // wrong would double the file on every write. 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(); const pg = tp.pg(); try pg.publish(.{ .seq = 1 }); var slot = try pg.alloc_pages(1); // fresh, above the mark const before = slot; for (0..10) |_| { const p = try pg.page_mut_cow(&slot); @memset(p, 0x42); } try testing.expectEqual(before, slot); } test "a recycled page is written in place rather than copied again" { // Mutation check: put `slot.* >= self.stable_pages` back in `page_mut_cow` // in place of `is_unpublished`. Red -- a recycled page has a low page number // and would be copied and freed all over again, so nothing is ever really // reused. That is not a hypothetical: it is what the churn gate measured as // a data file growing linearly and without bound, at 7.2x live data and // still climbing when the run was stopped. 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(); const pg = tp.pg(); // Free a page and let it walk all the way to reusable. const doomed = try pg.alloc_pages(1); @memset(pg.page_mut(doomed), 0xAA); try pg.publish(.{ .seq = 1 }); try pg.free_pages(doomed, 1); try pg.publish(.{ .seq = 2 }); try pg.publish(.{ .seq = 3 }); try testing.expect(pg.free_ready_pages() >= 1); // Recycling hands it back, below the stable mark. var slot = try pg.alloc_pages(1); try testing.expectEqual(doomed, slot); try testing.expect(slot < pg.stable_pages); @memset(pg.page_mut(slot), 0xBB); // A write through the COW path must land in place: the page is not part of // any published image, whatever its number. const tail_before = pg.alloc_tail; const before = slot; const p = try pg.page_mut_cow(&slot); @memset(p, 0xCC); try testing.expectEqual(before, slot); try testing.expectEqual(tail_before, pg.alloc_tail); try testing.expectEqual(@as(u8, 0xCC), pg.page(slot)[0]); } test "one-page requests do not carve up the runs the extents need" { // Mutation check: make `take_free` first-fit again (take the first extent // with `pages >= n`). Red -- the single-page allocations below shave the // large run down and the extent request has to grow the file. // // This is the shape the churn gate hit: copy-on-write asks for one page // thousands of times per generation while the document slab asks for // 2048-page extents, so first fit dismantled every run before an extent // could use it. The free list drained to empty every generation and the file // still grew by the whole write volume. 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(); const pg = tp.pg(); // A large run, plus enough single-page holes to serve the small requests. // The holes are kept apart from each other and from the run by pages that // are never freed -- otherwise coalescing merges the lot into one run and the // test stops being about fit at all. const run = try pg.alloc_pages(64); _ = try pg.alloc_pages(1); // separator, never freed var holes: [8]u32 = undefined; for (&holes) |*h| { h.* = try pg.alloc_pages(1); _ = try pg.alloc_pages(1); // separator, never freed } try pg.publish(.{ .seq = 1 }); try pg.free_pages(run, 64); for (holes) |h| try pg.free_pages(h, 1); try pg.publish(.{ .seq = 2 }); try pg.publish(.{ .seq = 3 }); // Eight one-page allocations must come out of the eight holes. for (0..holes.len) |_| _ = try pg.alloc_pages(1); const tail_before = pg.alloc_tail; // So the run is still whole and the extent request is served from it. const reused = try pg.alloc_pages(64); try testing.expectEqual(run, reused); try testing.expectEqual(tail_before, pg.alloc_tail); } test "a slab run comes off the free list aligned, or not at all" { // `alloc_slab_run` is a second allocation policy in the same allocator, so // what it must not do is as important as what it must: // // - what it hands out starts and ends on a system-page boundary, because // that is the granularity writeback tears at and the granularity // reclamation gives back at; // - a run too short to be worth an extent is left alone; // - the pieces it trims off stay on the free list rather than leaking. // // Mutation checks: drop the `alignForward`/`alignBackward` and the first // assertion goes red on the odd-page run; drop the `usable < min_pages` // test and the short run is handed out. 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(); const pg = tp.pg(); const spp: u32 = pages_per_map_align; // A long run deliberately starting one 4 KiB page past a boundary, and a // short one, kept apart so coalescing cannot merge them. _ = try pg.alloc_pages(std.mem.alignForward(u32, pg.alloc_tail, spp) - pg.alloc_tail + 1); const long = try pg.alloc_pages(400); _ = try pg.alloc_pages(1); // separator, never freed const short = try pg.alloc_pages(8); _ = try pg.alloc_pages(1); // separator, never freed try testing.expect(long % spp != 0); try pg.publish(.{ .seq = 1 }); try pg.free_pages(long, 400); try pg.free_pages(short, 8); try pg.publish(.{ .seq = 2 }); try pg.publish(.{ .seq = 3 }); const ready_before = pg.free_ready_pages(); var hold: Reservation = .{}; try pg.reserve_pages(&hold, 256); const run = pg.alloc_slab_run(&hold, 64, 256) orelse return error.TestUnexpectedResult; pg.release_reservation(&hold); try testing.expectEqual(@as(u32, 0), run.first % spp); try testing.expectEqual(@as(u32, 0), run.pages % spp); try testing.expectEqual(@as(u32, 256), run.pages); // It came out of the long run, not the short one, and past its unaligned // first page. try testing.expect(run.first > long); try testing.expect(run.first < long + 400); // Everything not handed out is still on the list. try testing.expectEqual(ready_before - 256, pg.free_ready_pages()); // The short run is below the floor and stays where it is, whatever is asked // of it; nothing else is left long enough either. var hold2: Reservation = .{}; try pg.reserve_pages(&hold2, 256); try testing.expect(pg.alloc_slab_run(&hold2, 200, 256) == null); pg.release_reservation(&hold2); try testing.expectEqual(ready_before - 256, pg.free_ready_pages()); } 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 // copying nodes concurrently were appending to one `ArrayList` unserialized // while a checkpoint moved it out from under them. // // Pages are conserved across the rotation and across coalescing, so the sum // over all three lists is the invariant to assert. Probabilistic by nature, // as any test of a data race is: it says nothing when green and is only // evidence when red. Mutation check: drop the lock from `free_pages` and // this fails or crashes within a few runs. const gpa = testing.allocator; var threaded: std.Io.Threaded = .init(gpa, .{}); defer threaded.deinit(); const io = threaded.io(); var tp = try TmpPager.init(io, 64 << 20); defer tp.deinit(); const freers = 4; const per_freer = 200; // One page each, allocated up front so no fiber is also growing the file. var pages: [freers * per_freer]u32 = undefined; for (&pages) |*p| p.* = try tp.pg().alloc_pages(1); try tp.pg().publish(.{ .seq = 1 }); const Worker = struct { fn freer(p: *Pager, run: []const u32) error{Canceled}!void { for (run) |page| p.free_pages(page, 1) catch return error.Canceled; } fn publisher(p: *Pager, seq: *std.atomic.Value(u64)) error{Canceled}!void { for (0..8) |_| { p.publish(.{ .seq = seq.fetchAdd(1, .monotonic) }) catch return error.Canceled; } } }; var seq = std.atomic.Value(u64).init(2); var group: std.Io.Group = .init; defer group.cancel(io); for (0..freers) |i| { group.async(io, Worker.freer, .{ tp.pg(), pages[i * per_freer ..][0..per_freer] }); } group.async(io, Worker.publisher, .{ tp.pg(), &seq }); try group.await(io); // Page identity, not a total: the publisher's own free-list streams are // allocated *off this list*, so a plain count would be short by however many // publishes found a fit. Every page still on the list must therefore be one // the freers put there, exactly once -- a lost or half-written append shows // up as a duplicate or as a page nobody freed, neither of which recycling // can produce. const lo = pages[0]; var seen = try std.DynamicBitSetUnmanaged.initEmpty(gpa, pages.len); defer seen.deinit(gpa); var on_list: usize = 0; for ([_][]const Extent{ tp.pg().free_ready.items, tp.pg().free_hold.items, tp.pg().free_pending.items, }) |list| for (list) |e| { for (0..e.pages) |i| { const p = e.first + @as(u32, @intCast(i)); // 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; } }; // The only pages missing are the ones a publish recycled into its stream, // and there were nine publishes at one page each. try testing.expect(pages.len - on_list <= 9); } test "freed pages that touch merge back into a usable run" { // Mutation check: drop the `coalesce_free_ready()` call from `publish`. Red // -- the four one-page frees below stay four separate holes and the run of // four has to come from the tail. Over a real workload the free list only // ever fragments, so an 8 MiB extent request never finds a home however much // free space has accumulated. 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(); const pg = tp.pg(); var pages: [4]u32 = undefined; for (&pages) |*x| x.* = try pg.alloc_pages(1); try pg.publish(.{ .seq = 1 }); // Freed out of order, one page at a time, which is how copy-on-write frees. try pg.free_pages(pages[2], 1); try pg.free_pages(pages[0], 1); try pg.free_pages(pages[3], 1); try pg.free_pages(pages[1], 1); try pg.publish(.{ .seq = 2 }); try pg.publish(.{ .seq = 3 }); const tail_before = pg.alloc_tail; const run = try pg.alloc_pages(4); try testing.expectEqual(pages[0], run); try testing.expectEqual(tail_before, pg.alloc_tail); } test "freed pages are recycled rather than growing the file" { // Copy-on-write abandons every page it touches, in every generation, so // without reuse a write-heavy workload grows the file by // `generations x touched_set` without bound. That is why PLAN amendment A2 // makes the free list a prerequisite rather than the defense-in-depth D6.2 // assumed. // // Mutation check: make alloc_pages_assume_reserved skip take_free and the // tail assertion goes red. 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(); const pg = tp.pg(); var slots: [32]u32 = undefined; for (&slots, 0..) |*slot, i| { slot.* = try pg.alloc_pages(1); @memset(pg.page_mut(slot.*), @truncate(i + 1)); } try pg.publish(.{ .seq = 1 }); // Three rounds of rewriting every page. The first COWs each one and frees // the original; two publishes later those originals come back. var tail_after_first: u32 = 0; for (0..3) |round| { for (&slots) |*slot| { const p = try pg.page_mut_cow(slot); @memset(p, @truncate(round)); } try pg.publish(.{ .seq = round + 2 }); if (round == 0) tail_after_first = pg.alloc_tail; } // By round three the freed pages are being handed back, so the tail has // stopped climbing by a full working set per round. const grew = pg.alloc_tail - tail_after_first; try testing.expect(grew < slots.len * 2); // And the data is still correct through the current slots. for (slots) |slot| try testing.expectEqual(@as(u8, 2), pg.page(slot)[0]); } test "the published image is made hardware read-only where the build allows it" { // The belt described in `protect_stable`. It exists for the mistake the // structural mechanism cannot catch: a caller writing through a `*Node` // obtained *before* the stable mark moved. // // What is asserted here is that the protection is really applied, not that a // violating write faults -- a SIGSEGV cannot be caught in-process, so // verifying the fault would need a child process, and the fault itself is // the OS's behaviour rather than this code's. An mprotect that silently // failed would leave a belt that looks present and does nothing, which is // the failure worth guarding against here. if (!protect_stable) return; // ReleaseFast: deliberately off 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(); const pg = tp.pg(); // Enough pages that the protected range spans at least one system page even // where those are 16 KiB. var i: usize = 0; while (i < 64) : (i += 1) { const p = try pg.alloc_pages(1); @memset(pg.page_mut(p), 0x31); } try pg.publish(.{ .seq = 1 }); try testing.expect(pg.protect_ok); // Reads through the protection still work. try testing.expectEqual(@as(u8, 0x31), pg.page(page_first_data)[0]); } test "an empty file is treated as absent, not as corruption" { // A create that died before its header landed must not look like a corrupt // database — the engine has to be able to open. 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(); const path = try testing.allocator.dupe(u8, tp.path); defer testing.allocator.free(path); tp.close(); { var f = try std.Io.Dir.cwd().openFile(io, path, .{ .mode = .read_write }); defer f.close(io); try f.setLength(io, 0); } tp.pager = try Pager.open(testing.allocator, io, path, .{ .uuid = 7, .reserve_bytes = 64 << 20 }); try testing.expect(tp.pg().fresh); }