From 1a62b1d09ff7ecbac54ef155b577a358b72e60e7 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Fri, 30 Oct 2026 08:58:48 -0700 Subject: [PATCH] filer: buffer recently fetched chunks so parallel readers do not refetch (#12254) * s3sink: upload through a concurrent-safe ReaderAt so parts do not thrash s3manager splits a seekable body into parallel section readers, and every part's ReadAt landed on the single chunk buffer in ChunkStreamReader: parts at different offsets kept evicting each other's chunk and refetching it whole, which produced massive read amplification against the volume servers and collapsed throughput. Build the upload body from filer.ChunkReadAt instead, the same concurrent-safe reader the mount and query paths use: each part's range is served from a shared reader cache with parallel chunk fetches and range-reads for clipped views, so every byte is fetched once per pass. Generated with [Devin](https://devin.ai) Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> * filer: buffer recently fetched chunks for concurrent readers * Revert "s3sink: upload through a concurrent-safe ReaderAt so parts do not thrash" This reverts commit be1ec2d7c7e3699293e8e6024cf646e6311302be. --------- Co-authored-by: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- weed/filer/stream.go | 70 ++++++++++++++++++++++---- weed/filer/stream_chunk_buffer_test.go | 58 +++++++++++++++++++++ 2 files changed, 118 insertions(+), 10 deletions(-) create mode 100644 weed/filer/stream_chunk_buffer_test.go diff --git a/weed/filer/stream.go b/weed/filer/stream.go index d31918a115..11cbdf7ade 100644 --- a/weed/filer/stream.go +++ b/weed/filer/stream.go @@ -2,6 +2,7 @@ package filer import ( "bytes" + "container/list" "context" "fmt" "io" @@ -372,17 +373,34 @@ func writeZero(w io.Writer, size int64) (err error) { } // ---------------- ChunkStreamReader ---------------------------------- + +// A reader buffers recently fetched chunks, not just the last one: readers +// consumed through io.SectionReader (e.g. parallel multipart uploads) keep a +// handful of part-sized windows hot at once, and a single shared buffer would +// refetch a whole chunk on nearly every ReadAt. The bounds cap the memory of +// tiny chunks as well as the total buffered bytes. +const maxBufferedChunks = 16 +const maxBufferedChunkBytes = 128 << 20 + +type bufferedChunk struct { + offset int64 + data []byte +} + type ChunkStreamReader struct { - head *Interval[*ChunkView] - chunkView *Interval[*ChunkView] - totalSize int64 - logicOffset int64 - buffer []byte - bufferOffset int64 - bufferLock sync.Mutex - chunk string - lookupFileId wdclient.LookupFileIdFunctionType - sourceErr error + head *Interval[*ChunkView] + chunkView *Interval[*ChunkView] + totalSize int64 + logicOffset int64 + buffer []byte + bufferOffset int64 + bufferLock sync.Mutex + chunk string + chunkBuffers map[int64]*list.Element + chunkBuffersLru *list.List + chunkBuffersBytes int64 + lookupFileId wdclient.LookupFileIdFunctionType + sourceErr error } var _ = io.ReadSeeker(&ChunkStreamReader{}) @@ -527,6 +545,14 @@ func (c *ChunkStreamReader) prepareBufferFor(offset int64) (err error) { } func (c *ChunkStreamReader) fetchChunkToBuffer(chunkView *ChunkView) error { + if elem, ok := c.chunkBuffers[chunkView.ViewOffset]; ok { + buffered := elem.Value.(*bufferedChunk) + c.chunkBuffersLru.MoveToFront(elem) + c.buffer = buffered.data + c.bufferOffset = buffered.offset + c.chunk = chunkView.FileId + return nil + } urlStrings, err := c.lookupFileId(context.Background(), chunkView.FileId) if err != nil { glog.V(1).Infof("operation LookupFileId %s failed, err: %v", chunkView.FileId, err) @@ -557,12 +583,33 @@ func (c *ChunkStreamReader) fetchChunkToBuffer(chunkView *ChunkView) error { c.buffer = buffer.Bytes() c.bufferOffset = chunkView.ViewOffset c.chunk = chunkView.FileId + c.rememberBufferedChunk(chunkView.ViewOffset, c.buffer) // glog.V(0).Infof("fetched %s [%d,%d)", chunkView.FileId, chunkView.ViewOffset, chunkView.ViewOffset+int64(chunkView.ViewSize)) return nil } +// rememberBufferedChunk caches a fetched chunk by its view offset and evicts +// the least recently used ones past the bounds. The chunk just inserted always +// survives eviction so the read that fetched it can proceed. +func (c *ChunkStreamReader) rememberBufferedChunk(offset int64, data []byte) { + if c.chunkBuffers == nil { + c.chunkBuffers = make(map[int64]*list.Element) + c.chunkBuffersLru = list.New() + } + c.chunkBuffers[offset] = c.chunkBuffersLru.PushFront(&bufferedChunk{offset: offset, data: data}) + c.chunkBuffersBytes += int64(len(data)) + for c.chunkBuffersLru.Len() > 1 && + (c.chunkBuffersLru.Len() > maxBufferedChunks || c.chunkBuffersBytes > maxBufferedChunkBytes) { + back := c.chunkBuffersLru.Back() + evicted := back.Value.(*bufferedChunk) + c.chunkBuffersLru.Remove(back) + delete(c.chunkBuffers, evicted.offset) + c.chunkBuffersBytes -= int64(len(evicted.data)) + } +} + func (c *ChunkStreamReader) rememberSourceError(err error) error { if c.sourceErr == nil { c.sourceErr = err @@ -589,6 +636,9 @@ func (c *ChunkStreamReader) Close() error { c.bufferLock.Lock() defer c.bufferLock.Unlock() c.buffer = nil + c.chunkBuffers = nil + c.chunkBuffersLru = nil + c.chunkBuffersBytes = 0 c.head = nil c.chunkView = nil return nil diff --git a/weed/filer/stream_chunk_buffer_test.go b/weed/filer/stream_chunk_buffer_test.go new file mode 100644 index 0000000000..9a015124c4 --- /dev/null +++ b/weed/filer/stream_chunk_buffer_test.go @@ -0,0 +1,58 @@ +package filer + +import ( + "bytes" + "context" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// Interleaved ReadAt calls across chunks must not refetch a chunk every time +// the read window moves: parallel consumers (e.g. multipart upload parts read +// through io.SectionReader) share the reader and would otherwise keep evicting +// one another's single buffer, refetching whole chunks on nearly every read. +func TestChunkStreamReaderReadAtReusesFetchedChunks(t *testing.T) { + contentA := bytes.Repeat([]byte("a"), 4096) + contentB := bytes.Repeat([]byte("b"), 4096) + var fetches atomic.Int32 + volume := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + fetches.Add(1) + switch strings.TrimPrefix(r.URL.Path, "/") { + case "1,aaaaaaaa": + _, _ = w.Write(contentA) + case "2,bbbbbbbb": + _, _ = w.Write(contentB) + } + })) + t.Cleanup(volume.Close) + + lookup := func(ctx context.Context, fileId string) ([]string, error) { + return []string{volume.URL + "/" + fileId}, nil + } + chunks := []*filer_pb.FileChunk{ + {FileId: "1,aaaaaaaa", Offset: 0, Size: uint64(len(contentA))}, + {FileId: "2,bbbbbbbb", Offset: int64(len(contentA)), Size: uint64(len(contentB))}, + } + reader := NewChunkStreamReaderFromLookup(context.Background(), lookup, chunks) + t.Cleanup(func() { _ = reader.Close() }) + + buf := make([]byte, 64) + for i := 0; i < 4; i++ { + for _, off := range []int64{0, int64(len(contentA))} { + _, err := reader.ReadAt(buf, off) + require.NoError(t, err) + } + } + assert.Equal(t, int32(2), fetches.Load(), "each chunk is fetched once, not once per read") + + _, err := reader.ReadAt(buf, 8) + require.NoError(t, err) + assert.Equal(t, contentA[8:72], buf) +}