dev/MultiforaDB
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

pager: a page reservation belongs to its consumer, not to the pager

Browse Source 
The promise `reserve_pages` makes was a single counter on the pager, and the
first concurrent benchmark since the data file landed aborted the server on
it, reliably, at four clients:

  assertion failed: page allocation overran reserve_pages' promise
    src/index.zig:955 in alloc_node
    src/db.zig:794  in upsert

Two upserts on different collections hold different collection locks, so they
run at the same time. Each ends by dropping "whatever is still promised" --
and `release_reservation` zeroed the shared counter, so the first to publish
released the second's promise while the second was still between its log
append and its supposedly infallible allocation. The tripwire fired, which is
the good outcome; the bad one is a growth that never happened and a store past
the mapped end.

This is PLAN risk 3 ("a shared pager makes alloc_tail and free_pending a
global mutex on every insert"), whose mitigation -- private pre-allocated runs
-- was never built. So: `pager.Reservation` is a per-consumer promise, held by
every Index, every Collection (for its doc slab) and the checkpoint, and each
one releases only its own. The pager keeps the sum, which is all `grow_to`
needs. `Engine.release_write_reservations` drops exactly the buckets one
upsert reserved through.

The allocator's own state -- the tail, the total, the free lists, the
unpublished set, file growth -- is now behind `alloc_lock`, taken
uncancelable. It is never held across the log append: that is precisely what
per-consumer reservations buy, and why group commit is unaffected.

  concurrent durable insertOne   4 clients  21697 docs/s   (was aborting)
                                16 clients  30678 docs/s

Mutation: make `release_reservation` zero `self.reserved_pages` again. Red on
the new pager test and on three command tests.
This commit is contained in:
Alexey Shakhmatov
2026-08-03 22:51:15 +03:00
parent 5228ed740a
commit 4b70ce6da9
3 changed files with 156 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

40
src/db.zig
View File

@@ -50,6 +50,10 @@ pub const Collection = struct {
/// rebuild. `slab_tail` cannot answer that -- it is an absolute file offset,
/// so it jumps forward whenever a fresh extent is taken.
slab_used: u64,
/// This collection's outstanding page promise, for the document slab. Per
/// collection because concurrent writers must not release each other's --
/// see `pager.Reservation`.
hold: pgr.Reservation,
/// Of those bytes, the ones still reachable. `slab_used - live_bytes` is
/// this collection's slab garbage, which only a rebuild reclaims. Kept per
/// collection so dropping one can move the right amount from the engine's
@@ -91,6 +95,7 @@ pub const Collection = struct {
.slab_end = 0,
.slab_used = 0,
.live_bytes = 0,
.hold = .{},
.indexes = .empty,
.id_index = undefined,
};
@@ -140,8 +145,8 @@ pub const Collection = struct {
slab_extent_pages,
(len + pgr.page_size - 1) / pgr.page_size,
));
try self.pager.reserve_pages(want_pages);
const first = self.pager.alloc_pages_assume_reserved(want_pages);
try self.pager.reserve_pages(&self.hold, want_pages);
const first = self.pager.alloc_pages_assume_reserved(&self.hold, want_pages);
try self.slab_extents.append(gpa, .{ .first = first, .pages = want_pages });
self.slab_tail = @as(u64, first) << pgr.page_shift;
self.slab_end = self.slab_tail + (@as(u64, want_pages) << pgr.page_shift);
@@ -261,6 +266,10 @@ pub const Engine = struct {
/// 16 KiB documents and one of 40 B documents look identical.
live_bytes: u64 = 0,
dead_bytes: u64 = 0,
/// The checkpoint's own page promise, for the catalog and free-list pages it
/// writes. Separate from any collection's for the same reason those are
/// separate from each other.
hold: pgr.Reservation = .{},
/// Set when the log has grown enough since the last checkpoint to be worth
/// reclaiming. Read by the write epilogue and the TTL monitor, both of which
/// run without holding a collection lock.
@@ -420,6 +429,15 @@ pub const Engine = struct {
/// own total (a rebuild appends through it too, and its copies are live by
/// definition); the engine's total only moves when a document actually
/// becomes live, which a rebuild's copies do not.
/// Drop the unclaimed part of every promise a write to this collection took:
/// the slab's and one per index. Called after the write is published, under
/// the same collection lock the reservations were taken under.
fn release_write_reservations(self: *Engine, coll: *Collection) void {
self.pager.release_reservation(&coll.hold);
self.pager.release_reservation(&coll.id_index.hold);
for (coll.indexes.items) |ix| self.pager.release_reservation(&ix.hold);
}
fn publish_doc_bytes(self: *Engine, coll: *Collection, bytes: []const u8) u64 {
const off = coll.slab_append(bytes);
self.live_bytes += bytes.len;
@@ -794,8 +812,10 @@ pub const Engine = struct {
b.ix.insert_entries(&b.built, off);
}
// The write is published; anything the reservations above did not claim
// is dead. Leaving it promised would grow the file on every write.
self.pager.release_reservation();
// is dead. Leaving it promised would grow the file on every write. Every
// consumer this upsert reserved through, and only those: another
// collection may be mid-write on another thread.
self.release_write_reservations(coll);
self.note_compact();
self.note_checkpoint();
}
@@ -919,7 +939,7 @@ pub const Engine = struct {
try ix.append_doc_entries(self.gpa, coll.doc_bytes(entry.off), entry.off);
}
_ = try ix.finish_bulk(self.gpa, true);
self.pager.release_reservation();
self.pager.release_reservation(&ix.hold);
// Reserve the collection slot, then persist and publish.
try coll.indexes.ensureUnusedCapacity(self.gpa, 1);
@@ -1284,7 +1304,7 @@ pub const Engine = struct {
const bytes = doc_bytes_in(self.pager, entry.off);
try coll.slab_reserve(self.gpa, bytes.len);
const new_off = coll.slab_append(bytes);
self.pager.release_reservation();
self.pager.release_reservation(&coll.hold);
try moved.append(self.gpa, .{ .off = new_off });
}
// Republish the offsets.
@@ -1313,7 +1333,7 @@ pub const Engine = struct {
// Duplicates are tolerated here for the same reason they are on open:
// refusing would make a maintenance task able to take the database down.
_ = ix.finish_bulk(self.gpa, false) catch |err| return err;
self.pager.release_reservation();
self.pager.release_reservation(&ix.hold);
}
/// Re-emit one collection's index specs and documents into the compacted
@@ -1367,7 +1387,7 @@ pub const Engine = struct {
};
}
// Tolerated, not enforced: the database must always open.
defer self.pager.release_reservation();
defer self.pager.release_reservation(&ix.hold);
if (try ix.finish_bulk(self.gpa, false)) {
std.debug.print(
"multiforadb: WARNING: unique index '{s}' has duplicate keys in existing " ++
@@ -1710,7 +1730,7 @@ pub const Engine = struct {
self.log_lock.unlock(self.io);
return err;
};
self.pager.release_reservation();
self.pager.release_reservation(&self.hold);
// The watermark is durable, so every record it covers is now
// redundant. Strictly after the publish: the other order loses data
// if a crash lands between them.
@@ -1844,7 +1864,7 @@ fn apply_record(ctx: *anyopaque, record: storage.Record, doc: *bson.Document) an
} else {
self.index_one(&coll.id_index, doc_bytes, off);
}
self.pager.release_reservation();
self.release_write_reservations(coll);
// The _id_ entry is added after replay, in build_all_indexes,
// together with the secondary indexes.
//

25
src/index.zig
View File

@@ -217,6 +217,10 @@ pub const Index = struct {
/// `Slot.off` means two different things depending on `spill` is unchanged,
/// only the second meaning moved.
ovf_extents: std.ArrayListUnmanaged(pgr.Extent),
/// This index's outstanding page promise. Per index rather than per pager
/// because two collections are written concurrently and each one's promise
/// has to survive the other's release -- see `pager.Reservation`.
hold: pgr.Reservation,
ovf_tail: u64,
ovf_end: u64,
/// Bulk-build staging: entries appended unsorted by append_doc_entries,
@@ -255,6 +259,7 @@ pub const Index = struct {
.pager = pager,
.node_pages = .empty,
.ovf_extents = .empty,
.hold = .{},
.ovf_tail = 0,
.ovf_end = 0,
.staging = .empty,
@@ -281,9 +286,9 @@ pub const Index = struct {
// Slot 0 is a dummy (0 is the null node id); the root is one empty
// leaf, so a fresh index is always a valid tree.
try self.node_pages.ensureUnusedCapacity(gpa, 2);
try pager.reserve_pages(2);
self.node_pages.appendAssumeCapacity(pager.alloc_pages_assume_reserved(1));
self.node_pages.appendAssumeCapacity(pager.alloc_pages_assume_reserved(1));
try pager.reserve_pages(&self.hold, 2);
self.node_pages.appendAssumeCapacity(pager.alloc_pages_assume_reserved(&self.hold, 1));
self.node_pages.appendAssumeCapacity(pager.alloc_pages_assume_reserved(&self.hold, 1));
self.page_mut(0).* = empty_node(0);
self.page_mut(1).* = empty_node(1);
self.root = 1;
@@ -435,7 +440,7 @@ pub const Index = struct {
// a slot in the id->page table, and the insertion after the log append
// must not be able to fail on either.
try self.node_pages.ensureUnusedCapacity(gpa, @intCast(extra_nodes));
try self.pager.reserve_pages(@intCast(extra_nodes));
try self.pager.reserve_pages(&self.hold, @intCast(extra_nodes));
try self.reserve_overflow(gpa, entries);
}
@@ -467,8 +472,8 @@ pub const Index = struct {
ovf_extent_pages,
(overflow_bytes + pgr.page_size - 1) / pgr.page_size,
));
try self.pager.reserve_pages(want_pages);
const first = self.pager.alloc_pages_assume_reserved(want_pages);
try self.pager.reserve_pages(&self.hold, want_pages);
const first = self.pager.alloc_pages_assume_reserved(&self.hold, want_pages);
try self.ovf_extents.append(gpa, .{ .first = first, .pages = want_pages });
self.ovf_tail = @as(u64, first) << pgr.page_shift;
self.ovf_end = self.ovf_tail + (@as(u64, want_pages) << pgr.page_shift);
@@ -593,9 +598,9 @@ pub const Index = struct {
self.ovf_tail = 0;
self.ovf_end = 0;
try self.node_pages.ensureUnusedCapacity(gpa, 2);
try self.pager.reserve_pages(2);
self.node_pages.appendAssumeCapacity(self.pager.alloc_pages_assume_reserved(1));
self.node_pages.appendAssumeCapacity(self.pager.alloc_pages_assume_reserved(1));
try self.pager.reserve_pages(&self.hold, 2);
self.node_pages.appendAssumeCapacity(self.pager.alloc_pages_assume_reserved(&self.hold, 1));
self.node_pages.appendAssumeCapacity(self.pager.alloc_pages_assume_reserved(&self.hold, 1));
self.page_mut(0).* = empty_node(0);
self.page_mut(1).* = empty_node(1);
self.root = 1;
@@ -952,7 +957,7 @@ pub const Index = struct {
/// honest response.
fn alloc_node(self: *Index) u32 {
assert_msg(self.node_pages.items.len < self.node_pages.capacity, "node allocation overran reserve_for's bound");
const p = self.pager.alloc_pages_assume_reserved(1);
const p = self.pager.alloc_pages_assume_reserved(&self.hold, 1);
self.node_pages.appendAssumeCapacity(p);
const id: u32 = @intCast(self.node_pages.items.len - 1);
self.page_mut(id).* = empty_node(0);

132
src/pager.zig
View File

@@ -169,6 +169,26 @@ pub const OpenOptions = struct {
reserve_bytes: usize = default_reserve_bytes,
};
/// One consumer's outstanding promise, in pages.
///
/// The promise used to be a single counter on the pager, and that is not
/// something concurrent writers can share. Two upserts on different
/// collections run at the same time (they hold different collection locks),
/// and each one ends by dropping "whatever is still promised" -- so the first
/// to finish zeroed the second's promise, and the second's supposedly
/// infallible allocation then tripped its own tripwire:
///
/// assertion failed: page allocation overran reserve_pages' promise
/// src/index.zig:955 in alloc_node
///
/// Reliably, at four concurrent clients, on the first benchmark run after the
/// data file landed (PLAN risk 3). Each consumer -- every index, every
/// collection's slab, the checkpoint -- now holds its own, and only ever
/// releases its own. The pager keeps the sum, which is all `grow_to` needs.
pub const Reservation = struct {
pages: u32 = 0,
};
pub const Pager = struct {
gpa: std.mem.Allocator,
io: std.Io,
@@ -196,6 +216,15 @@ pub const Pager = struct {
/// first one's allocation asserts. Which is exactly what happened, on a
/// 512 MB load, with the tripwire in alloc_node catching it.
reserved_pages: u32,
/// Serialises the allocator's bookkeeping: the tail, the reservation total,
/// the free lists, the unpublished set and file growth. Writers on different
/// collections hold different collection locks and allocate from this one
/// pager, so none of that can be a plain field (PLAN risk 3).
///
/// Taken uncancelable: the critical section is bookkeeping that leaves the
/// allocator inconsistent if abandoned half way, and it is never held across
/// the log append -- that is what per-consumer reservations buy.
alloc_lock: std.Io.Mutex,
/// True when this file was created by this open (no checkpoint to load).
fresh: bool,
@@ -297,6 +326,7 @@ pub const Pager = struct {
.file_pages = @intCast(existing_len / page_size),
.alloc_tail = page_first_data,
.reserved_pages = 0,
.alloc_lock = .init,
.fresh = created,
.loaded = .{},
.generation = 0,
@@ -444,14 +474,17 @@ pub const Pager = struct {
// -- allocation ---------------------------------------------------------
/// Hand out `n` contiguous pages, growing the file if needed.
/// Hand out `n` contiguous pages, growing the file if needed. For callers
/// with nothing to protect against failure -- copy-on-write, the catalog --
/// where reserving and claiming are one step.
pub fn alloc_pages(self: *Pager, n: u32) !u32 {
assert(n > 0);
try self.reserve_pages(n);
return self.alloc_pages_assume_reserved(n);
var hold: Reservation = .{};
try self.reserve_pages(&hold, n);
return self.alloc_pages_assume_reserved(&hold, n);
}
/// Guarantee that the next `n` pages can be handed out without failing.
/// Guarantee that `hold` can have `n` more pages handed out without failing.
/// Fallible, and meant to run *before* the log append on a write path, so
/// that publishing afterwards cannot fail — the invariant that keeps a
/// document from ever being live but unindexed.
@@ -459,25 +492,35 @@ pub const Pager = struct {
/// Nothing is dirtied here, so nothing is allocated on disk: `setLength`
/// leaves a sparse file, `ls -l` grows and `du` does not. That is what makes
/// a generous reservation cheap.
pub fn reserve_pages(self: *Pager, n: u32) !void {
// Additive: room for what is already promised *plus* this. Two
// consumers reserving before the same log append must both be able to
pub fn reserve_pages(self: *Pager, hold: *Reservation, n: u32) !void {
self.alloc_lock.lockUncancelable(self.io);
defer self.alloc_lock.unlock(self.io);
// Additive: room for every promise outstanding anywhere *plus* this one.
// Two consumers reserving before the same log append must both be able to
// rely on their promise.
try self.grow_to(self.alloc_tail + self.reserved_pages + n);
self.reserved_pages += n;
hold.pages += n;
}
/// Drop whatever is still promised but unclaimed.
/// Drop whatever `hold` still promises but has not claimed.
///
/// A reservation is scoped to one write: it is taken before the log append
/// so the publish afterwards cannot fail, and once the publish is done
/// anything unclaimed is dead. Without this the promise accumulates -- a
/// tree reservation covers the worst case of several splits and a typical
/// insert causes none, so `reserved_pages` grew by a handful per write and
/// dragged the file up with it. It showed as a 1.89 GB data file for 512 MB
/// of documents.
pub fn release_reservation(self: *Pager) void {
self.reserved_pages = 0;
/// insert causes none, so the total grew by a handful per write and dragged
/// the file up with it. It showed as a 1.89 GB data file for 512 MB of
/// documents.
pub fn release_reservation(self: *Pager, hold: *Reservation) void {
self.alloc_lock.lockUncancelable(self.io);
defer self.alloc_lock.unlock(self.io);
assert_msg(
self.reserved_pages >= hold.pages,
"a consumer released more pages than the pager had promised",
);
self.reserved_pages -= hold.pages;
hold.pages = 0;
}
/// Hand out `n` pages against a previous `reserve_pages`. Infallible.
@@ -540,17 +583,24 @@ pub const Pager = struct {
self.free_ready.items.len = w + 1;
}
pub fn alloc_pages_assume_reserved(self: *Pager, n: u32) u32 {
pub fn alloc_pages_assume_reserved(self: *Pager, hold: *Reservation, n: u32) u32 {
assert(n > 0);
self.alloc_lock.lockUncancelable(self.io);
defer self.alloc_lock.unlock(self.io);
assert_msg(
n <= hold.pages,
"page allocation overran reserve_pages' promise",
);
assert_msg(
n <= self.reserved_pages,
"page allocation overran reserve_pages' promise",
"page allocation overran the pager's total promise",
);
assert_msg(
self.alloc_tail + n <= self.mapped_pages,
"page allocation past the mapped end of the data file",
);
self.reserved_pages -= n;
hold.pages -= n;
// Reuse before growing. Without this the free list is decorative and the
// file grows without bound under churn, because copy-on-write abandons
// every page it touches in every generation (PLAN amendment A2).
@@ -841,6 +891,10 @@ pub const Pager = struct {
// rotate the free lists by one generation.
self.generation = wm.generation;
self.loaded = wm;
// The free lists and the unpublished set are allocator state, so the
// rotation below takes the same lock every allocation does.
self.alloc_lock.lockUncancelable(self.io);
defer self.alloc_lock.unlock(self.io);
self.stable_pages = self.alloc_tail;
self.protect_image();
try self.free_ready.appendSlice(self.gpa, self.free_hold.items);
@@ -1104,15 +1158,50 @@ test "two consumers reserving before one commit both keep their promise" {
// Consumer A reserves a few pages, then consumer B reserves a large extent
// and takes it -- exactly the order upsert uses.
try pg.reserve_pages(8);
try pg.reserve_pages(2048);
const b_first = pg.alloc_pages_assume_reserved(2048);
var a: Reservation = .{};
var b: Reservation = .{};
try pg.reserve_pages(&a, 8);
try pg.reserve_pages(&b, 2048);
const b_first = pg.alloc_pages_assume_reserved(&b, 2048);
// A's promise must have survived B's reservation *and* B's allocation.
const a_first = pg.alloc_pages_assume_reserved(8);
const a_first = pg.alloc_pages_assume_reserved(&a, 8);
try testing.expect(a_first >= b_first + 2048);
try testing.expectEqual(@as(u32, 0), pg.reserved_pages);
}
test "one consumer's release leaves another's promise intact" {
// Mutation check: make `release_reservation` zero `self.reserved_pages`
// (what it did when the promise was a single counter on the pager) and the
// allocation below goes red on `overran the pager's total promise`.
//
// Not hypothetical: two upserts on different collections hold different
// collection locks and run at the same time, so the first to publish
// released the second's promise out from under it. It aborted the server
// reliably at four concurrent clients (PLAN risk 3).
var threaded: std.Io.Threaded = .init_single_threaded;
defer threaded.deinit();
const io = threaded.io();
var tp = try TmpPager.init(io, 256 << 20);
defer tp.deinit();
const pg = tp.pg();
var writer_a: Reservation = .{};
var writer_b: Reservation = .{};
try pg.reserve_pages(&writer_a, 16);
try pg.reserve_pages(&writer_b, 16);
// A finishes its write and drops what it did not use.
_ = pg.alloc_pages_assume_reserved(&writer_a, 4);
pg.release_reservation(&writer_a);
try testing.expectEqual(@as(u32, 0), writer_a.pages);
// B's promise is untouched, and still claimable in full.
try testing.expectEqual(@as(u32, 16), writer_b.pages);
try testing.expectEqual(@as(u32, 16), pg.reserved_pages);
for (0..16) |_| _ = pg.alloc_pages_assume_reserved(&writer_b, 1);
try testing.expectEqual(@as(u32, 0), pg.reserved_pages);
}
test "a reservation makes the following allocation infallible" {
var threaded: std.Io.Threaded = .init_single_threaded;
defer threaded.deinit();
@@ -1121,10 +1210,11 @@ test "a reservation makes the following allocation infallible" {
defer tp.deinit();
const pg = tp.pg();
try pg.reserve_pages(64);
var hold: Reservation = .{};
try pg.reserve_pages(&hold, 64);
const before = pg.alloc_tail;
// Exactly the promised amount, one page at a time.
for (0..64) |_| _ = pg.alloc_pages_assume_reserved(1);
for (0..64) |_| _ = pg.alloc_pages_assume_reserved(&hold, 1);
try testing.expectEqual(before + 64, pg.alloc_tail);
}