//! Persisted order split into runs that never share a delta chain.
//!
//! A whole-store read sweep reconstructs every frame, and a delta frame is
//! only cheap when its base's plaintext is still in the reader's memo. One
//! pass in persisted order gets that for free: bases persist before their
//! deltas. Cutting persisted order into contiguous slices for several threads
//! loses it -- a slice meets most chains part-way through and rebuilds each
//! from its root -- and on a history whose chains run the length of the store
//! that rebuild costs more than the threads save.
//!
//! So the split here is by chain instead. Every frame belongs to the tree of
//! deltas grown from one full (non-delta) frame, its root; a run takes whole
//! trees, and walks its frames in persisted order. Within a run every base
//! still precedes its deltas, so a run's memo works exactly as the single
//! pass's does, over a fraction of the store.
//!
//! The split only decides who reads what. A frame whose recorded base cannot
//! be read is its own root here, and its reconstruction fails or succeeds on
//! its own wherever it lands: correctness never rests on the grouping.
const std = @import("std");
const logstore = @import("logstore.zig");
const store_workers = @import("store_workers.zig");
pub const Runs = struct {
/// Indexes into the persisted order the split was made from, grouped by
/// run and ascending within each. u32 halves the footprint of the one
/// array the split keeps; `split` refuses a store it cannot index.
members: []u32,
/// Run `r` is `members[starts[r]..starts[r + 1]]`.
starts: []usize,
pub fn count(self: *const Runs) usize {
return self.starts.len - 1;
}
pub fn of(self: *const Runs, run: usize) []const u32 {
return self.members[self.starts[run]..self.starts[run + 1]];
}
pub fn deinit(self: *Runs, gpa: std.mem.Allocator) void {
gpa.free(self.members);
gpa.free(self.starts);
self.* = undefined;
}
};
pub const SplitError = error{
OutOfMemory,
/// More frames than a u32 indexes. The caller reads such a store in one
/// pass instead.
TooManyFrames,
};
/// Split `rids`, a store's persisted order, into `run_count` runs of whole
/// delta trees. Finding each delta's base reads its frame header, which
/// `pool` spreads like any other read; persisted order is offset order, so
/// the base's offset names its place by binary search.
pub fn split(
gpa: std.mem.Allocator,
log: *logstore.LogStore,
rids: []const logstore.Rid,
run_count: usize,
pool: *store_workers.Pool,
) SplitError!Runs {
std.debug.assert(run_count >= 1);
if (rids.len > std.math.maxInt(u32)) return error.TooManyFrames;
// run[i] starts as the persisted index of frame i's base (or i itself
// for a root), then becomes frame i's run in one forward pass: a base
// always precedes its delta, so by the time frame i is reached its root's
// slot already holds the root's run. Roots are dealt to runs round-robin
// in the order they persist; the trees are many and small next to a run,
// so the deal evens out. One array, indexed by frame, read through the
// root.
const run = try gpa.alloc(u32, rids.len);
defer gpa.free(run);
var bases: Bases = .{ .log = log, .rids = rids, .base = run, .chunks = 0 };
bases.chunks = @max(1, @min(rids.len / Bases.chunk_len, pool.threadCount() * 4));
try pool.runEach(gpa, bases.chunks, &bases, Bases.run);
var next_run: u32 = 0;
for (run, 0..) |*r, i| {
if (r.* == i) {
r.* = next_run;
next_run = if (next_run + 1 == run_count) 0 else next_run + 1;
} else {
// The base's slot already holds its run, and a delta's run is
// its base's: by induction, its root's.
r.* = run[r.*];
}
}
const starts = try gpa.alloc(usize, run_count + 1);
errdefer gpa.free(starts);
@memset(starts, 0);
for (run) |r| starts[r + 1] += 1;
for (1..starts.len) |k| starts[k] += starts[k - 1];
const members = try gpa.alloc(u32, rids.len);
errdefer gpa.free(members);
const fill = try gpa.dupe(usize, starts[0..run_count]);
defer gpa.free(fill);
// Ascending i, so each run lists its frames in persisted order.
for (run, 0..) |r, i| {
members[fill[r]] = @intCast(i);
fill[r] += 1;
}
return .{ .members = members, .starts = starts };
}
const Bases = struct {
log: *logstore.LogStore,
rids: []const logstore.Rid,
/// Frame i's base index, or i for a root.
base: []u32,
chunks: usize,
const chunk_len: usize = 4096;
fn run(ctx: *anyopaque, chunk: usize) void {
const self: *Bases = @ptrCast(@alignCast(ctx));
const begin = self.rids.len * chunk / self.chunks;
const end = self.rids.len * (chunk + 1) / self.chunks;
for (begin..end) |i| self.base[i] = @intCast(self.baseIndex(i) orelse i);
}
/// The persisted index of frame `i`'s base, or null for a full frame and
/// for a base offset that is no earlier frame's start.
fn baseIndex(self: *const Bases, i: usize) ?usize {
if (self.log.kindOf(self.rids[i]) != .delta) return null;
const offset = self.log.deltaBaseOffsetOf(self.rids[i]) orelse return null;
var lo: usize = 0;
var hi: usize = i;
// Bounded: the interval halves every step.
while (lo < hi) {
const mid = lo + (hi - lo) / 2;
const at = self.log.offsetOf(self.rids[mid]);
if (at == offset) return mid;
if (at < offset) lo = mid + 1 else hi = mid;
}
return null;
}
};
test "a split keeps every delta tree in one run and every run in persisted order" {
const gpa = std.testing.allocator;
const io = std.testing.io;
var tmp = std.testing.tmpDir(.{});
defer tmp.cleanup();
const raw = try tmp.dir.createFile(io, "store", .{ .read = true });
defer raw.close(io);
var file_storage = @import("storage.zig").FileStorage.init(io, raw, 0);
var log = try logstore.LogStore.open(gpa, file_storage.storageFile(), 3);
defer log.deinit(gpa);
// Interleaved families: version v of every family persists before
// version v + 1 of any, so contiguous slices would cut every chain.
const families = 9;
var latest: [families]?logstore.Rid = @splat(null);
for (0..6) |v| for (0..families) |f| {
var content: std.ArrayList(u8) = .empty;
defer content.deinit(gpa);
for (0..64) |line| try content.print(gpa, "family {d} line {d}\n", .{ f, line });
try content.print(gpa, "version {d}\n", .{v});
var digest: [32]u8 = undefined;
std.crypto.hash.sha3.Sha3_256.hash(content.items, &digest, .{});
latest[f] = try log.putArtifact(&digest, content.items, latest[f]);
};
const rids = try log.ridsInPersistedOrder(gpa);
defer gpa.free(rids);
var threaded = std.Io.Threaded.init(gpa, .{});
defer threaded.deinit();
var pool: store_workers.Pool = .{ .io = threaded.io() };
pool.start(3);
defer pool.deinit();
var runs = try split(gpa, &log, rids, 4, &pool);
defer runs.deinit(gpa);
const run_of = try gpa.alloc(?usize, rids.len);
defer gpa.free(run_of);
@memset(run_of, null);
for (0..runs.count()) |r| {
var previous: ?u32 = null;
for (runs.of(r)) |i| {
if (previous) |p| try std.testing.expect(p < i);
previous = i;
try std.testing.expectEqual(@as(?usize, null), run_of[i]);
run_of[i] = r;
}
}
var deltas: usize = 0;
for (rids, 0..) |rid, i| {
try std.testing.expect(run_of[i] != null);
var buffer: [logstore.max_hash_len]u8 = undefined;
const hash = log.deltaBaseHashOf(rid, &buffer) orelse continue;
deltas += 1;
const base = std.mem.indexOfScalar(logstore.Rid, rids, log.lookup(hash).?).?;
// A delta and its base always share a run.
try std.testing.expectEqual(run_of[base], run_of[i]);
}
// The premise: the store really holds chains, across the interleaving.
try std.testing.expect(deltas > families);
}