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

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() interface{} {
+				return make([]byte, readPoolBufferSize)
+			},
+		},
 	}
 
 	// Initialize MRD cache if enabled
@@ -676,6 +685,9 @@ 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
 }
 
 ////////////////////////////////////////////////////////////////////////
@@ -3118,23 +3130,74 @@ func (fs *fileSystem) ReadFile(
 	}
 	// Serve the read.
 
+	req := gcsx.ReadRequest{
+		Offset: op.Offset,
+		Size:   op.Size,
+	}
+
+	useReadBufferPool := 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
+		}
+
+		var buffers [][]byte
+		remaining := op.Size
+		for remaining > 0 {
+			bufVal := fs.readBufferPool.Get()
+			poolBuf := bufVal.([]byte)
+			size := min(remaining, readPoolBufferSize)
+			buffers = append(buffers, poolBuf[:size])
+			remaining -= size
+		}
+		req.Buffers = buffers
+	} 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
+		if useReadBufferPool {
+			if len(resp.Data) > 0 {
+				op.Data = resp.Data
+				// Zero-copy hit: we did not use the pool buffers at all. Return all of them immediately.
+				for _, buf := range req.Buffers {
+					fs.readBufferPool.Put(buf[:cap(buf)])
+				}
+				req.Buffers = nil
+			} else {
+				op.Data = util.LimitBuffers(req.Buffers, resp.Size)
+				// Return unused buffers (from index len(op.Data) onwards) back to the pool immediately.
+				usedCount := len(op.Data)
+				for j := usedCount; j < len(req.Buffers); j++ {
+					fs.readBufferPool.Put(req.Buffers[j][:cap(req.Buffers[j])])
+				}
+				// Shrink req.Buffers so the callback only returns the used ones.
+				req.Buffers = req.Buffers[:usedCount]
+			}
+
+			op.Callback = func() {
+				if resp.Callback != nil {
+					resp.Callback()
+				}
+				for _, buf := range req.Buffers {
+					fs.readBufferPool.Put(buf[:cap(buf)])
+				}
+			}
+		} else {
+			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/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)
+}

internal/util/util.go

@@ -96,3 +96,21 @@ func BytesToHigherMiBs(bytes uint64) uint64 {
 	// Use integer arithmetic and bitwise shift to avoid slow float conversions
 	return (bytes + bytesInOneMiB - 1) >> 20
 }
+
+// LimitBuffers reslices the buffers slice in-place to represent the first 'limit'
+// bytes. It avoids allocating a new slice of slices by modifying the slice elements
+// in-place and returning a subslice of the original buffers slice.
+func LimitBuffers(buffers [][]byte, limit int) [][]byte {
+	if limit <= 0 {
+		return nil
+	}
+	var current int
+	for i, b := range buffers {
+		if current+len(b) >= limit {
+			buffers[i] = b[:limit-current]
+			return buffers[:i+1]
+		}
+		current += len(b)
+	}
+	return buffers
+}

internal/util/util_test.go

@@ -213,3 +213,57 @@ func (ts *UtilTest) TestBytesToHigherMiBs() {
 		assert.Equal(ts.T(), tc.mib, BytesToHigherMiBs(tc.bytes))
 	}
 }
+
+func (ts *UtilTest) TestLimitBuffers() {
+	cases := []struct {
+		name     string
+		buffers  [][]byte
+		limit    int
+		expected [][]byte
+	}{
+		{
+			name:     "limit zero",
+			buffers:  [][]byte{[]byte("hello"), []byte("world")},
+			limit:    0,
+			expected: nil,
+		},
+		{
+			name:     "limit negative",
+			buffers:  [][]byte{[]byte("hello"), []byte("world")},
+			limit:    -5,
+			expected: nil,
+		},
+		{
+			name:     "limit within first buffer",
+			buffers:  [][]byte{[]byte("hello"), []byte("world")},
+			limit:    3,
+			expected: [][]byte{[]byte("hel")},
+		},
+		{
+			name:     "limit at buffer boundary",
+			buffers:  [][]byte{[]byte("hello"), []byte("world")},
+			limit:    5,
+			expected: [][]byte{[]byte("hello")},
+		},
+		{
+			name:     "limit spanning multiple buffers",
+			buffers:  [][]byte{[]byte("hello"), []byte("world")},
+			limit:    8,
+			expected: [][]byte{[]byte("hello"), []byte("wor")},
+		},
+		{
+			name:     "limit greater than total length",
+			buffers:  [][]byte{[]byte("hello"), []byte("world")},
+			limit:    20,
+			expected: [][]byte{[]byte("hello"), []byte("world")},
+		},
+	}
+
+	for _, tc := range cases {
+		ts.Run(tc.name, func() {
+			result := LimitBuffers(tc.buffers, tc.limit)
+
+			assert.Equal(ts.T(), tc.expected, result)
+		})
+	}
+}

tools/cd_scripts/e2e_test.sh

@@ -426,7 +426,7 @@ TEST_DIR_PARALLEL_ZONAL=(
   mounting
   negative_stat_cache
   operations
-  rapid_appends
+  rapid_operations
   read_large_files
   # Reenable when b/461334834 is done.
   # readdirplus

tools/integration_tests/cloud_profiler/cloud_profiler_test.go

@@ -80,20 +80,7 @@ func TestMain(m *testing.M) {
 	// 1. Load and parse the common configuration.
 	cfg := test_suite.ReadConfigFile(setup.ConfigFile())
 	if len(cfg.CloudProfiler) == 0 {
-		log.Println("No configuration found for cloud profiler tests in config. Using flags instead.")
-
-		// Populate the config manually.
-		cfg.CloudProfiler = make([]test_suite.TestConfig, 1)
-		cfg.CloudProfiler[0].TestBucket = setup.TestBucket()
-		cfg.CloudProfiler[0].GKEMountedDirectory = setup.MountedDirectory()
-		cfg.CloudProfiler[0].Configs = make([]test_suite.ConfigItem, 1)
-		cfg.CloudProfiler[0].Configs[0].Flags = []string{
-			"--enable-cloud-profiler --cloud-profiler-cpu --cloud-profiler-heap --cloud-profiler-goroutines --cloud-profiler-mutex --cloud-profiler-allocated-heap",
-		}
-		testVersionFlag := fmt.Sprintf(" --cloud-profiler-label=%s", testVersionName)
-		testServiceNameFlag := fmt.Sprintf(" --cloud-profiler-service-name=%s", testServiceName)
-		cfg.CloudProfiler[0].Configs[0].Flags[0] = cfg.CloudProfiler[0].Configs[0].Flags[0] + testVersionFlag + testServiceNameFlag
-		cfg.CloudProfiler[0].Configs[0].Compatible = map[string]bool{"flat": true, "hns": true, "zonal": true}
+		log.Fatal("No configuration found for CloudProfiler in config file.")
 	} else if cfg.CloudProfiler[0].GKEMountedDirectory == "" {
 		if len(cfg.CloudProfiler[0].Configs[0].Flags) != 1 {
 			log.Fatalf("Expected exactly 1 config flag, got %d", len(cfg.CloudProfiler[0].Configs[0].Flags))

tools/integration_tests/improved_run_e2e_tests.sh

@@ -385,7 +385,7 @@ filter_array() {
 # Test packages for regional buckets.
 TEST_PACKAGES_FOR_RB=("${TEST_PACKAGES_COMMON[@]}" "inactive_stream_timeout" "cloud_profiler" "requester_pays_bucket")
 # Test packages for zonal buckets.
-TEST_PACKAGES_FOR_ZB=("${TEST_PACKAGES_COMMON[@]}" "rapid_appends" "unfinalized_object")
+TEST_PACKAGES_FOR_ZB=("${TEST_PACKAGES_COMMON[@]}" "rapid_operations" "unfinalized_object")
 # Test packages for TPC buckets.
 TEST_PACKAGES_FOR_TPC=("operations")