From 052e5ce34b8af7d6c20b0e4b4b194c050a84cc4b Mon Sep 17 00:00:00 2001 From: Mathieu Masson Date: Wed, 15 Jul 2026 08:37:35 +0200 Subject: [PATCH] repository: Add a GetBlobs primitive. * Adds a new GetBlobs primitive. This is meant as an optimization whenever we have multiple related blobs to fetch at once. The implementation will try to merge close blobs into one single big contiguous backend get request, reducing the cost of the request. * This is the bare minimum implementation, which should already land some real wins. More feature (parallelism, waste management etc) will come on top of this later. --- repository/getblobs.go | 158 +++++++++++++++ repository/getblobs_internal_test.go | 290 +++++++++++++++++++++++++++ repository/getblobs_test.go | 139 +++++++++++++ 3 files changed, 587 insertions(+) create mode 100644 repository/getblobs.go create mode 100644 repository/getblobs_internal_test.go create mode 100644 repository/getblobs_test.go diff --git a/repository/getblobs.go b/repository/getblobs.go new file mode 100644 index 00000000..e8e835d7 --- /dev/null +++ b/repository/getblobs.go @@ -0,0 +1,158 @@ +package repository + +import ( + "bytes" + "cmp" + "context" + "iter" + "slices" + "time" + + "github.com/PlakarKorp/kloset/objects" + "github.com/PlakarKorp/kloset/repository/state" + "github.com/PlakarKorp/kloset/resources" +) + +type BlobReq struct { + Type resources.Type + MAC objects.MAC +} + +// Output type of GetBlobs, embedding the requested blob (mac+type), and its +// associated data. If there was an error data is nil and the error is in the +// second part of the iterator. +type BlobResp struct { + BlobReq + Data []byte +} + +type GetBlobsOpts struct { +} + +const ( + // Default maximum range get size, 4 MB + maxSize = 4 << 20 +) + +type blobLocation struct { + BlobReq + state.Location +} + +type rangeRead struct { + loc state.Location // Actual range of data to get + blobs []blobLocation // blobs covered by that range. +} + +func (r *Repository) GetBlobs(ctx context.Context, blobs []BlobReq, opts *GetBlobsOpts) iter.Seq2[BlobResp, error] { + return func(yield func(BlobResp, error) bool) { + t0 := time.Now() + + locs := make([]blobLocation, 0, len(blobs)) + seen := make(map[BlobReq]struct{}) + + for _, b := range blobs { + if _, ok := seen[b]; ok { + continue + } + + seen[b] = struct{}{} + + loc, ok, err := r.state.GetSubpartForBlob(b.Type, b.MAC) + if err == nil && !ok { + err = ErrBlobNotFound + } + + if err != nil { + if !yield(BlobResp{b, nil}, err) { + return + } + + continue + } + + locs = append(locs, blobLocation{b, loc}) + } + + plan := readPlanner(locs, maxSize, 0.0) + defer func() { + r.Logger().Trace("repository", "GetBlobs(%d blobs): %d reads: %s", + len(blobs), len(plan), time.Since(t0)) + }() + + for _, g := range plan { + data, err := r.GetPackfileRange(g.loc) + + for _, b := range g.blobs { + if err != nil { + if !yield(BlobResp{b.BlobReq, nil}, err) { + return + } + + continue + } + + blobStart := b.Offset - g.loc.Offset + blobBytes := data[blobStart : blobStart+uint64(b.Length)] + blobData, err := r.decodeBuffer(blobBytes) + if err != nil { + if !yield(BlobResp{b.BlobReq, nil}, err) { + return + } + + continue + } + + if !yield(BlobResp{b.BlobReq, blobData}, nil) { + return + } + } + } + } +} + +// readPlanner tries to merge reads according to the policy given. +func readPlanner(locs []blobLocation, maxRdSize uint32, _ float64) []*rangeRead { + if len(locs) == 0 { + return nil + } + + // First sort by packfile then offset. + slices.SortFunc(locs, func(a, b blobLocation) int { + if c := bytes.Compare(a.Packfile[:], b.Packfile[:]); c == 0 { + return cmp.Compare(a.Offset, b.Offset) + } else { + return c + } + }) + + plan := make([]*rangeRead, 0) + + // append the first blob to simplify the following loop + rr := &rangeRead{ + loc: locs[0].Location, + blobs: []blobLocation{locs[0]}, + } + plan = append(plan, rr) + + for _, l := range locs[1:] { + curMerge := plan[len(plan)-1] + + if curMerge.loc.Packfile == l.Packfile { + size := (l.Offset + uint64(l.Length)) - curMerge.loc.Offset + if size <= uint64(maxRdSize) { + curMerge.loc.Length = uint32(size) + curMerge.blobs = append(curMerge.blobs, l) + continue + } + } + + rr = &rangeRead{ + loc: l.Location, + blobs: []blobLocation{l}, + } + plan = append(plan, rr) + } + + return plan +} diff --git a/repository/getblobs_internal_test.go b/repository/getblobs_internal_test.go new file mode 100644 index 00000000..fc5aeeb2 --- /dev/null +++ b/repository/getblobs_internal_test.go @@ -0,0 +1,290 @@ +package repository + +import ( + "math/rand/v2" + "testing" + + "github.com/PlakarKorp/kloset/objects" + "github.com/PlakarKorp/kloset/repository/state" + "github.com/PlakarKorp/kloset/resources" + "github.com/stretchr/testify/require" +) + +func bl(pf byte, offset uint64, length uint32) blobLocation { + return blobLocation{ + BlobReq{Type: resources.RT_CHUNK, MAC: objects.MAC{pf, byte(offset), byte(offset >> 8)}}, + state.Location{Packfile: objects.MAC{pf}, Offset: offset, Length: length}, + } +} + +func TestPlanReadsAdjacentMerge(t *testing.T) { + // Three contiguous blobs coalesce into a single zero-waste read. + plans := readPlanner([]blobLocation{ + bl(1, 200, 100), + bl(1, 0, 100), + bl(1, 100, 100), + }, maxSize, 0.3) + + require.Len(t, plans, 1) + require.Equal(t, uint64(0), plans[0].loc.Offset) + require.Equal(t, uint32(300), plans[0].loc.Length) + require.Len(t, plans[0].blobs, 3) +} + +func TestPlanReadsPackfileBoundary(t *testing.T) { + // Contiguous offsets in different packfiles never merge. + plans := readPlanner([]blobLocation{ + bl(1, 0, 100), + bl(2, 100, 100), + }, maxSize, 0.3) + + require.Len(t, plans, 2) +} + +func TestPlanReadsMaxReadSize(t *testing.T) { + // Merging stops when the merged read would exceed maxReadSize, and + // a single blob larger than maxReadSize still gets its own read. + plans := readPlanner([]blobLocation{ + bl(1, 0, 600), + bl(1, 600, 600), + bl(1, 1200, 2000), + }, 1000, 0.3) + + require.Len(t, plans, 3) + require.Equal(t, uint32(2000), plans[2].loc.Length) +} + +// checkPlan asserts the invariants any correct plan must satisfy, whatever +// the merge policy in force: +// +// 1. every input blob lands in exactly one plan (none dropped, none doubled); +// 2. a plan only contains blobs of its own packfile; +// 3. a plan's range covers every blob assigned to it, entirely; +// 4. a plan's range is tight: it starts at its first blob and ends at the +// end of its outermost blob — gaps between blobs are fetched, slack +// beyond them is not; +// 5. a multi-blob plan never exceeds maxReadSize (a single blob larger +// than maxReadSize is allowed a plan of its own). +func checkPlan(t *testing.T, input []blobLocation, plans []*rangeRead, maxReadSize uint32) { + t.Helper() + + seen := 0 + for i, p := range plans { + require.NotEmpty(t, p.blobs, "plan %d contains no blobs", i) + + planEnd := p.loc.Offset + uint64(p.loc.Length) + minOff, maxEnd := p.blobs[0].Offset, uint64(0) + for _, b := range p.blobs { + blobEnd := b.Offset + uint64(b.Length) + require.Equal(t, p.loc.Packfile, b.Packfile, + "plan %d: blob %x belongs to another packfile", i, b.MAC) + require.GreaterOrEqual(t, b.Offset, p.loc.Offset, + "plan %d: blob %x starts before the plan's range", i, b.MAC) + require.LessOrEqual(t, blobEnd, planEnd, + "plan %d: blob %x overruns the plan's range", i, b.MAC) + minOff = min(minOff, b.Offset) + maxEnd = max(maxEnd, blobEnd) + seen++ + } + + require.Equal(t, minOff, p.loc.Offset, + "plan %d fetches slack before its first blob", i) + require.Equal(t, maxEnd, planEnd, + "plan %d fetches slack past its outermost blob", i) + + if len(p.blobs) > 1 { + require.LessOrEqual(t, p.loc.Length, maxReadSize, + "plan %d: merged read exceeds maxReadSize", i) + } + } + require.Equal(t, len(input), seen, + "every input blob must appear in exactly one plan") +} + +// checkMaximal asserts that no two consecutive same-packfile plans could +// have been merged: their combined span must exceed maxReadSize. Unlike +// checkPlan this is specific to the everything-permitted policy +// (maxWaste 1.0) — with a real waste policy a split can be legitimate +// below the size cap. It catches the failure mode checkPlan cannot: +// a planner drifting back toward one read per blob would produce +// perfectly valid, tiny plans, and defeat the whole point. +// +// It assumes plans come out sorted by (packfile, offset), which the +// sequential greedy emission over sorted input guarantees. +func checkMaximal(t *testing.T, plans []*rangeRead, maxReadSize uint32) { + t.Helper() + + for i := 1; i < len(plans); i++ { + prev, cur := plans[i-1], plans[i] + if prev.loc.Packfile != cur.loc.Packfile { + continue + } + span := cur.loc.Offset + uint64(cur.loc.Length) - prev.loc.Offset + require.Greater(t, span, uint64(maxReadSize), + "plans %d and %d span %d bytes together and should have been merged", + i-1, i, span) + } +} + +func TestPlanReadsGapCountsTowardMaxReadSize(t *testing.T) { + // Two 100-byte blobs 800 bytes apart: the merged READ would be + // 1000 bytes even though only 200 are wanted. With maxReadSize + // 500 they must not merge — the cap bounds the fetched span, + // gaps included, not the sum of blob lengths. + input := []blobLocation{ + bl(1, 0, 100), + bl(1, 900, 100), + } + plans := readPlanner(input, 500, 1.0) + + require.Len(t, plans, 2) + checkPlan(t, input, plans, 500) +} + +func TestPlanReadsGapExtendsLength(t *testing.T) { + // Two blobs with a 100-byte gap, everything permitted: one plan, + // whose length spans the gap (300), not the sum of blob lengths + // (200) — otherwise the read misses the second blob's tail. + input := []blobLocation{ + bl(1, 0, 100), + bl(1, 200, 100), + } + plans := readPlanner(input, maxSize, 1.0) + + require.Len(t, plans, 1) + require.Equal(t, uint64(0), plans[0].loc.Offset) + require.Equal(t, uint32(300), plans[0].loc.Length) + checkPlan(t, input, plans, maxSize) +} + +func TestPlanReadsMaxReadSizeExact(t *testing.T) { + // A merge landing exactly on maxReadSize is allowed: the cap is + // inclusive ("reads up to maxReadSize"), not exclusive. + input := []blobLocation{ + bl(1, 0, 400), + bl(1, 600, 400), + } + plans := readPlanner(input, 1000, 1.0) + + require.Len(t, plans, 1) + require.Equal(t, uint32(1000), plans[0].loc.Length) + checkPlan(t, input, plans, 1000) +} + +func TestPlanReadsOversizedBlobDoesNotBlockNeighbours(t *testing.T) { + // A blob larger than maxReadSize gets a plan of its own, and the + // blobs after it still merge with each other. + input := []blobLocation{ + bl(1, 0, 2000), + bl(1, 2000, 100), + bl(1, 2100, 100), + } + plans := readPlanner(input, 1000, 1.0) + + require.Len(t, plans, 2) + require.Equal(t, uint32(2000), plans[0].loc.Length) + require.Equal(t, uint32(200), plans[1].loc.Length) + checkPlan(t, input, plans, 1000) +} + +func TestPlanReadsInterleavedPackfiles(t *testing.T) { + // Unsorted input alternating between two packfiles: blobs regroup + // per packfile and merge within it. + input := []blobLocation{ + bl(2, 100, 100), + bl(1, 100, 100), + bl(2, 0, 100), + bl(1, 0, 100), + } + plans := readPlanner(input, maxSize, 1.0) + + require.Len(t, plans, 2) + for _, p := range plans { + require.Equal(t, uint64(0), p.loc.Offset) + require.Equal(t, uint32(200), p.loc.Length) + } + checkPlan(t, input, plans, maxSize) +} + +func TestPlanReadsSingle(t *testing.T) { + input := []blobLocation{bl(1, 42, 100)} + plans := readPlanner(input, maxSize, 1.0) + + require.Len(t, plans, 1) + require.Equal(t, uint64(42), plans[0].loc.Offset) + require.Equal(t, uint32(100), plans[0].loc.Length) + checkPlan(t, input, plans, maxSize) +} + +func TestPlanReadsStress(t *testing.T) { + // Deterministic pseudo-random batches laid out the way blobs + // really sit in packfiles — disjoint ranges, sometimes adjacent, + // sometimes gapped, occasionally oversized — checked purely + // against the checkPlan invariants. Policy-agnostic on purpose: + // this must keep passing unchanged once the waste policy lands. + rng := rand.New(rand.NewPCG(0x9E3779B9, 0x7F4A7C15)) + + for round := range 20 { + maxReadSize := uint32(1024 + rng.IntN(8192)) + + input := make([]blobLocation, 0, 512) + for pf := byte(1); pf <= 3; pf++ { + var offset uint64 + for range 150 + rng.IntN(50) { + if rng.IntN(3) > 0 { // 1 in 3 blobs is adjacent to the previous one + offset += uint64(rng.IntN(3000)) + } + length := uint32(1 + rng.IntN(2000)) + if rng.IntN(50) == 0 { // occasional oversized blob + length = maxReadSize + uint32(rng.IntN(1000)) + } + input = append(input, bl(pf, offset, length)) + offset += uint64(length) + } + } + rng.Shuffle(len(input), func(i, j int) { + input[i], input[j] = input[j], input[i] + }) + + plans := readPlanner(input, maxReadSize, 1.0) + if t.Failed() { + return + } + t.Logf("round %d: maxReadSize=%d, %d blobs -> %d plans", + round, maxReadSize, len(input), len(plans)) + checkPlan(t, input, plans, maxReadSize) + checkMaximal(t, plans, maxReadSize) + } +} + +// BenchmarkReadPlanner runs at the scale of the measured no-change +// kernel-backup log (~162k blobs over 20 packfiles): planning must stay +// trivial next to a single ~80ms store round trip. +func BenchmarkReadPlanner(b *testing.B) { + rng := rand.New(rand.NewPCG(0xDEADBEEF, 0xCAFEBABE)) + + input := make([]blobLocation, 0, 162_000) + for pf := byte(1); pf <= 20; pf++ { + var offset uint64 + for range 8_100 { + if rng.IntN(3) > 0 { + offset += uint64(rng.IntN(3000)) + } + length := uint32(1 + rng.IntN(2000)) + input = append(input, bl(pf, offset, length)) + offset += uint64(length) + } + } + rng.Shuffle(len(input), func(i, j int) { + input[i], input[j] = input[j], input[i] + }) + + scratch := make([]blobLocation, len(input)) + b.ReportAllocs() + for b.Loop() { + // readPlanner sorts its input in place; replanning already + // sorted input would flatter the numbers. + copy(scratch, input) + readPlanner(scratch, maxSize, 1.0) + } +} diff --git a/repository/getblobs_test.go b/repository/getblobs_test.go new file mode 100644 index 00000000..98b28c22 --- /dev/null +++ b/repository/getblobs_test.go @@ -0,0 +1,139 @@ +package repository_test + +import ( + "context" + "fmt" + "strings" + "testing" + + "github.com/PlakarKorp/kloset/objects" + "github.com/PlakarKorp/kloset/repository" + "github.com/PlakarKorp/kloset/resources" + ptesting "github.com/PlakarKorp/kloset/testing" + "github.com/stretchr/testify/require" +) + +// chunkRequests backs up a few files and returns a BlobReq per chunk +// of every resolved object, so GetBlobs can be exercised against blobs +// that really live in packfiles. +func chunkRequests(t *testing.T, repo *repository.Repository, paths ...string) []repository.BlobReq { + files := []ptesting.MockFile{ptesting.NewMockDir("/")} + for i, p := range paths { + files = append(files, ptesting.NewMockFile(p, 0644, + strings.Repeat(fmt.Sprintf("get-blobs payload %d ", i), 64))) + } + snap := ptesting.GenerateSnapshot(t, repo, files) + require.NotNil(t, snap) + + fs, err := snap.Filesystem() + require.NoError(t, err) + + var reqs []repository.BlobReq + for _, p := range paths { + entry, err := fs.GetEntry(p) + require.NoError(t, err) + require.NotNil(t, entry.ResolvedObject) + for _, c := range entry.ResolvedObject.Chunks { + reqs = append(reqs, repository.BlobReq{Type: resources.RT_CHUNK, MAC: c.ContentMAC}) + } + } + require.NotEmpty(t, reqs) + return reqs +} + +func TestGetBlobsRoundtrip(t *testing.T) { + repo := ptesting.GenerateRepository(t, nil, nil, nil) + reqs := chunkRequests(t, repo, "/a.txt", "/b.txt", "/c.txt") + + // Duplicate every request: set semantics must collapse them. + got := make(map[repository.BlobReq][]byte) + for res, err := range repo.GetBlobs(context.Background(), append(reqs, reqs...), nil) { + require.NoError(t, err) + req := repository.BlobReq{Type: res.Type, MAC: res.MAC} + _, dup := got[req] + require.False(t, dup, "blob %x delivered twice", res.MAC) + got[req] = res.Data + } + + distinct := make(map[repository.BlobReq]struct{}) + for _, req := range reqs { + distinct[req] = struct{}{} + } + require.Len(t, got, len(distinct)) + + // Every blob must byte-match the single-blob read path. + for req, data := range got { + expected, err := repo.GetBlobBytes(req.Type, req.MAC) + require.NoError(t, err) + require.Equal(t, expected, data) + } +} + +func TestGetBlobsMissingBlob(t *testing.T) { + repo := ptesting.GenerateRepository(t, nil, nil, nil) + reqs := chunkRequests(t, repo, "/a.txt") + + bogus := objects.MAC{0xDE, 0xAD, 0xBE, 0xEF} + reqs = append(reqs, repository.BlobReq{Type: resources.RT_CHUNK, MAC: bogus}) + + var missing, found int + for res, err := range repo.GetBlobs(context.Background(), reqs, nil) { + if res.MAC == bogus { + require.ErrorIs(t, err, repository.ErrBlobNotFound) + require.Nil(t, res.Data) + missing++ + } else { + require.NoError(t, err) + found++ + } + } + require.Equal(t, 1, missing) + require.Equal(t, len(reqs)-1, found) +} + +func TestGetBlobsAllMissing(t *testing.T) { + // A batch where NO request resolves: every result carries + // ErrBlobNotFound and the iterator terminates cleanly. + repo := ptesting.GenerateRepository(t, nil, nil, nil) + _ = chunkRequests(t, repo, "/a.txt") // populate the repo, request none of it + + reqs := []repository.BlobReq{ + {Type: resources.RT_CHUNK, MAC: objects.MAC{0xDE, 0xAD}}, + {Type: resources.RT_CHUNK, MAC: objects.MAC{0xBE, 0xEF}}, + } + + count := 0 + for res, err := range repo.GetBlobs(context.Background(), reqs, nil) { + require.ErrorIs(t, err, repository.ErrBlobNotFound) + require.Nil(t, res.Data) + count++ + } + require.Equal(t, len(reqs), count) +} + +func TestGetBlobsRangeReadError(t *testing.T) { + // Blobs resolve in the state but the packfile is gone from the + // store: every blob of the failed range yields its identity plus + // the read error — no panic, no phantom success. + repo := ptesting.GenerateRepository(t, nil, nil, nil) + reqs := chunkRequests(t, repo, "/a.txt", "/b.txt") + + for mac := range repo.ListPackfiles() { + require.NoError(t, repo.DeletePackfile(mac)) + } + + count := 0 + for res, err := range repo.GetBlobs(context.Background(), reqs, nil) { + require.Error(t, err) + require.Nil(t, res.Data) + count++ + } + require.Equal(t, len(reqs), count) +} + +func TestGetBlobsEmpty(t *testing.T) { + repo := ptesting.GenerateRepository(t, nil, nil, nil) + for res, err := range repo.GetBlobs(context.Background(), nil, nil) { + t.Fatalf("unexpected result %x (err %v)", res.MAC, err) + } +}