From 963ffcfb3e5f916b86b61a3d3534ba4a2c2ff6aa Mon Sep 17 00:00:00 2001 From: Abhishek Gupta Date: Tue, 7 Jul 2026 20:38:18 +0000 Subject: [PATCH 01/11] Add BufferPool interface and update ReadRequest --- internal/gcsx/reader.go | 20 ++++++++++++-------- 1 file changed, 12 insertions(+), 8 deletions(-) diff --git a/internal/gcsx/reader.go b/internal/gcsx/reader.go index c1eddb2556..b2144fc04b 100644 --- a/internal/gcsx/reader.go +++ b/internal/gcsx/reader.go @@ -26,14 +26,20 @@ import ( // to an alternative reader. var FallbackToAnotherReader = errors.New("fallback to another reader is required") +// BufferPool defines an interface for on-demand buffer allocation and recycling. +type BufferPool interface { + Get() []byte + Put([]byte) +} + // ReadRequest encapsulates the parameters for a read operation. 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 @@ -57,12 +63,10 @@ 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)) + if req.BufferPool != nil { + totalCapacity = req.Size } else { - for _, b := range req.Buffers { - totalCapacity += int64(len(b)) - } + totalCapacity = int64(len(req.Buffer)) } sizeToRead := req.Size From dfb8a356b41666e3930af61de17c7a094900fdb2 Mon Sep 17 00:00:00 2001 From: Abhishek Gupta Date: Tue, 7 Jul 2026 20:38:19 +0000 Subject: [PATCH 02/11] Implement lazy vectored writer and update kernel range reader --- .../kernel_readers/kernel_range_reader.go | 18 +- .../kernel_range_reader_test.go | 36 +- internal/gcsx/vectored_writer.go | 132 +++++-- internal/gcsx/vectored_writer_test.go | 341 +++++++++++++++++- 4 files changed, 469 insertions(+), 58 deletions(-) diff --git a/internal/gcsx/kernel_readers/kernel_range_reader.go b/internal/gcsx/kernel_readers/kernel_range_reader.go index 8c6a731e48..d73502066a 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 { + writer := gcsx.NewVectoredWriter(req.BufferPool, bytesToRead) var written int64 written, err = io.CopyN(writer, 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 = writer.Buffers() + resp.Callback = func() { writer.Release() } + } else { + writer.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..524ea7bd71 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" ) @@ -166,15 +167,35 @@ func (t *KernelRangeReaderTest) TestReadAt_ClobberedError() { t.bucket.AssertExpectations(t.T()) } -func (t *KernelRangeReaderTest) TestReadAt_MultipleBuffers_Success() { +type testBufferPool struct { + buffers [][]byte + idx int +} + +func (p *testBufferPool) Get() []byte { + if p.idx < len(p.buffers) { + b := p.buffers[p.idx] + p.idx++ + return b + } + return make([]byte, 1024) +} + +func (p *testBufferPool) Put(b []byte) {} + +func (t *KernelRangeReaderTest) TestReadAt_BufferPool_Success() { data := []byte("abcdefghij") // length 10 - req := &gcsx.ReadRequest{ - Buffers: [][]byte{ + pool := &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 +204,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/vectored_writer.go b/internal/gcsx/vectored_writer.go index 923b1d2802..32dcdfbad4 100644 --- a/internal/gcsx/vectored_writer.go +++ b/internal/gcsx/vectored_writer.go @@ -14,7 +14,15 @@ package gcsx -import "io" +import ( + "context" + "io" +) + +// OffsetReader is an interface for reading data at an offset with a context. +type OffsetReader interface { + Read(ctx context.Context, dst []byte, offset int64) (n int, err error) +} type bufferWithOffset struct { buffer []byte @@ -25,51 +33,79 @@ type bufferWithOffset struct { type VectoredWriter struct { buffers []bufferWithOffset index int + pool BufferPool + maxSize int64 + written int64 } -// 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 +// NewVectoredWriter creates a new VectoredWriter that allocates buffers on demand from pool. +func NewVectoredWriter(pool BufferPool, maxSize int64) *VectoredWriter { + return &VectoredWriter{ + buffers: make([]bufferWithOffset, 0, 2), + pool: pool, + maxSize: maxSize, } - return w +} + +// 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 (w *VectoredWriter) availableBuffer() []byte { + for { + if w.index < len(w.buffers) { + sb := &w.buffers[w.index] + if sb.offset < len(sb.buffer) { + return sb.buffer[sb.offset:] + } + w.index++ + continue + } + if w.pool == nil || w.written >= w.maxSize { + return nil + } + buf := w.pool.Get() + if len(buf) == 0 { + return nil + } + if int64(len(buf)) > w.maxSize-w.written { + buf = buf[:w.maxSize-w.written] + } + w.buffers = append(w.buffers, bufferWithOffset{buffer: buf}) + } +} + +// advance updates the offset of the current buffer and the total bytes written. +func (w *VectoredWriter) advance(n int) { + w.buffers[w.index].offset += n + w.written += int64(n) } func (w *VectoredWriter) Write(p []byte) (n int, err error) { for n < len(p) { - if w.index >= len(w.buffers) { + buf := w.availableBuffer() + if buf == nil { 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 + toCopy := copy(buf, p[n:]) + w.advance(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 +func (w *VectoredWriter) readIntoBuffers(readFn func(buf []byte) (int, error)) (n int64, err error) { + for { + buf := w.availableBuffer() + if buf == nil { + break } - var readNum int - readNum, err = r.Read(sb.buffer[sb.offset:]) - sb.offset += readNum - n += int64(readNum) - + readNum, err = readFn(buf) + if readNum > 0 { + w.advance(readNum) + n += int64(readNum) + } if err != nil { if err == io.EOF { err = nil @@ -79,3 +115,39 @@ func (w *VectoredWriter) ReadFrom(r io.Reader) (n int64, err error) { } return n, err } + +// 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) { + return w.readIntoBuffers(r.Read) +} + +// ReadFromOffset reads data from r starting at offset directly into the +// underlying buffers, avoiding intermediate allocations and double-copying. +func (w *VectoredWriter) ReadFromOffset(ctx context.Context, r OffsetReader, offset int64) (n int64, err error) { + return w.readIntoBuffers(func(buf []byte) (int, error) { + return r.Read(ctx, buf, offset+w.written) + }) +} + +// Buffers returns the slices of bytes actually written to. +func (w *VectoredWriter) Buffers() [][]byte { + res := make([][]byte, 0, len(w.buffers)) + for _, sb := range w.buffers { + if sb.offset > 0 { + res = append(res, sb.buffer[:sb.offset]) + } + } + return res +} + +// Release puts all allocated buffers back into the pool. +func (w *VectoredWriter) Release() { + if w.pool == nil { + return + } + for _, sb := range w.buffers { + w.pool.Put(sb.buffer) + } + w.buffers = nil +} diff --git a/internal/gcsx/vectored_writer_test.go b/internal/gcsx/vectored_writer_test.go index 0ce338db8b..52ee244c54 100644 --- a/internal/gcsx/vectored_writer_test.go +++ b/internal/gcsx/vectored_writer_test.go @@ -16,6 +16,7 @@ package gcsx import ( "bytes" + "context" "errors" "io" "testing" @@ -23,6 +24,29 @@ import ( "github.com/stretchr/testify/assert" ) +type testBufferPool struct { + buffers [][]byte + idx int + putBuffers [][]byte + returnNilOnExhaustion bool +} + +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 + } + return make([]byte, 1024) +} + +func (p *testBufferPool) Put(b []byte) { + p.putBuffers = append(p.putBuffers, b) +} + func TestVectoredWriter_Write(t *testing.T) { tests := []struct { name string @@ -62,7 +86,7 @@ func TestVectoredWriter_Write(t *testing.T) { input: []byte("hello"), expectedN: 5, expectedErr: nil, - expectedBufs: [][]byte{[]byte("hell"), []byte("o\x00\x00\x00")}, + expectedBufs: [][]byte{[]byte("hell"), []byte("o")}, }, { name: "Write more than capacity", @@ -91,19 +115,27 @@ func TestVectoredWriter_Write(t *testing.T) { input: []byte(""), expectedN: 0, expectedErr: nil, - expectedBufs: [][]byte{make([]byte, 5)}, + expectedBufs: [][]byte{}, }, } for _, tc := range tests { t.Run(tc.name, func(t *testing.T) { - w := NewVectoredWriter(tc.buffers) + 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 := NewVectoredWriter(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, tc.buffers) + assert.Equal(t, tc.expectedBufs, w.Buffers()) }) } } @@ -157,7 +189,7 @@ func TestVectoredWriter_ReadFrom(t *testing.T) { reader: bytes.NewReader([]byte("hello")), expectedN: 5, expectedErr: nil, - expectedBufs: [][]byte{[]byte("hell"), []byte("o\x00\x00\x00")}, + expectedBufs: [][]byte{[]byte("hell"), []byte("o")}, }, { name: "ReadFrom more than capacity", @@ -186,7 +218,7 @@ func TestVectoredWriter_ReadFrom(t *testing.T) { reader: bytes.NewReader([]byte("")), expectedN: 0, expectedErr: nil, - expectedBufs: [][]byte{make([]byte, 5)}, + expectedBufs: [][]byte{}, }, { name: "ReadFrom reader error", @@ -196,19 +228,298 @@ func TestVectoredWriter_ReadFrom(t *testing.T) { reader: &errorReader{err: testErr}, expectedN: 0, expectedErr: testErr, - expectedBufs: [][]byte{make([]byte, 5)}, + expectedBufs: [][]byte{}, }, } for _, tc := range tests { t.Run(tc.name, func(t *testing.T) { - w := NewVectoredWriter(tc.buffers) + 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 := NewVectoredWriter(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, tc.buffers) + assert.Equal(t, tc.expectedBufs, w.Buffers()) + }) + } +} + +type mockOffsetReader struct { + data []byte + err error +} + +func (m *mockOffsetReader) Read(ctx context.Context, 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 offset+int64(n) >= int64(len(m.data)) { + err = io.EOF + } + return n, err +} + +func TestVectoredWriter_ReadFromOffset(t *testing.T) { + testErr := errors.New("offset read error") + + tests := []struct { + name string + buffers [][]byte + reader *mockOffsetReader + offset int64 + maxSize int64 + expectedN int64 + expectedErr error + expectedBufs [][]byte + }{ + { + name: "ReadFromOffset from zero offset fully", + buffers: [][]byte{ + make([]byte, 3), + make([]byte, 2), + }, + reader: &mockOffsetReader{data: []byte("hello")}, + offset: 0, + maxSize: 5, + expectedN: 5, + expectedErr: nil, + expectedBufs: [][]byte{[]byte("hel"), []byte("lo")}, + }, + { + name: "ReadFromOffset from non-zero offset across multiple buffers", + buffers: [][]byte{ + make([]byte, 3), + make([]byte, 3), + }, + reader: &mockOffsetReader{data: []byte("0123456789")}, + offset: 3, + maxSize: 6, + expectedN: 6, + expectedErr: nil, + expectedBufs: [][]byte{[]byte("345"), []byte("678")}, + }, + { + name: "ReadFromOffset with reader error", + buffers: [][]byte{ + make([]byte, 5), + }, + reader: &mockOffsetReader{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 := NewVectoredWriter(pool, tc.maxSize) + + // Act + n, err := w.ReadFromOffset(context.Background(), 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 TestVectoredWriter_Release(t *testing.T) { + tests := []struct { + name string + pool *testBufferPool + maxSize int64 + writeInput []byte + releaseCount int + expectedPutCount int + expectedBufs [][]byte + }{ + { + 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, + expectedBufs: [][]byte{}, + }, + { + name: "release with nil pool is safe", + pool: nil, + maxSize: 100, + writeInput: nil, + releaseCount: 1, + expectedPutCount: 0, + expectedBufs: [][]byte{}, + }, + { + 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, + expectedBufs: [][]byte{}, + }, + { + 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, + expectedBufs: [][]byte{}, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + // Arrange + var pool BufferPool + if tc.pool != nil { + pool = tc.pool + } + w := NewVectoredWriter(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.Equal(t, tc.expectedBufs, w.Buffers()) + }) + } +} + +func TestVectoredWriter_OversizedBufferTruncation(t *testing.T) { + // Arrange + pool := &testBufferPool{ + buffers: [][]byte{ + make([]byte, 100), + }, + } + w := NewVectoredWriter(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 TestVectoredWriter_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 := NewVectoredWriter(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 TestVectoredWriter_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 := NewVectoredWriter(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()) }) } } @@ -216,13 +527,15 @@ func TestVectoredWriter_ReadFrom(t *testing.T) { var globalWriter *VectoredWriter func BenchmarkNewVectoredWriter(b *testing.B) { - buffers := [][]byte{ - make([]byte, 100), - make([]byte, 100), - make([]byte, 100), + pool := &testBufferPool{ + buffers: [][]byte{ + make([]byte, 100), + make([]byte, 100), + make([]byte, 100), + }, } b.ResetTimer() for i := 0; i < b.N; i++ { - globalWriter = NewVectoredWriter(buffers) + globalWriter = NewVectoredWriter(pool, 300) } } From d31b999edba06734004c2636687fe4ac74260bc0 Mon Sep 17 00:00:00 2001 From: Abhishek Gupta Date: Tue, 7 Jul 2026 20:38:19 +0000 Subject: [PATCH 03/11] Integrate read buffer pool in filesystem and add regional read tests --- internal/fs/fs.go | 56 ++++++++-- internal/fs/handle/file.go | 17 +++ internal/fs/large_read_regional_test.go | 134 ++++++++++++++++++++++++ 3 files changed, 197 insertions(+), 10 deletions(-) create mode 100644 internal/fs/large_read_regional_test.go diff --git a/internal/fs/fs.go b/internal/fs/fs.go index 258c5cbfd4..a4fcc74093 100644 --- a/internal/fs/fs.go +++ b/internal/fs/fs.go @@ -27,6 +27,7 @@ import ( "reflect" "slices" "strings" + "sync" "syscall" "time" @@ -61,6 +62,9 @@ import ( "github.com/spf13/viper" ) +// readPoolBufferSize is the size of each buffer in readBufferPool (1 MiB). +const readPoolBufferSize = util.MiB + type ServerConfig struct { // A clock used for cache expiration. It is *not* used for inode times, for // which we use the wall clock. @@ -225,6 +229,11 @@ 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: sync.Pool{ + New: func() any { + return new([readPoolBufferSize]byte) + }, + }, } // Initialize MRD cache if enabled @@ -676,6 +685,20 @@ type fileSystem struct { // mrdCache manages the cache of inactive MultiRangeDownloaders. mrdCache *lru.Cache + + // readBufferPool is a sync.Pool of readPoolBufferSize (1 MiB) buffers. + readBufferPool sync.Pool +} + +type bufferPoolWrapper struct { + pool *sync.Pool +} + +func (bp *bufferPoolWrapper) Get() []byte { + return bp.pool.Get().(*[readPoolBufferSize]byte)[:] +} +func (bp *bufferPoolWrapper) Put(buf []byte) { + bp.pool.Put((*[readPoolBufferSize]byte)(buf[:cap(buf)])) } //////////////////////////////////////////////////////////////////////// @@ -3118,23 +3141,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 = &bufferPoolWrapper{pool: &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..c33cf25277 100644 --- a/internal/fs/handle/file.go +++ b/internal/fs/handle/file.go @@ -289,6 +289,23 @@ 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() + if req.BufferPool != nil { + writer := gcsx.NewVectoredWriter(req.BufferPool, req.Size) + written, err := writer.ReadFromOffset(ctx, fh.inode, req.Offset) + if err == io.EOF && written > 0 { + err = nil + } + if written > 0 { + return gcsx.ReadResponse{ + Size: int(written), + Data: writer.Buffers(), + Callback: func() { writer.Release() }, + }, err + } + writer.Release() + return gcsx.ReadResponse{Size: int(written)}, err + } + n, err := fh.inode.Read(ctx, req.Buffer, req.Offset) return gcsx.ReadResponse{Size: n}, err } 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) +} From 22011eb887a072c6e4dded2cacca81c15f0edaea Mon Sep 17 00:00:00 2001 From: Abhishek Gupta Date: Wed, 8 Jul 2026 06:27:50 +0000 Subject: [PATCH 04/11] Refactoring code --- internal/fs/fs.go | 4 + internal/fs/fs_internal_test.go | 53 +++++++++++ internal/fs/handle/file.go | 27 +----- internal/fs/inode/file.go | 37 +++++--- internal/fs/inode/file_test.go | 127 +++++++++++++++++++++++--- internal/gcsx/vectored_writer.go | 21 ++--- internal/gcsx/vectored_writer_test.go | 27 +++--- 7 files changed, 223 insertions(+), 73 deletions(-) diff --git a/internal/fs/fs.go b/internal/fs/fs.go index a4fcc74093..cabcf7f178 100644 --- a/internal/fs/fs.go +++ b/internal/fs/fs.go @@ -697,7 +697,11 @@ type bufferPoolWrapper struct { func (bp *bufferPoolWrapper) Get() []byte { return bp.pool.Get().(*[readPoolBufferSize]byte)[:] } + func (bp *bufferPoolWrapper) Put(buf []byte) { + if cap(buf) != readPoolBufferSize { + return + } bp.pool.Put((*[readPoolBufferSize]byte)(buf[:cap(buf)])) } diff --git a/internal/fs/fs_internal_test.go b/internal/fs/fs_internal_test.go index 74ec63dc56..5d464b748f 100644 --- a/internal/fs/fs_internal_test.go +++ b/internal/fs/fs_internal_test.go @@ -15,6 +15,7 @@ package fs import ( + "sync" "testing" "github.com/googlecloudplatform/gcsfuse/v3/cfg" @@ -67,3 +68,55 @@ func TestCacheDirVolumeBlockSize(t *testing.T) { }) } } + +func TestBufferPoolWrapper_Get(t *testing.T) { + // Arrange + var pool sync.Pool + pool.New = func() any { + return new([readPoolBufferSize]byte) + } + bp := &bufferPoolWrapper{pool: &pool} + + // Act + buf := bp.Get() + + // Assert + assert.Equal(t, readPoolBufferSize, len(buf)) + assert.Equal(t, readPoolBufferSize, cap(buf)) +} + +func TestBufferPoolWrapper_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 + var pool sync.Pool + pool.New = func() any { + return new([readPoolBufferSize]byte) + } + bp := &bufferPoolWrapper{pool: &pool} + + // Act & Assert + assert.NotPanics(t, func() { + bp.Put(tc.buf) + }) + }) + } +} diff --git a/internal/fs/handle/file.go b/internal/fs/handle/file.go index c33cf25277..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,25 +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() - if req.BufferPool != nil { - writer := gcsx.NewVectoredWriter(req.BufferPool, req.Size) - written, err := writer.ReadFromOffset(ctx, fh.inode, req.Offset) - if err == io.EOF && written > 0 { - err = nil - } - if written > 0 { - return gcsx.ReadResponse{ - Size: int(written), - Data: writer.Buffers(), - Callback: func() { writer.Release() }, - }, err - } - writer.Release() - return gcsx.ReadResponse{Size: int(written)}, err - } - - 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() @@ -339,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/inode/file.go b/internal/fs/inode/file.go index 819cc26144..c02a6b36e1 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 { + writer := gcsx.NewVectoredWriter(req.BufferPool, req.Size) + var written int64 + written, err = writer.ReadFromAt(f.content, req.Offset) + n = int(written) + if n > 0 { + data = writer.Buffers() + callback = func() { writer.Release() } + } else { + writer.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..7bfa47aded 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,109 @@ func (t *FileTest) TestRead() { } } +type inodeTestBufferPool struct { + buffers [][]byte + idx int +} + +func (p *inodeTestBufferPool) Get() []byte { + if p.idx < len(p.buffers) { + b := p.buffers[p.idx] + p.idx++ + return b + } + return make([]byte, 2) +} + +func (p *inodeTestBufferPool) Put(b []byte) {} + +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: io.EOF, + }, + { + name: "AtEOF", + offset: 4, + size: 2, + expectedData: []string{}, + expectedSize: 0, + expectedErr: io.EOF, + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func() { + // Arrange + pool := &inodeTestBufferPool{ + buffers: [][]byte{make([]byte, 2), make([]byte, 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 +639,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 +674,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 +1529,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 +1883,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 +1902,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/gcsx/vectored_writer.go b/internal/gcsx/vectored_writer.go index 32dcdfbad4..252438943c 100644 --- a/internal/gcsx/vectored_writer.go +++ b/internal/gcsx/vectored_writer.go @@ -15,15 +15,9 @@ package gcsx import ( - "context" "io" ) -// OffsetReader is an interface for reading data at an offset with a context. -type OffsetReader interface { - Read(ctx context.Context, dst []byte, offset int64) (n int, err error) -} - type bufferWithOffset struct { buffer []byte offset int @@ -107,9 +101,6 @@ func (w *VectoredWriter) readIntoBuffers(readFn func(buf []byte) (int, error)) ( n += int64(readNum) } if err != nil { - if err == io.EOF { - err = nil - } break } } @@ -119,14 +110,18 @@ func (w *VectoredWriter) readIntoBuffers(readFn func(buf []byte) (int, error)) ( // 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) { - return w.readIntoBuffers(r.Read) + n, err = w.readIntoBuffers(r.Read) + if err == io.EOF { + err = nil + } + return n, err } -// ReadFromOffset reads data from r starting at offset directly into the +// ReadFromAt reads data from r starting at offset directly into the // underlying buffers, avoiding intermediate allocations and double-copying. -func (w *VectoredWriter) ReadFromOffset(ctx context.Context, r OffsetReader, offset int64) (n int64, err error) { +func (w *VectoredWriter) ReadFromAt(r io.ReaderAt, offset int64) (n int64, err error) { return w.readIntoBuffers(func(buf []byte) (int, error) { - return r.Read(ctx, buf, offset+w.written) + return r.ReadAt(buf, offset+w.written) }) } diff --git a/internal/gcsx/vectored_writer_test.go b/internal/gcsx/vectored_writer_test.go index 52ee244c54..85e48c5269 100644 --- a/internal/gcsx/vectored_writer_test.go +++ b/internal/gcsx/vectored_writer_test.go @@ -16,7 +16,6 @@ package gcsx import ( "bytes" - "context" "errors" "io" "testing" @@ -253,12 +252,12 @@ func TestVectoredWriter_ReadFrom(t *testing.T) { } } -type mockOffsetReader struct { +type mockReaderAt struct { data []byte err error } -func (m *mockOffsetReader) Read(ctx context.Context, dst []byte, offset int64) (n int, err error) { +func (m *mockReaderAt) ReadAt(dst []byte, offset int64) (n int, err error) { if m.err != nil { return 0, m.err } @@ -266,19 +265,19 @@ func (m *mockOffsetReader) Read(ctx context.Context, dst []byte, offset int64) ( return 0, io.EOF } n = copy(dst, m.data[offset:]) - if offset+int64(n) >= int64(len(m.data)) { + if n < len(dst) { err = io.EOF } return n, err } -func TestVectoredWriter_ReadFromOffset(t *testing.T) { - testErr := errors.New("offset read error") +func TestVectoredWriter_ReadFromAt(t *testing.T) { + testErr := errors.New("reader at error") tests := []struct { name string buffers [][]byte - reader *mockOffsetReader + reader *mockReaderAt offset int64 maxSize int64 expectedN int64 @@ -286,12 +285,12 @@ func TestVectoredWriter_ReadFromOffset(t *testing.T) { expectedBufs [][]byte }{ { - name: "ReadFromOffset from zero offset fully", + name: "ReadFromAt from zero offset fully", buffers: [][]byte{ make([]byte, 3), make([]byte, 2), }, - reader: &mockOffsetReader{data: []byte("hello")}, + reader: &mockReaderAt{data: []byte("hello")}, offset: 0, maxSize: 5, expectedN: 5, @@ -299,12 +298,12 @@ func TestVectoredWriter_ReadFromOffset(t *testing.T) { expectedBufs: [][]byte{[]byte("hel"), []byte("lo")}, }, { - name: "ReadFromOffset from non-zero offset across multiple buffers", + name: "ReadFromAt from non-zero offset across multiple buffers", buffers: [][]byte{ make([]byte, 3), make([]byte, 3), }, - reader: &mockOffsetReader{data: []byte("0123456789")}, + reader: &mockReaderAt{data: []byte("0123456789")}, offset: 3, maxSize: 6, expectedN: 6, @@ -312,11 +311,11 @@ func TestVectoredWriter_ReadFromOffset(t *testing.T) { expectedBufs: [][]byte{[]byte("345"), []byte("678")}, }, { - name: "ReadFromOffset with reader error", + name: "ReadFromAt with reader error", buffers: [][]byte{ make([]byte, 5), }, - reader: &mockOffsetReader{err: testErr}, + reader: &mockReaderAt{err: testErr}, offset: 0, maxSize: 5, expectedN: 0, @@ -332,7 +331,7 @@ func TestVectoredWriter_ReadFromOffset(t *testing.T) { w := NewVectoredWriter(pool, tc.maxSize) // Act - n, err := w.ReadFromOffset(context.Background(), tc.reader, tc.offset) + n, err := w.ReadFromAt(tc.reader, tc.offset) // Assert assert.Equal(t, tc.expectedN, n) From ec889b699f4e0a48db4a3872fcc1034325070e22 Mon Sep 17 00:00:00 2001 From: Abhishek Gupta Date: Wed, 8 Jul 2026 20:46:55 +0000 Subject: [PATCH 05/11] Addressed the comments --- internal/fs/inode/file_test.go | 20 +------ .../kernel_range_reader_test.go | 19 +----- internal/gcsx/test_buffer_pool.go | 48 +++++++++++++++ internal/gcsx/vectored_writer.go | 37 ++++++------ internal/gcsx/vectored_writer_test.go | 60 ++++++------------- 5 files changed, 90 insertions(+), 94 deletions(-) create mode 100644 internal/gcsx/test_buffer_pool.go diff --git a/internal/fs/inode/file_test.go b/internal/fs/inode/file_test.go index 7bfa47aded..f725b3c1b4 100644 --- a/internal/fs/inode/file_test.go +++ b/internal/fs/inode/file_test.go @@ -514,21 +514,6 @@ func (t *FileTest) TestRead() { } } -type inodeTestBufferPool struct { - buffers [][]byte - idx int -} - -func (p *inodeTestBufferPool) Get() []byte { - if p.idx < len(p.buffers) { - b := p.buffers[p.idx] - p.idx++ - return b - } - return make([]byte, 2) -} - -func (p *inodeTestBufferPool) Put(b []byte) {} func (t *FileTest) TestReadWithBufferPool() { assert.Equal(t.T(), "taco", t.initialContents) @@ -586,8 +571,9 @@ func (t *FileTest) TestReadWithBufferPool() { for _, tc := range testCases { t.Run(tc.name, func() { // Arrange - pool := &inodeTestBufferPool{ - buffers: [][]byte{make([]byte, 2), make([]byte, 2)}, + pool := &gcsx.TestBufferPool{ + Buffers: [][]byte{make([]byte, 2), make([]byte, 2)}, + DefaultBufferSize: 2, } req := &gcsx.ReadRequest{ BufferPool: pool, diff --git a/internal/gcsx/kernel_readers/kernel_range_reader_test.go b/internal/gcsx/kernel_readers/kernel_range_reader_test.go index 524ea7bd71..865c60515f 100644 --- a/internal/gcsx/kernel_readers/kernel_range_reader_test.go +++ b/internal/gcsx/kernel_readers/kernel_range_reader_test.go @@ -167,26 +167,11 @@ func (t *KernelRangeReaderTest) TestReadAt_ClobberedError() { t.bucket.AssertExpectations(t.T()) } -type testBufferPool struct { - buffers [][]byte - idx int -} - -func (p *testBufferPool) Get() []byte { - if p.idx < len(p.buffers) { - b := p.buffers[p.idx] - p.idx++ - return b - } - return make([]byte, 1024) -} - -func (p *testBufferPool) Put(b []byte) {} func (t *KernelRangeReaderTest) TestReadAt_BufferPool_Success() { data := []byte("abcdefghij") // length 10 - pool := &testBufferPool{ - buffers: [][]byte{ + pool := &gcsx.TestBufferPool{ + Buffers: [][]byte{ make([]byte, 3), make([]byte, 2), make([]byte, 4), 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_writer.go b/internal/gcsx/vectored_writer.go index 252438943c..fb7004682a 100644 --- a/internal/gcsx/vectored_writer.go +++ b/internal/gcsx/vectored_writer.go @@ -46,27 +46,26 @@ func NewVectoredWriter(pool BufferPool, maxSize int64) *VectoredWriter { // from the pool. Returns nil if no more buffers can be allocated (e.g. pool is nil // or maxSize is reached). func (w *VectoredWriter) availableBuffer() []byte { - for { - if w.index < len(w.buffers) { - sb := &w.buffers[w.index] - if sb.offset < len(sb.buffer) { - return sb.buffer[sb.offset:] - } - w.index++ - continue - } - if w.pool == nil || w.written >= w.maxSize { - return nil - } - buf := w.pool.Get() - if len(buf) == 0 { - return nil - } - if int64(len(buf)) > w.maxSize-w.written { - buf = buf[:w.maxSize-w.written] + if w.index < len(w.buffers) { + sb := &w.buffers[w.index] + if sb.offset < len(sb.buffer) { + return sb.buffer[sb.offset:] } - w.buffers = append(w.buffers, bufferWithOffset{buffer: buf}) + w.index++ + } + + if w.pool == nil || w.written >= w.maxSize { + return nil + } + buf := w.pool.Get() + if len(buf) == 0 { + return nil + } + if int64(len(buf)) > w.maxSize-w.written { + buf = buf[:w.maxSize-w.written] } + w.buffers = append(w.buffers, bufferWithOffset{buffer: buf}) + return buf } // advance updates the offset of the current buffer and the total bytes written. diff --git a/internal/gcsx/vectored_writer_test.go b/internal/gcsx/vectored_writer_test.go index 85e48c5269..cd36d4cfe5 100644 --- a/internal/gcsx/vectored_writer_test.go +++ b/internal/gcsx/vectored_writer_test.go @@ -23,28 +23,6 @@ import ( "github.com/stretchr/testify/assert" ) -type testBufferPool struct { - buffers [][]byte - idx int - putBuffers [][]byte - returnNilOnExhaustion bool -} - -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 - } - return make([]byte, 1024) -} - -func (p *testBufferPool) Put(b []byte) { - p.putBuffers = append(p.putBuffers, b) -} func TestVectoredWriter_Write(t *testing.T) { tests := []struct { @@ -126,7 +104,7 @@ func TestVectoredWriter_Write(t *testing.T) { } var pool BufferPool if len(tc.buffers) > 0 { - pool = &testBufferPool{buffers: tc.buffers} + pool = &TestBufferPool{Buffers: tc.buffers} } w := NewVectoredWriter(pool, maxSize) @@ -239,7 +217,7 @@ func TestVectoredWriter_ReadFrom(t *testing.T) { } var pool BufferPool if len(tc.buffers) > 0 { - pool = &testBufferPool{buffers: tc.buffers} + pool = &TestBufferPool{Buffers: tc.buffers} } w := NewVectoredWriter(pool, maxSize) @@ -327,7 +305,7 @@ func TestVectoredWriter_ReadFromAt(t *testing.T) { for _, tc := range tests { t.Run(tc.name, func(t *testing.T) { // Arrange - pool := &testBufferPool{buffers: tc.buffers} + pool := &TestBufferPool{Buffers: tc.buffers} w := NewVectoredWriter(pool, tc.maxSize) // Act @@ -344,7 +322,7 @@ func TestVectoredWriter_ReadFromAt(t *testing.T) { func TestVectoredWriter_Release(t *testing.T) { tests := []struct { name string - pool *testBufferPool + pool *TestBufferPool maxSize int64 writeInput []byte releaseCount int @@ -353,8 +331,8 @@ func TestVectoredWriter_Release(t *testing.T) { }{ { name: "release allocated buffers to pool and clear state", - pool: &testBufferPool{ - buffers: [][]byte{ + pool: &TestBufferPool{ + Buffers: [][]byte{ make([]byte, 10), make([]byte, 10), make([]byte, 10), @@ -377,8 +355,8 @@ func TestVectoredWriter_Release(t *testing.T) { }, { name: "release with no allocated buffers is safe", - pool: &testBufferPool{ - buffers: [][]byte{ + pool: &TestBufferPool{ + Buffers: [][]byte{ make([]byte, 10), }, }, @@ -390,8 +368,8 @@ func TestVectoredWriter_Release(t *testing.T) { }, { name: "release multiple times is safe and idempotent", - pool: &testBufferPool{ - buffers: [][]byte{ + pool: &TestBufferPool{ + Buffers: [][]byte{ make([]byte, 10), }, }, @@ -422,7 +400,7 @@ func TestVectoredWriter_Release(t *testing.T) { // Assert if tc.pool != nil { - assert.Equal(t, tc.expectedPutCount, len(tc.pool.putBuffers)) + assert.Equal(t, tc.expectedPutCount, len(tc.pool.PutBuffers)) } assert.Equal(t, tc.expectedBufs, w.Buffers()) }) @@ -431,8 +409,8 @@ func TestVectoredWriter_Release(t *testing.T) { func TestVectoredWriter_OversizedBufferTruncation(t *testing.T) { // Arrange - pool := &testBufferPool{ - buffers: [][]byte{ + pool := &TestBufferPool{ + Buffers: [][]byte{ make([]byte, 100), }, } @@ -472,8 +450,8 @@ func TestVectoredWriter_PoolExhaustion(t *testing.T) { for _, tc := range tests { t.Run(tc.name, func(t *testing.T) { // Arrange - pool := &testBufferPool{ - returnNilOnExhaustion: true, + pool := &TestBufferPool{ + ReturnNilOnExhaustion: true, } w := NewVectoredWriter(pool, 100) @@ -507,8 +485,8 @@ func TestVectoredWriter_ZeroOrNegativeMaxSize(t *testing.T) { for _, tc := range tests { t.Run(tc.name, func(t *testing.T) { // Arrange - pool := &testBufferPool{ - buffers: [][]byte{make([]byte, 10)}, + pool := &TestBufferPool{ + Buffers: [][]byte{make([]byte, 10)}, } w := NewVectoredWriter(pool, tc.maxSize) @@ -526,8 +504,8 @@ func TestVectoredWriter_ZeroOrNegativeMaxSize(t *testing.T) { var globalWriter *VectoredWriter func BenchmarkNewVectoredWriter(b *testing.B) { - pool := &testBufferPool{ - buffers: [][]byte{ + pool := &TestBufferPool{ + Buffers: [][]byte{ make([]byte, 100), make([]byte, 100), make([]byte, 100), From c0226f25cd1bccbdf6c64e0342d701365d31ef9e Mon Sep 17 00:00:00 2001 From: Abhishek Gupta Date: Thu, 9 Jul 2026 07:07:55 +0000 Subject: [PATCH 06/11] Minor changes --- internal/fs/fs.go | 8 ++++---- internal/fs/fs_internal_test.go | 8 ++++---- internal/fs/inode/file_test.go | 1 - .../kernel_readers/kernel_range_reader_test.go | 1 - internal/gcsx/reader.go | 16 ++++++---------- internal/gcsx/vectored_writer_test.go | 1 - 6 files changed, 14 insertions(+), 21 deletions(-) diff --git a/internal/fs/fs.go b/internal/fs/fs.go index cabcf7f178..32fce92ad3 100644 --- a/internal/fs/fs.go +++ b/internal/fs/fs.go @@ -690,15 +690,15 @@ type fileSystem struct { readBufferPool sync.Pool } -type bufferPoolWrapper struct { +type fixedSizeBufferPool struct { pool *sync.Pool } -func (bp *bufferPoolWrapper) Get() []byte { +func (bp *fixedSizeBufferPool) Get() []byte { return bp.pool.Get().(*[readPoolBufferSize]byte)[:] } -func (bp *bufferPoolWrapper) Put(buf []byte) { +func (bp *fixedSizeBufferPool) Put(buf []byte) { if cap(buf) != readPoolBufferSize { return } @@ -3161,7 +3161,7 @@ func (fs *fileSystem) ReadFile( return syscall.ENOTSUP } - req.BufferPool = &bufferPoolWrapper{pool: &fs.readBufferPool} + req.BufferPool = &fixedSizeBufferPool{pool: &fs.readBufferPool} } else { req.Buffer = op.Dst } diff --git a/internal/fs/fs_internal_test.go b/internal/fs/fs_internal_test.go index 5d464b748f..2215ade94e 100644 --- a/internal/fs/fs_internal_test.go +++ b/internal/fs/fs_internal_test.go @@ -69,13 +69,13 @@ func TestCacheDirVolumeBlockSize(t *testing.T) { } } -func TestBufferPoolWrapper_Get(t *testing.T) { +func TestFixedSizeBufferPool_Get(t *testing.T) { // Arrange var pool sync.Pool pool.New = func() any { return new([readPoolBufferSize]byte) } - bp := &bufferPoolWrapper{pool: &pool} + bp := &fixedSizeBufferPool{pool: &pool} // Act buf := bp.Get() @@ -85,7 +85,7 @@ func TestBufferPoolWrapper_Get(t *testing.T) { assert.Equal(t, readPoolBufferSize, cap(buf)) } -func TestBufferPoolWrapper_Put(t *testing.T) { +func TestFixedSizeBufferPool_Put(t *testing.T) { testCases := []struct { name string buf []byte @@ -111,7 +111,7 @@ func TestBufferPoolWrapper_Put(t *testing.T) { pool.New = func() any { return new([readPoolBufferSize]byte) } - bp := &bufferPoolWrapper{pool: &pool} + bp := &fixedSizeBufferPool{pool: &pool} // Act & Assert assert.NotPanics(t, func() { diff --git a/internal/fs/inode/file_test.go b/internal/fs/inode/file_test.go index f725b3c1b4..7d70b48e1c 100644 --- a/internal/fs/inode/file_test.go +++ b/internal/fs/inode/file_test.go @@ -514,7 +514,6 @@ func (t *FileTest) TestRead() { } } - func (t *FileTest) TestReadWithBufferPool() { assert.Equal(t.T(), "taco", t.initialContents) diff --git a/internal/gcsx/kernel_readers/kernel_range_reader_test.go b/internal/gcsx/kernel_readers/kernel_range_reader_test.go index 865c60515f..d573d0cbc1 100644 --- a/internal/gcsx/kernel_readers/kernel_range_reader_test.go +++ b/internal/gcsx/kernel_readers/kernel_range_reader_test.go @@ -167,7 +167,6 @@ func (t *KernelRangeReaderTest) TestReadAt_ClobberedError() { t.bucket.AssertExpectations(t.T()) } - func (t *KernelRangeReaderTest) TestReadAt_BufferPool_Success() { data := []byte("abcdefghij") // length 10 pool := &gcsx.TestBufferPool{ diff --git a/internal/gcsx/reader.go b/internal/gcsx/reader.go index b2144fc04b..1bd9f8ec23 100644 --- a/internal/gcsx/reader.go +++ b/internal/gcsx/reader.go @@ -62,20 +62,16 @@ 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 req.BufferPool != nil { - totalCapacity = req.Size - } else { - totalCapacity = int64(len(req.Buffer)) - } - sizeToRead := req.Size - if sizeToRead <= 0 || sizeToRead > totalCapacity { - sizeToRead = totalCapacity + if req.BufferPool == nil { + bufLen := int64(len(req.Buffer)) + if sizeToRead <= 0 || sizeToRead > bufLen { + sizeToRead = bufLen + } } if limit > 0 && sizeToRead > limit { - sizeToRead = limit + return limit } return sizeToRead } diff --git a/internal/gcsx/vectored_writer_test.go b/internal/gcsx/vectored_writer_test.go index cd36d4cfe5..12abeff995 100644 --- a/internal/gcsx/vectored_writer_test.go +++ b/internal/gcsx/vectored_writer_test.go @@ -23,7 +23,6 @@ import ( "github.com/stretchr/testify/assert" ) - func TestVectoredWriter_Write(t *testing.T) { tests := []struct { name string From 473547304bb401c99a3bc1357992eeaa8672cec9 Mon Sep 17 00:00:00 2001 From: Abhishek Gupta Date: Thu, 9 Jul 2026 12:15:53 +0000 Subject: [PATCH 07/11] Addressing feedback --- internal/fs/fs.go | 31 +--- internal/fs/fs_internal_test.go | 53 ------ internal/fs/inode/file.go | 10 +- internal/fs/inode/file_test.go | 4 +- internal/gcsx/buffer_pool.go | 62 +++++++ internal/gcsx/buffer_pool_test.go | 65 ++++++++ .../kernel_readers/kernel_range_reader.go | 10 +- .../kernel_range_reader_test.go | 6 + internal/gcsx/reader.go | 16 +- internal/gcsx/vectored_read_buffer.go | 153 ++++++++++++++++++ ...r_test.go => vectored_read_buffer_test.go} | 41 +++-- internal/gcsx/vectored_writer.go | 147 ----------------- 12 files changed, 323 insertions(+), 275 deletions(-) create mode 100644 internal/gcsx/buffer_pool.go create mode 100644 internal/gcsx/buffer_pool_test.go create mode 100644 internal/gcsx/vectored_read_buffer.go rename internal/gcsx/{vectored_writer_test.go => vectored_read_buffer_test.go} (91%) delete mode 100644 internal/gcsx/vectored_writer.go diff --git a/internal/fs/fs.go b/internal/fs/fs.go index 32fce92ad3..3951ffe750 100644 --- a/internal/fs/fs.go +++ b/internal/fs/fs.go @@ -27,7 +27,6 @@ import ( "reflect" "slices" "strings" - "sync" "syscall" "time" @@ -62,9 +61,6 @@ import ( "github.com/spf13/viper" ) -// readPoolBufferSize is the size of each buffer in readBufferPool (1 MiB). -const readPoolBufferSize = util.MiB - type ServerConfig struct { // A clock used for cache expiration. It is *not* used for inode times, for // which we use the wall clock. @@ -229,11 +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: sync.Pool{ - New: func() any { - return new([readPoolBufferSize]byte) - }, - }, + readBufferPool: gcsx.NewFixedSizeBufferPool(), } // Initialize MRD cache if enabled @@ -686,23 +678,8 @@ type fileSystem struct { // mrdCache manages the cache of inactive MultiRangeDownloaders. mrdCache *lru.Cache - // readBufferPool is a sync.Pool of readPoolBufferSize (1 MiB) buffers. - readBufferPool sync.Pool -} - -type fixedSizeBufferPool struct { - pool *sync.Pool -} - -func (bp *fixedSizeBufferPool) Get() []byte { - return bp.pool.Get().(*[readPoolBufferSize]byte)[:] -} - -func (bp *fixedSizeBufferPool) Put(buf []byte) { - if cap(buf) != readPoolBufferSize { - return - } - bp.pool.Put((*[readPoolBufferSize]byte)(buf[:cap(buf)])) + // readBufferPool is a pool of readPoolBufferSize (1 MiB) buffers. + readBufferPool gcsx.BufferPool } //////////////////////////////////////////////////////////////////////// @@ -3161,7 +3138,7 @@ func (fs *fileSystem) ReadFile( return syscall.ENOTSUP } - req.BufferPool = &fixedSizeBufferPool{pool: &fs.readBufferPool} + req.BufferPool = fs.readBufferPool } else { req.Buffer = op.Dst } diff --git a/internal/fs/fs_internal_test.go b/internal/fs/fs_internal_test.go index 2215ade94e..74ec63dc56 100644 --- a/internal/fs/fs_internal_test.go +++ b/internal/fs/fs_internal_test.go @@ -15,7 +15,6 @@ package fs import ( - "sync" "testing" "github.com/googlecloudplatform/gcsfuse/v3/cfg" @@ -68,55 +67,3 @@ func TestCacheDirVolumeBlockSize(t *testing.T) { }) } } - -func TestFixedSizeBufferPool_Get(t *testing.T) { - // Arrange - var pool sync.Pool - pool.New = func() any { - return new([readPoolBufferSize]byte) - } - bp := &fixedSizeBufferPool{pool: &pool} - - // 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 - var pool sync.Pool - pool.New = func() any { - return new([readPoolBufferSize]byte) - } - bp := &fixedSizeBufferPool{pool: &pool} - - // Act & Assert - assert.NotPanics(t, func() { - bp.Put(tc.buf) - }) - }) - } -} diff --git a/internal/fs/inode/file.go b/internal/fs/inode/file.go index c02a6b36e1..2149d58554 100644 --- a/internal/fs/inode/file.go +++ b/internal/fs/inode/file.go @@ -664,15 +664,15 @@ func (f *FileInode) Read( var callback func() if req.BufferPool != nil { - writer := gcsx.NewVectoredWriter(req.BufferPool, req.Size) + vBuf := gcsx.NewVectoredReadBuffer(req.BufferPool, req.Size) var written int64 - written, err = writer.ReadFromAt(f.content, req.Offset) + written, err = vBuf.ReadFromAt(f.content, req.Offset) n = int(written) if n > 0 { - data = writer.Buffers() - callback = func() { writer.Release() } + data = vBuf.Buffers() + callback = func() { vBuf.Release() } } else { - writer.Release() + vBuf.Release() } } else { n, err = f.content.ReadAt(req.Buffer, req.Offset) diff --git a/internal/fs/inode/file_test.go b/internal/fs/inode/file_test.go index 7d70b48e1c..57e7b5b387 100644 --- a/internal/fs/inode/file_test.go +++ b/internal/fs/inode/file_test.go @@ -555,7 +555,7 @@ func (t *FileTest) TestReadWithBufferPool() { size: 2, expectedData: []string{"o"}, expectedSize: 1, - expectedErr: io.EOF, + expectedErr: nil, }, { name: "AtEOF", @@ -563,7 +563,7 @@ func (t *FileTest) TestReadWithBufferPool() { size: 2, expectedData: []string{}, expectedSize: 0, - expectedErr: io.EOF, + expectedErr: nil, }, } 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 d73502066a..1bea4d5505 100644 --- a/internal/gcsx/kernel_readers/kernel_range_reader.go +++ b/internal/gcsx/kernel_readers/kernel_range_reader.go @@ -112,18 +112,18 @@ func (krr *KernelRangeReader) ReadAt(ctx context.Context, req *gcsx.ReadRequest) var n int if req.BufferPool != nil { - writer := gcsx.NewVectoredWriter(req.BufferPool, bytesToRead) + 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) if err == io.EOF && written > 0 { err = io.ErrUnexpectedEOF } if n > 0 { - resp.Data = writer.Buffers() - resp.Callback = func() { writer.Release() } + resp.Data = vBuf.Buffers() + resp.Callback = func() { vBuf.Release() } } else { - writer.Release() // Release immediately on 0-byte read or early error + vBuf.Release() // Release immediately on 0-byte read or early error } } else { n, err = io.ReadFull(reader, req.Buffer[:bytesToRead]) diff --git a/internal/gcsx/kernel_readers/kernel_range_reader_test.go b/internal/gcsx/kernel_readers/kernel_range_reader_test.go index d573d0cbc1..b443958ca5 100644 --- a/internal/gcsx/kernel_readers/kernel_range_reader_test.go +++ b/internal/gcsx/kernel_readers/kernel_range_reader_test.go @@ -77,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() @@ -93,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) @@ -107,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] @@ -125,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() @@ -141,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) @@ -153,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() diff --git a/internal/gcsx/reader.go b/internal/gcsx/reader.go index 1bd9f8ec23..1bd79cf152 100644 --- a/internal/gcsx/reader.go +++ b/internal/gcsx/reader.go @@ -26,12 +26,6 @@ import ( // to an alternative reader. var FallbackToAnotherReader = errors.New("fallback to another reader is required") -// BufferPool defines an interface for on-demand buffer allocation and recycling. -type BufferPool interface { - Get() []byte - Put([]byte) -} - // ReadRequest encapsulates the parameters for a read operation. type ReadRequest struct { // Buffer is provided by jacobsa/fuse and should be filled with data from the object. @@ -64,14 +58,10 @@ type ReadRequest struct { func (req *ReadRequest) GetReadSize(limit int64) int64 { sizeToRead := req.Size if req.BufferPool == nil { - bufLen := int64(len(req.Buffer)) - if sizeToRead <= 0 || sizeToRead > bufLen { - sizeToRead = bufLen - } + sizeToRead = min(sizeToRead, int64(len(req.Buffer))) } - - if limit > 0 && sizeToRead > limit { - return limit + if limit > 0 { + sizeToRead = min(sizeToRead, limit) } return sizeToRead } diff --git a/internal/gcsx/vectored_read_buffer.go b/internal/gcsx/vectored_read_buffer.go new file mode 100644 index 0000000000..a64b43ec1c --- /dev/null +++ b/internal/gcsx/vectored_read_buffer.go @@ -0,0 +1,153 @@ +// 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" +) + +// 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(); len(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 updates the length of the active buffer and the total bytes written. +func (v *VectoredReadBuffer) advance(n int) { + if len(v.buffers) == 0 || n <= 0 { + return + } + idx := len(v.buffers) - 1 + v.buffers[idx] = v.buffers[idx][:len(v.buffers[idx])+n] + v.written += int64(n) +} + +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:]) + v.advance(toCopy) + 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 { + v.advance(readNum) + 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_writer_test.go b/internal/gcsx/vectored_read_buffer_test.go similarity index 91% rename from internal/gcsx/vectored_writer_test.go rename to internal/gcsx/vectored_read_buffer_test.go index 12abeff995..b45f9d46ae 100644 --- a/internal/gcsx/vectored_writer_test.go +++ b/internal/gcsx/vectored_read_buffer_test.go @@ -23,7 +23,7 @@ import ( "github.com/stretchr/testify/assert" ) -func TestVectoredWriter_Write(t *testing.T) { +func TestVectoredReadBuffer_Write(t *testing.T) { tests := []struct { name string buffers [][]byte @@ -105,7 +105,7 @@ func TestVectoredWriter_Write(t *testing.T) { if len(tc.buffers) > 0 { pool = &TestBufferPool{Buffers: tc.buffers} } - w := NewVectoredWriter(pool, maxSize) + w := NewVectoredReadBuffer(pool, maxSize) n, err := w.Write(tc.input) @@ -124,7 +124,7 @@ func (r *errorReader) Read(p []byte) (n int, err error) { return 0, r.err } -func TestVectoredWriter_ReadFrom(t *testing.T) { +func TestVectoredReadBuffer_ReadFrom(t *testing.T) { testErr := errors.New("read error") tests := []struct { @@ -218,7 +218,7 @@ func TestVectoredWriter_ReadFrom(t *testing.T) { if len(tc.buffers) > 0 { pool = &TestBufferPool{Buffers: tc.buffers} } - w := NewVectoredWriter(pool, maxSize) + w := NewVectoredReadBuffer(pool, maxSize) n, err := w.ReadFrom(tc.reader) @@ -248,7 +248,7 @@ func (m *mockReaderAt) ReadAt(dst []byte, offset int64) (n int, err error) { return n, err } -func TestVectoredWriter_ReadFromAt(t *testing.T) { +func TestVectoredReadBuffer_ReadFromAt(t *testing.T) { testErr := errors.New("reader at error") tests := []struct { @@ -305,7 +305,7 @@ func TestVectoredWriter_ReadFromAt(t *testing.T) { t.Run(tc.name, func(t *testing.T) { // Arrange pool := &TestBufferPool{Buffers: tc.buffers} - w := NewVectoredWriter(pool, tc.maxSize) + w := NewVectoredReadBuffer(pool, tc.maxSize) // Act n, err := w.ReadFromAt(tc.reader, tc.offset) @@ -318,7 +318,7 @@ func TestVectoredWriter_ReadFromAt(t *testing.T) { } } -func TestVectoredWriter_Release(t *testing.T) { +func TestVectoredReadBuffer_Release(t *testing.T) { tests := []struct { name string pool *TestBufferPool @@ -326,7 +326,6 @@ func TestVectoredWriter_Release(t *testing.T) { writeInput []byte releaseCount int expectedPutCount int - expectedBufs [][]byte }{ { name: "release allocated buffers to pool and clear state", @@ -341,7 +340,6 @@ func TestVectoredWriter_Release(t *testing.T) { writeInput: []byte("0123456789abcde"), // 15 bytes -> 2 buffers allocated releaseCount: 1, expectedPutCount: 2, - expectedBufs: [][]byte{}, }, { name: "release with nil pool is safe", @@ -350,7 +348,6 @@ func TestVectoredWriter_Release(t *testing.T) { writeInput: nil, releaseCount: 1, expectedPutCount: 0, - expectedBufs: [][]byte{}, }, { name: "release with no allocated buffers is safe", @@ -363,7 +360,6 @@ func TestVectoredWriter_Release(t *testing.T) { writeInput: nil, // 0 bytes -> 0 buffers allocated releaseCount: 1, expectedPutCount: 0, - expectedBufs: [][]byte{}, }, { name: "release multiple times is safe and idempotent", @@ -376,7 +372,6 @@ func TestVectoredWriter_Release(t *testing.T) { writeInput: []byte("hello"), // 5 bytes -> 1 buffer allocated releaseCount: 2, expectedPutCount: 1, - expectedBufs: [][]byte{}, }, } @@ -387,7 +382,7 @@ func TestVectoredWriter_Release(t *testing.T) { if tc.pool != nil { pool = tc.pool } - w := NewVectoredWriter(pool, tc.maxSize) + w := NewVectoredReadBuffer(pool, tc.maxSize) if len(tc.writeInput) > 0 { _, _ = w.Write(tc.writeInput) } @@ -401,19 +396,19 @@ func TestVectoredWriter_Release(t *testing.T) { if tc.pool != nil { assert.Equal(t, tc.expectedPutCount, len(tc.pool.PutBuffers)) } - assert.Equal(t, tc.expectedBufs, w.Buffers()) + assert.Empty(t, w.Buffers()) }) } } -func TestVectoredWriter_OversizedBufferTruncation(t *testing.T) { +func TestVectoredReadBuffer_OversizedBufferTruncation(t *testing.T) { // Arrange pool := &TestBufferPool{ Buffers: [][]byte{ make([]byte, 100), }, } - w := NewVectoredWriter(pool, 25) + w := NewVectoredReadBuffer(pool, 25) // Act n, err := w.Write([]byte("this is a test string of length greater than 25 characters")) @@ -425,7 +420,7 @@ func TestVectoredWriter_OversizedBufferTruncation(t *testing.T) { assert.Equal(t, 25, len(w.Buffers()[0])) } -func TestVectoredWriter_PoolExhaustion(t *testing.T) { +func TestVectoredReadBuffer_PoolExhaustion(t *testing.T) { tests := []struct { name string isReadFrom bool @@ -452,7 +447,7 @@ func TestVectoredWriter_PoolExhaustion(t *testing.T) { pool := &TestBufferPool{ ReturnNilOnExhaustion: true, } - w := NewVectoredWriter(pool, 100) + w := NewVectoredReadBuffer(pool, 100) // Act var n int64 @@ -472,7 +467,7 @@ func TestVectoredWriter_PoolExhaustion(t *testing.T) { } } -func TestVectoredWriter_ZeroOrNegativeMaxSize(t *testing.T) { +func TestVectoredReadBuffer_ZeroOrNegativeMaxSize(t *testing.T) { tests := []struct { name string maxSize int64 @@ -487,7 +482,7 @@ func TestVectoredWriter_ZeroOrNegativeMaxSize(t *testing.T) { pool := &TestBufferPool{ Buffers: [][]byte{make([]byte, 10)}, } - w := NewVectoredWriter(pool, tc.maxSize) + w := NewVectoredReadBuffer(pool, tc.maxSize) // Act n, err := w.Write([]byte("hello")) @@ -500,9 +495,9 @@ func TestVectoredWriter_ZeroOrNegativeMaxSize(t *testing.T) { } } -var globalWriter *VectoredWriter +var globalBuffer *VectoredReadBuffer -func BenchmarkNewVectoredWriter(b *testing.B) { +func BenchmarkNewVectoredReadBuffer(b *testing.B) { pool := &TestBufferPool{ Buffers: [][]byte{ make([]byte, 100), @@ -512,6 +507,6 @@ func BenchmarkNewVectoredWriter(b *testing.B) { } b.ResetTimer() for i := 0; i < b.N; i++ { - globalWriter = NewVectoredWriter(pool, 300) + globalBuffer = NewVectoredReadBuffer(pool, 300) } } diff --git a/internal/gcsx/vectored_writer.go b/internal/gcsx/vectored_writer.go deleted file mode 100644 index fb7004682a..0000000000 --- a/internal/gcsx/vectored_writer.go +++ /dev/null @@ -1,147 +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 - pool BufferPool - maxSize int64 - written int64 -} - -// NewVectoredWriter creates a new VectoredWriter that allocates buffers on demand from pool. -func NewVectoredWriter(pool BufferPool, maxSize int64) *VectoredWriter { - return &VectoredWriter{ - buffers: make([]bufferWithOffset, 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 (w *VectoredWriter) availableBuffer() []byte { - if w.index < len(w.buffers) { - sb := &w.buffers[w.index] - if sb.offset < len(sb.buffer) { - return sb.buffer[sb.offset:] - } - w.index++ - } - - if w.pool == nil || w.written >= w.maxSize { - return nil - } - buf := w.pool.Get() - if len(buf) == 0 { - return nil - } - if int64(len(buf)) > w.maxSize-w.written { - buf = buf[:w.maxSize-w.written] - } - w.buffers = append(w.buffers, bufferWithOffset{buffer: buf}) - return buf -} - -// advance updates the offset of the current buffer and the total bytes written. -func (w *VectoredWriter) advance(n int) { - w.buffers[w.index].offset += n - w.written += int64(n) -} - -func (w *VectoredWriter) Write(p []byte) (n int, err error) { - for n < len(p) { - buf := w.availableBuffer() - if buf == nil { - return n, io.ErrShortWrite - } - toCopy := copy(buf, p[n:]) - w.advance(toCopy) - n += toCopy - } - return n, nil -} - -func (w *VectoredWriter) readIntoBuffers(readFn func(buf []byte) (int, error)) (n int64, err error) { - for { - buf := w.availableBuffer() - if buf == nil { - break - } - var readNum int - readNum, err = readFn(buf) - if readNum > 0 { - w.advance(readNum) - n += int64(readNum) - } - if err != nil { - break - } - } - return n, err -} - -// 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) { - n, err = w.readIntoBuffers(r.Read) - if err == io.EOF { - err = nil - } - return n, err -} - -// ReadFromAt reads data from r starting at offset directly into the -// underlying buffers, avoiding intermediate allocations and double-copying. -func (w *VectoredWriter) ReadFromAt(r io.ReaderAt, offset int64) (n int64, err error) { - return w.readIntoBuffers(func(buf []byte) (int, error) { - return r.ReadAt(buf, offset+w.written) - }) -} - -// Buffers returns the slices of bytes actually written to. -func (w *VectoredWriter) Buffers() [][]byte { - res := make([][]byte, 0, len(w.buffers)) - for _, sb := range w.buffers { - if sb.offset > 0 { - res = append(res, sb.buffer[:sb.offset]) - } - } - return res -} - -// Release puts all allocated buffers back into the pool. -func (w *VectoredWriter) Release() { - if w.pool == nil { - return - } - for _, sb := range w.buffers { - w.pool.Put(sb.buffer) - } - w.buffers = nil -} From 79977936cffa89852c0cfa600dbae36d69693a14 Mon Sep 17 00:00:00 2001 From: Abhishek Gupta Date: Thu, 9 Jul 2026 12:46:52 +0000 Subject: [PATCH 08/11] Passing req.size in UTs as well --- internal/fs/handle/file_test.go | 5 +++++ 1 file changed, 5 insertions(+) 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() From 46ca4f3843ddb8f5f95966ade2c58842d796c555 Mon Sep 17 00:00:00 2001 From: Abhishek Gupta Date: Thu, 9 Jul 2026 18:37:31 +0000 Subject: [PATCH 09/11] Minor changes --- internal/gcsx/vectored_read_buffer.go | 34 +++++++++++++++++++-------- 1 file changed, 24 insertions(+), 10 deletions(-) diff --git a/internal/gcsx/vectored_read_buffer.go b/internal/gcsx/vectored_read_buffer.go index a64b43ec1c..8432caa0b8 100644 --- a/internal/gcsx/vectored_read_buffer.go +++ b/internal/gcsx/vectored_read_buffer.go @@ -15,9 +15,12 @@ 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 @@ -60,7 +63,7 @@ func (v *VectoredReadBuffer) availableBuffer() []byte { // 2. Fetch new buffer from pool if v.pool != nil { - if buf := v.pool.Get(); len(buf) > 0 { + 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]) @@ -71,14 +74,21 @@ func (v *VectoredReadBuffer) availableBuffer() []byte { return nil } -// advance updates the length of the active buffer and the total bytes written. -func (v *VectoredReadBuffer) advance(n int) { - if len(v.buffers) == 0 || n <= 0 { - return +// 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 } - idx := len(v.buffers) - 1 - v.buffers[idx] = v.buffers[idx][:len(v.buffers[idx])+n] - v.written += int64(n) + 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) { @@ -88,7 +98,9 @@ func (v *VectoredReadBuffer) Write(p []byte) (n int, err error) { return n, io.ErrShortWrite } toCopy := copy(buf, p[n:]) - v.advance(toCopy) + if err := v.advance(toCopy); err != nil { + return n, err + } n += toCopy } return n, nil @@ -103,7 +115,9 @@ func (v *VectoredReadBuffer) readIntoBuffers(readFn func(buf []byte) (int, error var readNum int readNum, err = readFn(buf) if readNum > 0 { - v.advance(readNum) + if advErr := v.advance(readNum); advErr != nil { + return n, advErr + } n += int64(readNum) } if err != nil { From 5df2631dd64cb5739ab7fefc76ad5f97a495b7fa Mon Sep 17 00:00:00 2001 From: Abhishek Gupta Date: Fri, 10 Jul 2026 09:27:38 +0000 Subject: [PATCH 10/11] Changes to better structure the code and simplify vectoredReadBuffer code as well --- internal/{gcsx => buffer}/buffer_pool.go | 24 ++-- internal/{gcsx => buffer}/buffer_pool_test.go | 10 +- .../{gcsx => buffer}/vectored_read_buffer.go | 122 +++++++++--------- .../vectored_read_buffer_test.go | 31 ++--- internal/fs/fs.go | 5 +- internal/fs/inode/file.go | 3 +- internal/fs/inode/file_test.go | 2 +- .../kernel_readers/kernel_range_reader.go | 3 +- .../kernel_range_reader_test.go | 3 +- internal/gcsx/reader.go | 3 +- .../fake/fake_buffer_pool.go} | 10 +- 11 files changed, 114 insertions(+), 102 deletions(-) rename internal/{gcsx => buffer}/buffer_pool.go (64%) rename internal/{gcsx => buffer}/buffer_pool_test.go (87%) rename internal/{gcsx => buffer}/vectored_read_buffer.go (51%) rename internal/{gcsx => buffer}/vectored_read_buffer_test.go (95%) rename internal/{gcsx/test_buffer_pool.go => storage/fake/fake_buffer_pool.go} (85%) diff --git a/internal/gcsx/buffer_pool.go b/internal/buffer/buffer_pool.go similarity index 64% rename from internal/gcsx/buffer_pool.go rename to internal/buffer/buffer_pool.go index c852e5e06a..e9baae7722 100644 --- a/internal/gcsx/buffer_pool.go +++ b/internal/buffer/buffer_pool.go @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -package gcsx +package buffer import ( "sync" @@ -21,8 +21,8 @@ import ( "github.com/googlecloudplatform/gcsfuse/v3/internal/util" ) -// BufferPool defines an interface for on-demand buffer allocation and recycling. -type BufferPool interface { +// Pool defines an interface for on-demand buffer allocation and recycling. +type Pool interface { Get() []byte Put([]byte) } @@ -30,17 +30,17 @@ type BufferPool interface { // 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) +// Ensure FixedSizePool implements Pool at compile time. +var _ Pool = (*FixedSizePool)(nil) -type FixedSizeBufferPool struct { +type FixedSizePool 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{ +// NewFixedSizePool creates a new FixedSizePool with 1 MiB buffers. +func NewFixedSizePool() *FixedSizePool { + return &FixedSizePool{ Pool: &sync.Pool{ New: func() any { return new([readPoolBufferSize]byte) @@ -49,13 +49,13 @@ func NewFixedSizeBufferPool() *FixedSizeBufferPool { } } -func (bp *FixedSizeBufferPool) Get() []byte { +func (bp *FixedSizePool) Get() []byte { return bp.Pool.Get().(*[readPoolBufferSize]byte)[:] } -func (bp *FixedSizeBufferPool) Put(buf []byte) { +func (bp *FixedSizePool) Put(buf []byte) { if cap(buf) != readPoolBufferSize { - logger.Errorf("FixedSizeBufferPool::Put Buffer capacity does not match readPoolBufferSize: %d vs %d", cap(buf), readPoolBufferSize) + logger.Errorf("FixedSizePool::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/buffer/buffer_pool_test.go similarity index 87% rename from internal/gcsx/buffer_pool_test.go rename to internal/buffer/buffer_pool_test.go index a4a1c2cc10..f33ceb8c82 100644 --- a/internal/gcsx/buffer_pool_test.go +++ b/internal/buffer/buffer_pool_test.go @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -package gcsx +package buffer import ( "testing" @@ -20,9 +20,9 @@ import ( "github.com/stretchr/testify/assert" ) -func TestFixedSizeBufferPool_Get(t *testing.T) { +func TestFixedSizePool_Get(t *testing.T) { // Arrange - bp := NewFixedSizeBufferPool() + bp := NewFixedSizePool() // Act buf := bp.Get() @@ -32,7 +32,7 @@ func TestFixedSizeBufferPool_Get(t *testing.T) { assert.Equal(t, readPoolBufferSize, cap(buf)) } -func TestFixedSizeBufferPool_Put(t *testing.T) { +func TestFixedSizePool_Put(t *testing.T) { testCases := []struct { name string buf []byte @@ -54,7 +54,7 @@ func TestFixedSizeBufferPool_Put(t *testing.T) { for _, tc := range testCases { t.Run(tc.name, func(t *testing.T) { // Arrange - bp := NewFixedSizeBufferPool() + bp := NewFixedSizePool() // Act & Assert assert.NotPanics(t, func() { diff --git a/internal/gcsx/vectored_read_buffer.go b/internal/buffer/vectored_read_buffer.go similarity index 51% rename from internal/gcsx/vectored_read_buffer.go rename to internal/buffer/vectored_read_buffer.go index 8432caa0b8..71390053a0 100644 --- a/internal/gcsx/vectored_read_buffer.go +++ b/internal/buffer/vectored_read_buffer.go @@ -12,28 +12,40 @@ // See the License for the specific language governing permissions and // limitations under the License. -package gcsx +package buffer 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 +// VectoredReadBuffer accumulates data across a sequence of pooled byte buffers up to maxSize. +// It implements io.Writer and io.ReaderFrom, and provides ReadFromAt to support zero-allocation, +// zero-copy reads from random-access sources (io.ReaderAt). Buffers are allocated lazily from pool +// on demand as data is written or read. +// +// Invariant: For each slice b in buffers, len(b) tracks the number of valid populated (written or read) +// 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 holds the slices of bytes allocated from pool. + // For each buffer b in buffers, len(b) represents the bytes actually populated (written or read), + // while cap(b) represents the full capacity allocated from the pool. buffers [][]byte - pool BufferPool + + // pool is the underlying pool from which new byte buffers are allocated on demand. + // If nil, no new buffers will be allocated once existing buffer capacity is exhausted. + pool Pool + + // maxSize is the maximum total number of bytes allowed to be populated across all buffers. + // Once written reaches maxSize, subsequent writes will return io.ErrShortWrite or reads will stop. maxSize int64 + + // written tracks the cumulative number of bytes populated (written or read) into buffers so far. written int64 } // NewVectoredReadBuffer creates a new VectoredReadBuffer that allocates buffers on demand from pool. -func NewVectoredReadBuffer(pool BufferPool, maxSize int64) *VectoredReadBuffer { +func NewVectoredReadBuffer(pool Pool, maxSize int64) *VectoredReadBuffer { return &VectoredReadBuffer{ buffers: make([][]byte, 0, 2), pool: pool, @@ -52,16 +64,17 @@ func (v *VectoredReadBuffer) availableBuffer() []byte { avail := v.maxSize - v.written - // 1. Check if the active (last) buffer has remaining capacity + // 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] + lastBuf := v.buffers[len(v.buffers)-1] + if remCap := cap(lastBuf) - len(lastBuf); remCap > 0 { + // Return the unpopulated tail of lastBuf, clamped by remaining maxSize. + rem := int(min(int64(remCap), avail)) + return lastBuf[len(lastBuf) : len(lastBuf)+rem] } } - // 2. Fetch new buffer from pool + // 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(). @@ -74,78 +87,71 @@ func (v *VectoredReadBuffer) availableBuffer() []byte { 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) { +func (v *VectoredReadBuffer) Write(p []byte) (int, error) { + var totalWritten int + for totalWritten < len(p) { buf := v.availableBuffer() if buf == nil { - return n, io.ErrShortWrite + return totalWritten, io.ErrShortWrite } - toCopy := copy(buf, p[n:]) - if err := v.advance(toCopy); err != nil { - return n, err - } - n += toCopy + // Copy up to len(buf) (the available space in this chunk) bytes from p into the buffer. + bytesCopied := copy(buf, p[totalWritten:]) + + // Expand the active buffer's length to include the newly written bytes. + idx := len(v.buffers) - 1 + lastBuf := v.buffers[idx] + v.buffers[idx] = lastBuf[:len(lastBuf)+bytesCopied] + + // Track cumulative bytes written across all buffers and total bytes written in this call. + v.written += int64(bytesCopied) + totalWritten += bytesCopied } - return n, nil + return totalWritten, nil } -func (v *VectoredReadBuffer) readIntoBuffers(readFn func(buf []byte) (int, error)) (n int64, err error) { +func (v *VectoredReadBuffer) readIntoBuffers(readFn func(buf []byte) (int, error)) (int64, error) { + var totalRead int64 for { buf := v.availableBuffer() if buf == nil { - break + return totalRead, nil } - var readNum int - readNum, err = readFn(buf) - if readNum > 0 { - if advErr := v.advance(readNum); advErr != nil { - return n, advErr - } - n += int64(readNum) + // Read up to len(buf) (the available space in this chunk) bytes directly into the buffer. + bytesRead, err := readFn(buf) + if bytesRead > 0 { + // Expand the active buffer's length to include the newly read bytes. + idx := len(v.buffers) - 1 + lastBuf := v.buffers[idx] + v.buffers[idx] = lastBuf[:len(lastBuf)+bytesRead] + + // Track cumulative bytes read across all buffers and total bytes read in this call. + v.written += int64(bytesRead) + totalRead += int64(bytesRead) } if err != nil { - break + if err == io.EOF { + return totalRead, nil + } + return totalRead, err } } - 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) { +func (v *VectoredReadBuffer) ReadFrom(r io.Reader) (int64, 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) { +func (v *VectoredReadBuffer) ReadFromAt(r io.ReaderAt, offset int64) (int64, 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. diff --git a/internal/gcsx/vectored_read_buffer_test.go b/internal/buffer/vectored_read_buffer_test.go similarity index 95% rename from internal/gcsx/vectored_read_buffer_test.go rename to internal/buffer/vectored_read_buffer_test.go index b45f9d46ae..189acd2203 100644 --- a/internal/gcsx/vectored_read_buffer_test.go +++ b/internal/buffer/vectored_read_buffer_test.go @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -package gcsx +package buffer import ( "bytes" @@ -20,6 +20,7 @@ import ( "io" "testing" + "github.com/googlecloudplatform/gcsfuse/v3/internal/storage/fake" "github.com/stretchr/testify/assert" ) @@ -101,9 +102,9 @@ func TestVectoredReadBuffer_Write(t *testing.T) { for _, b := range tc.buffers { maxSize += int64(len(b)) } - var pool BufferPool + var pool Pool if len(tc.buffers) > 0 { - pool = &TestBufferPool{Buffers: tc.buffers} + pool = &fake.FakeBufferPool{Buffers: tc.buffers} } w := NewVectoredReadBuffer(pool, maxSize) @@ -214,9 +215,9 @@ func TestVectoredReadBuffer_ReadFrom(t *testing.T) { for _, b := range tc.buffers { maxSize += int64(len(b)) } - var pool BufferPool + var pool Pool if len(tc.buffers) > 0 { - pool = &TestBufferPool{Buffers: tc.buffers} + pool = &fake.FakeBufferPool{Buffers: tc.buffers} } w := NewVectoredReadBuffer(pool, maxSize) @@ -304,7 +305,7 @@ func TestVectoredReadBuffer_ReadFromAt(t *testing.T) { for _, tc := range tests { t.Run(tc.name, func(t *testing.T) { // Arrange - pool := &TestBufferPool{Buffers: tc.buffers} + pool := &fake.FakeBufferPool{Buffers: tc.buffers} w := NewVectoredReadBuffer(pool, tc.maxSize) // Act @@ -321,7 +322,7 @@ func TestVectoredReadBuffer_ReadFromAt(t *testing.T) { func TestVectoredReadBuffer_Release(t *testing.T) { tests := []struct { name string - pool *TestBufferPool + pool *fake.FakeBufferPool maxSize int64 writeInput []byte releaseCount int @@ -329,7 +330,7 @@ func TestVectoredReadBuffer_Release(t *testing.T) { }{ { name: "release allocated buffers to pool and clear state", - pool: &TestBufferPool{ + pool: &fake.FakeBufferPool{ Buffers: [][]byte{ make([]byte, 10), make([]byte, 10), @@ -351,7 +352,7 @@ func TestVectoredReadBuffer_Release(t *testing.T) { }, { name: "release with no allocated buffers is safe", - pool: &TestBufferPool{ + pool: &fake.FakeBufferPool{ Buffers: [][]byte{ make([]byte, 10), }, @@ -363,7 +364,7 @@ func TestVectoredReadBuffer_Release(t *testing.T) { }, { name: "release multiple times is safe and idempotent", - pool: &TestBufferPool{ + pool: &fake.FakeBufferPool{ Buffers: [][]byte{ make([]byte, 10), }, @@ -378,7 +379,7 @@ func TestVectoredReadBuffer_Release(t *testing.T) { for _, tc := range tests { t.Run(tc.name, func(t *testing.T) { // Arrange - var pool BufferPool + var pool Pool if tc.pool != nil { pool = tc.pool } @@ -403,7 +404,7 @@ func TestVectoredReadBuffer_Release(t *testing.T) { func TestVectoredReadBuffer_OversizedBufferTruncation(t *testing.T) { // Arrange - pool := &TestBufferPool{ + pool := &fake.FakeBufferPool{ Buffers: [][]byte{ make([]byte, 100), }, @@ -444,7 +445,7 @@ func TestVectoredReadBuffer_PoolExhaustion(t *testing.T) { for _, tc := range tests { t.Run(tc.name, func(t *testing.T) { // Arrange - pool := &TestBufferPool{ + pool := &fake.FakeBufferPool{ ReturnNilOnExhaustion: true, } w := NewVectoredReadBuffer(pool, 100) @@ -479,7 +480,7 @@ func TestVectoredReadBuffer_ZeroOrNegativeMaxSize(t *testing.T) { for _, tc := range tests { t.Run(tc.name, func(t *testing.T) { // Arrange - pool := &TestBufferPool{ + pool := &fake.FakeBufferPool{ Buffers: [][]byte{make([]byte, 10)}, } w := NewVectoredReadBuffer(pool, tc.maxSize) @@ -498,7 +499,7 @@ func TestVectoredReadBuffer_ZeroOrNegativeMaxSize(t *testing.T) { var globalBuffer *VectoredReadBuffer func BenchmarkNewVectoredReadBuffer(b *testing.B) { - pool := &TestBufferPool{ + pool := &fake.FakeBufferPool{ Buffers: [][]byte{ make([]byte, 100), make([]byte, 100), diff --git a/internal/fs/fs.go b/internal/fs/fs.go index 3951ffe750..bba60f291a 100644 --- a/internal/fs/fs.go +++ b/internal/fs/fs.go @@ -42,6 +42,7 @@ import ( "golang.org/x/sync/semaphore" "github.com/googlecloudplatform/gcsfuse/v3/cfg" + "github.com/googlecloudplatform/gcsfuse/v3/internal/buffer" "github.com/googlecloudplatform/gcsfuse/v3/internal/cache/file" "github.com/googlecloudplatform/gcsfuse/v3/internal/cache/file/downloader" "github.com/googlecloudplatform/gcsfuse/v3/internal/cache/lru" @@ -225,7 +226,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(), + readBufferPool: buffer.NewFixedSizePool(), } // Initialize MRD cache if enabled @@ -679,7 +680,7 @@ type fileSystem struct { mrdCache *lru.Cache // readBufferPool is a pool of readPoolBufferSize (1 MiB) buffers. - readBufferPool gcsx.BufferPool + readBufferPool buffer.Pool } //////////////////////////////////////////////////////////////////////// diff --git a/internal/fs/inode/file.go b/internal/fs/inode/file.go index 2149d58554..321d31f793 100644 --- a/internal/fs/inode/file.go +++ b/internal/fs/inode/file.go @@ -26,6 +26,7 @@ import ( "github.com/googlecloudplatform/gcsfuse/v3/cfg" "github.com/googlecloudplatform/gcsfuse/v3/internal/block" + "github.com/googlecloudplatform/gcsfuse/v3/internal/buffer" "github.com/googlecloudplatform/gcsfuse/v3/internal/bufferedwrites" "github.com/googlecloudplatform/gcsfuse/v3/internal/cache/lru" "github.com/googlecloudplatform/gcsfuse/v3/internal/contentcache" @@ -664,7 +665,7 @@ func (f *FileInode) Read( var callback func() if req.BufferPool != nil { - vBuf := gcsx.NewVectoredReadBuffer(req.BufferPool, req.Size) + vBuf := buffer.NewVectoredReadBuffer(req.BufferPool, req.Size) var written int64 written, err = vBuf.ReadFromAt(f.content, req.Offset) n = int(written) diff --git a/internal/fs/inode/file_test.go b/internal/fs/inode/file_test.go index 57e7b5b387..f9efda9130 100644 --- a/internal/fs/inode/file_test.go +++ b/internal/fs/inode/file_test.go @@ -570,7 +570,7 @@ func (t *FileTest) TestReadWithBufferPool() { for _, tc := range testCases { t.Run(tc.name, func() { // Arrange - pool := &gcsx.TestBufferPool{ + pool := &fake.FakeBufferPool{ Buffers: [][]byte{make([]byte, 2), make([]byte, 2)}, DefaultBufferSize: 2, } diff --git a/internal/gcsx/kernel_readers/kernel_range_reader.go b/internal/gcsx/kernel_readers/kernel_range_reader.go index 1bea4d5505..82e0cff08b 100644 --- a/internal/gcsx/kernel_readers/kernel_range_reader.go +++ b/internal/gcsx/kernel_readers/kernel_range_reader.go @@ -20,6 +20,7 @@ import ( "fmt" "io" + "github.com/googlecloudplatform/gcsfuse/v3/internal/buffer" "github.com/googlecloudplatform/gcsfuse/v3/internal/fs/gcsfuse_errors" "github.com/googlecloudplatform/gcsfuse/v3/internal/gcsx" "github.com/googlecloudplatform/gcsfuse/v3/internal/logger" @@ -112,7 +113,7 @@ func (krr *KernelRangeReader) ReadAt(ctx context.Context, req *gcsx.ReadRequest) var n int if req.BufferPool != nil { - vBuf := gcsx.NewVectoredReadBuffer(req.BufferPool, bytesToRead) + vBuf := buffer.NewVectoredReadBuffer(req.BufferPool, bytesToRead) var written int64 written, err = io.CopyN(vBuf, reader, bytesToRead) n = int(written) diff --git a/internal/gcsx/kernel_readers/kernel_range_reader_test.go b/internal/gcsx/kernel_readers/kernel_range_reader_test.go index b443958ca5..b59a0fe434 100644 --- a/internal/gcsx/kernel_readers/kernel_range_reader_test.go +++ b/internal/gcsx/kernel_readers/kernel_range_reader_test.go @@ -25,6 +25,7 @@ import ( "github.com/googlecloudplatform/gcsfuse/v3/internal/fs/gcsfuse_errors" "github.com/googlecloudplatform/gcsfuse/v3/internal/gcsx" "github.com/googlecloudplatform/gcsfuse/v3/internal/storage" + "github.com/googlecloudplatform/gcsfuse/v3/internal/storage/fake" "github.com/googlecloudplatform/gcsfuse/v3/internal/storage/gcs" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/mock" @@ -175,7 +176,7 @@ func (t *KernelRangeReaderTest) TestReadAt_ClobberedError() { func (t *KernelRangeReaderTest) TestReadAt_BufferPool_Success() { data := []byte("abcdefghij") // length 10 - pool := &gcsx.TestBufferPool{ + pool := &fake.FakeBufferPool{ Buffers: [][]byte{ make([]byte, 3), make([]byte, 2), diff --git a/internal/gcsx/reader.go b/internal/gcsx/reader.go index 1bd79cf152..df142ab99a 100644 --- a/internal/gcsx/reader.go +++ b/internal/gcsx/reader.go @@ -18,6 +18,7 @@ import ( "context" "errors" + "github.com/googlecloudplatform/gcsfuse/v3/internal/buffer" "github.com/googlecloudplatform/gcsfuse/v3/internal/storage/gcs" ) @@ -33,7 +34,7 @@ type ReadRequest struct { // 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 + BufferPool buffer.Pool // 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 diff --git a/internal/gcsx/test_buffer_pool.go b/internal/storage/fake/fake_buffer_pool.go similarity index 85% rename from internal/gcsx/test_buffer_pool.go rename to internal/storage/fake/fake_buffer_pool.go index 64117aa4bb..ed88833687 100644 --- a/internal/gcsx/test_buffer_pool.go +++ b/internal/storage/fake/fake_buffer_pool.go @@ -12,10 +12,10 @@ // See the License for the specific language governing permissions and // limitations under the License. -package gcsx +package fake -// TestBufferPool is a generic mock implementation of BufferPool used for testing across packages. -type TestBufferPool struct { +// FakeBufferPool is a generic fake implementation of buffer.Pool used for testing across packages. +type FakeBufferPool struct { Buffers [][]byte Idx int PutBuffers [][]byte @@ -26,7 +26,7 @@ type TestBufferPool struct { // 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 { +func (p *FakeBufferPool) Get() []byte { if p.Idx < len(p.Buffers) { b := p.Buffers[p.Idx] p.Idx++ @@ -43,6 +43,6 @@ func (p *TestBufferPool) Get() []byte { } // Put records the returned buffer in PutBuffers. -func (p *TestBufferPool) Put(b []byte) { +func (p *FakeBufferPool) Put(b []byte) { p.PutBuffers = append(p.PutBuffers, b) } From e6a06252bc65eb59885eba040ab42543bf935109 Mon Sep 17 00:00:00 2001 From: Abhishek Gupta Date: Mon, 13 Jul 2026 05:46:43 +0000 Subject: [PATCH 11/11] Moving the fake buffer pool implementation to it's correct package --- .../fake => buffer}/fake_buffer_pool.go | 4 ++-- internal/buffer/vectored_read_buffer_test.go | 23 +++++++++---------- internal/fs/inode/file_test.go | 3 ++- .../kernel_range_reader_test.go | 4 ++-- 4 files changed, 17 insertions(+), 17 deletions(-) rename internal/{storage/fake => buffer}/fake_buffer_pool.go (92%) diff --git a/internal/storage/fake/fake_buffer_pool.go b/internal/buffer/fake_buffer_pool.go similarity index 92% rename from internal/storage/fake/fake_buffer_pool.go rename to internal/buffer/fake_buffer_pool.go index ed88833687..8fbfb78575 100644 --- a/internal/storage/fake/fake_buffer_pool.go +++ b/internal/buffer/fake_buffer_pool.go @@ -12,9 +12,9 @@ // See the License for the specific language governing permissions and // limitations under the License. -package fake +package buffer -// FakeBufferPool is a generic fake implementation of buffer.Pool used for testing across packages. +// FakeBufferPool is a generic fake implementation of Pool used for testing across packages. type FakeBufferPool struct { Buffers [][]byte Idx int diff --git a/internal/buffer/vectored_read_buffer_test.go b/internal/buffer/vectored_read_buffer_test.go index 189acd2203..afc6cf923f 100644 --- a/internal/buffer/vectored_read_buffer_test.go +++ b/internal/buffer/vectored_read_buffer_test.go @@ -20,7 +20,6 @@ import ( "io" "testing" - "github.com/googlecloudplatform/gcsfuse/v3/internal/storage/fake" "github.com/stretchr/testify/assert" ) @@ -104,7 +103,7 @@ func TestVectoredReadBuffer_Write(t *testing.T) { } var pool Pool if len(tc.buffers) > 0 { - pool = &fake.FakeBufferPool{Buffers: tc.buffers} + pool = &FakeBufferPool{Buffers: tc.buffers} } w := NewVectoredReadBuffer(pool, maxSize) @@ -217,7 +216,7 @@ func TestVectoredReadBuffer_ReadFrom(t *testing.T) { } var pool Pool if len(tc.buffers) > 0 { - pool = &fake.FakeBufferPool{Buffers: tc.buffers} + pool = &FakeBufferPool{Buffers: tc.buffers} } w := NewVectoredReadBuffer(pool, maxSize) @@ -305,7 +304,7 @@ func TestVectoredReadBuffer_ReadFromAt(t *testing.T) { for _, tc := range tests { t.Run(tc.name, func(t *testing.T) { // Arrange - pool := &fake.FakeBufferPool{Buffers: tc.buffers} + pool := &FakeBufferPool{Buffers: tc.buffers} w := NewVectoredReadBuffer(pool, tc.maxSize) // Act @@ -322,7 +321,7 @@ func TestVectoredReadBuffer_ReadFromAt(t *testing.T) { func TestVectoredReadBuffer_Release(t *testing.T) { tests := []struct { name string - pool *fake.FakeBufferPool + pool *FakeBufferPool maxSize int64 writeInput []byte releaseCount int @@ -330,7 +329,7 @@ func TestVectoredReadBuffer_Release(t *testing.T) { }{ { name: "release allocated buffers to pool and clear state", - pool: &fake.FakeBufferPool{ + pool: &FakeBufferPool{ Buffers: [][]byte{ make([]byte, 10), make([]byte, 10), @@ -352,7 +351,7 @@ func TestVectoredReadBuffer_Release(t *testing.T) { }, { name: "release with no allocated buffers is safe", - pool: &fake.FakeBufferPool{ + pool: &FakeBufferPool{ Buffers: [][]byte{ make([]byte, 10), }, @@ -364,7 +363,7 @@ func TestVectoredReadBuffer_Release(t *testing.T) { }, { name: "release multiple times is safe and idempotent", - pool: &fake.FakeBufferPool{ + pool: &FakeBufferPool{ Buffers: [][]byte{ make([]byte, 10), }, @@ -404,7 +403,7 @@ func TestVectoredReadBuffer_Release(t *testing.T) { func TestVectoredReadBuffer_OversizedBufferTruncation(t *testing.T) { // Arrange - pool := &fake.FakeBufferPool{ + pool := &FakeBufferPool{ Buffers: [][]byte{ make([]byte, 100), }, @@ -445,7 +444,7 @@ func TestVectoredReadBuffer_PoolExhaustion(t *testing.T) { for _, tc := range tests { t.Run(tc.name, func(t *testing.T) { // Arrange - pool := &fake.FakeBufferPool{ + pool := &FakeBufferPool{ ReturnNilOnExhaustion: true, } w := NewVectoredReadBuffer(pool, 100) @@ -480,7 +479,7 @@ func TestVectoredReadBuffer_ZeroOrNegativeMaxSize(t *testing.T) { for _, tc := range tests { t.Run(tc.name, func(t *testing.T) { // Arrange - pool := &fake.FakeBufferPool{ + pool := &FakeBufferPool{ Buffers: [][]byte{make([]byte, 10)}, } w := NewVectoredReadBuffer(pool, tc.maxSize) @@ -499,7 +498,7 @@ func TestVectoredReadBuffer_ZeroOrNegativeMaxSize(t *testing.T) { var globalBuffer *VectoredReadBuffer func BenchmarkNewVectoredReadBuffer(b *testing.B) { - pool := &fake.FakeBufferPool{ + pool := &FakeBufferPool{ Buffers: [][]byte{ make([]byte, 100), make([]byte, 100), diff --git a/internal/fs/inode/file_test.go b/internal/fs/inode/file_test.go index f9efda9130..50fde5deee 100644 --- a/internal/fs/inode/file_test.go +++ b/internal/fs/inode/file_test.go @@ -26,6 +26,7 @@ import ( "time" "github.com/googlecloudplatform/gcsfuse/v3/cfg" + "github.com/googlecloudplatform/gcsfuse/v3/internal/buffer" "github.com/googlecloudplatform/gcsfuse/v3/internal/contentcache" "github.com/googlecloudplatform/gcsfuse/v3/internal/fs/gcsfuse_errors" "github.com/googlecloudplatform/gcsfuse/v3/internal/gcsx" @@ -570,7 +571,7 @@ func (t *FileTest) TestReadWithBufferPool() { for _, tc := range testCases { t.Run(tc.name, func() { // Arrange - pool := &fake.FakeBufferPool{ + pool := &buffer.FakeBufferPool{ Buffers: [][]byte{make([]byte, 2), make([]byte, 2)}, DefaultBufferSize: 2, } diff --git a/internal/gcsx/kernel_readers/kernel_range_reader_test.go b/internal/gcsx/kernel_readers/kernel_range_reader_test.go index b59a0fe434..d7d008c583 100644 --- a/internal/gcsx/kernel_readers/kernel_range_reader_test.go +++ b/internal/gcsx/kernel_readers/kernel_range_reader_test.go @@ -22,10 +22,10 @@ import ( "testing" storagev2 "cloud.google.com/go/storage" + "github.com/googlecloudplatform/gcsfuse/v3/internal/buffer" "github.com/googlecloudplatform/gcsfuse/v3/internal/fs/gcsfuse_errors" "github.com/googlecloudplatform/gcsfuse/v3/internal/gcsx" "github.com/googlecloudplatform/gcsfuse/v3/internal/storage" - "github.com/googlecloudplatform/gcsfuse/v3/internal/storage/fake" "github.com/googlecloudplatform/gcsfuse/v3/internal/storage/gcs" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/mock" @@ -176,7 +176,7 @@ func (t *KernelRangeReaderTest) TestReadAt_ClobberedError() { func (t *KernelRangeReaderTest) TestReadAt_BufferPool_Success() { data := []byte("abcdefghij") // length 10 - pool := &fake.FakeBufferPool{ + pool := &buffer.FakeBufferPool{ Buffers: [][]byte{ make([]byte, 3), make([]byte, 2),