From fb57f2b96c04ec8227f6f40b92cbcdf41c27cc2b Mon Sep 17 00:00:00 2001 From: Amy Xu Date: Mon, 9 Nov 2026 22:15:13 +0000 Subject: [PATCH 1/4] unpack largest layers first and add test to check largest layer is processed first Change-Id: I9c8b934f6d220134ca30dc9ace478f3d80bb8784 Signed-off-by: Amy Xu --- internal/imagecache/imagecache.go | 15 ++++- internal/imagecache/pull_gated_test.go | 87 +++++++++++++++++++++++++- 2 files changed, 98 insertions(+), 4 deletions(-) diff --git a/internal/imagecache/imagecache.go b/internal/imagecache/imagecache.go index d49112c6c9..648d385bd9 100644 --- a/internal/imagecache/imagecache.go +++ b/internal/imagecache/imagecache.go @@ -59,6 +59,7 @@ import ( "path/filepath" "runtime" "slices" + "sort" "strconv" "strings" "sync" @@ -631,10 +632,22 @@ func (s *Store) pullWith(ctx context.Context, parsedRef name.Reference, digest v } } + // Unpack largest layers first so a large layer does not queue behind small ones. + order := make([]int, len(layers)) + sizes := make([]int64, len(layers)) + for i, l := range layers { + order[i] = i + sizes[i], _ = l.Size() + } + sort.SliceStable(order, func(a, b int) bool { + return sizes[order[a]] > sizes[order[b]] + }) + layerDirs := make([]string, len(layers)) g, gctx := errgroup.WithContext(ctx) g.SetLimit(layerPullConcurrency) - for i, layer := range layers { + for _, i := range order { + layer := layers[i] g.Go(func() error { // This is the config's claim. Unpack hashes the content to check // it, since a remote layer's DiffID() would just repeat the config. diff --git a/internal/imagecache/pull_gated_test.go b/internal/imagecache/pull_gated_test.go index 445c577b3d..ef7a16d75a 100644 --- a/internal/imagecache/pull_gated_test.go +++ b/internal/imagecache/pull_gated_test.go @@ -26,6 +26,8 @@ import ( "net/http" "os" "path/filepath" + "slices" + "strconv" "strings" "sync" "testing" @@ -37,7 +39,8 @@ import ( // gatedRegistry is an in-memory registry whose blob GETs block while their // digest hex is gated. type gatedRegistry struct { - host string + host string + arrived chan string mu sync.Mutex gates map[string]chan struct{} @@ -45,20 +48,28 @@ type gatedRegistry struct { func newGatedRegistry(t *testing.T) *gatedRegistry { t.Helper() - g := &gatedRegistry{gates: map[string]chan struct{}{}} + g := &gatedRegistry{ + arrived: make(chan string, 16), + gates: map[string]chan struct{}{}, + } _, g.host = newWrappedRegistry(t, func(inner http.Handler) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if r.Method == http.MethodGet && strings.Contains(r.URL.Path, "/blobs/") { g.mu.Lock() - var ch chan struct{} + var ( + matched string + ch chan struct{} + ) for hex, c := range g.gates { if strings.Contains(r.URL.Path, hex) { + matched = hex ch = c break } } g.mu.Unlock() if ch != nil { + g.arrived <- matched <-ch } } @@ -389,3 +400,73 @@ func TestDetachedPullBoundedByPullTimeout(t *testing.T) { t.Fatalf("EnsureImage on a wedged pull = %v, want context.DeadlineExceeded", err) } } + +// A large layer at the end of the manifest must start in the first +// layerPullConcurrency wave rather than queueing behind smaller layers, while +// LayerDirs still returns layers in manifest order. +func TestPullStartsLargestLayersFirst(t *testing.T) { + reg := newGatedRegistry(t) + layers := make([]v1.Layer, layerPullConcurrency+1) + releases := make([]func(), len(layers)) + for i := range layerPullConcurrency { + layers[i] = layerFromEntries(t, []tarEntry{ + {name: strconv.Itoa(i), typeflag: tar.TypeReg, mode: 0o644, body: strconv.Itoa(i)}, + }) + releases[i] = reg.gate(t, layers[i]) + } + // Make the final layer incompressible enough to have the largest compressed blob. + largeBody := make([]byte, 4096) + for i := range largeBody { + largeBody[i] = byte(i*31 + i>>3) + } + large := layerFromEntries(t, []tarEntry{ + {name: strconv.Itoa(layerPullConcurrency), typeflag: tar.TypeReg, mode: 0o644, body: string(largeBody)}, + }) + layers[layerPullConcurrency] = large + releases[layerPullConcurrency] = reg.gate(t, large) + largeDigest, err := large.Digest() + if err != nil { + t.Fatalf("large layer digest: %v", err) + } + + ref := reg.host + "/test/lpt:latest" + pushImage(t, ref, v1.Config{}, layers...) + + store := newTestStore(t) + type result struct { + img *Image + err error + } + done := make(chan result, 1) + go func() { + img, err := store.EnsureImage(context.Background(), ref) + done <- result{img: img, err: err} + }() + + // All 5 layers are gated, so at most layerPullConcurrency (4) blob GETs can + // arrive before a gate opens. The largest layer (manifest index 4) must be + // among them. + firstWave := make([]string, layerPullConcurrency) + for i := range firstWave { + firstWave[i] = <-reg.arrived + } + if !slices.Contains(firstWave, largeDigest.Hex) { + t.Errorf("first %d layer pulls = %v, want largest layer %s started in first wave", layerPullConcurrency, firstWave, largeDigest.Hex) + } + + for _, release := range releases { + release() + } + res := <-done + if res.err != nil { + t.Fatalf("EnsureImage: %v", res.err) + } + if len(res.img.LayerDirs) != len(layers) { + t.Fatalf("len(LayerDirs) = %d, want %d", len(res.img.LayerDirs), len(layers)) + } + for i, dir := range res.img.LayerDirs { + if _, err := os.Stat(filepath.Join(dir, layerFSDirName, strconv.Itoa(i))); err != nil { + t.Errorf("LayerDirs[%d] missing entry %q: %v", i, strconv.Itoa(i), err) + } + } +} From 95ed0e4623649437bcc68d2b19724a251913d664 Mon Sep 17 00:00:00 2001 From: Amy Xu Date: Mon, 9 Nov 2026 22:47:11 +0000 Subject: [PATCH 2/4] use the existing pattern instead of a results struct Change-Id: I60ae80f216eed307cf76ede3526608f1747ad11a Signed-off-by: Amy Xu --- internal/imagecache/pull_gated_test.go | 23 ++++++++++------------- 1 file changed, 10 insertions(+), 13 deletions(-) diff --git a/internal/imagecache/pull_gated_test.go b/internal/imagecache/pull_gated_test.go index ef7a16d75a..172b826818 100644 --- a/internal/imagecache/pull_gated_test.go +++ b/internal/imagecache/pull_gated_test.go @@ -433,14 +433,12 @@ func TestPullStartsLargestLayersFirst(t *testing.T) { pushImage(t, ref, v1.Config{}, layers...) store := newTestStore(t) - type result struct { - img *Image - err error - } - done := make(chan result, 1) + done := make(chan error, 1) + var img *Image go func() { - img, err := store.EnsureImage(context.Background(), ref) - done <- result{img: img, err: err} + var err error + img, err = store.EnsureImage(context.Background(), ref) + done <- err }() // All 5 layers are gated, so at most layerPullConcurrency (4) blob GETs can @@ -457,14 +455,13 @@ func TestPullStartsLargestLayersFirst(t *testing.T) { for _, release := range releases { release() } - res := <-done - if res.err != nil { - t.Fatalf("EnsureImage: %v", res.err) + if err := <-done; err != nil { + t.Fatalf("EnsureImage: %v", err) } - if len(res.img.LayerDirs) != len(layers) { - t.Fatalf("len(LayerDirs) = %d, want %d", len(res.img.LayerDirs), len(layers)) + if len(img.LayerDirs) != len(layers) { + t.Fatalf("len(LayerDirs) = %d, want %d", len(img.LayerDirs), len(layers)) } - for i, dir := range res.img.LayerDirs { + for i, dir := range img.LayerDirs { if _, err := os.Stat(filepath.Join(dir, layerFSDirName, strconv.Itoa(i))); err != nil { t.Errorf("LayerDirs[%d] missing entry %q: %v", i, strconv.Itoa(i), err) } From b8780ed651e082868616714d417efaf75a8a9aec Mon Sep 17 00:00:00 2001 From: Amy Xu Date: Tue, 10 Nov 2026 01:07:27 +0000 Subject: [PATCH 3/4] pull sorting into helper, update test to do exact order match Change-Id: I04d5162c09d874c8317babd1c86fb2bc2b0f734a Signed-off-by: Amy Xu --- internal/imagecache/imagecache.go | 28 ++++++------ internal/imagecache/pull_gated_test.go | 59 ++++++++++++++------------ 2 files changed, 48 insertions(+), 39 deletions(-) diff --git a/internal/imagecache/imagecache.go b/internal/imagecache/imagecache.go index 648d385bd9..47aa6b65bf 100644 --- a/internal/imagecache/imagecache.go +++ b/internal/imagecache/imagecache.go @@ -632,21 +632,10 @@ func (s *Store) pullWith(ctx context.Context, parsedRef name.Reference, digest v } } - // Unpack largest layers first so a large layer does not queue behind small ones. - order := make([]int, len(layers)) - sizes := make([]int64, len(layers)) - for i, l := range layers { - order[i] = i - sizes[i], _ = l.Size() - } - sort.SliceStable(order, func(a, b int) bool { - return sizes[order[a]] > sizes[order[b]] - }) - layerDirs := make([]string, len(layers)) g, gctx := errgroup.WithContext(ctx) g.SetLimit(layerPullConcurrency) - for _, i := range order { + for _, i := range layerPullOrder(layers) { layer := layers[i] g.Go(func() error { // This is the config's claim. Unpack hashes the content to check @@ -707,6 +696,21 @@ func (s *Store) pullWith(ctx context.Context, parsedRef name.Reference, digest v return &Image{Digest: digest, Config: cfgFile.Config, LayerDirs: layerDirs}, nil } +// layerPullOrder returns manifest indices sorted by descending compressed size, +// preserving manifest order on ties so the largest layers unpack first. +func layerPullOrder(layers []v1.Layer) []int { + order := make([]int, len(layers)) + sizes := make([]int64, len(layers)) + for i, l := range layers { + order[i] = i + sizes[i], _ = l.Size() + } + sort.SliceStable(order, func(a, b int) bool { + return sizes[order[a]] > sizes[order[b]] + }) + return order +} + // pinLayers marks the layers as in use by an in-flight pull and returns the // matching unpin. Taken by each pull for all of its layers before any // ensureLayer call, outside the layer flight, so a pull that joins another diff --git a/internal/imagecache/pull_gated_test.go b/internal/imagecache/pull_gated_test.go index 172b826818..a9ceda98ac 100644 --- a/internal/imagecache/pull_gated_test.go +++ b/internal/imagecache/pull_gated_test.go @@ -27,7 +27,6 @@ import ( "os" "path/filepath" "slices" - "strconv" "strings" "sync" "testing" @@ -401,32 +400,38 @@ func TestDetachedPullBoundedByPullTimeout(t *testing.T) { } } -// A large layer at the end of the manifest must start in the first -// layerPullConcurrency wave rather than queueing behind smaller layers, while -// LayerDirs still returns layers in manifest order. +// Larger layers at the end of the manifest must start in the first +// layerPullConcurrency wave ahead of smaller layers, breaking ties in manifest +// order while LayerDirs still returns layers in manifest order. func TestPullStartsLargestLayersFirst(t *testing.T) { reg := newGatedRegistry(t) - layers := make([]v1.Layer, layerPullConcurrency+1) - releases := make([]func(), len(layers)) - for i := range layerPullConcurrency { + specs := []struct { + name string + size int + }{ + {"a", 1 << 11}, + {"b", 1 << 11}, // ties with "a"; manifest order picks "a" before "b" + {"c", 1 << 16}, + {"d", 1 << 18}, + {"e", 1 << 20}, + } + layers := make([]v1.Layer, len(specs)) + digests := make([]string, len(specs)) + releases := make([]func(), len(specs)) + for i, s := range specs { layers[i] = layerFromEntries(t, []tarEntry{ - {name: strconv.Itoa(i), typeflag: tar.TypeReg, mode: 0o644, body: strconv.Itoa(i)}, + {name: s.name, typeflag: tar.TypeReg, mode: 0o644, body: strings.Repeat(s.name, s.size)}, }) + d, err := layers[i].Digest() + if err != nil { + t.Fatalf("layer %s digest: %v", s.name, err) + } + digests[i] = d.Hex releases[i] = reg.gate(t, layers[i]) } - // Make the final layer incompressible enough to have the largest compressed blob. - largeBody := make([]byte, 4096) - for i := range largeBody { - largeBody[i] = byte(i*31 + i>>3) - } - large := layerFromEntries(t, []tarEntry{ - {name: strconv.Itoa(layerPullConcurrency), typeflag: tar.TypeReg, mode: 0o644, body: string(largeBody)}, - }) - layers[layerPullConcurrency] = large - releases[layerPullConcurrency] = reg.gate(t, large) - largeDigest, err := large.Digest() - if err != nil { - t.Fatalf("large layer digest: %v", err) + + if got, want := layerPullOrder(layers), []int{4, 3, 2, 0, 1}; !slices.Equal(got, want) { + t.Fatalf("layerPullOrder = %v, want %v", got, want) } ref := reg.host + "/test/lpt:latest" @@ -442,14 +447,14 @@ func TestPullStartsLargestLayersFirst(t *testing.T) { }() // All 5 layers are gated, so at most layerPullConcurrency (4) blob GETs can - // arrive before a gate opens. The largest layer (manifest index 4) must be - // among them. + // arrive before a gate opens: "c", "d", "e" by size, plus "a" ahead of + // equal-sized "b" by manifest order. firstWave := make([]string, layerPullConcurrency) for i := range firstWave { firstWave[i] = <-reg.arrived } - if !slices.Contains(firstWave, largeDigest.Hex) { - t.Errorf("first %d layer pulls = %v, want largest layer %s started in first wave", layerPullConcurrency, firstWave, largeDigest.Hex) + if slices.Contains(firstWave, digests[1]) { + t.Errorf("first %d layer pulls = %v, want layer b (%s) queued behind larger layers c..e and earlier tie a", layerPullConcurrency, firstWave, digests[1]) } for _, release := range releases { @@ -462,8 +467,8 @@ func TestPullStartsLargestLayersFirst(t *testing.T) { t.Fatalf("len(LayerDirs) = %d, want %d", len(img.LayerDirs), len(layers)) } for i, dir := range img.LayerDirs { - if _, err := os.Stat(filepath.Join(dir, layerFSDirName, strconv.Itoa(i))); err != nil { - t.Errorf("LayerDirs[%d] missing entry %q: %v", i, strconv.Itoa(i), err) + if _, err := os.Stat(filepath.Join(dir, layerFSDirName, specs[i].name)); err != nil { + t.Errorf("LayerDirs[%d] missing entry %q: %v", i, specs[i].name, err) } } } From cc25665e73353e916e20a683f22b1aa341db6e59 Mon Sep 17 00:00:00 2001 From: Amy Xu Date: Tue, 10 Nov 2026 05:07:22 +0000 Subject: [PATCH 4/4] materialize errors when calling size, make arrivals optional Change-Id: I7fcf3c044be39217f5b027efae52094b072b6986 Signed-off-by: Amy Xu --- internal/imagecache/imagecache.go | 26 ++++++++++------ internal/imagecache/pull_gated_test.go | 42 ++++++++++++++++++-------- 2 files changed, 47 insertions(+), 21 deletions(-) diff --git a/internal/imagecache/imagecache.go b/internal/imagecache/imagecache.go index 47aa6b65bf..fee23d81a2 100644 --- a/internal/imagecache/imagecache.go +++ b/internal/imagecache/imagecache.go @@ -48,6 +48,7 @@ package imagecache import ( + "cmp" "context" "encoding/json" "errors" @@ -59,7 +60,6 @@ import ( "path/filepath" "runtime" "slices" - "sort" "strconv" "strings" "sync" @@ -635,7 +635,11 @@ func (s *Store) pullWith(ctx context.Context, parsedRef name.Reference, digest v layerDirs := make([]string, len(layers)) g, gctx := errgroup.WithContext(ctx) g.SetLimit(layerPullConcurrency) - for _, i := range layerPullOrder(layers) { + order, err := layerPullOrder(layers) + if err != nil { + return nil, err + } + for _, i := range order { layer := layers[i] g.Go(func() error { // This is the config's claim. Unpack hashes the content to check @@ -696,19 +700,23 @@ func (s *Store) pullWith(ctx context.Context, parsedRef name.Reference, digest v return &Image{Digest: digest, Config: cfgFile.Config, LayerDirs: layerDirs}, nil } -// layerPullOrder returns manifest indices sorted by descending compressed size, -// preserving manifest order on ties so the largest layers unpack first. -func layerPullOrder(layers []v1.Layer) []int { +// layerPullOrder returns layer indices by descending compressed size, so the +// largest layers start first. Ties keep manifest order. +func layerPullOrder(layers []v1.Layer) ([]int, error) { order := make([]int, len(layers)) sizes := make([]int64, len(layers)) for i, l := range layers { + size, err := l.Size() + if err != nil { + return nil, fmt.Errorf("while reading layer %d size: %w", i, err) + } order[i] = i - sizes[i], _ = l.Size() + sizes[i] = size } - sort.SliceStable(order, func(a, b int) bool { - return sizes[order[a]] > sizes[order[b]] + slices.SortStableFunc(order, func(a, b int) int { + return cmp.Compare(sizes[b], sizes[a]) }) - return order + return order, nil } // pinLayers marks the layers as in use by an in-flight pull and returns the diff --git a/internal/imagecache/pull_gated_test.go b/internal/imagecache/pull_gated_test.go index a9ceda98ac..a1dfbba83c 100644 --- a/internal/imagecache/pull_gated_test.go +++ b/internal/imagecache/pull_gated_test.go @@ -38,19 +38,16 @@ import ( // gatedRegistry is an in-memory registry whose blob GETs block while their // digest hex is gated. type gatedRegistry struct { - host string - arrived chan string + host string - mu sync.Mutex - gates map[string]chan struct{} + mu sync.Mutex + gates map[string]chan struct{} + arrived chan string // nil until arrivals is called } func newGatedRegistry(t *testing.T) *gatedRegistry { t.Helper() - g := &gatedRegistry{ - arrived: make(chan string, 16), - gates: map[string]chan struct{}{}, - } + g := &gatedRegistry{gates: map[string]chan struct{}{}} _, g.host = newWrappedRegistry(t, func(inner http.Handler) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if r.Method == http.MethodGet && strings.Contains(r.URL.Path, "/blobs/") { @@ -66,9 +63,12 @@ func newGatedRegistry(t *testing.T) *gatedRegistry { break } } + arrived := g.arrived g.mu.Unlock() if ch != nil { - g.arrived <- matched + if arrived != nil { + arrived <- matched + } <-ch } } @@ -78,6 +78,15 @@ func newGatedRegistry(t *testing.T) *gatedRegistry { return g } +// arrivals returns a channel that receives the digest hex of each gated blob +// GET as it reaches its gate. +func (g *gatedRegistry) arrivals() <-chan string { + g.mu.Lock() + defer g.mu.Unlock() + g.arrived = make(chan string, 16) + return g.arrived +} + // gate blocks future GETs of the layer's compressed blob until release. func (g *gatedRegistry) gate(t *testing.T, layer v1.Layer) (release func()) { t.Helper() @@ -430,13 +439,18 @@ func TestPullStartsLargestLayersFirst(t *testing.T) { releases[i] = reg.gate(t, layers[i]) } - if got, want := layerPullOrder(layers), []int{4, 3, 2, 0, 1}; !slices.Equal(got, want) { - t.Fatalf("layerPullOrder = %v, want %v", got, want) + order, err := layerPullOrder(layers) + if err != nil { + t.Fatalf("layerPullOrder: %v", err) + } + if want := []int{4, 3, 2, 0, 1}; !slices.Equal(order, want) { + t.Fatalf("layerPullOrder = %v, want %v", order, want) } ref := reg.host + "/test/lpt:latest" pushImage(t, ref, v1.Config{}, layers...) + arrived := reg.arrivals() store := newTestStore(t) done := make(chan error, 1) var img *Image @@ -451,7 +465,11 @@ func TestPullStartsLargestLayersFirst(t *testing.T) { // equal-sized "b" by manifest order. firstWave := make([]string, layerPullConcurrency) for i := range firstWave { - firstWave[i] = <-reg.arrived + select { + case firstWave[i] = <-arrived: + case err := <-done: + t.Fatalf("EnsureImage returned before %d layer pulls arrived: %v", layerPullConcurrency, err) + } } if slices.Contains(firstWave, digests[1]) { t.Errorf("first %d layer pulls = %v, want layer b (%s) queued behind larger layers c..e and earlier tie a", layerPullConcurrency, firstWave, digests[1])