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) +}