feat(kernel range reader): Add buffer pooling support for large reads on regional buckets

internal/fs/fs.go

@@ -225,6 +225,7 @@ func NewFileSystem(ctx context.Context, serverCfg *ServerConfig) (fuseutil.FileS
 		globalMaxWriteBlocksSem:    semaphore.NewWeighted(serverCfg.NewConfig.Write.GlobalMaxBlocks),
 		globalMaxReadBlocksSem:     semaphore.NewWeighted(serverCfg.NewConfig.Read.GlobalMaxBlocks),
 		globalMetadataPrefetchSem:  semaphore.NewWeighted(serverCfg.NewConfig.MetadataCache.MetadataPrefetchMaxWorkers),
+		readBufferPool:             gcsx.NewFixedSizeBufferPool(),
 	}
 
 	// Initialize MRD cache if enabled
@@ -676,6 +677,9 @@ type fileSystem struct {
 
 	// mrdCache manages the cache of inactive MultiRangeDownloaders.
 	mrdCache *lru.Cache
+
+	// readBufferPool is a pool of readPoolBufferSize (1 MiB) buffers.
+	readBufferPool gcsx.BufferPool
 }
 
 ////////////////////////////////////////////////////////////////////////
@@ -3118,23 +3122,36 @@ func (fs *fileSystem) ReadFile(
 	}
 	// Serve the read.
 
+	req := gcsx.ReadRequest{
+		Offset: op.Offset,
+		Size:   op.Size,
+	}
+
+	useReadBufferPool := op.Size > 0 && op.Dst == nil
+	if useReadBufferPool {
+		if !fs.newConfig.FileSystem.EnableKernelReader || fh.Inode().Bucket().BucketType().IsRapid() {
+			logger.Errorf("ReadFile: buffer pool allocation is only supported for regional buckets with"+
+				" kernel reader enabled (EnableKernelReader: %v, IsRapid: %v)",
+				fs.newConfig.FileSystem.EnableKernelReader,
+				fh.Inode().Bucket().BucketType().IsRapid())
+			fh.Inode().Unlock()
+			return syscall.ENOTSUP
+		}
+
+		req.BufferPool = fs.readBufferPool
+	} else {
+		req.Buffer = op.Dst
+	}
+
 	if fs.newConfig.FileSystem.EnableKernelReader {
 		var resp gcsx.ReadResponse
-		req := &gcsx.ReadRequest{
-			Buffer: op.Dst,
-			Offset: op.Offset,
-		}
-		resp, err = fh.ReadWithKernelReader(ctx, req)
+		resp, err = fh.ReadWithKernelReader(ctx, &req)
 		op.BytesRead = resp.Size
 		op.Data = resp.Data
 		op.Callback = resp.Callback
 	} else if fs.newConfig.EnableNewReader {
 		var resp gcsx.ReadResponse
-		req := &gcsx.ReadRequest{
-			Buffer: op.Dst,
-			Offset: op.Offset,
-		}
-		resp, err = fh.ReadWithReadManager(ctx, req, fs.sequentialReadSizeMb)
+		resp, err = fh.ReadWithReadManager(ctx, &req, fs.sequentialReadSizeMb)
 		op.BytesRead = resp.Size
 		op.Data = resp.Data
 		op.Callback = resp.Callback

internal/fs/handle/file.go

@@ -209,8 +209,7 @@ func (fh *FileHandle) ReadWithReadManager(ctx context.Context, req *gcsx.ReadReq
 	if !fh.inode.SourceGenerationIsAuthoritative() {
 		// Read from inode if source generation is not authoratative
 		defer fh.inode.Unlock()
-		n, err := fh.inode.Read(ctx, req.Buffer, req.Offset)
-		return gcsx.ReadResponse{Size: n}, err
+		return fh.inode.Read(ctx, req)
 	}
 
 	fh.lockHandleAndRelockInode(true)
@@ -289,8 +288,7 @@ func (fh *FileHandle) ReadWithKernelReader(ctx context.Context, req *gcsx.ReadRe
 	if !fh.inode.SourceGenerationIsAuthoritative() {
 		// Read from inode if source generation is not authoritative.
 		defer fh.inode.Unlock()
-		n, err := fh.inode.Read(ctx, req.Buffer, req.Offset)
-		return gcsx.ReadResponse{Size: n}, err
+		return fh.inode.Read(ctx, req)
 	}
 	fh.inode.Unlock()
 
@@ -322,8 +320,8 @@ func (fh *FileHandle) Read(ctx context.Context, dst []byte, offset int64, sequen
 	// we have one.
 	if !fh.inode.SourceGenerationIsAuthoritative() {
 		defer fh.inode.Unlock()
-		n, err = fh.inode.Read(ctx, dst, offset)
-		return dst, n, err
+		res, err := fh.inode.Read(ctx, &gcsx.ReadRequest{Buffer: dst, Offset: offset})
+		return dst, res.Size, err
 	}
 
 	fh.lockHandleAndRelockInode(true)

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()

internal/fs/inode/file.go

@@ -638,16 +638,15 @@ 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.
 //
 // LOCKS_REQUIRED(f.mu)
 func (f *FileInode) Read(
 	ctx context.Context,
-	dst []byte,
-	offset int64) (n int, err error) {
+	req *gcsx.ReadRequest) (res gcsx.ReadResponse, err error) {
 	if f.bwh != nil {
 		err = fmt.Errorf("unexpected read call for %q when streaming write is in progress for it", f.Name().LocalName())
 		return
@@ -660,18 +659,34 @@ func (f *FileInode) Read(
 		return
 	}
 
-	// Read from the local content, propagating io.EOF.
-	n, err = f.content.ReadAt(dst, offset)
-	switch {
-	case err == io.EOF:
-		return
+	var n int
+	var data [][]byte
+	var callback func()
+
+	if req.BufferPool != nil {
+		vBuf := gcsx.NewVectoredReadBuffer(req.BufferPool, req.Size)
+		var written int64
+		written, err = vBuf.ReadFromAt(f.content, req.Offset)
+		n = int(written)
+		if n > 0 {
+			data = vBuf.Buffers()
+			callback = func() { vBuf.Release() }
+		} else {
+			vBuf.Release()
+		}
+	} else {
+		n, err = f.content.ReadAt(req.Buffer, req.Offset)
+	}
 
-	case err != nil:
+	if err != nil && err != io.EOF {
 		err = fmt.Errorf("content.ReadAt: %w", err)
-		return
 	}
 
-	return
+	return gcsx.ReadResponse{
+		Size:     n,
+		Data:     data,
+		Callback: callback,
+	}, err
 }
 
 func getMetricOpenMode(openMode util.OpenMode) metrics.OpenMode {

internal/fs/inode/file_test.go

@@ -501,8 +501,8 @@ func (t *FileTest) TestRead() {
 	for _, tc := range testCases {
 		desc := fmt.Sprintf("offset: %d, size: %d", tc.offset, tc.size)
 		data := make([]byte, tc.size)
-		n, err := t.in.Read(t.ctx, data, tc.offset)
-		data = data[:n]
+		res, err := t.in.Read(t.ctx, &gcsx.ReadRequest{Buffer: data, Offset: tc.offset})
+		data = data[:res.Size]
 
 		// Ignore EOF.
 		if err == io.EOF {
@@ -514,6 +514,94 @@ func (t *FileTest) TestRead() {
 	}
 }
 
+func (t *FileTest) TestReadWithBufferPool() {
+	assert.Equal(t.T(), "taco", t.initialContents)
+
+	testCases := []struct {
+		name         string
+		offset       int64
+		size         int64
+		expectedData []string
+		expectedSize int
+		expectedErr  error
+	}{
+		{
+			name:         "ExactFit_Full",
+			offset:       0,
+			size:         4,
+			expectedData: []string{"ta", "co"},
+			expectedSize: 4,
+			expectedErr:  nil,
+		},
+		{
+			name:         "ExactFit_Partial",
+			offset:       0,
+			size:         2,
+			expectedData: []string{"ta"},
+			expectedSize: 2,
+			expectedErr:  nil,
+		},
+		{
+			name:         "ExactFit_Remaining",
+			offset:       1,
+			size:         3,
+			expectedData: []string{"ac", "o"},
+			expectedSize: 3,
+			expectedErr:  nil,
+		},
+		{
+			name:         "PastEOF",
+			offset:       3,
+			size:         2,
+			expectedData: []string{"o"},
+			expectedSize: 1,
+			expectedErr:  nil,
+		},
+		{
+			name:         "AtEOF",
+			offset:       4,
+			size:         2,
+			expectedData: []string{},
+			expectedSize: 0,
+			expectedErr:  nil,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func() {
+			// Arrange
+			pool := &gcsx.TestBufferPool{
+				Buffers:           [][]byte{make([]byte, 2), make([]byte, 2)},
+				DefaultBufferSize: 2,
+			}
+			req := &gcsx.ReadRequest{
+				BufferPool: pool,
+				Offset:     tc.offset,
+				Size:       tc.size,
+			}
+
+			// Act
+			res, err := t.in.Read(t.ctx, req)
+
+			// Assert
+			assert.ErrorIs(t.T(), err, tc.expectedErr)
+			assert.Equal(t.T(), tc.expectedSize, res.Size)
+			if res.Size > 0 {
+				var actualStrings []string
+				for _, b := range res.Data {
+					actualStrings = append(actualStrings, string(b))
+				}
+				assert.Equal(t.T(), tc.expectedData, actualStrings)
+				require.NotNil(t.T(), res.Callback)
+				res.Callback()
+			} else {
+				assert.Empty(t.T(), res.Data)
+				assert.Nil(t.T(), res.Callback)
+			}
+		})
+	}
+}
+
 func (t *FileTest) TestWrite() {
 	var err error
 
@@ -536,14 +624,14 @@ func (t *FileTest) TestWrite() {
 
 	// Read back the content.
 	var buf [1024]byte
-	n, err := t.in.Read(t.ctx, buf[:], 0)
+	res, err := t.in.Read(t.ctx, &gcsx.ReadRequest{Buffer: buf[:], Offset: 0})
 
 	if err == io.EOF {
 		err = nil
 	}
 
 	assert.Nil(t.T(), err)
-	assert.Equal(t.T(), "pacoburrito", string(buf[:n]))
+	assert.Equal(t.T(), "pacoburrito", string(buf[:res.Size]))
 
 	// Check attributes.
 	attrs, err := t.in.Attributes(t.ctx, true)
@@ -571,14 +659,14 @@ func (t *FileTest) TestTruncate() {
 
 	// Read the contents.
 	var buf [1024]byte
-	n, err := t.in.Read(t.ctx, buf[:], 0)
+	res, err := t.in.Read(t.ctx, &gcsx.ReadRequest{Buffer: buf[:], Offset: 0})
 
 	if err == io.EOF {
 		err = nil
 	}
 
 	assert.Nil(t.T(), err)
-	assert.Equal(t.T(), "ta", string(buf[:n]))
+	assert.Equal(t.T(), "ta", string(buf[:res.Size]))
 
 	// Check attributes.
 	attrs, err = t.in.Attributes(t.ctx, true)
@@ -1426,7 +1514,7 @@ func (t *FileTest) TestSetMtime_ContentClean() {
 	var attrs fuseops.InodeAttributes
 
 	// Cause the content to be faulted in.
-	_, err = t.in.Read(t.ctx, make([]byte, 1), 0)
+	_, err = t.in.Read(t.ctx, &gcsx.ReadRequest{Buffer: make([]byte, 1), Offset: 0})
 	assert.Nil(t.T(), err)
 
 	// Set mtime.
@@ -1780,13 +1868,13 @@ func (t *FileTest) TestReadFileWhenStreamingWritesAreEnabled() {
 			// Flush is required before reading an object for which BWH is open.
 			assert.NoError(t.T(), t.in.Flush(context.Background()))
 
-			n, err := t.in.Read(t.ctx, data, 0)
+			res, err := t.in.Read(t.ctx, &gcsx.ReadRequest{Buffer: data, Offset: 0})
 
 			if tc.performWrite {
-				assert.Equal(t.T(), len(data), n)
+				assert.Equal(t.T(), len(data), res.Size)
 				require.NoError(t.T(), err)
 			} else {
-				assert.Equal(t.T(), 0, n)
+				assert.Equal(t.T(), 0, res.Size)
 				require.Error(t.T(), err)
 				assert.ErrorIs(t.T(), err, io.EOF)
 			}
@@ -1799,9 +1887,9 @@ func (t *FileTest) TestReadEmptyGCSFileWhenStreamingWritesAreNotInProgress() {
 	t.in.config = &cfg.Config{Write: *getWriteConfig()}
 	data := make([]byte, 10)
 
-	n, err := t.in.Read(t.ctx, data, 0)
+	res, err := t.in.Read(t.ctx, &gcsx.ReadRequest{Buffer: data, Offset: 0})
 
-	assert.Equal(t.T(), 0, n)
+	assert.Equal(t.T(), 0, res.Size)
 	require.Error(t.T(), err)
 	assert.Contains(t.T(), err.Error(), "EOF")
 }