From be1ec2d7c7e3699293e8e6024cf646e6311302be Mon Sep 17 00:00:00 2001 From: "user.email" Date: Fri, 30 Oct 2026 05:32:42 -0700 Subject: [PATCH] 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> --- weed/replication/sink/s3sink/s3_sink.go | 46 ++++++++++++++++---- weed/replication/sink/s3sink/s3_sink_test.go | 42 ++++++++++++++++++ 2 files changed, 80 insertions(+), 8 deletions(-) diff --git a/weed/replication/sink/s3sink/s3_sink.go b/weed/replication/sink/s3sink/s3_sink.go index 8f365a6e5a..ce3a5c741b 100644 --- a/weed/replication/sink/s3sink/s3_sink.go +++ b/weed/replication/sink/s3sink/s3_sink.go @@ -1,8 +1,12 @@ package S3Sink import ( + "bytes" + "context" "encoding/base64" "fmt" + "io" + "math" "net/url" "strconv" "strings" @@ -22,6 +26,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/replication/sink" "github.com/seaweedfs/seaweedfs/weed/replication/source" "github.com/seaweedfs/seaweedfs/weed/util" + "github.com/seaweedfs/seaweedfs/weed/util/chunk_cache" ) type S3Sink struct { @@ -164,15 +169,21 @@ func (s3sink *S3Sink) CreateEntry(key string, entry *filer_pb.Entry, signatures return nil } - reader := filer.NewFileReader(s3sink.filerSource, entry) + reader, closeReader, err := s3sink.entryReader(entry) + if err != nil { + return err + } // Decrypt SSE-encrypted objects so the destination receives plaintext decryptedReader, err := repl_util.MaybeDecryptReader(reader, entry) if err != nil { - repl_util.CloseReader(reader) + closeReader() return fmt.Errorf("decrypt SSE object: %w", err) } - defer repl_util.CloseMaybeDecryptedReader(reader, decryptedReader) + defer func() { + repl_util.CloseMaybeDecryptedReader(reader, decryptedReader) + closeReader() + }() // Create an uploader with the session and custom options uploader := s3manager.NewUploaderWithClient(s3sink.conn, func(u *s3manager.Uploader) { @@ -220,16 +231,35 @@ func (s3sink *S3Sink) CreateEntry(key string, entry *filer_pb.Entry, signatures uploadInput.ContentMD5 = aws.String(base64.StdEncoding.EncodeToString([]byte(entry.Attributes.Md5))) } _, err = uploader.Upload(&uploadInput) - if err != nil { - if sourceErr := filer.ReaderSourceError(reader); sourceErr != nil { - return fmt.Errorf("read source %s: %w", key, sourceErr) - } - } return err } +// entryReader returns a reader over the entry's content for the uploader. +// Chunked entries are read through a concurrent-safe ReaderAt with its own +// chunk cache, so parallel part readers each fetch the ranges they need; a +// single stream reader buffers only one chunk and concurrent parts would keep +// evicting it and refetching from the volume servers. +func (s3sink *S3Sink) entryReader(entry *filer_pb.Entry) (reader io.Reader, closeFn func(), err error) { + if len(entry.Content) > 0 { + return bytes.NewReader(entry.Content), func() {}, nil + } + + lookupFileIdFn := filer.LookupFn(s3sink.filerSource) + fileSize := int64(filer.FileSize(entry)) + chunkViews, err := filer.ViewFromChunks(context.Background(), lookupFileIdFn, entry.GetChunks(), 0, math.MaxInt64) + if err != nil { + return nil, nil, fmt.Errorf("resolve chunks: %w", err) + } + + concurrency := max(s3sink.uploaderConcurrency, 1) + filer.DefaultPrefetchCount + readerAt := filer.NewChunkReaderAtFromClient(context.Background(), + filer.NewReaderCache(concurrency, chunk_cache.NewChunkCacheInMemory(int64(concurrency)), lookupFileIdFn, nil), + chunkViews, fileSize, filer.DefaultPrefetchCount) + return io.NewSectionReader(readerAt, 0, fileSize), func() { readerAt.Close() }, nil +} + func (s3sink *S3Sink) UpdateEntry(key string, oldEntry *filer_pb.Entry, newParentPath string, newEntry *filer_pb.Entry, deleteIncludeChunks bool, signatures []int32) (foundExistingEntry bool, err error) { key = cleanKey(key) return true, s3sink.CreateEntry(key, newEntry, signatures) diff --git a/weed/replication/sink/s3sink/s3_sink_test.go b/weed/replication/sink/s3sink/s3_sink_test.go index 63bf8373ab..6ffdcbda76 100644 --- a/weed/replication/sink/s3sink/s3_sink_test.go +++ b/weed/replication/sink/s3sink/s3_sink_test.go @@ -94,6 +94,48 @@ func newCapturingS3Sink(t *testing.T) (*S3Sink, *capturePutRoundTripper) { return s, rt } +func TestS3SinkEntryReaderSupportsConcurrentParts(t *testing.T) { + sink := &S3Sink{uploaderConcurrency: 8} + entry := &filer_pb.Entry{ + Attributes: &filer_pb.FuseAttributes{FileSize: 100}, + Chunks: []*filer_pb.FileChunk{ + {FileId: "1,01637037d6", Size: 100}, + }, + } + + reader, closeFn, err := sink.entryReader(entry) + require.NoError(t, err) + defer closeFn() + + // s3manager only splits a body into parallel part readers when it is both + // an io.ReaderAt and an io.Seeker; the reader must be safe for concurrent + // ReadAt so parts do not evict one another's buffered chunk. + _, isReaderAt := reader.(io.ReaderAt) + seeker, isSeeker := reader.(io.Seeker) + require.True(t, isReaderAt, "upload body should support concurrent ReadAt") + require.True(t, isSeeker, "upload body should be seekable") + + size, err := seeker.Seek(0, io.SeekEnd) + require.NoError(t, err) + require.Equal(t, int64(100), size) +} + +func TestS3SinkEntryReaderForInlineContent(t *testing.T) { + sink := &S3Sink{uploaderConcurrency: 8} + entry := &filer_pb.Entry{ + Attributes: &filer_pb.FuseAttributes{FileSize: 4}, + Content: []byte("data"), + } + + reader, closeFn, err := sink.entryReader(entry) + require.NoError(t, err) + defer closeFn() + + content, err := io.ReadAll(reader) + require.NoError(t, err) + require.Equal(t, []byte("data"), content) +} + func TestS3SinkCreateEntryPassesMimeAsContentType(t *testing.T) { sink, rt := newCapturingS3Sink(t) entry := &filer_pb.Entry{