diff --git a/weed/s3api/s3api_object_handlers_copy.go b/weed/s3api/s3api_object_handlers_copy.go index ea728b067..bad328a12 100644 --- a/weed/s3api/s3api_object_handlers_copy.go +++ b/weed/s3api/s3api_object_handlers_copy.go @@ -1379,16 +1379,7 @@ func (s3a *S3ApiServer) prepareChunkCopy(sourceFileId, dstPath string, expectedD // uploadChunkData uploads chunk data to the destination using common upload logic // isCompressed indicates if the data is already compressed and should not be compressed again func (s3a *S3ApiServer) uploadChunkData(chunkData []byte, assignResult *filer_pb.AssignVolumeResponse, isCompressed bool) error { - dstUrl := fmt.Sprintf("http://%s/%s", assignResult.Location.Url, assignResult.FileId) - - uploadOption := &operation.UploadOption{ - UploadUrl: dstUrl, - Cipher: false, // Data is already encrypted if source had CipherKey; don't re-encrypt - IsInputCompressed: isCompressed, - MimeType: "", - PairMap: nil, - Jwt: security.EncodedJwt(assignResult.Auth), - } + uploadOption := newChunkUploadOption(chunkData, assignResult, isCompressed) uploader, err := operation.NewUploader() if err != nil { return fmt.Errorf("create uploader: %w", err) @@ -1401,6 +1392,33 @@ func (s3a *S3ApiServer) uploadChunkData(chunkData []byte, assignResult *filer_pb return nil } +// multipartFramingOverhead reserves space for the multipart wrapper +// upload_content writes around chunkData (boundary + Content-Disposition + +// optional Content-Type/Content-Encoding/Content-MD5 headers + trailing +// boundary). Real-world overhead is a few hundred bytes; rounding to 1 KiB +// avoids a single grow on the buffer we hand to the multipart writer. +const multipartFramingOverhead = 1024 + +// newChunkUploadOption builds the operation.UploadOption used by every +// chunk-copy upload. It always sets BytesBuffer to a fresh, per-call buffer +// so upload_content does not fall back to the package-global +// valyala/bytebufferpool — that pool retains every high-water buffer for the +// process's lifetime, and under concurrent UploadPartCopy load it hoarded +// one chunk-sized buffer per concurrent upload (see #6541). The per-call +// buffer is GC'd as soon as the upload returns. +func newChunkUploadOption(chunkData []byte, assignResult *filer_pb.AssignVolumeResponse, isCompressed bool) *operation.UploadOption { + dstUrl := fmt.Sprintf("http://%s/%s", assignResult.Location.Url, assignResult.FileId) + return &operation.UploadOption{ + UploadUrl: dstUrl, + Cipher: false, // Data is already encrypted if source had CipherKey; don't re-encrypt + IsInputCompressed: isCompressed, + MimeType: "", + PairMap: nil, + Jwt: security.EncodedJwt(assignResult.Auth), + BytesBuffer: bytes.NewBuffer(make([]byte, 0, len(chunkData)+multipartFramingOverhead)), + } +} + // downloadChunkData downloads chunk data from the source URL func (s3a *S3ApiServer) downloadChunkData(srcUrl, fileId string, offset, size int64, cipherKey []byte) ([]byte, error) { jwt := filer.JwtForVolumeServer(fileId) diff --git a/weed/s3api/s3api_object_handlers_copy_chunk_upload_test.go b/weed/s3api/s3api_object_handlers_copy_chunk_upload_test.go new file mode 100644 index 000000000..913ceb268 --- /dev/null +++ b/weed/s3api/s3api_object_handlers_copy_chunk_upload_test.go @@ -0,0 +1,77 @@ +package s3api + +import ( + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" +) + +// TestNewChunkUploadOption_AvoidsBytePool is a regression guard for the +// s3-side retention amplifier in https://github.com/seaweedfs/seaweedfs/issues/6541. +// +// operation.upload_content falls back to the package-global +// valyala/bytebufferpool (which retains high-water buffers for the +// process's lifetime) whenever UploadOption.BytesBuffer is nil. Concurrent +// UploadPartCopy calls then hoard one chunk-sized buffer per concurrent +// upload in that pool, and RSS never recedes. The chunk-copy path must +// therefore always provide its own buffer; this test pins that contract. +func TestNewChunkUploadOption_AvoidsBytePool(t *testing.T) { + cases := []struct { + name string + chunkLen int + }{ + {"empty chunk", 0}, + {"small chunk", 1024}, + {"typical 8 MiB chunk", 8 * 1024 * 1024}, + {"large 64 MiB chunk", 64 * 1024 * 1024}, + } + + assign := &filer_pb.AssignVolumeResponse{ + FileId: "1,foo", + Location: &filer_pb.Location{ + Url: "127.0.0.1:8080", + PublicUrl: "127.0.0.1:8080", + }, + Auth: "", + } + + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + data := make([]byte, c.chunkLen) + opt := newChunkUploadOption(data, assign, false) + + if opt.BytesBuffer == nil { + t.Fatalf("BytesBuffer must be non-nil so chunk uploads bypass " + + "the package-global bytebufferpool (regression: #6541)") + } + if got, want := opt.BytesBuffer.Cap(), c.chunkLen+multipartFramingOverhead; got < want { + t.Errorf("BytesBuffer cap=%d, want >=%d (one alloc-and-grow "+ + "per upload defeats the purpose of pre-sizing)", got, want) + } + if opt.BytesBuffer.Len() != 0 { + t.Errorf("BytesBuffer not empty: len=%d", opt.BytesBuffer.Len()) + } + if opt.Cipher { + t.Errorf("Cipher must be false: chunk-copy data is already " + + "encrypted if the source had a CipherKey") + } + }) + } +} + +// TestNewChunkUploadOption_PerCallIsolation verifies each call returns a +// distinct buffer. Sharing one buffer across concurrent uploads would +// corrupt the multipart bodies under concurrent UploadPartCopy. +func TestNewChunkUploadOption_PerCallIsolation(t *testing.T) { + assign := &filer_pb.AssignVolumeResponse{ + FileId: "1,foo", + Location: &filer_pb.Location{Url: "127.0.0.1:8080"}, + } + + o1 := newChunkUploadOption(nil, assign, false) + o2 := newChunkUploadOption(nil, assign, false) + if o1.BytesBuffer == o2.BytesBuffer { + t.Fatalf("newChunkUploadOption must hand each caller a distinct buffer; " + + "sharing one across concurrent uploads would corrupt the multipart bodies") + } +} diff --git a/weed/util/buffer_pool/sync_pool.go b/weed/util/buffer_pool/sync_pool.go index b97274691..d0b4f589f 100644 --- a/weed/util/buffer_pool/sync_pool.go +++ b/weed/util/buffer_pool/sync_pool.go @@ -5,6 +5,16 @@ import ( "sync" ) +// maxRetainedBufferCap caps the capacity of buffers we hand back to the +// sync.Pool. Buffers grown past this (e.g. by a 64 MiB chunk upload through +// volume.PostHandler -> needle.ParseUpload -> bytes.Buffer.ReadFrom) are +// dropped instead of pooled, so the underlying byte array becomes garbage +// and is collected. Without this cap the pool effectively hoards every +// high-water buffer for the process's lifetime — see #6541, where Harbor's +// concurrent UploadPartCopy filled the pool with 64 MiB buffers and RSS +// never receded. +const maxRetainedBufferCap = 4 * 1024 * 1024 + var syncPool = sync.Pool{ New: func() interface{} { return new(bytes.Buffer) @@ -16,5 +26,13 @@ func SyncPoolGetBuffer() *bytes.Buffer { } func SyncPoolPutBuffer(buffer *bytes.Buffer) { + if buffer == nil { + return + } + if buffer.Cap() > maxRetainedBufferCap { + // Drop the buffer; let GC reclaim the oversized backing array. + return + } + buffer.Reset() syncPool.Put(buffer) } diff --git a/weed/util/buffer_pool/sync_pool_test.go b/weed/util/buffer_pool/sync_pool_test.go new file mode 100644 index 000000000..8eee077ac --- /dev/null +++ b/weed/util/buffer_pool/sync_pool_test.go @@ -0,0 +1,92 @@ +package buffer_pool + +import ( + "bytes" + "testing" +) + +// TestSyncPoolPutBuffer_DropsOversized is a regression guard for the volume-side +// retention amplifier in https://github.com/seaweedfs/seaweedfs/issues/6541. +// +// volume.PostHandler -> needle.ParseUpload -> bytes.Buffer.ReadFrom grows the +// buffer to chunk size. If we Put that buffer back as-is, sync.Pool keeps it +// at the grown capacity for the rest of the process's lifetime. With +// concurrent UploadPartCopy load that fills the pool with N × chunk-size +// backing arrays that never shrink — exactly the "RSS never recedes" pattern +// reported in the issue. +// +// We verify a Put + Get round trip can never round-trip a buffer larger than +// maxRetainedBufferCap, regardless of how big it grew while in use. +func TestSyncPoolPutBuffer_DropsOversized(t *testing.T) { + // Drain the pool so we start from a deterministic point. sync.Pool may + // still hold cached entries on other Ps, but for the small/big-cap + // distinction below that doesn't matter — we assert an invariant on + // every Get, not the identity of a specific buffer. + for i := 0; i < 64; i++ { + _ = SyncPoolGetBuffer() + } + + big := &bytes.Buffer{} + big.Grow(maxRetainedBufferCap * 4) // simulate a large upload buffer + if got := big.Cap(); got <= maxRetainedBufferCap { + t.Fatalf("test setup: big.Cap=%d should exceed threshold %d", + got, maxRetainedBufferCap) + } + SyncPoolPutBuffer(big) + + // Any number of Gets must not return a buffer with cap > threshold. + // (If big had been retained, sync.Pool's per-P cache would surface it + // on the very next Get on this goroutine.) + for i := 0; i < 16; i++ { + got := SyncPoolGetBuffer() + if cap := got.Cap(); cap > maxRetainedBufferCap { + t.Fatalf("Get %d returned buffer with cap=%d, exceeds threshold %d "+ + "(regression: oversized buffers retained in pool?)", + i, cap, maxRetainedBufferCap) + } + } +} + +// TestSyncPoolPutBuffer_KeepsRightSized verifies the cap is one-sided: we +// still pool reasonably-sized buffers so the common case (small uploads, +// header parsing) doesn't pay an alloc per request. +func TestSyncPoolPutBuffer_KeepsRightSized(t *testing.T) { + for i := 0; i < 64; i++ { + _ = SyncPoolGetBuffer() + } + + small := &bytes.Buffer{} + small.Grow(maxRetainedBufferCap / 2) + smallCap := small.Cap() + SyncPoolPutBuffer(small) + + // We don't assert pointer identity (sync.Pool can hand back any cached + // buffer), but the previously-Put buffer should appear among Gets on + // the same goroutine in the absence of a GC. If our cap-policy ever + // regresses to dropping right-sized buffers, this test starts seeing + // only fresh (cap=0) buffers and fails. + sawPooled := false + for i := 0; i < 8; i++ { + got := SyncPoolGetBuffer() + if got.Cap() == smallCap { + sawPooled = true + break + } + } + if !sawPooled { + t.Fatalf("right-sized buffer (cap=%d) never came back from the pool "+ + "(regression: cap policy too aggressive?)", smallCap) + } +} + +// TestSyncPoolPutBuffer_NilSafe documents that Put tolerates a nil buffer. +// The volume server defers Put on a buffer obtained via Get, but defensive +// callers in other paths may Put(nil); we should not panic. +func TestSyncPoolPutBuffer_NilSafe(t *testing.T) { + defer func() { + if r := recover(); r != nil { + t.Fatalf("SyncPoolPutBuffer(nil) panicked: %v", r) + } + }() + SyncPoolPutBuffer(nil) +}