From f16741194472fddb198660f27c3d35d0babfa629 Mon Sep 17 00:00:00 2001 From: Travis Rich Date: Tue, 20 Oct 2026 18:42:25 -0400 Subject: [PATCH] Layout F: reads in fewer round trips #65 bounded F's reads at a cost in round trips, which showed on staging: an fsck of a 204k-record F version ran past six minutes (its job was canceled) where today's layout's took a few. - An index read skips layers whose keys miss its own (from the manifest, no reads), then takes 16 at a time: a collection's usual ~10 runs in one round. - A leaf's index pages: 8 in flight, not one ahead. - Pack reads take up to 64 blocks a request (16). - Readers of a whole leaf (bodyLineStream, fsck, sync, copyRecords) take a few MB of records a lookup (leafChunks), not 256 records: a leaf of small records is one lookup. --- packages/protocol/src/repo/build.ts | 4 +-- packages/protocol/src/repo/fsck.ts | 9 ++---- packages/protocol/src/repo/layouts/kit.ts | 34 +++++++++++++++------ packages/protocol/src/repo/layouts/types.ts | 28 +++++++++++++++++ packages/protocol/src/repo/repo.ts | 8 ++--- packages/protocol/src/repo/sync.ts | 7 ++--- packages/protocol/test/repo/layouts.test.ts | 3 +- 7 files changed, 65 insertions(+), 28 deletions(-) diff --git a/packages/protocol/src/repo/build.ts b/packages/protocol/src/repo/build.ts index bcac509..006584a 100644 --- a/packages/protocol/src/repo/build.ts +++ b/packages/protocol/src/repo/build.ts @@ -47,6 +47,7 @@ import { tracksTypes, typeFileRefs, } from './file-sets.js' +import { leafChunks } from './layouts/index.js' import { bodyOfRecord, dropRecordBody, @@ -200,8 +201,7 @@ async function copyRecords( for await (const leaf of leaves(source, root)) { const node = await source.node(leaf.hash) const entries = (node.kind === 'leaf' ? node.entries : []) as RecordEntry[] - for (let i = 0; i < entries.length; i += 256) { - const at = Array.from({ length: Math.min(256, entries.length - i) }, (_, j) => i + j) + for (const at of leafChunks(entries)) { const lines = await repo.records[from].linesAt({ hash: leaf.hash, type, entries }, at) at.forEach((k, j) => writer.add(type, entries[k]!, lines[j]!)) await writer.drain?.() diff --git a/packages/protocol/src/repo/fsck.ts b/packages/protocol/src/repo/fsck.ts index 089c1e6..5d5efbc 100644 --- a/packages/protocol/src/repo/fsck.ts +++ b/packages/protocol/src/repo/fsck.ts @@ -23,7 +23,7 @@ import { mergeTree } from '../tree/merge.js' import { fileTree, recordTree, type RecordEntry } from '../tree/node.js' import { diffTrees, leaves, newNodes } from '../tree/read.js' import { verifyTree } from '../tree/verify.js' -import type { RecordSet } from './layouts/index.js' +import { leafChunks, type RecordSet } from './layouts/index.js' import { type PublicKeyInfo, readCollectionInfo, @@ -36,7 +36,7 @@ import { keys, type Leaf, type Repo, RepoSource } from './repo.js' /** * Read a leaf's records, each checked against its entry (the repository is opened - * untrusted). Layout F reads them a few hundred at a time: a leaf of large + * untrusted). Layout F reads them a few MB at a time (leafChunks): a leaf of large * records is never held whole. */ async function checkLeaf(repo: Repo, leaf: Leaf, set: RecordSet): Promise { @@ -44,10 +44,7 @@ async function checkLeaf(repo: Repo, leaf: Leaf, set: RecordSet): Promise await repo.bodyLines(leaf, set) return } - for (let i = 0; i < leaf.entries.length; i += 256) { - const at = Array.from({ length: Math.min(256, leaf.entries.length - i) }, (_, j) => i + j) - await repo.linesAt(leaf, at, set) - } + for (const at of leafChunks(leaf.entries)) await repo.linesAt(leaf, at, set) } export interface FsckOptions { diff --git a/packages/protocol/src/repo/layouts/kit.ts b/packages/protocol/src/repo/layouts/kit.ts index f7c7e82..0805263 100644 --- a/packages/protocol/src/repo/layouts/kit.ts +++ b/packages/protocol/src/repo/layouts/kit.ts @@ -50,10 +50,18 @@ export const mergeRead = (inputs: number) => MERGE_READ_MIN, Math.min(MERGE_READ, Math.floor(MERGE_READ_BUDGET / Math.max(1, inputs))), ) -/** Blocks of one pack a read takes in one request, at most (neighbours, gaps of up to two). */ -const READ_BLOCKS = 16 -/** Layers an index read works through at once, newest first. */ -const READ_LAYERS = 4 +/** + * Blocks of one pack a read takes in one request, at most (neighbours, gaps of up to + * two): a leaf of large records in a request or two, about 1 MB compressed each. + */ +const READ_BLOCKS = 64 +/** + * Layers an index read works through at once, newest first, of those whose keys + * overlap it: the ~10 runs a collection usually has in one round of reads. + */ +const READ_LAYERS = 16 +/** Pages of one layer a leaf's read has in flight (a typed leaf's range is a few). */ +const READ_PAGES = 8 export const newId = () => crypto.randomUUID().replace(/-/g, '') @@ -686,12 +694,12 @@ export class LayerReader { } const lo = pageAt(dir, maybe[0]!) const hi = Math.max(lo, pageAt(dir, maybe.at(-1)!)) - // One page ahead: the next page's read overlaps this one's scan. - let ahead = this.#page(layer.id, dir, lo) + // READ_PAGES in flight, taken in order: a range of a few pages in one round trip. + const ahead: Promise<[string, string][]>[] = [] + let next = lo for (let p = lo; p <= hi; p++) { - const page = await ahead - if (p < hi) ahead = this.#page(layer.id, dir, p + 1) - keep(page) + while (next <= hi && next < p + READ_PAGES) ahead.push(this.#page(layer.id, dir, next++)) + keep(await ahead.shift()!) } return out } @@ -1207,7 +1215,13 @@ export class LayeredIndex { async values(m: Manifest, keys: Iterable): Promise> { const sorted = [...new Set(keys)].sort(compareUtf8) const out = new Map() - const layers = m.layers.filter(this.filter).sort((a, b) => b.seq - a.seq) + if (!sorted.length) return out + // Only layers whose keys overlap the read's (from the manifest: no reads). + const lo = sorted[0]! + const hi = sorted.at(-1)! + const layers = m.layers + .filter((l) => this.filter(l) && compareUtf8(l.min, hi) <= 0 && compareUtf8(l.max, lo) >= 0) + .sort((a, b) => b.seq - a.seq) for (let i = 0; i < layers.length && out.size < sorted.length; i += READ_LAYERS) { const left = out.size ? sorted.filter((k) => !out.has(k)) : sorted const got = await Promise.all( diff --git a/packages/protocol/src/repo/layouts/types.ts b/packages/protocol/src/repo/layouts/types.ts index 783457e..253d008 100644 --- a/packages/protocol/src/repo/layouts/types.ts +++ b/packages/protocol/src/repo/layouts/types.ts @@ -85,5 +85,33 @@ export type HeldRun = PendingRun export const recordKey = (type: string, id: string, hash: string) => `${type}\u0000${id}\u0000${hash}` +/** Record bytes a reader of a leaf takes at once (layout F): a leaf of small records in one read. */ +export const LEAF_CHUNK_BYTES = 4 * 1024 * 1024 + +/** + * A leaf's entries as runs of indexes of at most LEAF_CHUNK_BYTES of records (one at + * least): each run is one index lookup, so a leaf of small records is read in one, + * and one of large records is never held whole. + */ +export function leafChunks( + entries: readonly RecordEntry[], + maxBytes = LEAF_CHUNK_BYTES, +): number[][] { + const out: number[][] = [] + let run: number[] = [] + let bytes = 0 + entries.forEach((e, i) => { + if (run.length && bytes + e.size > maxBytes) { + out.push(run) + run = [] + bytes = 0 + } + run.push(i) + bytes += e.size + }) + if (run.length) out.push(run) + return out +} + /** The key a record's versions share (`recordKey` without the hash): what a filter holds. */ export const recordIdKey = (type: string, id: string) => `${type}\u0000${id}` diff --git a/packages/protocol/src/repo/repo.ts b/packages/protocol/src/repo/repo.ts index 6140226..d288e63 100644 --- a/packages/protocol/src/repo/repo.ts +++ b/packages/protocol/src/repo/repo.ts @@ -45,6 +45,7 @@ import { compareUtf8 } from '../utf8.js' import { chunkLines, gunzipChunks, gunzipStream, gunzipText, gzip, splitLines } from './gzip.js' import { type HeldRun, + leafChunks, type LayoutName, MissingRecordError, type Placed, @@ -497,11 +498,8 @@ export class Repo { // The memory cache holds lines resolved. const hit = opts.pointers && this.trusted ? undefined : this.lru.get(this.#ck(key)) if (this.records && !hit) { - // A packed layout: 256 records at a time, so a leaf of large records is never held whole. - for (let i = 0; i < leaf.entries.length; i += 256) { - const at = Array.from({ length: Math.min(256, leaf.entries.length - i) }, (_, j) => i + j) - yield* await this.linesAt(leaf, at, opts.set) - } + // A packed layout: a few MB of records at a time, so a leaf of large records is never held whole. + for (const at of leafChunks(leaf.entries)) yield* await this.linesAt(leaf, at, opts.set) return } if (hit || !this.trusted) { diff --git a/packages/protocol/src/repo/sync.ts b/packages/protocol/src/repo/sync.ts index 2b845d7..65141a7 100644 --- a/packages/protocol/src/repo/sync.ts +++ b/packages/protocol/src/repo/sync.ts @@ -40,6 +40,7 @@ import { versionHash, } from '../format.js' import { gunzipText, gzip, splitLines } from './gzip.js' +import { leafChunks } from './layouts/index.js' import { IntegrityError, isNodeDelta, @@ -231,15 +232,13 @@ export async function* treeObjects( yield { ...node, resumeKey: n.lastKey } continue } - // Built a member per 256 records, so the leaf is never held raw: as stored, a + // Built a member per few MB of records, so the leaf is never held raw: as stored, a // body is gzip members, and only its compressed size is held (as a stored one is). const store = repo.records[tree.set] const leafOf = { hash: n.hash, type: tree.slug!, entries } const members: Uint8Array[] = [] - for (let i = 0; i < entries.length; i += 256) { - const at = Array.from({ length: Math.min(256, entries.length - i) }, (_, j) => i + j) + for (const at of leafChunks(entries)) members.push(await gzip((await store.linesAt(leafOf, at)).join('\n') + '\n')) - } yield { key: keys.body(n.hash), bytes: concatBytes(members), resumeKey: n.lastKey } continue } diff --git a/packages/protocol/test/repo/layouts.test.ts b/packages/protocol/test/repo/layouts.test.ts index 1052308..2c5b689 100644 --- a/packages/protocol/test/repo/layouts.test.ts +++ b/packages/protocol/test/repo/layouts.test.ts @@ -988,7 +988,8 @@ describe("layout F's reads and merges hold a bounded amount", () => { expect(await layout.leafLines({ hash: 'leaf', type: 'T', entries })).toEqual( entries.map((_, i) => `line ${i}`), ) - expect(n.most).toBeLessThanOrEqual(4) + // Sixteen at a time: never all forty. + expect(n.most).toBeLessThanOrEqual(16) }) it('plans a reclaim only once merges bring the runs down', async () => {