diff --git a/internal/fs/fs.go b/internal/fs/fs.go index 258c5cbfd4..3951ffe750 100644 --- a/internal/fs/fs.go +++ b/internal/fs/fs.go @@ -225,6 +225,7 @@ func NewFileSystem(ctx context.Context, serverCfg *ServerConfig) (fuseutil.FileS globalMaxWriteBlocksSem: semaphore.NewWeighted(serverCfg.NewConfig.Write.GlobalMaxBlocks), globalMaxReadBlocksSem: semaphore.NewWeighted(serverCfg.NewConfig.Read.GlobalMaxBlocks), globalMetadataPrefetchSem: semaphore.NewWeighted(serverCfg.NewConfig.MetadataCache.MetadataPrefetchMaxWorkers), + readBufferPool: gcsx.NewFixedSizeBufferPool(), } // Initialize MRD cache if enabled @@ -676,6 +677,9 @@ type fileSystem struct { // mrdCache manages the cache of inactive MultiRangeDownloaders. mrdCache *lru.Cache + + // readBufferPool is a pool of readPoolBufferSize (1 MiB) buffers. + readBufferPool gcsx.BufferPool } //////////////////////////////////////////////////////////////////////// @@ -3118,23 +3122,36 @@ func (fs *fileSystem) ReadFile( } // Serve the read. + req := gcsx.ReadRequest{ + Offset: op.Offset, + Size: op.Size, + } + + useReadBufferPool := op.Size > 0 && op.Dst == nil + if useReadBufferPool { + if !fs.newConfig.FileSystem.EnableKernelReader || fh.Inode().Bucket().BucketType().IsRapid() { + logger.Errorf("ReadFile: buffer pool allocation is only supported for regional buckets with"+ + " kernel reader enabled (EnableKernelReader: %v, IsRapid: %v)", + fs.newConfig.FileSystem.EnableKernelReader, + fh.Inode().Bucket().BucketType().IsRapid()) + fh.Inode().Unlock() + return syscall.ENOTSUP + } + + req.BufferPool = fs.readBufferPool + } else { + req.Buffer = op.Dst + } + if fs.newConfig.FileSystem.EnableKernelReader { var resp gcsx.ReadResponse - req := &gcsx.ReadRequest{ - Buffer: op.Dst, - Offset: op.Offset, - } - resp, err = fh.ReadWithKernelReader(ctx, req) + resp, err = fh.ReadWithKernelReader(ctx, &req) op.BytesRead = resp.Size op.Data = resp.Data op.Callback = resp.Callback } else if fs.newConfig.EnableNewReader { var resp gcsx.ReadResponse - req := &gcsx.ReadRequest{ - Buffer: op.Dst, - Offset: op.Offset, - } - resp, err = fh.ReadWithReadManager(ctx, req, fs.sequentialReadSizeMb) + resp, err = fh.ReadWithReadManager(ctx, &req, fs.sequentialReadSizeMb) op.BytesRead = resp.Size op.Data = resp.Data op.Callback = resp.Callback diff --git a/internal/fs/handle/file.go b/internal/fs/handle/file.go index dc2dc4ab9c..2ebdcd4d3e 100644 --- a/internal/fs/handle/file.go +++ b/internal/fs/handle/file.go @@ -209,8 +209,7 @@ func (fh *FileHandle) ReadWithReadManager(ctx context.Context, req *gcsx.ReadReq if !fh.inode.SourceGenerationIsAuthoritative() { // Read from inode if source generation is not authoratative defer fh.inode.Unlock() - n, err := fh.inode.Read(ctx, req.Buffer, req.Offset) - return gcsx.ReadResponse{Size: n}, err + return fh.inode.Read(ctx, req) } fh.lockHandleAndRelockInode(true) @@ -289,8 +288,7 @@ func (fh *FileHandle) ReadWithKernelReader(ctx context.Context, req *gcsx.ReadRe if !fh.inode.SourceGenerationIsAuthoritative() { // Read from inode if source generation is not authoritative. defer fh.inode.Unlock() - n, err := fh.inode.Read(ctx, req.Buffer, req.Offset) - return gcsx.ReadResponse{Size: n}, err + return fh.inode.Read(ctx, req) } fh.inode.Unlock() @@ -322,8 +320,8 @@ func (fh *FileHandle) Read(ctx context.Context, dst []byte, offset int64, sequen // we have one. if !fh.inode.SourceGenerationIsAuthoritative() { defer fh.inode.Unlock() - n, err = fh.inode.Read(ctx, dst, offset) - return dst, n, err + res, err := fh.inode.Read(ctx, &gcsx.ReadRequest{Buffer: dst, Offset: offset}) + return dst, res.Size, err } fh.lockHandleAndRelockInode(true) diff --git a/internal/fs/handle/file_test.go b/internal/fs/handle/file_test.go index a83b071805..eccc11ba23 100644 --- a/internal/fs/handle/file_test.go +++ b/internal/fs/handle/file_test.go @@ -701,6 +701,7 @@ func (t *fileTest) Test_ReadWithKernelReader_Success() { req := &gcsx.ReadRequest{ Buffer: buf, Offset: 0, + Size: int64(len(buf)), } // ReadWithKernelReader requires the inode lock to be held by the caller, and unlocks it internally. fh.inode.Lock() @@ -756,6 +757,7 @@ func (t *fileTest) Test_ReadWithKernelReader_NotAuthoritative() { req := &gcsx.ReadRequest{ Buffer: buf, Offset: 0, + Size: int64(len(buf)), } // ReadWithKernelReader requires the inode lock to be held on entry. fh.inode.Lock() @@ -810,6 +812,7 @@ func (t *fileTest) Test_ReadWithKernelReader_NotAuthoritative_ReadError() { req := &gcsx.ReadRequest{ Buffer: buf, Offset: -1, // Negative offset will trigger a read error from the local temp file + Size: int64(len(buf)), } // ReadWithKernelReader requires the inode lock to be held on entry. fh.inode.Lock() @@ -857,6 +860,7 @@ func (t *fileTest) Test_ReadWithKernelReader_NilReader() { req := &gcsx.ReadRequest{ Buffer: make([]byte, 4), Offset: 0, + Size: 4, } // ReadWithKernelReader requires the inode lock to be held on entry. fh.inode.Lock() @@ -921,6 +925,7 @@ func (t *fileTest) Test_ReadWithKernelReader_ReadAtError() { req := &gcsx.ReadRequest{ Buffer: buf, Offset: 0, + Size: int64(len(buf)), } // ReadWithKernelReader requires the inode lock to be held on entry. fh.inode.Lock() diff --git a/internal/fs/inode/file.go b/internal/fs/inode/file.go index 819cc26144..2149d58554 100644 --- a/internal/fs/inode/file.go +++ b/internal/fs/inode/file.go @@ -638,7 +638,7 @@ func (f *FileInode) Bucket() *gcsx.SyncerBucket { return f.bucket } -// Serve a read for this file with semantics matching io.ReaderAt. +// Serve a read for this file with semantics matching gcsx.Reader.ReadAt. // // The caller may be better off reading directly from GCS when // f.SourceGenerationIsAuthoritative() is true. @@ -646,8 +646,7 @@ func (f *FileInode) Bucket() *gcsx.SyncerBucket { // LOCKS_REQUIRED(f.mu) func (f *FileInode) Read( ctx context.Context, - dst []byte, - offset int64) (n int, err error) { + req *gcsx.ReadRequest) (res gcsx.ReadResponse, err error) { if f.bwh != nil { err = fmt.Errorf("unexpected read call for %q when streaming write is in progress for it", f.Name().LocalName()) return @@ -660,18 +659,34 @@ func (f *FileInode) Read( return } - // Read from the local content, propagating io.EOF. - n, err = f.content.ReadAt(dst, offset) - switch { - case err == io.EOF: - return + var n int + var data [][]byte + var callback func() + + if req.BufferPool != nil { + vBuf := gcsx.NewVectoredReadBuffer(req.BufferPool, req.Size) + var written int64 + written, err = vBuf.ReadFromAt(f.content, req.Offset) + n = int(written) + if n > 0 { + data = vBuf.Buffers() + callback = func() { vBuf.Release() } + } else { + vBuf.Release() + } + } else { + n, err = f.content.ReadAt(req.Buffer, req.Offset) + } - case err != nil: + if err != nil && err != io.EOF { err = fmt.Errorf("content.ReadAt: %w", err) - return } - return + return gcsx.ReadResponse{ + Size: n, + Data: data, + Callback: callback, + }, err } func getMetricOpenMode(openMode util.OpenMode) metrics.OpenMode { diff --git a/internal/fs/inode/file_test.go b/internal/fs/inode/file_test.go index 0aba6cebf7..57e7b5b387 100644 --- a/internal/fs/inode/file_test.go +++ b/internal/fs/inode/file_test.go @@ -501,8 +501,8 @@ func (t *FileTest) TestRead() { for _, tc := range testCases { desc := fmt.Sprintf("offset: %d, size: %d", tc.offset, tc.size) data := make([]byte, tc.size) - n, err := t.in.Read(t.ctx, data, tc.offset) - data = data[:n] + res, err := t.in.Read(t.ctx, &gcsx.ReadRequest{Buffer: data, Offset: tc.offset}) + data = data[:res.Size] // Ignore EOF. if err == io.EOF { @@ -514,6 +514,94 @@ func (t *FileTest) TestRead() { } } +func (t *FileTest) TestReadWithBufferPool() { + assert.Equal(t.T(), "taco", t.initialContents) + + testCases := []struct { + name string + offset int64 + size int64 + expectedData []string + expectedSize int + expectedErr error + }{ + { + name: "ExactFit_Full", + offset: 0, + size: 4, + expectedData: []string{"ta", "co"}, + expectedSize: 4, + expectedErr: nil, + }, + { + name: "ExactFit_Partial", + offset: 0, + size: 2, + expectedData: []string{"ta"}, + expectedSize: 2, + expectedErr: nil, + }, + { + name: "ExactFit_Remaining", + offset: 1, + size: 3, + expectedData: []string{"ac", "o"}, + expectedSize: 3, + expectedErr: nil, + }, + { + name: "PastEOF", + offset: 3, + size: 2, + expectedData: []string{"o"}, + expectedSize: 1, + expectedErr: nil, + }, + { + name: "AtEOF", + offset: 4, + size: 2, + expectedData: []string{}, + expectedSize: 0, + expectedErr: nil, + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func() { + // Arrange + pool := &gcsx.TestBufferPool{ + Buffers: [][]byte{make([]byte, 2), make([]byte, 2)}, + DefaultBufferSize: 2, + } + req := &gcsx.ReadRequest{ + BufferPool: pool, + Offset: tc.offset, + Size: tc.size, + } + + // Act + res, err := t.in.Read(t.ctx, req) + + // Assert + assert.ErrorIs(t.T(), err, tc.expectedErr) + assert.Equal(t.T(), tc.expectedSize, res.Size) + if res.Size > 0 { + var actualStrings []string + for _, b := range res.Data { + actualStrings = append(actualStrings, string(b)) + } + assert.Equal(t.T(), tc.expectedData, actualStrings) + require.NotNil(t.T(), res.Callback) + res.Callback() + } else { + assert.Empty(t.T(), res.Data) + assert.Nil(t.T(), res.Callback) + } + }) + } +} + func (t *FileTest) TestWrite() { var err error @@ -536,14 +624,14 @@ func (t *FileTest) TestWrite() { // Read back the content. var buf [1024]byte - n, err := t.in.Read(t.ctx, buf[:], 0) + res, err := t.in.Read(t.ctx, &gcsx.ReadRequest{Buffer: buf[:], Offset: 0}) if err == io.EOF { err = nil } assert.Nil(t.T(), err) - assert.Equal(t.T(), "pacoburrito", string(buf[:n])) + assert.Equal(t.T(), "pacoburrito", string(buf[:res.Size])) // Check attributes. attrs, err := t.in.Attributes(t.ctx, true) @@ -571,14 +659,14 @@ func (t *FileTest) TestTruncate() { // Read the contents. var buf [1024]byte - n, err := t.in.Read(t.ctx, buf[:], 0) + res, err := t.in.Read(t.ctx, &gcsx.ReadRequest{Buffer: buf[:], Offset: 0}) if err == io.EOF { err = nil } assert.Nil(t.T(), err) - assert.Equal(t.T(), "ta", string(buf[:n])) + assert.Equal(t.T(), "ta", string(buf[:res.Size])) // Check attributes. attrs, err = t.in.Attributes(t.ctx, true) @@ -1426,7 +1514,7 @@ func (t *FileTest) TestSetMtime_ContentClean() { var attrs fuseops.InodeAttributes // Cause the content to be faulted in. - _, err = t.in.Read(t.ctx, make([]byte, 1), 0) + _, err = t.in.Read(t.ctx, &gcsx.ReadRequest{Buffer: make([]byte, 1), Offset: 0}) assert.Nil(t.T(), err) // Set mtime. @@ -1780,13 +1868,13 @@ func (t *FileTest) TestReadFileWhenStreamingWritesAreEnabled() { // Flush is required before reading an object for which BWH is open. assert.NoError(t.T(), t.in.Flush(context.Background())) - n, err := t.in.Read(t.ctx, data, 0) + res, err := t.in.Read(t.ctx, &gcsx.ReadRequest{Buffer: data, Offset: 0}) if tc.performWrite { - assert.Equal(t.T(), len(data), n) + assert.Equal(t.T(), len(data), res.Size) require.NoError(t.T(), err) } else { - assert.Equal(t.T(), 0, n) + assert.Equal(t.T(), 0, res.Size) require.Error(t.T(), err) assert.ErrorIs(t.T(), err, io.EOF) } @@ -1799,9 +1887,9 @@ func (t *FileTest) TestReadEmptyGCSFileWhenStreamingWritesAreNotInProgress() { t.in.config = &cfg.Config{Write: *getWriteConfig()} data := make([]byte, 10) - n, err := t.in.Read(t.ctx, data, 0) + res, err := t.in.Read(t.ctx, &gcsx.ReadRequest{Buffer: data, Offset: 0}) - assert.Equal(t.T(), 0, n) + assert.Equal(t.T(), 0, res.Size) require.Error(t.T(), err) assert.Contains(t.T(), err.Error(), "EOF") } diff --git a/internal/fs/large_read_regional_test.go b/internal/fs/large_read_regional_test.go new file mode 100644 index 0000000000..6b9d478200 --- /dev/null +++ b/internal/fs/large_read_regional_test.go @@ -0,0 +1,134 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package fs_test + +import ( + "context" + "syscall" + "testing" + + "github.com/googlecloudplatform/gcsfuse/v3/cfg" + "github.com/googlecloudplatform/gcsfuse/v3/internal/fs" + "github.com/googlecloudplatform/gcsfuse/v3/internal/storage/fake" + "github.com/googlecloudplatform/gcsfuse/v3/internal/storage/gcs" + "github.com/googlecloudplatform/gcsfuse/v3/metrics" + "github.com/googlecloudplatform/gcsfuse/v3/tracing" + "github.com/jacobsa/fuse/fuseops" + "github.com/jacobsa/fuse/fuseutil" + "github.com/jacobsa/timeutil" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func setupLargeRead(ctx context.Context, t *testing.T, isZonal bool, enableKernelReader bool, fileContent string) (fuseutil.FileSystem, *fuseops.ReadFileOp) { + t.Helper() + bucketName := "regional-bucket" + if isZonal { + bucketName = "zonal-bucket" + } + bucket := fake.NewFakeBucket(timeutil.RealClock(), bucketName, gcs.BucketType{Zonal: isZonal, Hierarchical: false}) + + serverCfg := &fs.ServerConfig{ + NewConfig: &cfg.Config{ + Write: cfg.WriteConfig{ + GlobalMaxBlocks: 1, + }, + FileSystem: cfg.FileSystemConfig{ + EnableKernelReader: enableKernelReader, + }, + }, + CacheClock: &timeutil.SimulatedClock{}, + BucketName: bucketName, + BucketManager: &fakeBucketManager{ + buckets: map[string]gcs.Bucket{ + bucketName: bucket, + }, + }, + SequentialReadSizeMb: 200, + TraceHandle: tracing.NewOTELTracer(), + MetricHandle: metrics.NewNoopMetrics(), + } + + server, err := fs.NewFileSystem(ctx, serverCfg) + require.NoError(t, err) + + fileName := "large-file" + createWithContents(ctx, t, bucket, fileName, fileContent) + lookUpOp := &fuseops.LookUpInodeOp{ + Parent: fuseops.RootInodeID, + Name: fileName, + } + err = server.LookUpInode(ctx, lookUpOp) + require.NoError(t, err) + + openOp := &fuseops.OpenFileOp{ + Inode: lookUpOp.Entry.Child, + } + err = server.OpenFile(ctx, openOp) + require.NoError(t, err) + + readOp := &fuseops.ReadFileOp{ + Inode: lookUpOp.Entry.Child, + Handle: openOp.Handle, + Offset: 0, + Size: int64(len(fileContent)), + Dst: nil, + } + + return server, readOp +} + +func TestLargeReadRegional_Success(t *testing.T) { + // Arrange + ctx := context.Background() + fileContent := "regional bucket content for vectored read verification." + server, readOp := setupLargeRead(ctx, t, false, true, fileContent) + + // Act + err := server.ReadFile(ctx, readOp) + + // Assert + require.NoError(t, err) + assert.Equal(t, len(fileContent), readOp.BytesRead) + require.Len(t, readOp.Data, 1) + assert.Equal(t, []byte(fileContent), readOp.Data[0]) + if readOp.Callback != nil { + readOp.Callback() + } +} + +func TestLargeReadRegional_NotSupportedForZonal(t *testing.T) { + // Arrange + ctx := context.Background() + server, readOp := setupLargeRead(ctx, t, true, true, "zonal bucket content.") + + // Act + err := server.ReadFile(ctx, readOp) + + // Assert + assert.ErrorIs(t, err, syscall.ENOTSUP) +} + +func TestLargeReadRegional_NotSupportedWhenKernelReaderDisabled(t *testing.T) { + // Arrange + ctx := context.Background() + server, readOp := setupLargeRead(ctx, t, false, false, "regional bucket content.") + + // Act + err := server.ReadFile(ctx, readOp) + + // Assert + assert.ErrorIs(t, err, syscall.ENOTSUP) +} diff --git a/internal/gcsx/buffer_pool.go b/internal/gcsx/buffer_pool.go new file mode 100644 index 0000000000..c852e5e06a --- /dev/null +++ b/internal/gcsx/buffer_pool.go @@ -0,0 +1,62 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package gcsx + +import ( + "sync" + + "github.com/googlecloudplatform/gcsfuse/v3/internal/logger" + "github.com/googlecloudplatform/gcsfuse/v3/internal/util" +) + +// BufferPool defines an interface for on-demand buffer allocation and recycling. +type BufferPool interface { + Get() []byte + Put([]byte) +} + +// readPoolBufferSize is the size of each buffer in readBufferPool (1 MiB). +const readPoolBufferSize = util.MiB + +// Ensure FixedSizeBufferPool implements BufferPool at compile time. +var _ BufferPool = (*FixedSizeBufferPool)(nil) + +type FixedSizeBufferPool struct { + // Pool is the underlying sync.Pool storing fixed-size byte slices. + Pool *sync.Pool +} + +// NewFixedSizeBufferPool creates a new FixedSizeBufferPool with 1 MiB buffers. +func NewFixedSizeBufferPool() *FixedSizeBufferPool { + return &FixedSizeBufferPool{ + Pool: &sync.Pool{ + New: func() any { + return new([readPoolBufferSize]byte) + }, + }, + } +} + +func (bp *FixedSizeBufferPool) Get() []byte { + return bp.Pool.Get().(*[readPoolBufferSize]byte)[:] +} + +func (bp *FixedSizeBufferPool) Put(buf []byte) { + if cap(buf) != readPoolBufferSize { + logger.Errorf("FixedSizeBufferPool::Put Buffer capacity does not match readPoolBufferSize: %d vs %d", cap(buf), readPoolBufferSize) + return + } + bp.Pool.Put((*[readPoolBufferSize]byte)(buf[:cap(buf)])) +} diff --git a/internal/gcsx/buffer_pool_test.go b/internal/gcsx/buffer_pool_test.go new file mode 100644 index 0000000000..a4a1c2cc10 --- /dev/null +++ b/internal/gcsx/buffer_pool_test.go @@ -0,0 +1,65 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package gcsx + +import ( + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestFixedSizeBufferPool_Get(t *testing.T) { + // Arrange + bp := NewFixedSizeBufferPool() + + // Act + buf := bp.Get() + + // Assert + assert.Equal(t, readPoolBufferSize, len(buf)) + assert.Equal(t, readPoolBufferSize, cap(buf)) +} + +func TestFixedSizeBufferPool_Put(t *testing.T) { + testCases := []struct { + name string + buf []byte + }{ + { + name: "ValidBuffer", + buf: make([]byte, readPoolBufferSize), + }, + { + name: "CapacityLessThanPoolSize", + buf: make([]byte, 10), + }, + { + name: "CapacityGreaterThanPoolSize", + buf: make([]byte, readPoolBufferSize+10), + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + // Arrange + bp := NewFixedSizeBufferPool() + + // Act & Assert + assert.NotPanics(t, func() { + bp.Put(tc.buf) + }) + }) + } +} diff --git a/internal/gcsx/kernel_readers/kernel_range_reader.go b/internal/gcsx/kernel_readers/kernel_range_reader.go index 8c6a731e48..1bea4d5505 100644 --- a/internal/gcsx/kernel_readers/kernel_range_reader.go +++ b/internal/gcsx/kernel_readers/kernel_range_reader.go @@ -70,7 +70,7 @@ func (krr *KernelRangeReader) ReadAt(ctx context.Context, req *gcsx.ReadRequest) } // If the destination buffer is empty, there's nothing to read. - if len(req.Buffer) == 0 && len(req.Buffers) == 0 { + if len(req.Buffer) == 0 && req.BufferPool == nil { return resp, nil } @@ -111,18 +111,22 @@ func (krr *KernelRangeReader) ReadAt(ctx context.Context, req *gcsx.ReadRequest) }() var n int - if len(req.Buffers) == 0 { - n, err = io.ReadFull(reader, req.Buffer[:bytesToRead]) - } else { - writer := gcsx.NewVectoredWriter(req.Buffers) + if req.BufferPool != nil { + vBuf := gcsx.NewVectoredReadBuffer(req.BufferPool, bytesToRead) var written int64 - written, err = io.CopyN(writer, reader, bytesToRead) + written, err = io.CopyN(vBuf, reader, bytesToRead) n = int(written) - // Align error behavior with io.ReadFull: if EOF is encountered after - // reading some but not all of the requested bytes, return ErrUnexpectedEOF. if err == io.EOF && written > 0 { err = io.ErrUnexpectedEOF } + if n > 0 { + resp.Data = vBuf.Buffers() + resp.Callback = func() { vBuf.Release() } + } else { + vBuf.Release() // Release immediately on 0-byte read or early error + } + } else { + n, err = io.ReadFull(reader, req.Buffer[:bytesToRead]) } resp.Size = n diff --git a/internal/gcsx/kernel_readers/kernel_range_reader_test.go b/internal/gcsx/kernel_readers/kernel_range_reader_test.go index 010ef6874b..b443958ca5 100644 --- a/internal/gcsx/kernel_readers/kernel_range_reader_test.go +++ b/internal/gcsx/kernel_readers/kernel_range_reader_test.go @@ -28,6 +28,7 @@ import ( "github.com/googlecloudplatform/gcsfuse/v3/internal/storage/gcs" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" "github.com/stretchr/testify/suite" ) @@ -76,6 +77,7 @@ func (t *KernelRangeReaderTest) TestReadAt_Success() { req := &gcsx.ReadRequest{ Buffer: make([]byte, 5), Offset: 0, + Size: 5, } mockReader := &mockStorageReader{io.NopCloser(bytes.NewReader(data))} t.bucket.On("NewReaderWithReadHandle", mock.Anything, mock.Anything).Return(mockReader, nil).Once() @@ -92,6 +94,7 @@ func (t *KernelRangeReaderTest) TestReadAt_EOF() { req := &gcsx.ReadRequest{ Buffer: make([]byte, 5), Offset: 100, // Equal to object size + Size: 5, } resp, err := t.reader.ReadAt(context.Background(), req) @@ -106,6 +109,7 @@ func (t *KernelRangeReaderTest) TestReadAt_PartialRead() { req := &gcsx.ReadRequest{ Buffer: make([]byte, 10), Offset: 5, + Size: 10, } // Expected read range: [5, 10), size = 5 expectedData := data[5:10] @@ -124,6 +128,7 @@ func (t *KernelRangeReaderTest) TestReadAt_NewReaderError() { req := &gcsx.ReadRequest{ Buffer: make([]byte, 5), Offset: 0, + Size: 5, } expectedErr := io.ErrUnexpectedEOF t.bucket.On("NewReaderWithReadHandle", mock.Anything, mock.Anything).Return(nil, expectedErr).Once() @@ -140,6 +145,7 @@ func (t *KernelRangeReaderTest) TestReadAt_NilObject() { req := &gcsx.ReadRequest{ Buffer: make([]byte, 5), Offset: 0, + Size: 5, } resp, err := t.reader.ReadAt(context.Background(), req) @@ -152,6 +158,7 @@ func (t *KernelRangeReaderTest) TestReadAt_ClobberedError() { req := &gcsx.ReadRequest{ Buffer: make([]byte, 5), Offset: 0, + Size: 5, } gcsErr := &gcs.NotFoundError{Err: errors.New("not found")} t.bucket.On("NewReaderWithReadHandle", mock.Anything, mock.Anything).Return(nil, gcsErr).Once() @@ -166,15 +173,19 @@ func (t *KernelRangeReaderTest) TestReadAt_ClobberedError() { t.bucket.AssertExpectations(t.T()) } -func (t *KernelRangeReaderTest) TestReadAt_MultipleBuffers_Success() { +func (t *KernelRangeReaderTest) TestReadAt_BufferPool_Success() { data := []byte("abcdefghij") // length 10 - req := &gcsx.ReadRequest{ + pool := &gcsx.TestBufferPool{ Buffers: [][]byte{ make([]byte, 3), make([]byte, 2), make([]byte, 4), }, - Offset: 0, + } + req := &gcsx.ReadRequest{ + BufferPool: pool, + Offset: 0, + Size: 9, } mockReader := &mockStorageReader{io.NopCloser(bytes.NewReader(data))} t.bucket.On("NewReaderWithReadHandle", mock.Anything, mock.Anything).Return(mockReader, nil).Once() @@ -183,8 +194,9 @@ func (t *KernelRangeReaderTest) TestReadAt_MultipleBuffers_Success() { assert.NoError(t.T(), err) assert.Equal(t.T(), 9, resp.Size) // Sum of buffers capacity is 3+2+4 = 9 - assert.Equal(t.T(), "abc", string(req.Buffers[0])) - assert.Equal(t.T(), "de", string(req.Buffers[1])) - assert.Equal(t.T(), "fghi", string(req.Buffers[2])) + require.Len(t.T(), resp.Data, 3) + assert.Equal(t.T(), "abc", string(resp.Data[0])) + assert.Equal(t.T(), "de", string(resp.Data[1])) + assert.Equal(t.T(), "fghi", string(resp.Data[2])) t.bucket.AssertExpectations(t.T()) } diff --git a/internal/gcsx/reader.go b/internal/gcsx/reader.go index c1eddb2556..1bd79cf152 100644 --- a/internal/gcsx/reader.go +++ b/internal/gcsx/reader.go @@ -31,9 +31,9 @@ type ReadRequest struct { // Buffer is provided by jacobsa/fuse and should be filled with data from the object. Buffer []byte - // Buffers is a slice of slices of bytes that should be filled sequentially with data. - // If Buffers is non-empty, readers should populate Buffers instead of Buffer. - Buffers [][]byte + // BufferPool provides on-demand buffer allocation for vectored reads when Dst is nil. + // If BufferPool is non-nil, readers should allocate buffers on demand and set ReadResponse.Callback. + BufferPool BufferPool // Offset specifies the starting position in the object from where data should be read. // Note: This value should not be modified by any reader. It is used by the @@ -56,22 +56,12 @@ type ReadRequest struct { // GetReadSize calculates the size to read based on the request capacity and the provided limit. func (req *ReadRequest) GetReadSize(limit int64) int64 { - var totalCapacity int64 - if len(req.Buffers) == 0 { - totalCapacity = int64(len(req.Buffer)) - } else { - for _, b := range req.Buffers { - totalCapacity += int64(len(b)) - } - } - sizeToRead := req.Size - if sizeToRead <= 0 || sizeToRead > totalCapacity { - sizeToRead = totalCapacity + if req.BufferPool == nil { + sizeToRead = min(sizeToRead, int64(len(req.Buffer))) } - - if limit > 0 && sizeToRead > limit { - sizeToRead = limit + if limit > 0 { + sizeToRead = min(sizeToRead, limit) } return sizeToRead } diff --git a/internal/gcsx/test_buffer_pool.go b/internal/gcsx/test_buffer_pool.go new file mode 100644 index 0000000000..64117aa4bb --- /dev/null +++ b/internal/gcsx/test_buffer_pool.go @@ -0,0 +1,48 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package gcsx + +// TestBufferPool is a generic mock implementation of BufferPool used for testing across packages. +type TestBufferPool struct { + Buffers [][]byte + Idx int + PutBuffers [][]byte + ReturnNilOnExhaustion bool + DefaultBufferSize int +} + +// Get returns the next available buffer from Buffers. If all buffers have been consumed, +// it returns nil if ReturnNilOnExhaustion is true, or a newly allocated buffer of size +// DefaultBufferSize (or 1024 if DefaultBufferSize is <= 0). +func (p *TestBufferPool) Get() []byte { + if p.Idx < len(p.Buffers) { + b := p.Buffers[p.Idx] + p.Idx++ + return b + } + if p.ReturnNilOnExhaustion { + return nil + } + size := p.DefaultBufferSize + if size <= 0 { + size = 1024 + } + return make([]byte, size) +} + +// Put records the returned buffer in PutBuffers. +func (p *TestBufferPool) Put(b []byte) { + p.PutBuffers = append(p.PutBuffers, b) +} diff --git a/internal/gcsx/vectored_read_buffer.go b/internal/gcsx/vectored_read_buffer.go new file mode 100644 index 0000000000..8432caa0b8 --- /dev/null +++ b/internal/gcsx/vectored_read_buffer.go @@ -0,0 +1,167 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package gcsx + +import ( + "errors" + "io" +) + +var ErrInvalidAdvance = errors.New("VectoredReadBuffer: Invalid Buffer Advance") + +// VectoredReadBuffer implements io.Writer and writes data sequentially into a slice of byte buffers. +// Invariant: For each slice b in buffers, len(b) tracks the number of valid written bytes, +// and cap(b) tracks the total capacity allocated from the pool. We never use 3-index slicing +// to shrink cap(b) so that Release() can safely return full-capacity buffers back to the pool. +type VectoredReadBuffer struct { + buffers [][]byte + pool BufferPool + maxSize int64 + written int64 +} + +// NewVectoredReadBuffer creates a new VectoredReadBuffer that allocates buffers on demand from pool. +func NewVectoredReadBuffer(pool BufferPool, maxSize int64) *VectoredReadBuffer { + return &VectoredReadBuffer{ + buffers: make([][]byte, 0, 2), + pool: pool, + maxSize: maxSize, + } +} + +// availableBuffer returns a slice of the current buffer available for writing. +// If the current buffer is full or no buffers exist, it allocates a new buffer +// from the pool. Returns nil if no more buffers can be allocated (e.g. pool is nil +// or maxSize is reached). +func (v *VectoredReadBuffer) availableBuffer() []byte { + if v.written >= v.maxSize { + return nil + } + + avail := v.maxSize - v.written + + // 1. Check if the active (last) buffer has remaining capacity + if len(v.buffers) > 0 { + last := v.buffers[len(v.buffers)-1] + if cap(last) > len(last) { + rem := int(min(int64(cap(last)-len(last)), avail)) + return last[len(last) : len(last)+rem] + } + } + + // 2. Fetch new buffer from pool + if v.pool != nil { + if buf := v.pool.Get(); cap(buf) > 0 { + // Append with length 0, preserving full pool capacity for Release(). + v.buffers = append(v.buffers, buf[:0]) + + // Clamp the returned slice length (NOT capacity) to remaining maxSize. + return buf[:min(int64(cap(buf)), avail)] + } + } + return nil +} + +// advance increases the length of the active buffer by n bytes and updates total written bytes. +// It returns ErrInvalidAdvance if n is negative, exceeds available capacity, or no buffers exist. +func (v *VectoredReadBuffer) advance(n int) error { + if n == 0 { + return nil + } + if len(v.buffers) > 0 { + idx := len(v.buffers) - 1 + if n > 0 && n+len(v.buffers[idx]) <= cap(v.buffers[idx]) { + v.buffers[idx] = v.buffers[idx][:len(v.buffers[idx])+n] + v.written += int64(n) + return nil + } + } + return ErrInvalidAdvance +} + +func (v *VectoredReadBuffer) Write(p []byte) (n int, err error) { + for n < len(p) { + buf := v.availableBuffer() + if buf == nil { + return n, io.ErrShortWrite + } + toCopy := copy(buf, p[n:]) + if err := v.advance(toCopy); err != nil { + return n, err + } + n += toCopy + } + return n, nil +} + +func (v *VectoredReadBuffer) readIntoBuffers(readFn func(buf []byte) (int, error)) (n int64, err error) { + for { + buf := v.availableBuffer() + if buf == nil { + break + } + var readNum int + readNum, err = readFn(buf) + if readNum > 0 { + if advErr := v.advance(readNum); advErr != nil { + return n, advErr + } + n += int64(readNum) + } + if err != nil { + break + } + } + if err == io.EOF { + err = nil + } + return n, err +} + +// ReadFrom implements io.ReaderFrom. It reads data from r directly into the +// underlying buffers, avoiding intermediate allocations and double-copying. +func (v *VectoredReadBuffer) ReadFrom(r io.Reader) (n int64, err error) { + return v.readIntoBuffers(r.Read) +} + +// ReadFromAt reads data from r starting at offset directly into the +// underlying buffers, avoiding intermediate allocations and double-copying. +func (v *VectoredReadBuffer) ReadFromAt(r io.ReaderAt, offset int64) (n int64, err error) { + return v.readIntoBuffers(func(buf []byte) (int, error) { + return r.ReadAt(buf, offset+v.written) + }) +} + +// Buffers returns the slices of bytes actually written to. +// This executes in O(1) time with 0 allocations. +func (v *VectoredReadBuffer) Buffers() [][]byte { + // If the last buffer was allocated by availableBuffer() but nothing was written to it, + // exclude it from the returned slice without reallocating. + if len(v.buffers) > 0 && len(v.buffers[len(v.buffers)-1]) == 0 { + return v.buffers[:len(v.buffers)-1] + } + return v.buffers +} + +// Release puts all allocated buffers back into the pool. +func (v *VectoredReadBuffer) Release() { + if v.pool == nil { + return + } + for _, b := range v.buffers { + v.pool.Put(b[:cap(b)]) + } + v.buffers = nil +} diff --git a/internal/gcsx/vectored_read_buffer_test.go b/internal/gcsx/vectored_read_buffer_test.go new file mode 100644 index 0000000000..b45f9d46ae --- /dev/null +++ b/internal/gcsx/vectored_read_buffer_test.go @@ -0,0 +1,512 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package gcsx + +import ( + "bytes" + "errors" + "io" + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestVectoredReadBuffer_Write(t *testing.T) { + tests := []struct { + name string + buffers [][]byte + input []byte + expectedN int + expectedErr error + expectedBufs [][]byte + }{ + { + name: "Write to single buffer fully", + buffers: [][]byte{ + make([]byte, 5), + }, + input: []byte("hello"), + expectedN: 5, + expectedErr: nil, + expectedBufs: [][]byte{[]byte("hello")}, + }, + { + name: "Write to multiple buffers fully", + buffers: [][]byte{ + make([]byte, 3), + make([]byte, 2), + }, + input: []byte("hello"), + expectedN: 5, + expectedErr: nil, + expectedBufs: [][]byte{[]byte("hel"), []byte("lo")}, + }, + { + name: "Write with remaining space", + buffers: [][]byte{ + make([]byte, 4), + make([]byte, 4), + }, + input: []byte("hello"), + expectedN: 5, + expectedErr: nil, + expectedBufs: [][]byte{[]byte("hell"), []byte("o")}, + }, + { + name: "Write more than capacity", + buffers: [][]byte{ + make([]byte, 2), + make([]byte, 2), + }, + input: []byte("hello"), + expectedN: 4, + expectedErr: io.ErrShortWrite, + expectedBufs: [][]byte{[]byte("he"), []byte("ll")}, + }, + { + name: "Write to empty buffers list", + buffers: [][]byte{}, + input: []byte("hello"), + expectedN: 0, + expectedErr: io.ErrShortWrite, + expectedBufs: [][]byte{}, + }, + { + name: "Write empty data", + buffers: [][]byte{ + make([]byte, 5), + }, + input: []byte(""), + expectedN: 0, + expectedErr: nil, + expectedBufs: [][]byte{}, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + var maxSize int64 + for _, b := range tc.buffers { + maxSize += int64(len(b)) + } + var pool BufferPool + if len(tc.buffers) > 0 { + pool = &TestBufferPool{Buffers: tc.buffers} + } + w := NewVectoredReadBuffer(pool, maxSize) + + n, err := w.Write(tc.input) + + assert.Equal(t, tc.expectedN, n) + assert.Equal(t, tc.expectedErr, err) + assert.Equal(t, tc.expectedBufs, w.Buffers()) + }) + } +} + +type errorReader struct { + err error +} + +func (r *errorReader) Read(p []byte) (n int, err error) { + return 0, r.err +} + +func TestVectoredReadBuffer_ReadFrom(t *testing.T) { + testErr := errors.New("read error") + + tests := []struct { + name string + buffers [][]byte + reader io.Reader + expectedN int64 + expectedErr error + expectedBufs [][]byte + }{ + { + name: "ReadFrom single buffer fully", + buffers: [][]byte{ + make([]byte, 5), + }, + reader: bytes.NewReader([]byte("hello")), + expectedN: 5, + expectedErr: nil, + expectedBufs: [][]byte{[]byte("hello")}, + }, + { + name: "ReadFrom multiple buffers fully", + buffers: [][]byte{ + make([]byte, 3), + make([]byte, 2), + }, + reader: bytes.NewReader([]byte("hello")), + expectedN: 5, + expectedErr: nil, + expectedBufs: [][]byte{[]byte("hel"), []byte("lo")}, + }, + { + name: "ReadFrom with remaining space", + buffers: [][]byte{ + make([]byte, 4), + make([]byte, 4), + }, + reader: bytes.NewReader([]byte("hello")), + expectedN: 5, + expectedErr: nil, + expectedBufs: [][]byte{[]byte("hell"), []byte("o")}, + }, + { + name: "ReadFrom more than capacity", + buffers: [][]byte{ + make([]byte, 2), + make([]byte, 2), + }, + reader: bytes.NewReader([]byte("hello")), + expectedN: 4, + expectedErr: nil, + expectedBufs: [][]byte{[]byte("he"), []byte("ll")}, + }, + { + name: "ReadFrom to empty buffers list", + buffers: [][]byte{}, + reader: bytes.NewReader([]byte("hello")), + expectedN: 0, + expectedErr: nil, + expectedBufs: [][]byte{}, + }, + { + name: "ReadFrom empty reader", + buffers: [][]byte{ + make([]byte, 5), + }, + reader: bytes.NewReader([]byte("")), + expectedN: 0, + expectedErr: nil, + expectedBufs: [][]byte{}, + }, + { + name: "ReadFrom reader error", + buffers: [][]byte{ + make([]byte, 5), + }, + reader: &errorReader{err: testErr}, + expectedN: 0, + expectedErr: testErr, + expectedBufs: [][]byte{}, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + var maxSize int64 + for _, b := range tc.buffers { + maxSize += int64(len(b)) + } + var pool BufferPool + if len(tc.buffers) > 0 { + pool = &TestBufferPool{Buffers: tc.buffers} + } + w := NewVectoredReadBuffer(pool, maxSize) + + n, err := w.ReadFrom(tc.reader) + + assert.Equal(t, tc.expectedN, n) + assert.Equal(t, tc.expectedErr, err) + assert.Equal(t, tc.expectedBufs, w.Buffers()) + }) + } +} + +type mockReaderAt struct { + data []byte + err error +} + +func (m *mockReaderAt) ReadAt(dst []byte, offset int64) (n int, err error) { + if m.err != nil { + return 0, m.err + } + if offset >= int64(len(m.data)) { + return 0, io.EOF + } + n = copy(dst, m.data[offset:]) + if n < len(dst) { + err = io.EOF + } + return n, err +} + +func TestVectoredReadBuffer_ReadFromAt(t *testing.T) { + testErr := errors.New("reader at error") + + tests := []struct { + name string + buffers [][]byte + reader *mockReaderAt + offset int64 + maxSize int64 + expectedN int64 + expectedErr error + expectedBufs [][]byte + }{ + { + name: "ReadFromAt from zero offset fully", + buffers: [][]byte{ + make([]byte, 3), + make([]byte, 2), + }, + reader: &mockReaderAt{data: []byte("hello")}, + offset: 0, + maxSize: 5, + expectedN: 5, + expectedErr: nil, + expectedBufs: [][]byte{[]byte("hel"), []byte("lo")}, + }, + { + name: "ReadFromAt from non-zero offset across multiple buffers", + buffers: [][]byte{ + make([]byte, 3), + make([]byte, 3), + }, + reader: &mockReaderAt{data: []byte("0123456789")}, + offset: 3, + maxSize: 6, + expectedN: 6, + expectedErr: nil, + expectedBufs: [][]byte{[]byte("345"), []byte("678")}, + }, + { + name: "ReadFromAt with reader error", + buffers: [][]byte{ + make([]byte, 5), + }, + reader: &mockReaderAt{err: testErr}, + offset: 0, + maxSize: 5, + expectedN: 0, + expectedErr: testErr, + expectedBufs: [][]byte{}, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + // Arrange + pool := &TestBufferPool{Buffers: tc.buffers} + w := NewVectoredReadBuffer(pool, tc.maxSize) + + // Act + n, err := w.ReadFromAt(tc.reader, tc.offset) + + // Assert + assert.Equal(t, tc.expectedN, n) + assert.Equal(t, tc.expectedErr, err) + assert.Equal(t, tc.expectedBufs, w.Buffers()) + }) + } +} + +func TestVectoredReadBuffer_Release(t *testing.T) { + tests := []struct { + name string + pool *TestBufferPool + maxSize int64 + writeInput []byte + releaseCount int + expectedPutCount int + }{ + { + name: "release allocated buffers to pool and clear state", + pool: &TestBufferPool{ + Buffers: [][]byte{ + make([]byte, 10), + make([]byte, 10), + make([]byte, 10), + }, + }, + maxSize: 30, + writeInput: []byte("0123456789abcde"), // 15 bytes -> 2 buffers allocated + releaseCount: 1, + expectedPutCount: 2, + }, + { + name: "release with nil pool is safe", + pool: nil, + maxSize: 100, + writeInput: nil, + releaseCount: 1, + expectedPutCount: 0, + }, + { + name: "release with no allocated buffers is safe", + pool: &TestBufferPool{ + Buffers: [][]byte{ + make([]byte, 10), + }, + }, + maxSize: 10, + writeInput: nil, // 0 bytes -> 0 buffers allocated + releaseCount: 1, + expectedPutCount: 0, + }, + { + name: "release multiple times is safe and idempotent", + pool: &TestBufferPool{ + Buffers: [][]byte{ + make([]byte, 10), + }, + }, + maxSize: 10, + writeInput: []byte("hello"), // 5 bytes -> 1 buffer allocated + releaseCount: 2, + expectedPutCount: 1, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + // Arrange + var pool BufferPool + if tc.pool != nil { + pool = tc.pool + } + w := NewVectoredReadBuffer(pool, tc.maxSize) + if len(tc.writeInput) > 0 { + _, _ = w.Write(tc.writeInput) + } + + // Act + for i := 0; i < tc.releaseCount; i++ { + w.Release() + } + + // Assert + if tc.pool != nil { + assert.Equal(t, tc.expectedPutCount, len(tc.pool.PutBuffers)) + } + assert.Empty(t, w.Buffers()) + }) + } +} + +func TestVectoredReadBuffer_OversizedBufferTruncation(t *testing.T) { + // Arrange + pool := &TestBufferPool{ + Buffers: [][]byte{ + make([]byte, 100), + }, + } + w := NewVectoredReadBuffer(pool, 25) + + // Act + n, err := w.Write([]byte("this is a test string of length greater than 25 characters")) + + // Assert + assert.Equal(t, 25, n) + assert.Equal(t, io.ErrShortWrite, err) + assert.Equal(t, 1, len(w.Buffers())) + assert.Equal(t, 25, len(w.Buffers()[0])) +} + +func TestVectoredReadBuffer_PoolExhaustion(t *testing.T) { + tests := []struct { + name string + isReadFrom bool + expectedN int64 + expectedErr error + }{ + { + name: "Write with pool exhaustion returns short write", + isReadFrom: false, + expectedN: 0, + expectedErr: io.ErrShortWrite, + }, + { + name: "ReadFrom with pool exhaustion stops cleanly", + isReadFrom: true, + expectedN: 0, + expectedErr: nil, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + // Arrange + pool := &TestBufferPool{ + ReturnNilOnExhaustion: true, + } + w := NewVectoredReadBuffer(pool, 100) + + // Act + var n int64 + var err error + if tc.isReadFrom { + n, err = w.ReadFrom(bytes.NewReader([]byte("hello"))) + } else { + var nw int + nw, err = w.Write([]byte("hello")) + n = int64(nw) + } + + // Assert + assert.Equal(t, tc.expectedN, n) + assert.Equal(t, tc.expectedErr, err) + }) + } +} + +func TestVectoredReadBuffer_ZeroOrNegativeMaxSize(t *testing.T) { + tests := []struct { + name string + maxSize int64 + }{ + {name: "zero maxSize", maxSize: 0}, + {name: "negative maxSize", maxSize: -10}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + // Arrange + pool := &TestBufferPool{ + Buffers: [][]byte{make([]byte, 10)}, + } + w := NewVectoredReadBuffer(pool, tc.maxSize) + + // Act + n, err := w.Write([]byte("hello")) + + // Assert + assert.Equal(t, 0, n) + assert.Equal(t, io.ErrShortWrite, err) + assert.Empty(t, w.Buffers()) + }) + } +} + +var globalBuffer *VectoredReadBuffer + +func BenchmarkNewVectoredReadBuffer(b *testing.B) { + pool := &TestBufferPool{ + Buffers: [][]byte{ + make([]byte, 100), + make([]byte, 100), + make([]byte, 100), + }, + } + b.ResetTimer() + for i := 0; i < b.N; i++ { + globalBuffer = NewVectoredReadBuffer(pool, 300) + } +} diff --git a/internal/gcsx/vectored_writer.go b/internal/gcsx/vectored_writer.go deleted file mode 100644 index 923b1d2802..0000000000 --- a/internal/gcsx/vectored_writer.go +++ /dev/null @@ -1,81 +0,0 @@ -// Copyright 2026 Google LLC -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -package gcsx - -import "io" - -type bufferWithOffset struct { - buffer []byte - offset int -} - -// VectoredWriter implements io.Writer and writes data sequentially into a slice of bufferWithOffset. -type VectoredWriter struct { - buffers []bufferWithOffset - index int -} - -// NewVectoredWriter creates a new VectoredWriter that writes into the provided buffers. -func NewVectoredWriter(buffers [][]byte) *VectoredWriter { - w := &VectoredWriter{buffers: make([]bufferWithOffset, len(buffers))} - for i, b := range buffers { - w.buffers[i].buffer = b - } - return w -} - -func (w *VectoredWriter) Write(p []byte) (n int, err error) { - for n < len(p) { - if w.index >= len(w.buffers) { - return n, io.ErrShortWrite - } - sb := &w.buffers[w.index] - avail := len(sb.buffer) - sb.offset - if avail <= 0 { - w.index++ - continue - } - toCopy := min(len(p)-n, avail) - copy(sb.buffer[sb.offset:sb.offset+toCopy], p[n:n+toCopy]) - sb.offset += toCopy - n += toCopy - } - return n, nil -} - -// ReadFrom implements io.ReaderFrom. It reads data from r directly into the -// underlying buffers, avoiding intermediate allocations and double-copying. -func (w *VectoredWriter) ReadFrom(r io.Reader) (n int64, err error) { - for w.index < len(w.buffers) { - sb := &w.buffers[w.index] - if sb.offset >= len(sb.buffer) { - w.index++ - continue - } - - var readNum int - readNum, err = r.Read(sb.buffer[sb.offset:]) - sb.offset += readNum - n += int64(readNum) - - if err != nil { - if err == io.EOF { - err = nil - } - break - } - } - return n, err -} diff --git a/internal/gcsx/vectored_writer_test.go b/internal/gcsx/vectored_writer_test.go deleted file mode 100644 index 0ce338db8b..0000000000 --- a/internal/gcsx/vectored_writer_test.go +++ /dev/null @@ -1,228 +0,0 @@ -// Copyright 2026 Google LLC -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -package gcsx - -import ( - "bytes" - "errors" - "io" - "testing" - - "github.com/stretchr/testify/assert" -) - -func TestVectoredWriter_Write(t *testing.T) { - tests := []struct { - name string - buffers [][]byte - input []byte - expectedN int - expectedErr error - expectedBufs [][]byte - }{ - { - name: "Write to single buffer fully", - buffers: [][]byte{ - make([]byte, 5), - }, - input: []byte("hello"), - expectedN: 5, - expectedErr: nil, - expectedBufs: [][]byte{[]byte("hello")}, - }, - { - name: "Write to multiple buffers fully", - buffers: [][]byte{ - make([]byte, 3), - make([]byte, 2), - }, - input: []byte("hello"), - expectedN: 5, - expectedErr: nil, - expectedBufs: [][]byte{[]byte("hel"), []byte("lo")}, - }, - { - name: "Write with remaining space", - buffers: [][]byte{ - make([]byte, 4), - make([]byte, 4), - }, - input: []byte("hello"), - expectedN: 5, - expectedErr: nil, - expectedBufs: [][]byte{[]byte("hell"), []byte("o\x00\x00\x00")}, - }, - { - name: "Write more than capacity", - buffers: [][]byte{ - make([]byte, 2), - make([]byte, 2), - }, - input: []byte("hello"), - expectedN: 4, - expectedErr: io.ErrShortWrite, - expectedBufs: [][]byte{[]byte("he"), []byte("ll")}, - }, - { - name: "Write to empty buffers list", - buffers: [][]byte{}, - input: []byte("hello"), - expectedN: 0, - expectedErr: io.ErrShortWrite, - expectedBufs: [][]byte{}, - }, - { - name: "Write empty data", - buffers: [][]byte{ - make([]byte, 5), - }, - input: []byte(""), - expectedN: 0, - expectedErr: nil, - expectedBufs: [][]byte{make([]byte, 5)}, - }, - } - - for _, tc := range tests { - t.Run(tc.name, func(t *testing.T) { - w := NewVectoredWriter(tc.buffers) - - n, err := w.Write(tc.input) - - assert.Equal(t, tc.expectedN, n) - assert.Equal(t, tc.expectedErr, err) - assert.Equal(t, tc.expectedBufs, tc.buffers) - }) - } -} - -type errorReader struct { - err error -} - -func (r *errorReader) Read(p []byte) (n int, err error) { - return 0, r.err -} - -func TestVectoredWriter_ReadFrom(t *testing.T) { - testErr := errors.New("read error") - - tests := []struct { - name string - buffers [][]byte - reader io.Reader - expectedN int64 - expectedErr error - expectedBufs [][]byte - }{ - { - name: "ReadFrom single buffer fully", - buffers: [][]byte{ - make([]byte, 5), - }, - reader: bytes.NewReader([]byte("hello")), - expectedN: 5, - expectedErr: nil, - expectedBufs: [][]byte{[]byte("hello")}, - }, - { - name: "ReadFrom multiple buffers fully", - buffers: [][]byte{ - make([]byte, 3), - make([]byte, 2), - }, - reader: bytes.NewReader([]byte("hello")), - expectedN: 5, - expectedErr: nil, - expectedBufs: [][]byte{[]byte("hel"), []byte("lo")}, - }, - { - name: "ReadFrom with remaining space", - buffers: [][]byte{ - make([]byte, 4), - make([]byte, 4), - }, - reader: bytes.NewReader([]byte("hello")), - expectedN: 5, - expectedErr: nil, - expectedBufs: [][]byte{[]byte("hell"), []byte("o\x00\x00\x00")}, - }, - { - name: "ReadFrom more than capacity", - buffers: [][]byte{ - make([]byte, 2), - make([]byte, 2), - }, - reader: bytes.NewReader([]byte("hello")), - expectedN: 4, - expectedErr: nil, - expectedBufs: [][]byte{[]byte("he"), []byte("ll")}, - }, - { - name: "ReadFrom to empty buffers list", - buffers: [][]byte{}, - reader: bytes.NewReader([]byte("hello")), - expectedN: 0, - expectedErr: nil, - expectedBufs: [][]byte{}, - }, - { - name: "ReadFrom empty reader", - buffers: [][]byte{ - make([]byte, 5), - }, - reader: bytes.NewReader([]byte("")), - expectedN: 0, - expectedErr: nil, - expectedBufs: [][]byte{make([]byte, 5)}, - }, - { - name: "ReadFrom reader error", - buffers: [][]byte{ - make([]byte, 5), - }, - reader: &errorReader{err: testErr}, - expectedN: 0, - expectedErr: testErr, - expectedBufs: [][]byte{make([]byte, 5)}, - }, - } - - for _, tc := range tests { - t.Run(tc.name, func(t *testing.T) { - w := NewVectoredWriter(tc.buffers) - - n, err := w.ReadFrom(tc.reader) - - assert.Equal(t, tc.expectedN, n) - assert.Equal(t, tc.expectedErr, err) - assert.Equal(t, tc.expectedBufs, tc.buffers) - }) - } -} - -var globalWriter *VectoredWriter - -func BenchmarkNewVectoredWriter(b *testing.B) { - buffers := [][]byte{ - make([]byte, 100), - make([]byte, 100), - make([]byte, 100), - } - b.ResetTimer() - for i := 0; i < b.N; i++ { - globalWriter = NewVectoredWriter(buffers) - } -}