From 80a6f5fde12687fc596c654811e0e092dfbeec18 Mon Sep 17 00:00:00 2001 From: Elizabeth Worstell Date: Mon, 20 Jul 2026 15:07:01 -0700 Subject: [PATCH] perf(cache): fan out large ranged reads through the remote tier Generalize the S3 object-window adapter into objectWindow, a byte window over any pinned object revision, and use it to apply the large-range policy to Remote ranged reads: ranges of 8 MiB or more pay one Stat to learn the object's size and ETag, then split into parallel 4 MiB sub-range requests pinned with If-Match. Chunks stay below the remote's own S3 fan-out threshold so sub-ranges never fan out twice, and memory is bounded by the reader's reorder window (2*concurrency*chunkSize). Ranges the policy cannot pin or split delegate to a single stream. --- internal/cache/object_window.go | 62 +++++++++++++++ internal/cache/remote.go | 63 +++++++++++++++ internal/cache/remote_test.go | 122 ++++++++++++++++++++++++++++++ internal/cache/s3_parallel_get.go | 59 ++++----------- 4 files changed, 263 insertions(+), 43 deletions(-) create mode 100644 internal/cache/object_window.go diff --git a/internal/cache/object_window.go b/internal/cache/object_window.go new file mode 100644 index 0000000..d734609 --- /dev/null +++ b/internal/cache/object_window.go @@ -0,0 +1,62 @@ +package cache + +import ( + "context" + "fmt" + "io" + "net/http" + + "github.com/alecthomas/errors" + + "github.com/block/cachew/client" +) + +// objectWindow adapts a byte window of a pinned object revision to +// [client.RangeReader], letting ParallelGet drive parallel range downloads. +// Range offsets are relative to the window; openRange receives absolute +// object offsets and must pin every request to the window's revision, so +// discovery and sub-range requests can never splice revisions. Response +// headers are synthesized from the pinned values: the backend enforces the +// real preconditions, surfacing violations as read errors. The window's +// identity is bound at construction, so the Key argument is ignored. +type objectWindow struct { + openRange func(ctx context.Context, start, length int64) (io.ReadCloser, error) + start int64 // window offset within the object + length int64 // window size in bytes + etag string +} + +// newCacheObjectWindow returns a window over [start, start+length) of the +// pinned revision of key in c, served by ETag-conditional ranged Opens. +func newCacheObjectWindow(c client.RangeReader, key Key, start, length int64, etag string) *objectWindow { + return &objectWindow{ + openRange: func(ctx context.Context, start, length int64) (io.ReadCloser, error) { + rc, _, err := c.Open(ctx, key, Range(start, start+length), IfMatch(etag)) + return rc, errors.WithStack(err) + }, + start: start, + length: length, + etag: etag, + } +} + +func (w *objectWindow) Open(ctx context.Context, _ Key, opts ...Option) (io.ReadCloser, http.Header, error) { + start, length, outcome := NewRequestOptions(opts...).ResolveRange(w.length, w.etag) + headers := http.Header{} + if w.etag != "" { + headers.Set(ETagKey, w.etag) + } + switch outcome { + case RangeNotSatisfiable: + headers.Set("Content-Range", fmt.Sprintf("bytes */%d", w.length)) + return nil, headers, errors.WithStack(ErrRangeNotSatisfiable) + case RangePartial: + headers.Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", start, start+length-1, w.length)) + case RangeFull: + } + rc, err := w.openRange(ctx, w.start+start, length) + if err != nil { + return nil, nil, err + } + return rc, headers, nil +} diff --git a/internal/cache/remote.go b/internal/cache/remote.go index 5a8ca58..85d7911 100644 --- a/internal/cache/remote.go +++ b/internal/cache/remote.go @@ -3,7 +3,9 @@ package cache import ( "context" "io" + "math" "net/http" + "strconv" "time" "github.com/alecthomas/errors" @@ -38,11 +40,72 @@ func (r *Remote) Namespace(namespace Namespace) Cache { return &Remote{c: r.c.Namespace(namespace)} } +// Remote ranged reads at least 2*remoteRangeChunkSize long fan out into +// parallel sub-range requests, because a single HTTP stream is limited to a +// fraction of the available bandwidth. Chunks stay below the remote's own S3 +// large-range threshold (2*minRangePartSize) so each sub-range maps to a +// single upstream request rather than fanning out twice. Memory is bounded by +// the reader's reorder window: 2*concurrency*chunkSize per read. +const ( + remoteRangeChunkSize = minRangePartSize + remoteRangeConcurrency = 8 +) + func (r *Remote) Open(ctx context.Context, key Key, opts ...Option) (io.ReadCloser, http.Header, error) { + if remoteRangeMayFanOut(NewRequestOptions(opts...)) { + return r.parallelRangedOpen(ctx, key, opts) + } rc, h, err := r.c.Open(ctx, key, opts...) return rc, h, errors.WithStack(err) } +// remoteRangeMayFanOut reports whether the request's Range could span enough +// bytes to be worth a Stat plus parallel fan-out. The raw range spec is +// resolved against an unbounded object, so explicit and suffix ranges report +// their exact requested length and open-ended ranges an effectively infinite +// one; If-Range gating needs the stored ETag and is deferred to the resolve +// against the real object. +func remoteRangeMayFanOut(ro RequestOptions) bool { + if ro.Range == "" { + return false + } + ro.IfRange = "" + _, length, outcome := ro.ResolveRange(math.MaxInt64, "") + return outcome == RangePartial && length >= 2*remoteRangeChunkSize +} + +// parallelRangedOpen serves a large ranged read with parallel sub-range +// requests pinned to the stored ETag. A preliminary Stat resolves the +// request's conditionals and range against the object's real size and +// revision; requests the policy cannot pin or split degrade to a single +// delegated stream. +func (r *Remote) parallelRangedOpen(ctx context.Context, key Key, opts []Option) (io.ReadCloser, http.Header, error) { + headers, err := r.c.Stat(ctx, key, opts...) + if err != nil { + return nil, headers, errors.WithStack(err) + } + size, sizeErr := strconv.ParseInt(headers.Get("Content-Length"), 10, 64) + etag := headers.Get(ETagKey) + if sizeErr != nil || etag == "" { + rc, h, err := r.c.Open(ctx, key, opts...) + return rc, h, errors.WithStack(err) + } + start, length, partial, rangeErr := rangeShortCircuit(headers, size, opts) + if rangeErr != nil { + return nil, headers, errors.WithStack(rangeErr) + } + if !partial || length < 2*remoteRangeChunkSize { + rc, h, err := r.c.Open(ctx, key, opts...) + return rc, h, errors.WithStack(err) + } + window := newCacheObjectWindow(r.c, key, start, length, etag) + rc, err := client.ParallelGetReader(ctx, window, Key{}, remoteRangeChunkSize, remoteRangeConcurrency) + if err != nil { + return nil, nil, errors.WithStack(err) + } + return rc, headers, nil +} + func (r *Remote) Stat(ctx context.Context, key Key, opts ...Option) (http.Header, error) { return errors.WithStack2(r.c.Stat(ctx, key, opts...)) } diff --git a/internal/cache/remote_test.go b/internal/cache/remote_test.go index 0761900..cff71f8 100644 --- a/internal/cache/remote_test.go +++ b/internal/cache/remote_test.go @@ -1,11 +1,14 @@ package cache_test import ( + "bytes" "io" "log/slog" + "math/rand" "net/http" "net/http/httptest" "os" + "sync/atomic" "testing" "time" @@ -82,6 +85,125 @@ func TestRemoteInvalidateSkipsRemoteAuthoritativeTier(t *testing.T) { assert.Equal(t, content, data) } +// countingHandler wraps an http.Handler, counting HEAD requests and ranged +// GETs, and how many ranged GETs lack an If-Match revision pin. +type countingHandler struct { + http.Handler + heads atomic.Int32 + rangedGets atomic.Int32 + unpinnedGets atomic.Int32 +} + +func (h *countingHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { + switch { + case r.Method == http.MethodHead: + h.heads.Add(1) + case r.Method == http.MethodGet && r.Header.Get("Range") != "": + h.rangedGets.Add(1) + if r.Header.Get("If-Match") == "" { + h.unpinnedGets.Add(1) + } + } + h.Handler.ServeHTTP(w, r) +} + +func newCountingRemote(t *testing.T) (cache.Cache, *countingHandler) { + t.Helper() + ctx := t.Context() + _, ctx = logging.Configure(ctx, logging.Config{Level: slog.LevelError}) + memCache, err := cache.NewMemory(ctx, cache.MemoryConfig{LimitMB: 64, MaxTTL: time.Hour}) + assert.NoError(t, err) + t.Cleanup(func() { memCache.Close() }) + + mux := http.NewServeMux() + _, err = strategy.NewAPIV1(ctx, struct{}{}, memCache, mux) + assert.NoError(t, err) + counting := &countingHandler{Handler: mux} + ts := httptest.NewServer(counting) + t.Cleanup(ts.Close) + + remote := cache.NewRemote(ts.URL, nil).Namespace("test") + t.Cleanup(func() { remote.Close() }) + return remote, counting +} + +func TestRemoteLargeRangedReadFansOut(t *testing.T) { + ctx := t.Context() + remote, counting := newCountingRemote(t) + + data := make([]byte, 12<<20) + _, err := rand.New(rand.NewSource(42)).Read(data) + assert.NoError(t, err) + + key := cache.NewKey("remote-fan-out") + w, err := remote.Create(ctx, key, nil, time.Hour) + assert.NoError(t, err) + _, err = w.Write(data) + assert.NoError(t, err) + assert.NoError(t, w.Close()) + + r, headers, err := remote.Open(ctx, key, cache.Range(1<<20, 11<<20)) + assert.NoError(t, err) + got, err := io.ReadAll(r) + assert.NoError(t, err) + assert.NoError(t, r.Close()) + + assert.True(t, bytes.Equal(data[1<<20:11<<20], got), "reassembled range differs from original") + assert.Equal(t, "bytes 1048576-11534335/12582912", headers.Get("Content-Range")) + assert.Equal(t, "10485760", headers.Get("Content-Length")) + assert.True(t, counting.rangedGets.Load() >= 3, + "expected sub-range fan-out, got %d ranged GETs", counting.rangedGets.Load()) + assert.Equal(t, int32(0), counting.unpinnedGets.Load(), "every sub-range request must carry If-Match") +} + +func TestRemoteSmallRangedReadStaysSingleStream(t *testing.T) { + ctx := t.Context() + remote, counting := newCountingRemote(t) + + data := bytes.Repeat([]byte("abcdefgh"), 1<<20/8) + key := cache.NewKey("remote-small-range") + w, err := remote.Create(ctx, key, nil, time.Hour) + assert.NoError(t, err) + _, err = w.Write(data) + assert.NoError(t, err) + assert.NoError(t, w.Close()) + + r, _, err := remote.Open(ctx, key, cache.Range(16, 4096)) + assert.NoError(t, err) + got, err := io.ReadAll(r) + assert.NoError(t, err) + assert.NoError(t, r.Close()) + + assert.True(t, bytes.Equal(data[16:4096], got), "range body differs from original") + assert.Equal(t, int32(0), counting.heads.Load(), "small ranges must not pay a Stat") + assert.Equal(t, int32(1), counting.rangedGets.Load(), "small ranges must use a single request") +} + +func TestRemoteLargeRangedReadIfRangeMissServesFullBody(t *testing.T) { + ctx := t.Context() + remote, counting := newCountingRemote(t) + + data := make([]byte, 9<<20) + _, err := rand.New(rand.NewSource(7)).Read(data) + assert.NoError(t, err) + + key := cache.NewKey("remote-if-range-miss") + w, err := remote.Create(ctx, key, nil, time.Hour) + assert.NoError(t, err) + _, err = w.Write(data) + assert.NoError(t, err) + assert.NoError(t, w.Close()) + + r, _, err := remote.Open(ctx, key, cache.Range(0, 9<<20), cache.IfRange(`"other-revision"`)) + assert.NoError(t, err) + got, err := io.ReadAll(r) + assert.NoError(t, err) + assert.NoError(t, r.Close()) + + assert.True(t, bytes.Equal(data, got), "If-Range miss must serve the full representation") + assert.Equal(t, int32(1), counting.rangedGets.Load(), "an unpinnable range must degrade to a single stream") +} + func TestRemoteCacheSoak(t *testing.T) { if os.Getenv("SOAK_TEST") == "" { t.Skip("Skipping soak test; set SOAK_TEST=1 to run") diff --git a/internal/cache/s3_parallel_get.go b/internal/cache/s3_parallel_get.go index 112fc14..1ee6ad8 100644 --- a/internal/cache/s3_parallel_get.go +++ b/internal/cache/s3_parallel_get.go @@ -2,9 +2,7 @@ package cache import ( "context" - "fmt" "io" - "net/http" "github.com/alecthomas/errors" "github.com/minio/minio-go/v7" @@ -23,7 +21,7 @@ const minRangePartSize int64 = 4 << 20 func (s *S3) parallelGetReader(ctx context.Context, bucket, objectName string, size int64, etag string) (io.ReadCloser, error) { chunkSize := int64(s.config.DownloadPartSizeMB) << 20 // #nosec G115 -- DownloadPartSizeMB is a small operator-supplied tuning value. if concurrency := int(s.config.DownloadConcurrency); size > chunkSize && concurrency > 1 { // #nosec G115 -- DownloadConcurrency is a small operator-supplied tuning value. - window := &s3ObjectWindow{s3: s, bucket: bucket, objectName: objectName, start: 0, length: size, etag: etag} + window := s.objectWindow(bucket, objectName, 0, size, etag) return client.ParallelGetReader(ctx, window, Key{}, chunkSize, concurrency) //nolint:wrapcheck } obj, err := s.client.GetObject(ctx, bucket, objectName, minio.GetObjectOptions{}) @@ -46,10 +44,24 @@ func (s *S3) rangedGetReader(ctx context.Context, bucket, objectName string, sta } chunkSize := (length + concurrency - 1) / concurrency chunkSize = min(max(chunkSize, minRangePartSize), int64(s.config.DownloadPartSizeMB)<<20) // #nosec G115 -- DownloadPartSizeMB is a small operator-supplied tuning value. - window := &s3ObjectWindow{s3: s, bucket: bucket, objectName: objectName, start: start, length: length, etag: etag} + window := s.objectWindow(bucket, objectName, start, length, etag) return client.ParallelGetReader(ctx, window, Key{}, chunkSize, int(concurrency)) //nolint:wrapcheck } +// objectWindow returns a window over [start, start+length) of the pinned S3 +// object revision, served by direct sub-range GETs so chunk requests bypass +// S3.Open's per-call stat and range policy. +func (s *S3) objectWindow(bucket, objectName string, start, length int64, etag string) *objectWindow { + return &objectWindow{ + openRange: func(ctx context.Context, start, length int64) (io.ReadCloser, error) { + return s.rangeGetReader(ctx, bucket, objectName, start, length, etag) + }, + start: start, + length: length, + etag: etag, + } +} + // rangeGetReader returns an io.ReadCloser for a single byte range of an S3 // object, pinned to etag (when non-empty) so the read sees a consistent // object revision. @@ -69,42 +81,3 @@ func (s *S3) rangeGetReader(ctx context.Context, bucket, objectName string, star } return &s3Reader{obj: obj}, nil } - -// s3ObjectWindow adapts a byte window of a pinned S3 object revision to -// [client.RangeReader], letting ParallelGet drive parallel S3 downloads. -// Range offsets are relative to the window, and every request carries an -// If-Match for the pinned etag regardless of the supplied options, so -// discovery and sub-range requests can never splice revisions. Response -// headers are synthesized from the pinned values: minio enforces the real -// preconditions (SetRange, SetMatchETag) at the protocol level, surfacing -// violations as read errors. The window's identity is bound in the struct, so -// the Key argument is ignored. -type s3ObjectWindow struct { - s3 *S3 - bucket string - objectName string - start int64 // window offset within the object - length int64 // window size in bytes - etag string -} - -func (w *s3ObjectWindow) Open(ctx context.Context, _ Key, opts ...Option) (io.ReadCloser, http.Header, error) { - start, length, outcome := NewRequestOptions(opts...).ResolveRange(w.length, w.etag) - headers := http.Header{} - if w.etag != "" { - headers.Set(ETagKey, w.etag) - } - switch outcome { - case RangeNotSatisfiable: - headers.Set("Content-Range", fmt.Sprintf("bytes */%d", w.length)) - return nil, headers, errors.WithStack(ErrRangeNotSatisfiable) - case RangePartial: - headers.Set("Content-Range", fmt.Sprintf("bytes %d-%d/%d", start, start+length-1, w.length)) - case RangeFull: - } - rc, err := w.s3.rangeGetReader(ctx, w.bucket, w.objectName, w.start+start, length, w.etag) - if err != nil { - return nil, nil, err - } - return rc, headers, nil -}