mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-10-31 01:10:41 +01:00
filer: fetch only the part of a chunk the view covers
Replaces the PinRandomMode size heuristic with a per-chunk coverage rule. ViewFromVisibleIntervals already clips chunk views to the request window, so a view that is not IsFullChunk() is one the request only partially needs; fetch it as a range regardless of the detected read pattern. This closes the holes a request-size pin left open: ranges larger than SeqTolerance no longer revert to whole-chunk downloads once their buffers look sequential, and ranges that fully cover a chunk keep the shared whole-chunk path instead of fetching 256KiB slices piecemeal. Prefetch (MaybeCache) skips clipped views so it cannot amplify a range read either. PinRandomMode is dropped: no caller needs it once coverage drives the fetch choice. Range fetches route through fetchChunkDataFn so tests observe them the same way as whole-chunk downloads.
This commit is contained in:
1 parent
58aa9acfb0
commit
dd75add62f
8 files changed
+173
-50
No files matched your search
@@ -380,15 +380,6 @@ func fetchWholeChunk(ctx context.Context, bytesBuffer *bytes.Buffer, lookupFileI
|
||||
})
|
||||
}
|
||||
|
||||
func fetchChunkRange(ctx context.Context, buffer []byte, lookupFileIdFn wdclient.LookupFileIdFunctionType, fileId string, cipherKey []byte, isGzipped bool, offset int64, refreshUrls util_http.RefreshUrlsFunc) (int, error) {
|
||||
urlStrings, err := lookupFileIdFn(ctx, fileId)
|
||||
if err != nil {
|
||||
glog.ErrorfCtx(ctx, "operation LookupFileId %s failed, err: %v", fileId, err)
|
||||
return 0, err
|
||||
}
|
||||
return util_http.RetriedFetchChunkData(ctx, buffer, urlStrings, cipherKey, isGzipped, false, offset, fileId, refreshUrls)
|
||||
}
|
||||
|
||||
// retriedStreamFetchChunkData streams a chunk from the first location that
|
||||
// answers. refreshUrls may be nil; when a location failed and a later one
|
||||
// answered, it is called so the reads that follow start from a fresh list.
|
||||
|
||||
+13
-9
@@ -180,12 +180,6 @@ func (c *ChunkReadAt) ReleaseStream() {
|
||||
c.readerCache.releaseStream(&c.stream)
|
||||
}
|
||||
|
||||
// PinRandomMode pins this reader to range fetches; used by callers that
|
||||
// already know the request is a partial read, e.g. a small ranged GET.
|
||||
func (c *ChunkReadAt) PinRandomMode() {
|
||||
c.readerPattern.PinRandomMode()
|
||||
}
|
||||
|
||||
func (c *ChunkReadAt) ReadAt(p []byte, offset int64) (n int, err error) {
|
||||
|
||||
c.readerPattern.MonitorReadAt(offset, len(p))
|
||||
@@ -355,14 +349,17 @@ func (c *ChunkReadAt) doReadAt(ctx context.Context, p []byte, offset int64) (n i
|
||||
|
||||
func (c *ChunkReadAt) readChunkSliceAt(ctx context.Context, buffer []byte, chunkView *ChunkView, nextChunkViews *Interval[*ChunkView], offset uint64) (n int, err error) {
|
||||
|
||||
if c.readerPattern.IsRandomMode() {
|
||||
// A view clipped to part of its chunk (e.g. the edge of a ranged GET,
|
||||
// whose views ViewFromVisibleIntervals clips to the request) only ever
|
||||
// needs that part: fetch it as a range no matter the detected pattern.
|
||||
// Fetching the chunk whole would multiply volume-server reads.
|
||||
if !chunkView.IsFullChunk() || c.readerPattern.IsRandomMode() {
|
||||
c.readerCache.releaseStream(&c.stream)
|
||||
n, err := c.readerCache.chunkCache.ReadChunkAt(buffer, chunkView.FileId, offset)
|
||||
if n > 0 {
|
||||
return n, err
|
||||
}
|
||||
return fetchChunkRange(ctx, buffer, c.readerCache.lookupFileIdFn, chunkView.FileId, chunkView.CipherKey, chunkView.IsGzipped, int64(offset),
|
||||
refreshUrls(ctx, c.readerCache.cacheInvalidator, c.readerCache.lookupFileIdFn, chunkView.FileId))
|
||||
return c.readerCache.fetchChunkRange(ctx, buffer, chunkView, int64(offset))
|
||||
}
|
||||
|
||||
shouldCache := (uint64(chunkView.ViewOffset) + chunkView.ChunkSize) <= c.readerCache.chunkCache.GetMaxFilePartSizeInCache()
|
||||
@@ -386,6 +383,13 @@ func (c *ChunkReadAt) readChunkSliceAt(ctx context.Context, buffer []byte, chunk
|
||||
// readChunkSliceAtForParallel is a simplified version for parallel chunk fetching
|
||||
// It doesn't update lastChunkFid or trigger prefetch (handled by the caller)
|
||||
func (c *ChunkReadAt) readChunkSliceAtForParallel(ctx context.Context, buffer []byte, chunkView *ChunkView, offset uint64) (n int, err error) {
|
||||
if !chunkView.IsFullChunk() {
|
||||
n, err = c.readerCache.chunkCache.ReadChunkAt(buffer, chunkView.FileId, offset)
|
||||
if n > 0 {
|
||||
return n, err
|
||||
}
|
||||
return c.readerCache.fetchChunkRange(ctx, buffer, chunkView, int64(offset))
|
||||
}
|
||||
shouldCache := (uint64(chunkView.ViewOffset) + chunkView.ChunkSize) <= c.readerCache.chunkCache.GetMaxFilePartSizeInCache()
|
||||
return c.readerCache.ReadChunkAt(ctx, buffer, chunkView.FileId, chunkView.CipherKey, chunkView.IsGzipped, int64(offset), int(chunkView.ChunkSize), shouldCache)
|
||||
}
|
||||
|
||||
@@ -208,6 +208,143 @@ func TestChunkStreamConcurrentReadsOnOneReader(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
type recordedFetch struct {
|
||||
fileId string
|
||||
isFullChunk bool
|
||||
offset int64
|
||||
size int
|
||||
}
|
||||
|
||||
// fetchRecorder stubs the volume fetch and records how each chunk was
|
||||
// requested: isFullChunk=false is a range fetch of just the view's slice,
|
||||
// isFullChunk=true is a whole-chunk download into the shared cache.
|
||||
func fetchRecorder(rc *ReaderCache) (fetches *[]recordedFetch) {
|
||||
var mu sync.Mutex
|
||||
recorded := &[]recordedFetch{}
|
||||
rc.fetchChunkDataFn = func(_ context.Context, buffer []byte, _ []string, _ []byte, _ bool, isFullChunk bool, offset int64, fileId string, _ util_http.RefreshUrlsFunc) (int, error) {
|
||||
mu.Lock()
|
||||
*recorded = append(*recorded, recordedFetch{fileId, isFullChunk, offset, len(buffer)})
|
||||
mu.Unlock()
|
||||
for i := range buffer {
|
||||
buffer[i] = fileId[len(fileId)-1]
|
||||
}
|
||||
return len(buffer), nil
|
||||
}
|
||||
return recorded
|
||||
}
|
||||
|
||||
// A reader whose views are clipped to a request window — how the S3 gateway
|
||||
// builds a ranged GET — must fetch only the covered part of each chunk:
|
||||
// clipped edge views take range fetches, a fully covered chunk keeps the
|
||||
// shared whole-chunk path. This is what keeps a ranged GET larger than a
|
||||
// buffer from multiplying volume-server reads (issue #11564), without giving
|
||||
// up whole-chunk caching where the whole chunk is actually wanted.
|
||||
func TestChunkReadAtClippedViewsFetchOnlyCoveredParts(t *testing.T) {
|
||||
const chunkSize = 64 << 10
|
||||
|
||||
rc := NewReaderCache(64, (*chunk_cache.TieredChunkCache)(nil), func(context.Context, string) ([]string, error) {
|
||||
return []string{"unused"}, nil
|
||||
}, nil)
|
||||
defer rc.destroy()
|
||||
fetches := fetchRecorder(rc)
|
||||
|
||||
// Window [56KiB, 136KiB): tail of chunk0, all of chunk1, head of chunk2.
|
||||
views := NewIntervalList[*ChunkView]()
|
||||
views.AppendInterval(&Interval[*ChunkView]{
|
||||
StartOffset: chunkSize - 8<<10,
|
||||
StopOffset: chunkSize,
|
||||
Value: &ChunkView{FileId: "chunk0", OffsetInChunk: chunkSize - 8<<10, ViewSize: 8 << 10, ViewOffset: chunkSize - 8<<10, ChunkSize: chunkSize},
|
||||
})
|
||||
views.AppendInterval(&Interval[*ChunkView]{
|
||||
StartOffset: chunkSize,
|
||||
StopOffset: 2 * chunkSize,
|
||||
Value: &ChunkView{FileId: "chunk1", ViewSize: chunkSize, ViewOffset: chunkSize, ChunkSize: chunkSize},
|
||||
})
|
||||
views.AppendInterval(&Interval[*ChunkView]{
|
||||
StartOffset: 2 * chunkSize,
|
||||
StopOffset: 2*chunkSize + 8<<10,
|
||||
Value: &ChunkView{FileId: "chunk2", ViewSize: 8 << 10, ViewOffset: 2 * chunkSize, ChunkSize: chunkSize},
|
||||
})
|
||||
|
||||
reader := NewChunkReaderAtFromClient(context.Background(), rc, views, 3*chunkSize, 0)
|
||||
buf := make([]byte, chunkSize+16<<10)
|
||||
if n, err := reader.ReadAt(buf, chunkSize-8<<10); err != nil || n != len(buf) {
|
||||
t.Fatalf("window read: n=%d err=%v", n, err)
|
||||
}
|
||||
// buf holds [56KiB, 136KiB): chunk0's tail, chunk1, chunk2's head.
|
||||
for i, b := range buf {
|
||||
want := byte('1')
|
||||
if i < 8<<10 {
|
||||
want = '0'
|
||||
} else if i >= 8<<10+chunkSize {
|
||||
want = '2'
|
||||
}
|
||||
if b != want {
|
||||
t.Fatalf("buf[%d]=%q, want %q", i, b, want)
|
||||
}
|
||||
}
|
||||
|
||||
want := []recordedFetch{
|
||||
{fileId: "chunk0", isFullChunk: false, offset: chunkSize - 8<<10, size: 8 << 10},
|
||||
{fileId: "chunk1", isFullChunk: true, offset: 0, size: chunkSize},
|
||||
{fileId: "chunk2", isFullChunk: false, offset: 0, size: 8 << 10},
|
||||
}
|
||||
got := map[string]recordedFetch{}
|
||||
for _, f := range *fetches {
|
||||
if _, dup := got[f.fileId]; dup {
|
||||
t.Fatalf("chunk %s fetched more than once: %+v", f.fileId, *fetches)
|
||||
}
|
||||
got[f.fileId] = f
|
||||
}
|
||||
for _, w := range want {
|
||||
if g, ok := got[w.fileId]; !ok {
|
||||
t.Fatalf("chunk %s never fetched: %+v", w.fileId, *fetches)
|
||||
} else if g != w {
|
||||
t.Fatalf("chunk %s fetched as %+v, want %+v", w.fileId, g, w)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// The regression from issue #11564: a ranged GET sitting inside one big
|
||||
// chunk. Every buffer of the request must stay a range fetch — none may
|
||||
// escalate into a whole-chunk download once the reads look sequential.
|
||||
func TestChunkReadAtRangeInsideOneChunkStaysRangeFetch(t *testing.T) {
|
||||
const chunkSize = 1 << 20
|
||||
const sliceSize = 16 << 10
|
||||
|
||||
rc := NewReaderCache(64, (*chunk_cache.TieredChunkCache)(nil), func(context.Context, string) ([]string, error) {
|
||||
return []string{"unused"}, nil
|
||||
}, nil)
|
||||
defer rc.destroy()
|
||||
fetches := fetchRecorder(rc)
|
||||
|
||||
// Range [32KiB, 96KiB) inside one 1MiB chunk: a single clipped view.
|
||||
views := NewIntervalList[*ChunkView]()
|
||||
views.AppendInterval(&Interval[*ChunkView]{
|
||||
StartOffset: 32 << 10,
|
||||
StopOffset: 96 << 10,
|
||||
Value: &ChunkView{FileId: "chunk0", OffsetInChunk: 32 << 10, ViewSize: 64 << 10, ViewOffset: 32 << 10, ChunkSize: chunkSize},
|
||||
})
|
||||
|
||||
reader := NewChunkReaderAtFromClient(context.Background(), rc, views, chunkSize, 0)
|
||||
for offset := int64(32 << 10); offset < 96<<10; offset += sliceSize {
|
||||
buf := make([]byte, sliceSize)
|
||||
if n, err := reader.ReadAt(buf, offset); err != nil || n != sliceSize {
|
||||
t.Fatalf("read at %d: n=%d err=%v", offset, n, err)
|
||||
}
|
||||
}
|
||||
|
||||
if len(*fetches) != 4 {
|
||||
t.Fatalf("got %d fetches, want 4 range fetches: %+v", len(*fetches), *fetches)
|
||||
}
|
||||
for i, f := range *fetches {
|
||||
wantOffset := int64(32<<10) + int64(i)*sliceSize
|
||||
if f.isFullChunk || f.offset != wantOffset || f.size != sliceSize {
|
||||
t.Fatalf("fetch %d = %+v, want range fetch offset=%d size=%d", i, f, wantOffset, sliceSize)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// A chunk a stream is positioned in must outlast downloader-limit eviction:
|
||||
// otherwise a busy cache drops the buffer mid-stream and forces a refetch.
|
||||
func TestChunkReadAtPinnedChunkSurvivesEviction(t *testing.T) {
|
||||
|
||||
@@ -100,6 +100,11 @@ func (rc *ReaderCache) MaybeCache(chunkViews *Interval[*ChunkView], count int) {
|
||||
// abort when slots are filled
|
||||
return
|
||||
}
|
||||
if !chunkView.IsFullChunk() {
|
||||
// the view is clipped to part of the chunk, so the rest is never
|
||||
// read; prefetching it whole would download bytes nobody needs
|
||||
continue
|
||||
}
|
||||
|
||||
// glog.V(4).Infof("prefetch %s offset %d", chunkView.FileId, chunkView.ViewOffset)
|
||||
// cache this chunk if not yet
|
||||
@@ -114,6 +119,20 @@ func (rc *ReaderCache) MaybeCache(chunkViews *Interval[*ChunkView], count int) {
|
||||
return
|
||||
}
|
||||
|
||||
// fetchChunkRange downloads only [offset, offset+len(buffer)) of a chunk,
|
||||
// for views clipped to part of their chunk and for random-mode reads. It
|
||||
// goes through fetchChunkDataFn so tests observe range fetches the same way
|
||||
// they observe whole-chunk downloads.
|
||||
func (rc *ReaderCache) fetchChunkRange(ctx context.Context, buffer []byte, chunkView *ChunkView, offset int64) (int, error) {
|
||||
urlStrings, err := rc.lookupFileIdFn(ctx, chunkView.FileId)
|
||||
if err != nil {
|
||||
glog.ErrorfCtx(ctx, "operation LookupFileId %s failed, err: %v", chunkView.FileId, err)
|
||||
return 0, err
|
||||
}
|
||||
return rc.fetchChunkDataFn(ctx, buffer, urlStrings, chunkView.CipherKey, chunkView.IsGzipped, false, offset, chunkView.FileId,
|
||||
refreshUrls(ctx, rc.cacheInvalidator, rc.lookupFileIdFn, chunkView.FileId))
|
||||
}
|
||||
|
||||
// chunkStream is one sequential reader's position in a shared ReaderCache.
|
||||
// The chunk it is reading stays pinned until the stream reads it to the end or
|
||||
// moves elsewhere, so another stream finishing or leaving the same chunk does
|
||||
|
||||
@@ -72,7 +72,7 @@ func TestReaderCacheBudgetInFlight(t *testing.T) {
|
||||
return len(buffer), nil
|
||||
}
|
||||
if prefetch {
|
||||
rc.MaybeCache(&Interval[*ChunkView]{Value: &ChunkView{FileId: "chunk", ChunkSize: 3 << 10}}, 1)
|
||||
rc.MaybeCache(&Interval[*ChunkView]{Value: &ChunkView{FileId: "chunk", ViewSize: 3 << 10, ChunkSize: 3 << 10}}, 1)
|
||||
} else {
|
||||
readers.Add(1)
|
||||
go func() {
|
||||
@@ -180,7 +180,7 @@ func TestReaderCacheFailedPrefetchReleasesBudget(t *testing.T) {
|
||||
rc.fetchChunkDataFn = func(_ context.Context, _ []byte, _ []string, _ []byte, _ bool, _ bool, _ int64, _ string, _ util_http.RefreshUrlsFunc) (int, error) {
|
||||
return 0, fmt.Errorf("fetch failed")
|
||||
}
|
||||
rc.MaybeCache(&Interval[*ChunkView]{Value: &ChunkView{FileId: "failed", ChunkSize: 1024}}, 1)
|
||||
rc.MaybeCache(&Interval[*ChunkView]{Value: &ChunkView{FileId: "failed", ViewSize: 1024, ChunkSize: 1024}}, 1)
|
||||
deadline := time.Now().Add(5 * time.Second)
|
||||
for {
|
||||
rc.Lock()
|
||||
@@ -278,7 +278,7 @@ func TestReaderCachePrefetchBufferDroppedAfterRead(t *testing.T) {
|
||||
buffer[0] = 42
|
||||
return len(buffer), nil
|
||||
}
|
||||
rc.MaybeCache(&Interval[*ChunkView]{Value: &ChunkView{FileId: "chunk", ChunkSize: 4 << 10}}, 1)
|
||||
rc.MaybeCache(&Interval[*ChunkView]{Value: &ChunkView{FileId: "chunk", ViewSize: 4 << 10, ChunkSize: 4 << 10}}, 1)
|
||||
|
||||
buf := make([]byte, 4<<10)
|
||||
n, err := rc.ReadChunkAt(context.Background(), buf, "chunk", nil, false, 0, 4<<10, false)
|
||||
|
||||
@@ -7,7 +7,6 @@ import (
|
||||
type ReaderPattern struct {
|
||||
isSequentialCounter int64
|
||||
readFrontier int64 // highest (offset+size) observed across reads
|
||||
pinnedRandom int64 // set when the caller already knows the reads are a partial range
|
||||
}
|
||||
|
||||
const ModeChangeLimit = 3
|
||||
@@ -66,12 +65,5 @@ func (rp *ReaderPattern) MonitorReadAt(offset int64, size int) {
|
||||
}
|
||||
|
||||
func (rp *ReaderPattern) IsRandomMode() bool {
|
||||
return atomic.LoadInt64(&rp.pinnedRandom) != 0 || atomic.LoadInt64(&rp.isSequentialCounter) < 0
|
||||
}
|
||||
|
||||
// PinRandomMode pins the pattern to random for readers that already know
|
||||
// they serve a partial range, e.g. a small ranged GET: contiguous buffer
|
||||
// reads within the request must not escalate into whole-chunk fetches.
|
||||
func (rp *ReaderPattern) PinRandomMode() {
|
||||
atomic.StoreInt64(&rp.pinnedRandom, 1)
|
||||
return atomic.LoadInt64(&rp.isSequentialCounter) < 0
|
||||
}
|
||||
@@ -122,17 +122,3 @@ func TestReaderPatternRangedReadStaysRandom(t *testing.T) {
|
||||
t.Fatal("sustained sequential reads should restore sequential mode")
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
// A reader pinned random (a known small ranged request) must stay random
|
||||
// for every buffer of the request, near or far.
|
||||
func TestReaderPatternPinnedRandom(t *testing.T) {
|
||||
rp := NewReaderPattern()
|
||||
rp.PinRandomMode()
|
||||
for i := int64(0); i < 10; i++ {
|
||||
rp.MonitorReadAt(i*256*1024, 256*1024) // contiguous from the start
|
||||
if !rp.IsRandomMode() {
|
||||
t.Fatalf("pinned reader flipped to sequential at read %d", i+1)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1104,12 +1104,6 @@ func (s3a *S3ApiServer) streamFromVolumeServers(w http.ResponseWriter, r *http.R
|
||||
chunkViews := filer.ViewFromVisibleIntervals(visibleIntervals, offset, size)
|
||||
reader := filer.NewChunkReaderAtFromClient(ctx, s3a.readerCache, chunkViews, totalSize, filer.DefaultPrefetchCount)
|
||||
defer reader.ReleaseStream()
|
||||
// A small ranged request reads back less than a whole chunk; pin random
|
||||
// mode so every buffer fetch stays a range read instead of downloading
|
||||
// each covered chunk in full.
|
||||
if isRangeRequest && size <= filer.SeqTolerance {
|
||||
reader.PinRandomMode()
|
||||
}
|
||||
streamPrepTime = time.Since(tStreamPrep)
|
||||
|
||||
// A cached chunk whose volume server is down, or whose needle was evicted and
|
||||
|
||||
Reference in new issue
Block a user