diff --git a/src/db.zig b/src/db.zig index 1ca867d..d197603 100644 --- a/src/db.zig +++ b/src/db.zig @@ -165,7 +165,13 @@ pub const Collection = struct { // an extent recycled off the free list starts *below* the mark and is // still writable. Asking the mark meant every recycled extent was thrown // away after one document, so churn never reused anything. - if (self.pager.is_unpublished_at(self.slab_tail) and self.slab_tail + len <= self.slab_end) return; + // + // Room is checked from the *rounded-up* cursor rather than the cursor + // itself, so the round-up `slab_append` may have to do is guaranteed to + // fit. Without that the append's own re-check could discover it needs a + // fresh extent, which is fallible, after the log record is already + // durable. Costs under one system page per extent. + if (self.pager.is_unpublished_at(self.slab_tail) and self.appendable_end(len)) return; // The page holding the tail is frozen, but the *rest* of the extent is // not: nothing above the live cursor is referenced by the image or by an // index. So skip to the next system page and keep the extent, instead of @@ -179,8 +185,8 @@ pub const Collection = struct { // collections: the data file reached 11.8x the live data and grew by // ~335 MB per checkpoint, heading for DatabaseTooLarge at around 6 GB of // real data. - const resumed = std.mem.alignForward(u64, self.slab_tail, pgr.map_align); - if (resumed + len <= self.slab_end) { + if (self.appendable_end(len)) { + const resumed = std.mem.alignForward(u64, self.slab_tail, pgr.map_align); self.pager.mark_appendable(resumed, self.slab_end); self.slab_tail = resumed; return; @@ -198,9 +204,34 @@ pub const Collection = struct { self.slab_end = self.slab_tail + (@as(u64, want_pages) << pgr.page_shift); } + /// Whether a document of `len` bytes fits in this extent even if the cursor + /// first has to be rounded up to a system page. The reservation and the + /// append both ask this, so they agree on what "there is room" means. + fn appendable_end(self: *const Collection, len: usize) bool { + return std.mem.alignForward(u64, self.slab_tail, pgr.map_align) + len <= self.slab_end; + } + /// Copy `bytes` into the slab and return its absolute file offset. /// Infallible: slab_reserve must have run for at least this many bytes. fn slab_append(self: *Collection, bytes: []const u8) u64 { + // The cursor was checked in `slab_reserve`, but a checkpoint can have + // published since -- the reservation runs before the log append and this + // runs after it, with an fsync in between. `publish` clears the whole + // unpublished set, so a cursor that was writable then can be inside the + // frozen image now, and the copy below would store into it: a bus error + // where the protection is compiled in, and a silent overwrite of durable + // data in ReleaseFast, where it is not. + // + // Re-arming is infallible because `slab_reserve` measured its room from + // the rounded-up cursor. The pager's append lock holds off the next + // publish for the rest of this function, so the answer stays true. + self.pager.lock_append(); + defer self.pager.unlock_append(); + if (!self.pager.is_unpublished_at(self.slab_tail)) { + const resumed = std.mem.alignForward(u64, self.slab_tail, pgr.map_align); + self.pager.mark_appendable(resumed, self.slab_end); + self.slab_tail = resumed; + } assert_msg( self.slab_tail + bytes.len <= self.slab_end, "document append overran the slab reservation", diff --git a/src/pager.zig b/src/pager.zig index 485131c..5d51f8d 100644 --- a/src/pager.zig +++ b/src/pager.zig @@ -241,6 +241,21 @@ pub const Pager = struct { /// 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, @@ -358,6 +373,7 @@ pub const Pager = struct { .alloc_tail = page_first_data, .reserved_pages = 0, .alloc_lock = .init, + .append_lock = .init, .fresh = created, .loaded = .{}, .generation = 0, @@ -529,6 +545,18 @@ pub const Pager = struct { 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 @@ -1037,6 +1065,15 @@ pub const Pager = struct { // 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