From 2e9a36e750653b39e9bb7ffab5379b0e79dd3844 Mon Sep 17 00:00:00 2001 From: Abhishek Gupta Date: Tue, 7 Jul 2026 05:35:49 +0000 Subject: [PATCH 1/4] Changes to use a sync pool to allocate buffers for larger reads --- internal/fs/fs.go | 87 ++++++++++++--- internal/fs/large_read_regional_test.go | 134 ++++++++++++++++++++++++ internal/util/util.go | 18 ++++ internal/util/util_test.go | 54 ++++++++++ 4 files changed, 281 insertions(+), 12 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..917ca09585 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() 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 diff --git a/internal/fs/large_read_regional_test.go b/internal/fs/large_read_regional_test.go new file mode 100644 index 0000000000..6b9d478200 --- /dev/null +++ b/internal/fs/large_read_regional_test.go @@ -0,0 +1,134 @@ +// Copyright 2026 Google LLC +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package fs_test + +import ( + "context" + "syscall" + "testing" + + "github.com/googlecloudplatform/gcsfuse/v3/cfg" + "github.com/googlecloudplatform/gcsfuse/v3/internal/fs" + "github.com/googlecloudplatform/gcsfuse/v3/internal/storage/fake" + "github.com/googlecloudplatform/gcsfuse/v3/internal/storage/gcs" + "github.com/googlecloudplatform/gcsfuse/v3/metrics" + "github.com/googlecloudplatform/gcsfuse/v3/tracing" + "github.com/jacobsa/fuse/fuseops" + "github.com/jacobsa/fuse/fuseutil" + "github.com/jacobsa/timeutil" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func setupLargeRead(ctx context.Context, t *testing.T, isZonal bool, enableKernelReader bool, fileContent string) (fuseutil.FileSystem, *fuseops.ReadFileOp) { + t.Helper() + bucketName := "regional-bucket" + if isZonal { + bucketName = "zonal-bucket" + } + bucket := fake.NewFakeBucket(timeutil.RealClock(), bucketName, gcs.BucketType{Zonal: isZonal, Hierarchical: false}) + + serverCfg := &fs.ServerConfig{ + NewConfig: &cfg.Config{ + Write: cfg.WriteConfig{ + GlobalMaxBlocks: 1, + }, + FileSystem: cfg.FileSystemConfig{ + EnableKernelReader: enableKernelReader, + }, + }, + CacheClock: &timeutil.SimulatedClock{}, + BucketName: bucketName, + BucketManager: &fakeBucketManager{ + buckets: map[string]gcs.Bucket{ + bucketName: bucket, + }, + }, + SequentialReadSizeMb: 200, + TraceHandle: tracing.NewOTELTracer(), + MetricHandle: metrics.NewNoopMetrics(), + } + + server, err := fs.NewFileSystem(ctx, serverCfg) + require.NoError(t, err) + + fileName := "large-file" + createWithContents(ctx, t, bucket, fileName, fileContent) + lookUpOp := &fuseops.LookUpInodeOp{ + Parent: fuseops.RootInodeID, + Name: fileName, + } + err = server.LookUpInode(ctx, lookUpOp) + require.NoError(t, err) + + openOp := &fuseops.OpenFileOp{ + Inode: lookUpOp.Entry.Child, + } + err = server.OpenFile(ctx, openOp) + require.NoError(t, err) + + readOp := &fuseops.ReadFileOp{ + Inode: lookUpOp.Entry.Child, + Handle: openOp.Handle, + Offset: 0, + Size: int64(len(fileContent)), + Dst: nil, + } + + return server, readOp +} + +func TestLargeReadRegional_Success(t *testing.T) { + // Arrange + ctx := context.Background() + fileContent := "regional bucket content for vectored read verification." + server, readOp := setupLargeRead(ctx, t, false, true, fileContent) + + // Act + err := server.ReadFile(ctx, readOp) + + // Assert + require.NoError(t, err) + assert.Equal(t, len(fileContent), readOp.BytesRead) + require.Len(t, readOp.Data, 1) + assert.Equal(t, []byte(fileContent), readOp.Data[0]) + if readOp.Callback != nil { + readOp.Callback() + } +} + +func TestLargeReadRegional_NotSupportedForZonal(t *testing.T) { + // Arrange + ctx := context.Background() + server, readOp := setupLargeRead(ctx, t, true, true, "zonal bucket content.") + + // Act + err := server.ReadFile(ctx, readOp) + + // Assert + assert.ErrorIs(t, err, syscall.ENOTSUP) +} + +func TestLargeReadRegional_NotSupportedWhenKernelReaderDisabled(t *testing.T) { + // Arrange + ctx := context.Background() + server, readOp := setupLargeRead(ctx, t, false, false, "regional bucket content.") + + // Act + err := server.ReadFile(ctx, readOp) + + // Assert + assert.ErrorIs(t, err, syscall.ENOTSUP) +} diff --git a/internal/util/util.go b/internal/util/util.go index 74164b836e..96812d2150 100644 --- a/internal/util/util.go +++ b/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 +} diff --git a/internal/util/util_test.go b/internal/util/util_test.go index c30363e0f4..c8f4fc2804 100644 --- a/internal/util/util_test.go +++ b/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) + }) + } +} From c314677e884c22e24592b0098e0e41245d9e7fad Mon Sep 17 00:00:00 2001 From: Abhishek Gupta Date: Tue, 7 Jul 2026 07:42:14 +0000 Subject: [PATCH 2/4] Rough changes to enable krr --- cfg/config_util.go | 14 +++++++ cmd/mount.go | 14 +++++++ go.mod | 2 +- go.sum | 4 +- internal/fs/fs.go | 60 +++++++++++++++++++++++++++ internal/kernelparams/kernelparams.go | 7 ++++ 6 files changed, 98 insertions(+), 3 deletions(-) diff --git a/cfg/config_util.go b/cfg/config_util.go index 5f8e3e3241..3c682c8860 100644 --- a/cfg/config_util.go +++ b/cfg/config_util.go @@ -41,11 +41,25 @@ func DefaultMaxBackground() int { return min(max(12, 2*runtime.NumCPU()), maxBackgroundLimit) } +func DefaultMaxBackgroundForNonRapid() int { + return 96 +} + func DefaultCongestionThreshold() int { // 75 % of DefaultMaxBackground return (3 * DefaultMaxBackground()) / 4 } +func DefaultCongestionThresholdForNonRapid() int { + // 75 % of DefaultMaxBackgroundForNonRapid + return (3 * DefaultMaxBackgroundForNonRapid()) / 4 +} + +func DefaultMaxReadAheadKbForNonRapid() int { + // 128 Kib + return 128 * 1024 +} + func DefaultMaxParallelDownloads() int { return max(16, 2*runtime.NumCPU()) } diff --git a/cmd/mount.go b/cmd/mount.go index 8745d76811..51ae076d7d 100644 --- a/cmd/mount.go +++ b/cmd/mount.go @@ -20,6 +20,7 @@ import ( "time" "github.com/googlecloudplatform/gcsfuse/v3/cfg" + "github.com/googlecloudplatform/gcsfuse/v3/internal/kernelparams" "github.com/googlecloudplatform/gcsfuse/v3/internal/mount" "github.com/googlecloudplatform/gcsfuse/v3/internal/storage" "github.com/googlecloudplatform/gcsfuse/v3/metrics" @@ -146,6 +147,15 @@ be interacting with the file system.`) fsName := fsName(bucketName) + // Apply pre mount kernel settings in non-GKE environments for non dynamic mounts when kernel reader is enabled. + if !isDynamicMount(bucketName) && !cfg.IsGKEEnvironment(mountPoint) && newConfig.FileSystem.EnableKernelReader { + kernelparamsManager := kernelparams.NewKernelParamsManager() + if kernelparams.ShouldUpdateMaxPagesLimit(int(newConfig.FileSystem.FuseMaxPagesLimit)) { + kernelparamsManager.SetMaxPagesLimit(int(newConfig.FileSystem.FuseMaxPagesLimit)) + kernelparamsManager.ApplyNonGKE(mountPoint) + } + } + // Mount the file system. logger.Infof("Mounting file system %q...", fsName) @@ -189,6 +199,10 @@ func getFuseMountConfig(fsName string, newConfig *cfg.Config) *fuse.MountConfig EnableAsyncReads: newConfig.FileSystem.EnableKernelReader, } + if newConfig.FileSystem.FuseMaxPagesLimit != 0 { + mountCfg.MaxPages = uint16(newConfig.FileSystem.FuseMaxPagesLimit) + } + if newConfig.Logging.WireLog != "" { wireLog, err := os.Create(string(newConfig.Logging.WireLog)) if err == nil { diff --git a/go.mod b/go.mod index 728f2abb27..9c0b5e5ad8 100644 --- a/go.mod +++ b/go.mod @@ -17,7 +17,7 @@ require ( github.com/google/uuid v1.6.0 github.com/googleapis/gax-go/v2 v2.22.0 github.com/jacobsa/daemonize v0.0.0-20240917082746-f35568b6c3ec - github.com/jacobsa/fuse v0.0.0-20260302145937-f1ba38d60fdf + github.com/jacobsa/fuse v0.0.0-20260630194014-a124548f6da7 github.com/jacobsa/oglematchers v0.0.0-20150720000706-141901ea67cd github.com/jacobsa/oglemock v0.0.0-20150831005832-e94d794d06ff github.com/jacobsa/ogletest v0.0.0-20170503003838-80d50a735a11 diff --git a/go.sum b/go.sum index ccdf39387f..ade4575450 100644 --- a/go.sum +++ b/go.sum @@ -134,8 +134,8 @@ github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2 github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw= github.com/jacobsa/daemonize v0.0.0-20240917082746-f35568b6c3ec h1:xsRGrfdnjvJtEMD2ouh8gOGIeDF9LrgXjo+9Q69RVzI= github.com/jacobsa/daemonize v0.0.0-20240917082746-f35568b6c3ec/go.mod h1:Ip4fOwzCrnDVuluHBd7FXIMb7SHOKfkt9/UDrYSZvqI= -github.com/jacobsa/fuse v0.0.0-20260302145937-f1ba38d60fdf h1:1FpPcJSf6jjJGvIltaLwJCpbFCMI9bVUCAAxUSxqWnY= -github.com/jacobsa/fuse v0.0.0-20260302145937-f1ba38d60fdf/go.mod h1:fcpw1yk/suvFhB8rT9P+pst+NLboWsBLky9csooKjPc= +github.com/jacobsa/fuse v0.0.0-20260630194014-a124548f6da7 h1:Vk7KHtuE6XOWJ5Qfnx3rQnXqPIPOdG5LXCbL95IbxOw= +github.com/jacobsa/fuse v0.0.0-20260630194014-a124548f6da7/go.mod h1:fcpw1yk/suvFhB8rT9P+pst+NLboWsBLky9csooKjPc= github.com/jacobsa/oglematchers v0.0.0-20150720000706-141901ea67cd h1:9GCSedGjMcLZCrusBZuo4tyKLpKUPenUUqi34AkuFmA= github.com/jacobsa/oglematchers v0.0.0-20150720000706-141901ea67cd/go.mod h1:TlmyIZDpGmwRoTWiakdr+HA1Tukze6C6XbRVidYq02M= github.com/jacobsa/oglemock v0.0.0-20150831005832-e94d794d06ff h1:2xRHTvkpJ5zJmglXLRqHiZQNjUoOkhUyhTAhEQvPAWw= diff --git a/internal/fs/fs.go b/internal/fs/fs.go index 917ca09585..9f9362ca24 100644 --- a/internal/fs/fs.go +++ b/internal/fs/fs.go @@ -158,6 +158,60 @@ type ServerConfig struct { Notifier *fuse.Notifier } +func shouldEnableKernelReaderForNonRapidBuckets(serverCfg *ServerConfig, bucketType gcs.BucketType) bool { + if bucketType.IsRapid() { + return false + } + + // Do not enable if file cache or buffered read are enabled explicitly by user + // or if kernel does not support increasing max pages limit beyond 1MiB. + isFileCacheSet := serverCfg.ViperConfig.IsSet("cache-dir") + isBufferedReadSet := serverCfg.ViperConfig.IsSet("read.enable-buffered-read") + return /*serverCfg.NewConfig.Profile == "aiml-serving" &&*/ !isFileCacheSet && !isBufferedReadSet && kernelparams.SupportsFuseMaxPagesLimit() +} + +func applyRegionalKernelReaderOptimizations(serverCfg *ServerConfig, bucketType gcs.BucketType, optimizedFlags map[string]cfg.OptimizationResult) { + if serverCfg.NewConfig.DisableAutoconfig || bucketType.IsRapid() { + return + } + + // 1. Automatically enable kernel reader for non-rapid buckets if appropriate. + if !serverCfg.ViperConfig.IsSet("file-system.enable-kernel-reader") && !optimizedFlags["file-system.enable-kernel-reader"].Optimized { + if shouldEnableKernelReaderForNonRapidBuckets(serverCfg, bucketType) { + serverCfg.NewConfig.FileSystem.EnableKernelReader = true + optimizedFlags["file-system.enable-kernel-reader"] = cfg.OptimizationResult{ + Optimized: true, + FinalValue: true, + } + logger.Info("GCSFuse Config: Automatically enabled enable-kernel-reader for aiml-serving profile on flat/hns bucket.") + } + } + + // 2. If kernel reader is enabled, apply parameter optimizations cleanly using a table-driven loop. + if serverCfg.NewConfig.FileSystem.EnableKernelReader { + type flagOpt struct { + name string + val int64 + setter func(int64) + } + opts := []flagOpt{ + {"file-system.max-read-ahead-kb", int64(cfg.DefaultMaxReadAheadKbForNonRapid()), func(v int64) { serverCfg.NewConfig.FileSystem.MaxReadAheadKb = v }}, + {"file-system.max-background", int64(cfg.DefaultMaxBackgroundForNonRapid()), func(v int64) { serverCfg.NewConfig.FileSystem.MaxBackground = v }}, + {"file-system.congestion-threshold", int64(cfg.DefaultCongestionThresholdForNonRapid()), func(v int64) { serverCfg.NewConfig.FileSystem.CongestionThreshold = v }}, + {"file-system.fuse-max-pages-limit", 16 * gcsx.MiB / int64(os.Getpagesize()), func(v int64) { serverCfg.NewConfig.FileSystem.FuseMaxPagesLimit = v }}, + } + for _, opt := range opts { + if !serverCfg.ViperConfig.IsSet(opt.name) && !optimizedFlags[opt.name].Optimized { + opt.setter(opt.val) + optimizedFlags[opt.name] = cfg.OptimizationResult{ + Optimized: true, + FinalValue: opt.val, + } + } + } + } +} + // Create a fuse file system server according to the supplied configuration. func NewFileSystem(ctx context.Context, serverCfg *ServerConfig) (fuseutil.FileSystem, error) { // Check permissions bits. @@ -280,6 +334,12 @@ func NewFileSystem(ctx context.Context, serverCfg *ServerConfig) (fuseutil.FileS optimizedFlags := serverCfg.NewConfig.ApplyOptimizations(serverCfg.ViperConfig, &cfg.OptimizationInput{ BucketType: bucketTypeEnum, }) + if optimizedFlags == nil { + optimizedFlags = make(map[string]cfg.OptimizationResult) + } + + applyRegionalKernelReaderOptimizations(serverCfg, bucketType, optimizedFlags) + if len(optimizedFlags) > 0 { logger.Info("GCSFuse Config", "Applied optimizations for bucket-type: ", bucketTypeEnum, "Full Config", optimizedFlags) optimizedFlagNames := slices.Collect(maps.Keys(optimizedFlags)) diff --git a/internal/kernelparams/kernelparams.go b/internal/kernelparams/kernelparams.go index b5c0d9464a..893ccab536 100644 --- a/internal/kernelparams/kernelparams.go +++ b/internal/kernelparams/kernelparams.go @@ -43,6 +43,13 @@ var readMaxPagesLimitFunc = func() (int, error) { return val, nil } +// SupportsFuseMaxPagesLimit checks if the host kernel supports configuring +// the FUSE max_pages_limit attribute by reading the procfs entry. +func SupportsFuseMaxPagesLimit() bool { + _, err := readMaxPagesLimitFunc() + return err == nil +} + // KernelParamsManager wraps KernelParamsConfig with a mutex to ensure thread safety. type KernelParamsManager struct { *KernelParamsConfig From 356337e9d8bd406852a7522b394b35f02611299a Mon Sep 17 00:00:00 2001 From: Abhishek Gupta Date: Tue, 7 Jul 2026 07:47:36 +0000 Subject: [PATCH 3/4] Refactoring code to make it cleaner --- cfg/config.go | 1 + cfg/config_util.go | 65 ++++++++++++++++++++++++++++++++++++++++++++++ cfg/types.go | 14 ++++++++++ internal/fs/fs.go | 63 +++----------------------------------------- 4 files changed, 83 insertions(+), 60 deletions(-) diff --git a/cfg/config.go b/cfg/config.go index 2edccc1168..b93960ef42 100644 --- a/cfg/config.go +++ b/cfg/config.go @@ -408,6 +408,7 @@ func (c *Config) ApplyOptimizations(v *viper.Viper, input *OptimizationInput) ma } } } + c.applyRegionalKernelReaderOptimizations(v, input, optimizedFlags) return optimizedFlags } diff --git a/cfg/config_util.go b/cfg/config_util.go index 3c682c8860..26b8de6ff7 100644 --- a/cfg/config_util.go +++ b/cfg/config_util.go @@ -20,6 +20,8 @@ import ( "runtime" "strings" "time" + + "github.com/spf13/viper" ) const ( @@ -117,3 +119,66 @@ func GetBucketType(hierarchical, zonal, pirlo bool) BucketType { } return BucketTypeFlat } + +func DefaultFuseMaxPagesLimitForNonRapid() int { + // 16 MiB maximum request size based on the kernel page size. + return (16 * 1024 * 1024) / kernelPageSize +} + +// IsNonRapid returns true for regional and multi-regional bucket types (flat and hierarchical). +func (bt BucketType) IsNonRapid() bool { + return bt == BucketTypeFlat || bt == BucketTypeHierarchical +} + +func (c *Config) shouldEnableKernelReaderForNonRapidBuckets(v *viper.Viper, input *OptimizationInput) bool { + if input == nil || !input.BucketType.IsNonRapid() || v == nil { + return false + } + + // Do not enable if file cache or buffered read are enabled explicitly by user + // or if kernel does not support increasing max pages limit beyond 1MiB. + isFileCacheSet := v.IsSet("cache-dir") + isBufferedReadSet := v.IsSet("read.enable-buffered-read") + return /*c.Profile == "aiml-serving" &&*/ !isFileCacheSet && !isBufferedReadSet && input.HasCapability(CapabilityFuseMaxPagesLimit) +} + +func (c *Config) applyRegionalKernelReaderOptimizations(v *viper.Viper, input *OptimizationInput, optimizedFlags map[string]OptimizationResult) { + if c.DisableAutoconfig || v == nil || input == nil || !input.BucketType.IsNonRapid() { + return + } + + // 1. Automatically enable kernel reader for non-rapid buckets if appropriate. + if !v.IsSet("file-system.enable-kernel-reader") && !optimizedFlags["file-system.enable-kernel-reader"].Optimized { + if c.shouldEnableKernelReaderForNonRapidBuckets(v, input) { + c.FileSystem.EnableKernelReader = true + optimizedFlags["file-system.enable-kernel-reader"] = OptimizationResult{ + Optimized: true, + FinalValue: true, + } + } + } + + // 2. If kernel reader is enabled, apply parameter optimizations cleanly using a table-driven loop. + if c.FileSystem.EnableKernelReader { + type flagOpt struct { + name string + val int64 + setter func(int64) + } + opts := []flagOpt{ + {"file-system.max-read-ahead-kb", int64(DefaultMaxReadAheadKbForNonRapid()), func(v int64) { c.FileSystem.MaxReadAheadKb = v }}, + {"file-system.max-background", int64(DefaultMaxBackgroundForNonRapid()), func(v int64) { c.FileSystem.MaxBackground = v }}, + {"file-system.congestion-threshold", int64(DefaultCongestionThresholdForNonRapid()), func(v int64) { c.FileSystem.CongestionThreshold = v }}, + {"file-system.fuse-max-pages-limit", int64(DefaultFuseMaxPagesLimitForNonRapid()), func(v int64) { c.FileSystem.FuseMaxPagesLimit = v }}, + } + for _, opt := range opts { + if !v.IsSet(opt.name) && !optimizedFlags[opt.name].Optimized { + opt.setter(opt.val) + optimizedFlags[opt.name] = OptimizationResult{ + Optimized: true, + FinalValue: opt.val, + } + } + } + } +} diff --git a/cfg/types.go b/cfg/types.go index d9b334be91..f3060134ef 100644 --- a/cfg/types.go +++ b/cfg/types.go @@ -144,6 +144,20 @@ type OptimizationInput struct { // BucketType specifies the GCS bucket type. // An empty string means no bucket-type-based optimization should be applied. BucketType BucketType + + // Capabilities stores runtime or system capabilities that may influence optimizations + // (e.g., whether the host kernel supports specific attributes like max_pages_limit). + Capabilities map[string]bool +} + +const ( + // CapabilityFuseMaxPagesLimit indicates if the host kernel supports configuring FUSE max_pages_limit. + CapabilityFuseMaxPagesLimit = "fuse_max_pages_limit" +) + +// HasCapability checks if a runtime capability is present and true. +func (o *OptimizationInput) HasCapability(cap string) bool { + return o != nil && o.Capabilities != nil && o.Capabilities[cap] } // BucketType represents the type of GCS bucket. diff --git a/internal/fs/fs.go b/internal/fs/fs.go index 9f9362ca24..b9d9409044 100644 --- a/internal/fs/fs.go +++ b/internal/fs/fs.go @@ -158,60 +158,6 @@ type ServerConfig struct { Notifier *fuse.Notifier } -func shouldEnableKernelReaderForNonRapidBuckets(serverCfg *ServerConfig, bucketType gcs.BucketType) bool { - if bucketType.IsRapid() { - return false - } - - // Do not enable if file cache or buffered read are enabled explicitly by user - // or if kernel does not support increasing max pages limit beyond 1MiB. - isFileCacheSet := serverCfg.ViperConfig.IsSet("cache-dir") - isBufferedReadSet := serverCfg.ViperConfig.IsSet("read.enable-buffered-read") - return /*serverCfg.NewConfig.Profile == "aiml-serving" &&*/ !isFileCacheSet && !isBufferedReadSet && kernelparams.SupportsFuseMaxPagesLimit() -} - -func applyRegionalKernelReaderOptimizations(serverCfg *ServerConfig, bucketType gcs.BucketType, optimizedFlags map[string]cfg.OptimizationResult) { - if serverCfg.NewConfig.DisableAutoconfig || bucketType.IsRapid() { - return - } - - // 1. Automatically enable kernel reader for non-rapid buckets if appropriate. - if !serverCfg.ViperConfig.IsSet("file-system.enable-kernel-reader") && !optimizedFlags["file-system.enable-kernel-reader"].Optimized { - if shouldEnableKernelReaderForNonRapidBuckets(serverCfg, bucketType) { - serverCfg.NewConfig.FileSystem.EnableKernelReader = true - optimizedFlags["file-system.enable-kernel-reader"] = cfg.OptimizationResult{ - Optimized: true, - FinalValue: true, - } - logger.Info("GCSFuse Config: Automatically enabled enable-kernel-reader for aiml-serving profile on flat/hns bucket.") - } - } - - // 2. If kernel reader is enabled, apply parameter optimizations cleanly using a table-driven loop. - if serverCfg.NewConfig.FileSystem.EnableKernelReader { - type flagOpt struct { - name string - val int64 - setter func(int64) - } - opts := []flagOpt{ - {"file-system.max-read-ahead-kb", int64(cfg.DefaultMaxReadAheadKbForNonRapid()), func(v int64) { serverCfg.NewConfig.FileSystem.MaxReadAheadKb = v }}, - {"file-system.max-background", int64(cfg.DefaultMaxBackgroundForNonRapid()), func(v int64) { serverCfg.NewConfig.FileSystem.MaxBackground = v }}, - {"file-system.congestion-threshold", int64(cfg.DefaultCongestionThresholdForNonRapid()), func(v int64) { serverCfg.NewConfig.FileSystem.CongestionThreshold = v }}, - {"file-system.fuse-max-pages-limit", 16 * gcsx.MiB / int64(os.Getpagesize()), func(v int64) { serverCfg.NewConfig.FileSystem.FuseMaxPagesLimit = v }}, - } - for _, opt := range opts { - if !serverCfg.ViperConfig.IsSet(opt.name) && !optimizedFlags[opt.name].Optimized { - opt.setter(opt.val) - optimizedFlags[opt.name] = cfg.OptimizationResult{ - Optimized: true, - FinalValue: opt.val, - } - } - } - } -} - // Create a fuse file system server according to the supplied configuration. func NewFileSystem(ctx context.Context, serverCfg *ServerConfig) (fuseutil.FileSystem, error) { // Check permissions bits. @@ -333,13 +279,10 @@ func NewFileSystem(ctx context.Context, serverCfg *ServerConfig) (fuseutil.FileS bucketTypeEnum := cfg.GetBucketType(bucketType.Hierarchical, bucketType.Zonal, bucketType.Pirlo != gcs.PirloStateNone) optimizedFlags := serverCfg.NewConfig.ApplyOptimizations(serverCfg.ViperConfig, &cfg.OptimizationInput{ BucketType: bucketTypeEnum, + Capabilities: map[string]bool{ + cfg.CapabilityFuseMaxPagesLimit: kernelparams.SupportsFuseMaxPagesLimit(), + }, }) - if optimizedFlags == nil { - optimizedFlags = make(map[string]cfg.OptimizationResult) - } - - applyRegionalKernelReaderOptimizations(serverCfg, bucketType, optimizedFlags) - if len(optimizedFlags) > 0 { logger.Info("GCSFuse Config", "Applied optimizations for bucket-type: ", bucketTypeEnum, "Full Config", optimizedFlags) optimizedFlagNames := slices.Collect(maps.Keys(optimizedFlags)) From 2f1bb5abbe5e1cd2485bb20b0804284b9586016c Mon Sep 17 00:00:00 2001 From: Abhishek Gupta Date: Thu, 9 Jul 2026 09:05:32 +0000 Subject: [PATCH 4/4] Changes to further simplify the default logic --- cfg/config.go | 58 ++++--- cfg/config_test.go | 159 +++++++++++++----- cfg/config_util.go | 86 +++++----- cfg/config_util_test.go | 12 +- cfg/optimize.go | 2 +- cfg/params.yaml | 38 +++-- cfg/types.go | 8 +- cmd/config_validation_test.go | 6 +- cmd/root_test.go | 36 ++-- tools/config-gen/main.go | 6 +- tools/config-gen/parser.go | 2 +- tools/config-gen/parser_test.go | 6 + tools/config-gen/templates/config_test.tpl | 2 +- .../zonal_bucket_optimization_test.go | 4 +- 14 files changed, 262 insertions(+), 163 deletions(-) diff --git a/cfg/config.go b/cfg/config.go index b93960ef42..d4fe3b42f1 100644 --- a/cfg/config.go +++ b/cfg/config.go @@ -28,22 +28,18 @@ import ( var AllFlagOptimizationRules = map[string]shared.OptimizationRules{"file-system.congestion-threshold": { BucketTypeOptimization: []shared.BucketTypeOptimization{ { - BucketType: "zonal", - Value: int64(DefaultCongestionThreshold()), + BucketType: "rapid", + Value: int64(BucketTypeRapid.DefaultCongestionThreshold()), }, { - BucketType: "pirlo", - Value: int64(DefaultCongestionThreshold()), + BucketType: "non-rapid", + Value: int64(BucketTypeNonRapid.DefaultCongestionThreshold()), }, }, }, "file-system.enable-kernel-reader": { BucketTypeOptimization: []shared.BucketTypeOptimization{ { - BucketType: "zonal", - Value: bool(true), - }, - { - BucketType: "pirlo", + BucketType: "rapid", Value: bool(true), }, }, @@ -69,6 +65,17 @@ var AllFlagOptimizationRules = map[string]shared.OptimizationRules{"file-system. Value: bool(true), }, }, +}, "file-system.fuse-max-pages-limit": { + BucketTypeOptimization: []shared.BucketTypeOptimization{ + { + BucketType: "rapid", + Value: int64(BucketTypeRapid.DefaultFuseMaxPagesLimit()), + }, + { + BucketType: "non-rapid", + Value: int64(BucketTypeNonRapid.DefaultFuseMaxPagesLimit()), + }, + }, }, "implicit-dirs": { MachineBasedOptimization: []shared.MachineBasedOptimization{ { @@ -100,23 +107,23 @@ var AllFlagOptimizationRules = map[string]shared.OptimizationRules{"file-system. }, "file-system.max-background": { BucketTypeOptimization: []shared.BucketTypeOptimization{ { - BucketType: "zonal", - Value: int64(DefaultMaxBackground()), + BucketType: "rapid", + Value: int64(BucketTypeRapid.DefaultMaxBackground()), }, { - BucketType: "pirlo", - Value: int64(DefaultMaxBackground()), + BucketType: "non-rapid", + Value: int64(BucketTypeNonRapid.DefaultMaxBackground()), }, }, }, "file-system.max-read-ahead-kb": { BucketTypeOptimization: []shared.BucketTypeOptimization{ { - BucketType: "zonal", - Value: int64(16384), + BucketType: "rapid", + Value: int64(BucketTypeRapid.DefaultMaxReadAheadKb()), }, { - BucketType: "pirlo", - Value: int64(16384), + BucketType: "non-rapid", + Value: int64(BucketTypeNonRapid.DefaultMaxReadAheadKb()), }, }, }, "metadata-cache.negative-ttl-secs": { @@ -300,6 +307,18 @@ func (c *Config) ApplyOptimizations(v *viper.Viper, input *OptimizationInput) ma } } } + if !v.IsSet("file-system.fuse-max-pages-limit") { + rules := AllFlagOptimizationRules["file-system.fuse-max-pages-limit"] + result := getOptimizedValue(&rules, c.FileSystem.FuseMaxPagesLimit, profileName, machineType, input, machineTypeToGroupMap) + if result.Optimized { + if val, ok := result.FinalValue.(int64); ok { + if c.FileSystem.FuseMaxPagesLimit != val { + c.FileSystem.FuseMaxPagesLimit = val + optimizedFlags["file-system.fuse-max-pages-limit"] = result + } + } + } + } if !v.IsSet("implicit-dirs") { rules := AllFlagOptimizationRules["implicit-dirs"] result := getOptimizedValue(&rules, c.ImplicitDirs, profileName, machineType, input, machineTypeToGroupMap) @@ -408,7 +427,6 @@ func (c *Config) ApplyOptimizations(v *viper.Viper, input *OptimizationInput) ma } } } - c.applyRegionalKernelReaderOptimizations(v, input, optimizedFlags) return optimizedFlags } @@ -1130,7 +1148,7 @@ func BuildFlagSet(flagSet *pflag.FlagSet) error { flagSet.StringP("file-cache-include-regex", "", "", "Include file paths (in the format bucket_name/object_key) specified by this regex for file caching.") - flagSet.IntP("file-cache-max-parallel-downloads", "", DefaultMaxParallelDownloads(), "Sets an uber limit of number of concurrent file download requests that are made across all files.") + flagSet.IntP("file-cache-max-parallel-downloads", "", BucketTypeRapid.DefaultMaxParallelDownloads(), "Sets an uber limit of number of concurrent file download requests that are made across all files.") flagSet.IntP("file-cache-max-size-mb", "", -1, "Maximum size of the file-cache in MiBs") @@ -1158,7 +1176,7 @@ func BuildFlagSet(flagSet *pflag.FlagSet) error { flagSet.BoolP("foreground", "", false, "Stay in the foreground after mounting.") - flagSet.IntP("fuse-max-pages-limit", "", DefaultFuseMaxPagesLimit(), "Sets the limit for the maximum number of pages that fuse can process in a single request. This is a global, machine-level configuration that applies across all mounts. To prevent lowering the limits of other FUSE filesystems, the host's limit is only updated if the specified value is greater than the current system limit.") + flagSet.IntP("fuse-max-pages-limit", "", BucketTypeRapid.DefaultFuseMaxPagesLimit(), "Sets the limit for the maximum number of pages that fuse can process in a single request. This is a global, machine-level configuration that applies across all mounts. To prevent lowering the limits of other FUSE filesystems, the host's limit is only updated if the specified value is greater than the current system limit.") if err := flagSet.MarkHidden("fuse-max-pages-limit"); err != nil { return err diff --git a/cfg/config_test.go b/cfg/config_test.go index 834cd52429..bbb7832234 100644 --- a/cfg/config_test.go +++ b/cfg/config_test.go @@ -41,7 +41,7 @@ func TestApplyOptimizations(t *testing.T) { "file-system.congestion-threshold": 98765, "machine-type": "a2-megagpu-16g", }, - input: &OptimizationInput{BucketType: BucketTypeZonal}, + input: &OptimizationInput{BucketType: BucketTypeRapid}, expectOptimized: false, expectedValue: int64(98765), }, @@ -56,20 +56,20 @@ func TestApplyOptimizations(t *testing.T) { expectedValue: 0, }, { - name: "bucket_type_zonal", + name: "bucket_type_rapid", config: Config{Profile: ""}, userSetFlags: map[string]any{}, - input: &OptimizationInput{BucketType: BucketTypeZonal}, + input: &OptimizationInput{BucketType: BucketTypeRapid}, expectOptimized: true, - expectedValue: DefaultCongestionThreshold(), + expectedValue: BucketTypeRapid.DefaultCongestionThreshold(), }, { - name: "bucket_type_pirlo", + name: "bucket_type_non-rapid", config: Config{Profile: ""}, userSetFlags: map[string]any{}, - input: &OptimizationInput{BucketType: BucketTypePirlo}, + input: &OptimizationInput{BucketType: BucketTypeNonRapid}, expectOptimized: true, - expectedValue: DefaultCongestionThreshold(), + expectedValue: BucketTypeNonRapid.DefaultCongestionThreshold(), }, } @@ -81,7 +81,7 @@ func TestApplyOptimizations(t *testing.T) { if tc.name == "user_set" { c.FileSystem.CongestionThreshold = tc.expectedValue.(int64) } else { - c.FileSystem.CongestionThreshold = 0 + c.FileSystem.CongestionThreshold = int64(0) } v := viper.New() @@ -118,7 +118,7 @@ func TestApplyOptimizations(t *testing.T) { "file-system.enable-kernel-reader": true, "machine-type": "a2-megagpu-16g", }, - input: &OptimizationInput{BucketType: BucketTypeZonal}, + input: &OptimizationInput{BucketType: BucketTypeRapid}, expectOptimized: false, expectedValue: true, }, @@ -133,18 +133,10 @@ func TestApplyOptimizations(t *testing.T) { expectedValue: false, }, { - name: "bucket_type_zonal", - config: Config{Profile: ""}, - userSetFlags: map[string]any{}, - input: &OptimizationInput{BucketType: BucketTypeZonal}, - expectOptimized: true, - expectedValue: true, - }, - { - name: "bucket_type_pirlo", + name: "bucket_type_rapid", config: Config{Profile: ""}, userSetFlags: map[string]any{}, - input: &OptimizationInput{BucketType: BucketTypePirlo}, + input: &OptimizationInput{BucketType: BucketTypeRapid}, expectOptimized: true, expectedValue: true, }, @@ -158,7 +150,7 @@ func TestApplyOptimizations(t *testing.T) { if tc.name == "user_set" { c.FileSystem.EnableKernelReader = tc.expectedValue.(bool) } else { - c.FileSystem.EnableKernelReader = false + c.FileSystem.EnableKernelReader = bool(false) } v := viper.New() @@ -237,7 +229,7 @@ func TestApplyOptimizations(t *testing.T) { if tc.name == "user_set" { c.FileCache.CacheFileForRangeRead = tc.expectedValue.(bool) } else { - c.FileCache.CacheFileForRangeRead = false + c.FileCache.CacheFileForRangeRead = bool(false) } v := viper.New() @@ -314,7 +306,7 @@ func TestApplyOptimizations(t *testing.T) { if tc.name == "user_set" { c.Write.FinalizeFileForRapid = tc.expectedValue.(bool) } else { - c.Write.FinalizeFileForRapid = false + c.Write.FinalizeFileForRapid = bool(false) } v := viper.New() @@ -334,6 +326,83 @@ func TestApplyOptimizations(t *testing.T) { }) } }) + // Tests for file-system.fuse-max-pages-limit + t.Run("file-system.fuse-max-pages-limit", func(t *testing.T) { + testCases := []struct { + name string + config Config + userSetFlags map[string]any + input *OptimizationInput + expectOptimized bool + expectedValue any + }{ + { + name: "user_set", + config: Config{}, + userSetFlags: map[string]any{ + "file-system.fuse-max-pages-limit": 98765, + "machine-type": "a2-megagpu-16g", + }, + input: &OptimizationInput{BucketType: BucketTypeRapid}, + expectOptimized: false, + expectedValue: int64(98765), + }, + { + name: "no_optimization", + config: Config{Profile: "non_existent_profile"}, + userSetFlags: map[string]any{ + "machine-type": "low-end-machine", + }, + input: nil, + expectOptimized: false, + expectedValue: BucketTypeRapid.DefaultFuseMaxPagesLimit(), + }, + { + name: "bucket_type_rapid", + config: Config{Profile: ""}, + userSetFlags: map[string]any{}, + input: &OptimizationInput{BucketType: BucketTypeRapid}, + expectOptimized: false, + expectedValue: BucketTypeRapid.DefaultFuseMaxPagesLimit(), + }, + { + name: "bucket_type_non-rapid", + config: Config{Profile: ""}, + userSetFlags: map[string]any{}, + input: &OptimizationInput{BucketType: BucketTypeNonRapid}, + expectOptimized: true, + expectedValue: BucketTypeNonRapid.DefaultFuseMaxPagesLimit(), + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + // We need a copy of the config for each test case. + c := tc.config + // Set the default or non-default value on the config object. + if tc.name == "user_set" { + c.FileSystem.FuseMaxPagesLimit = tc.expectedValue.(int64) + } else { + c.FileSystem.FuseMaxPagesLimit = int64(BucketTypeRapid.DefaultFuseMaxPagesLimit()) + } + + v := viper.New() + for key, val := range tc.userSetFlags { + v.Set(key, val) + } + + optimizedFlags := c.ApplyOptimizations(v, tc.input) + + if tc.expectOptimized { + assert.Contains(t, optimizedFlags, "file-system.fuse-max-pages-limit") + } else { + assert.NotContains(t, optimizedFlags, "file-system.fuse-max-pages-limit") + } + // Use EqualValues to handle the int vs int64 type mismatch for default values. + assert.EqualValues(t, tc.expectedValue, c.FileSystem.FuseMaxPagesLimit) + }) + } + }) // Tests for implicit-dirs t.Run("implicit-dirs", func(t *testing.T) { testCases := []struct { @@ -430,7 +499,7 @@ func TestApplyOptimizations(t *testing.T) { if tc.name == "user_set" { c.ImplicitDirs = tc.expectedValue.(bool) } else { - c.ImplicitDirs = false + c.ImplicitDirs = bool(false) } v := viper.New() @@ -501,7 +570,7 @@ func TestApplyOptimizations(t *testing.T) { if tc.name == "user_set" { c.FileSystem.KernelListCacheTtlSecs = tc.expectedValue.(int64) } else { - c.FileSystem.KernelListCacheTtlSecs = 0 + c.FileSystem.KernelListCacheTtlSecs = int64(0) } v := viper.New() @@ -538,7 +607,7 @@ func TestApplyOptimizations(t *testing.T) { "file-system.max-background": 98765, "machine-type": "a2-megagpu-16g", }, - input: &OptimizationInput{BucketType: BucketTypeZonal}, + input: &OptimizationInput{BucketType: BucketTypeRapid}, expectOptimized: false, expectedValue: int64(98765), }, @@ -553,20 +622,20 @@ func TestApplyOptimizations(t *testing.T) { expectedValue: 0, }, { - name: "bucket_type_zonal", + name: "bucket_type_rapid", config: Config{Profile: ""}, userSetFlags: map[string]any{}, - input: &OptimizationInput{BucketType: BucketTypeZonal}, + input: &OptimizationInput{BucketType: BucketTypeRapid}, expectOptimized: true, - expectedValue: DefaultMaxBackground(), + expectedValue: BucketTypeRapid.DefaultMaxBackground(), }, { - name: "bucket_type_pirlo", + name: "bucket_type_non-rapid", config: Config{Profile: ""}, userSetFlags: map[string]any{}, - input: &OptimizationInput{BucketType: BucketTypePirlo}, + input: &OptimizationInput{BucketType: BucketTypeNonRapid}, expectOptimized: true, - expectedValue: DefaultMaxBackground(), + expectedValue: BucketTypeNonRapid.DefaultMaxBackground(), }, } @@ -578,7 +647,7 @@ func TestApplyOptimizations(t *testing.T) { if tc.name == "user_set" { c.FileSystem.MaxBackground = tc.expectedValue.(int64) } else { - c.FileSystem.MaxBackground = 0 + c.FileSystem.MaxBackground = int64(0) } v := viper.New() @@ -615,7 +684,7 @@ func TestApplyOptimizations(t *testing.T) { "file-system.max-read-ahead-kb": 98765, "machine-type": "a2-megagpu-16g", }, - input: &OptimizationInput{BucketType: BucketTypeZonal}, + input: &OptimizationInput{BucketType: BucketTypeRapid}, expectOptimized: false, expectedValue: int64(98765), }, @@ -630,20 +699,20 @@ func TestApplyOptimizations(t *testing.T) { expectedValue: 0, }, { - name: "bucket_type_zonal", + name: "bucket_type_rapid", config: Config{Profile: ""}, userSetFlags: map[string]any{}, - input: &OptimizationInput{BucketType: BucketTypeZonal}, + input: &OptimizationInput{BucketType: BucketTypeRapid}, expectOptimized: true, - expectedValue: 16384, + expectedValue: BucketTypeRapid.DefaultMaxReadAheadKb(), }, { - name: "bucket_type_pirlo", + name: "bucket_type_non-rapid", config: Config{Profile: ""}, userSetFlags: map[string]any{}, - input: &OptimizationInput{BucketType: BucketTypePirlo}, + input: &OptimizationInput{BucketType: BucketTypeNonRapid}, expectOptimized: true, - expectedValue: 16384, + expectedValue: BucketTypeNonRapid.DefaultMaxReadAheadKb(), }, } @@ -655,7 +724,7 @@ func TestApplyOptimizations(t *testing.T) { if tc.name == "user_set" { c.FileSystem.MaxReadAheadKb = tc.expectedValue.(int64) } else { - c.FileSystem.MaxReadAheadKb = 0 + c.FileSystem.MaxReadAheadKb = int64(0) } v := viper.New() @@ -771,7 +840,7 @@ func TestApplyOptimizations(t *testing.T) { if tc.name == "user_set" { c.MetadataCache.NegativeTtlSecs = tc.expectedValue.(int64) } else { - c.MetadataCache.NegativeTtlSecs = 5 + c.MetadataCache.NegativeTtlSecs = int64(5) } v := viper.New() @@ -887,7 +956,7 @@ func TestApplyOptimizations(t *testing.T) { if tc.name == "user_set" { c.MetadataCache.TtlSecs = tc.expectedValue.(int64) } else { - c.MetadataCache.TtlSecs = 60 + c.MetadataCache.TtlSecs = int64(60) } v := viper.New() @@ -996,7 +1065,7 @@ func TestApplyOptimizations(t *testing.T) { if tc.name == "user_set" { c.FileSystem.RenameDirLimit = tc.expectedValue.(int64) } else { - c.FileSystem.RenameDirLimit = 0 + c.FileSystem.RenameDirLimit = int64(0) } v := viper.New() @@ -1112,7 +1181,7 @@ func TestApplyOptimizations(t *testing.T) { if tc.name == "user_set" { c.MetadataCache.StatCacheMaxSizeMb = tc.expectedValue.(int64) } else { - c.MetadataCache.StatCacheMaxSizeMb = 34 + c.MetadataCache.StatCacheMaxSizeMb = int64(34) } v := viper.New() @@ -1201,7 +1270,7 @@ func TestApplyOptimizations(t *testing.T) { if tc.name == "user_set" { c.Write.GlobalMaxBlocks = tc.expectedValue.(int64) } else { - c.Write.GlobalMaxBlocks = 4 + c.Write.GlobalMaxBlocks = int64(4) } v := viper.New() diff --git a/cfg/config_util.go b/cfg/config_util.go index 26b8de6ff7..cbab6ee510 100644 --- a/cfg/config_util.go +++ b/cfg/config_util.go @@ -33,36 +33,32 @@ const ( var kernelPageSize int = os.Getpagesize() -func DefaultFuseMaxPagesLimit() int { - // The default limit is calculated to achieve a 1 MiB maximum request size - // based on the kernel page size. - return (1024 * 1024) / kernelPageSize -} - -func DefaultMaxBackground() int { +func (bt BucketType) DefaultMaxBackground() int { + if bt.IsNonRapid() { + return 96 + } return min(max(12, 2*runtime.NumCPU()), maxBackgroundLimit) } -func DefaultMaxBackgroundForNonRapid() int { - return 96 +func (bt BucketType) DefaultCongestionThreshold() int { + return (3 * bt.DefaultMaxBackground()) / 4 } -func DefaultCongestionThreshold() int { - // 75 % of DefaultMaxBackground - return (3 * DefaultMaxBackground()) / 4 -} - -func DefaultCongestionThresholdForNonRapid() int { - // 75 % of DefaultMaxBackgroundForNonRapid - return (3 * DefaultMaxBackgroundForNonRapid()) / 4 +func (bt BucketType) DefaultFuseMaxPagesLimit() int { + if bt.IsNonRapid() { + return (16 * 1024 * 1024) / kernelPageSize + } + return (1024 * 1024) / kernelPageSize } -func DefaultMaxReadAheadKbForNonRapid() int { - // 128 Kib - return 128 * 1024 +func (bt BucketType) DefaultMaxReadAheadKb() int64 { + if bt.IsNonRapid() { + return 128 * 1024 + } + return 16 * 1024 } -func DefaultMaxParallelDownloads() int { +func (bt BucketType) DefaultMaxParallelDownloads() int { return max(16, 2*runtime.NumCPU()) } @@ -120,26 +116,29 @@ func GetBucketType(hierarchical, zonal, pirlo bool) BucketType { return BucketTypeFlat } -func DefaultFuseMaxPagesLimitForNonRapid() int { - // 16 MiB maximum request size based on the kernel page size. - return (16 * 1024 * 1024) / kernelPageSize +// IsRapid returns true for rapid bucket types (zonal and pirlo). +func (bt BucketType) IsRapid() bool { + return bt == BucketTypeZonal || bt == BucketTypePirlo || bt == BucketTypeRapid } // IsNonRapid returns true for regional and multi-regional bucket types (flat and hierarchical). func (bt BucketType) IsNonRapid() bool { - return bt == BucketTypeFlat || bt == BucketTypeHierarchical + return bt == BucketTypeFlat || bt == BucketTypeHierarchical || bt == BucketTypeNonRapid } -func (c *Config) shouldEnableKernelReaderForNonRapidBuckets(v *viper.Viper, input *OptimizationInput) bool { - if input == nil || !input.BucketType.IsNonRapid() || v == nil { - return false +// Matches returns true if this BucketType matches the target rule BucketType, +// including group matches for "rapid" and "non-rapid". +func (bt BucketType) Matches(target BucketType) bool { + if bt == target { + return true } - - // Do not enable if file cache or buffered read are enabled explicitly by user - // or if kernel does not support increasing max pages limit beyond 1MiB. - isFileCacheSet := v.IsSet("cache-dir") - isBufferedReadSet := v.IsSet("read.enable-buffered-read") - return /*c.Profile == "aiml-serving" &&*/ !isFileCacheSet && !isBufferedReadSet && input.HasCapability(CapabilityFuseMaxPagesLimit) + if target == BucketTypeRapid && bt.IsRapid() { + return true + } + if target == BucketTypeNonRapid && bt.IsNonRapid() { + return true + } + return false } func (c *Config) applyRegionalKernelReaderOptimizations(v *viper.Viper, input *OptimizationInput, optimizedFlags map[string]OptimizationResult) { @@ -147,16 +146,7 @@ func (c *Config) applyRegionalKernelReaderOptimizations(v *viper.Viper, input *O return } - // 1. Automatically enable kernel reader for non-rapid buckets if appropriate. - if !v.IsSet("file-system.enable-kernel-reader") && !optimizedFlags["file-system.enable-kernel-reader"].Optimized { - if c.shouldEnableKernelReaderForNonRapidBuckets(v, input) { - c.FileSystem.EnableKernelReader = true - optimizedFlags["file-system.enable-kernel-reader"] = OptimizationResult{ - Optimized: true, - FinalValue: true, - } - } - } + // 1. TODO: Automatically enable kernel reader for non-rapid buckets if appropriate. // 2. If kernel reader is enabled, apply parameter optimizations cleanly using a table-driven loop. if c.FileSystem.EnableKernelReader { @@ -166,10 +156,10 @@ func (c *Config) applyRegionalKernelReaderOptimizations(v *viper.Viper, input *O setter func(int64) } opts := []flagOpt{ - {"file-system.max-read-ahead-kb", int64(DefaultMaxReadAheadKbForNonRapid()), func(v int64) { c.FileSystem.MaxReadAheadKb = v }}, - {"file-system.max-background", int64(DefaultMaxBackgroundForNonRapid()), func(v int64) { c.FileSystem.MaxBackground = v }}, - {"file-system.congestion-threshold", int64(DefaultCongestionThresholdForNonRapid()), func(v int64) { c.FileSystem.CongestionThreshold = v }}, - {"file-system.fuse-max-pages-limit", int64(DefaultFuseMaxPagesLimitForNonRapid()), func(v int64) { c.FileSystem.FuseMaxPagesLimit = v }}, + {"file-system.max-read-ahead-kb", input.BucketType.DefaultMaxReadAheadKb(), func(v int64) { c.FileSystem.MaxReadAheadKb = v }}, + {"file-system.max-background", int64(input.BucketType.DefaultMaxBackground()), func(v int64) { c.FileSystem.MaxBackground = v }}, + {"file-system.congestion-threshold", int64(input.BucketType.DefaultCongestionThreshold()), func(v int64) { c.FileSystem.CongestionThreshold = v }}, + {"file-system.fuse-max-pages-limit", int64(input.BucketType.DefaultFuseMaxPagesLimit()), func(v int64) { c.FileSystem.FuseMaxPagesLimit = v }}, } for _, opt := range opts { if !v.IsSet(opt.name) && !optimizedFlags[opt.name].Optimized { diff --git a/cfg/config_util_test.go b/cfg/config_util_test.go index a0d2f5ebd3..67eb4c9b6f 100644 --- a/cfg/config_util_test.go +++ b/cfg/config_util_test.go @@ -22,17 +22,19 @@ import ( ) func Test_DefaultMaxBackground(t *testing.T) { - assert.GreaterOrEqual(t, DefaultMaxBackground(), 12) - assert.LessOrEqual(t, DefaultMaxBackground(), maxBackgroundLimit) + assert.GreaterOrEqual(t, BucketTypeRapid.DefaultMaxBackground(), 12) + assert.LessOrEqual(t, BucketTypeRapid.DefaultMaxBackground(), maxBackgroundLimit) + assert.Equal(t, 96, BucketTypeNonRapid.DefaultMaxBackground()) } func Test_DefaultCongestionThreshold(t *testing.T) { - assert.GreaterOrEqual(t, DefaultCongestionThreshold(), 9) - assert.LessOrEqual(t, DefaultCongestionThreshold(), 144) // 75% of maxBackgroundLimit + assert.GreaterOrEqual(t, BucketTypeRapid.DefaultCongestionThreshold(), 9) + assert.LessOrEqual(t, BucketTypeRapid.DefaultCongestionThreshold(), 144) // 75% of maxBackgroundLimit + assert.Equal(t, 72, BucketTypeNonRapid.DefaultCongestionThreshold()) } func Test_DefaultMaxParallelDownloads(t *testing.T) { - assert.GreaterOrEqual(t, DefaultMaxParallelDownloads(), 16) + assert.GreaterOrEqual(t, BucketTypeRapid.DefaultMaxParallelDownloads(), 16) } func TestIsFileCacheEnabled(t *testing.T) { diff --git a/cfg/optimize.go b/cfg/optimize.go index f346db7133..9cd2045ce1 100644 --- a/cfg/optimize.go +++ b/cfg/optimize.go @@ -163,7 +163,7 @@ func getOptimizedValue( // 3. If no profile is set, and no machine-type optimization applies, then check for bucket-type optimization. if input != nil && input.BucketType.IsValid() { for _, bto := range rules.BucketTypeOptimization { - if BucketType(bto.BucketType) == input.BucketType { + if input.BucketType.Matches(BucketType(bto.BucketType)) { return OptimizationResult{ FinalValue: bto.Value, OptimizationReason: fmt.Sprintf("bucket-type %q", input.BucketType), diff --git a/cfg/params.yaml b/cfg/params.yaml index b297b14247..01598c1d82 100644 --- a/cfg/params.yaml +++ b/cfg/params.yaml @@ -338,7 +338,7 @@ params: flag-name: "file-cache-max-parallel-downloads" type: "int" usage: "Sets an uber limit of number of concurrent file download requests that are made across all files." - default: "DefaultMaxParallelDownloads()" + default: "BucketTypeRapid.DefaultMaxParallelDownloads()" - config-path: "file-cache.max-size-mb" flag-name: "file-cache-max-size-mb" @@ -377,10 +377,10 @@ params: hide-flag: true optimizations: bucket-type-optimization: - - bucket-type: "zonal" - value: "DefaultCongestionThreshold()" - - bucket-type: "pirlo" - value: "DefaultCongestionThreshold()" + - bucket-type: "rapid" + value: "BucketTypeRapid.DefaultCongestionThreshold()" + - bucket-type: "non-rapid" + value: "BucketTypeNonRapid.DefaultCongestionThreshold()" - config-path: "file-system.dir-mode" flag-name: "dir-mode" @@ -403,9 +403,7 @@ params: hide-flag: true optimizations: bucket-type-optimization: - - bucket-type: "zonal" - value: true - - bucket-type: "pirlo" + - bucket-type: "rapid" value: true - config-path: "file-system.experimental-enable-dentry-cache" @@ -455,8 +453,14 @@ params: single request. This is a global, machine-level configuration that applies across all mounts. To prevent lowering the limits of other FUSE filesystems, the host's limit is only updated if the specified value is greater than the current system limit. - default: "DefaultFuseMaxPagesLimit()" + default: "BucketTypeRapid.DefaultFuseMaxPagesLimit()" hide-flag: true + optimizations: + bucket-type-optimization: + - bucket-type: "rapid" + value: "BucketTypeRapid.DefaultFuseMaxPagesLimit()" + - bucket-type: "non-rapid" + value: "BucketTypeNonRapid.DefaultFuseMaxPagesLimit()" - config-path: "file-system.fuse-options" flag-name: "o" @@ -523,10 +527,10 @@ params: hide-flag: true optimizations: bucket-type-optimization: - - bucket-type: "zonal" - value: "DefaultMaxBackground()" - - bucket-type: "pirlo" - value: "DefaultMaxBackground()" + - bucket-type: "rapid" + value: "BucketTypeRapid.DefaultMaxBackground()" + - bucket-type: "non-rapid" + value: "BucketTypeNonRapid.DefaultMaxBackground()" - config-path: "file-system.max-read-ahead-kb" flag-name: "max-read-ahead-kb" @@ -539,10 +543,10 @@ params: hide-flag: true optimizations: bucket-type-optimization: - - bucket-type: "zonal" - value: 16384 # 16 MiB - - bucket-type: "pirlo" - value: 16384 # 16 MiB + - bucket-type: "rapid" + value: "BucketTypeRapid.DefaultMaxReadAheadKb()" + - bucket-type: "non-rapid" + value: "BucketTypeNonRapid.DefaultMaxReadAheadKb()" - config-path: "file-system.rename-dir-limit" flag-name: "rename-dir-limit" diff --git a/cfg/types.go b/cfg/types.go index f3060134ef..a911c997d1 100644 --- a/cfg/types.go +++ b/cfg/types.go @@ -175,9 +175,15 @@ const ( // BucketTypeFlat represents a flat (regional or multi-regional) bucket. BucketTypeFlat BucketType = "flat" + + // BucketTypeRapid represents a rapid bucket group (zonal and pirlo). + BucketTypeRapid BucketType = "rapid" + + // BucketTypeNonRapid represents a non-rapid bucket group (flat and hierarchical). + BucketTypeNonRapid BucketType = "non-rapid" ) // IsValid returns true if the BucketType is one of the defined valid types. func (bt BucketType) IsValid() bool { - return bt == BucketTypeZonal || bt == BucketTypePirlo || bt == BucketTypeHierarchical || bt == BucketTypeFlat + return bt == BucketTypeZonal || bt == BucketTypePirlo || bt == BucketTypeHierarchical || bt == BucketTypeFlat || bt == BucketTypeRapid || bt == BucketTypeNonRapid } diff --git a/cmd/config_validation_test.go b/cmd/config_validation_test.go index b347c5ddc8..82ca748cf3 100644 --- a/cmd/config_validation_test.go +++ b/cmd/config_validation_test.go @@ -588,7 +588,7 @@ func TestValidateConfigFile_FileSystemConfigSuccessful(t *testing.T) { configFile: "testdata/empty_file.yaml", expectedConfig: &cfg.Config{ FileSystem: cfg.FileSystemConfig{ - FuseMaxPagesLimit: int64(cfg.DefaultFuseMaxPagesLimit()), + FuseMaxPagesLimit: int64(cfg.BucketTypeRapid.DefaultFuseMaxPagesLimit()), DirMode: 0755, DisableParallelDirops: false, FileMode: 0644, @@ -609,7 +609,7 @@ func TestValidateConfigFile_FileSystemConfigSuccessful(t *testing.T) { configFile: "testdata/file_system_config/unset_file_system_config.yaml", expectedConfig: &cfg.Config{ FileSystem: cfg.FileSystemConfig{ - FuseMaxPagesLimit: int64(cfg.DefaultFuseMaxPagesLimit()), + FuseMaxPagesLimit: int64(cfg.BucketTypeRapid.DefaultFuseMaxPagesLimit()), DirMode: 0755, DisableParallelDirops: false, FileMode: 0644, @@ -630,7 +630,7 @@ func TestValidateConfigFile_FileSystemConfigSuccessful(t *testing.T) { configFile: "testdata/valid_config.yaml", expectedConfig: &cfg.Config{ FileSystem: cfg.FileSystemConfig{ - FuseMaxPagesLimit: int64(cfg.DefaultFuseMaxPagesLimit()), + FuseMaxPagesLimit: int64(cfg.BucketTypeRapid.DefaultFuseMaxPagesLimit()), DirMode: 0777, DisableParallelDirops: true, FileMode: 0666, diff --git a/cmd/root_test.go b/cmd/root_test.go index 2174bc5d79..3d2eb60bc3 100644 --- a/cmd/root_test.go +++ b/cmd/root_test.go @@ -955,7 +955,7 @@ func TestArgsParsing_FileSystemFlags(t *testing.T) { ExperimentalEnableDentryCache: false, ExperimentalEnableReaddirplus: false, FileMode: 0644, - FuseMaxPagesLimit: int64(cfg.DefaultFuseMaxPagesLimit()), + FuseMaxPagesLimit: int64(cfg.BucketTypeRapid.DefaultFuseMaxPagesLimit()), FuseOptions: []string{}, Gid: -1, IgnoreInterrupts: true, @@ -983,7 +983,7 @@ func TestArgsParsing_FileSystemFlags(t *testing.T) { args: []string{"gcsfuse", "--dir-mode=0777", "--disable-parallel-dirops", "--experimental-enable-dentry-cache", "--experimental-enable-readdirplus", "--file-mode=0666", "--o", "ro", "--gid=7", "--ignore-interrupts=false", "--kernel-list-cache-ttl-secs=300", "--rename-dir-limit=10", "--temp-dir=~/temp", "--uid=8", "abc", "pqr"}, expectedConfig: &cfg.Config{ FileSystem: cfg.FileSystemConfig{ - FuseMaxPagesLimit: int64(cfg.DefaultFuseMaxPagesLimit()), + FuseMaxPagesLimit: int64(cfg.BucketTypeRapid.DefaultFuseMaxPagesLimit()), DirMode: 0777, DisableParallelDirops: true, ExperimentalEnableDentryCache: true, @@ -1006,7 +1006,7 @@ func TestArgsParsing_FileSystemFlags(t *testing.T) { args: []string{"gcsfuse", "--dir-mode=777", "--file-mode=666", "abc", "pqr"}, expectedConfig: &cfg.Config{ FileSystem: cfg.FileSystemConfig{ - FuseMaxPagesLimit: int64(cfg.DefaultFuseMaxPagesLimit()), + FuseMaxPagesLimit: int64(cfg.BucketTypeRapid.DefaultFuseMaxPagesLimit()), DirMode: 0777, DisableParallelDirops: false, ExperimentalEnableDentryCache: false, @@ -1029,7 +1029,7 @@ func TestArgsParsing_FileSystemFlags(t *testing.T) { args: []string{"gcsfuse", "--dir-mode=777", "--machine-type=a3-highgpu-8g", "--disable-autoconfig=false", "--file-mode=666", "abc", "pqr"}, expectedConfig: &cfg.Config{ FileSystem: cfg.FileSystemConfig{ - FuseMaxPagesLimit: int64(cfg.DefaultFuseMaxPagesLimit()), + FuseMaxPagesLimit: int64(cfg.BucketTypeRapid.DefaultFuseMaxPagesLimit()), DirMode: 0777, DisableParallelDirops: false, ExperimentalEnableDentryCache: false, @@ -1054,7 +1054,7 @@ func TestArgsParsing_FileSystemFlags(t *testing.T) { args: []string{"gcsfuse", "--dir-mode=777", "--machine-type=a3-highgpu-8g", "--disable-autoconfig=true", "--file-mode=666", "abc", "pqr"}, expectedConfig: &cfg.Config{ FileSystem: cfg.FileSystemConfig{ - FuseMaxPagesLimit: int64(cfg.DefaultFuseMaxPagesLimit()), + FuseMaxPagesLimit: int64(cfg.BucketTypeRapid.DefaultFuseMaxPagesLimit()), DirMode: 0777, DisableParallelDirops: false, ExperimentalEnableDentryCache: false, @@ -1079,7 +1079,7 @@ func TestArgsParsing_FileSystemFlags(t *testing.T) { args: []string{"gcsfuse", "--dir-mode=777", "--machine-type=a3-highgpu-8g", "--disable-autoconfig=false", "--rename-dir-limit=15000", "--file-mode=666", "abc", "pqr"}, expectedConfig: &cfg.Config{ FileSystem: cfg.FileSystemConfig{ - FuseMaxPagesLimit: int64(cfg.DefaultFuseMaxPagesLimit()), + FuseMaxPagesLimit: int64(cfg.BucketTypeRapid.DefaultFuseMaxPagesLimit()), DirMode: 0777, DisableParallelDirops: false, ExperimentalEnableDentryCache: false, @@ -1159,7 +1159,7 @@ func TestArgsParsing_FileSystemFlags(t *testing.T) { args: []string{"gcsfuse", "--experimental-o-direct", "abc", "pqr"}, expectedConfig: &cfg.Config{ FileSystem: cfg.FileSystemConfig{ - FuseMaxPagesLimit: int64(cfg.DefaultFuseMaxPagesLimit()), + FuseMaxPagesLimit: int64(cfg.BucketTypeRapid.DefaultFuseMaxPagesLimit()), DirMode: 0755, FileMode: 0644, FuseOptions: []string{}, @@ -1176,7 +1176,7 @@ func TestArgsParsing_FileSystemFlags(t *testing.T) { args: []string{"gcsfuse", "--experimental-enable-pirlo", "abc", "pqr"}, expectedConfig: &cfg.Config{ FileSystem: cfg.FileSystemConfig{ - FuseMaxPagesLimit: int64(cfg.DefaultFuseMaxPagesLimit()), + FuseMaxPagesLimit: int64(cfg.BucketTypeRapid.DefaultFuseMaxPagesLimit()), DirMode: 0755, FileMode: 0644, FuseOptions: []string{}, @@ -1193,7 +1193,7 @@ func TestArgsParsing_FileSystemFlags(t *testing.T) { args: []string{"gcsfuse", "--max-read-ahead-kb=1024", "abc", "pqr"}, expectedConfig: &cfg.Config{ FileSystem: cfg.FileSystemConfig{ - FuseMaxPagesLimit: int64(cfg.DefaultFuseMaxPagesLimit()), + FuseMaxPagesLimit: int64(cfg.BucketTypeRapid.DefaultFuseMaxPagesLimit()), DirMode: 0755, FileMode: 0644, FuseOptions: []string{}, @@ -1211,7 +1211,7 @@ func TestArgsParsing_FileSystemFlags(t *testing.T) { args: []string{"gcsfuse", "abc", "pqr"}, expectedConfig: &cfg.Config{ FileSystem: cfg.FileSystemConfig{ - FuseMaxPagesLimit: int64(cfg.DefaultFuseMaxPagesLimit()), + FuseMaxPagesLimit: int64(cfg.BucketTypeRapid.DefaultFuseMaxPagesLimit()), DirMode: 0755, FileMode: 0644, FuseOptions: []string{}, @@ -1228,7 +1228,7 @@ func TestArgsParsing_FileSystemFlags(t *testing.T) { args: []string{"gcsfuse", "--max-background=512", "abc", "pqr"}, expectedConfig: &cfg.Config{ FileSystem: cfg.FileSystemConfig{ - FuseMaxPagesLimit: int64(cfg.DefaultFuseMaxPagesLimit()), + FuseMaxPagesLimit: int64(cfg.BucketTypeRapid.DefaultFuseMaxPagesLimit()), DirMode: 0755, FileMode: 0644, FuseOptions: []string{}, @@ -1246,7 +1246,7 @@ func TestArgsParsing_FileSystemFlags(t *testing.T) { args: []string{"gcsfuse", "abc", "pqr"}, expectedConfig: &cfg.Config{ FileSystem: cfg.FileSystemConfig{ - FuseMaxPagesLimit: int64(cfg.DefaultFuseMaxPagesLimit()), + FuseMaxPagesLimit: int64(cfg.BucketTypeRapid.DefaultFuseMaxPagesLimit()), DirMode: 0755, FileMode: 0644, FuseOptions: []string{}, @@ -1264,7 +1264,7 @@ func TestArgsParsing_FileSystemFlags(t *testing.T) { args: []string{"gcsfuse", "--congestion-threshold=256", "abc", "pqr"}, expectedConfig: &cfg.Config{ FileSystem: cfg.FileSystemConfig{ - FuseMaxPagesLimit: int64(cfg.DefaultFuseMaxPagesLimit()), + FuseMaxPagesLimit: int64(cfg.BucketTypeRapid.DefaultFuseMaxPagesLimit()), DirMode: 0755, FileMode: 0644, FuseOptions: []string{}, @@ -1282,7 +1282,7 @@ func TestArgsParsing_FileSystemFlags(t *testing.T) { args: []string{"gcsfuse", "abc", "pqr"}, expectedConfig: &cfg.Config{ FileSystem: cfg.FileSystemConfig{ - FuseMaxPagesLimit: int64(cfg.DefaultFuseMaxPagesLimit()), + FuseMaxPagesLimit: int64(cfg.BucketTypeRapid.DefaultFuseMaxPagesLimit()), DirMode: 0755, FileMode: 0644, FuseOptions: []string{}, @@ -1300,7 +1300,7 @@ func TestArgsParsing_FileSystemFlags(t *testing.T) { args: []string{"gcsfuse", "--enable-kernel-reader", "abc", "pqr"}, expectedConfig: &cfg.Config{ FileSystem: cfg.FileSystemConfig{ - FuseMaxPagesLimit: int64(cfg.DefaultFuseMaxPagesLimit()), + FuseMaxPagesLimit: int64(cfg.BucketTypeRapid.DefaultFuseMaxPagesLimit()), DirMode: 0755, FileMode: 0644, FuseOptions: []string{}, @@ -1318,7 +1318,7 @@ func TestArgsParsing_FileSystemFlags(t *testing.T) { args: []string{"gcsfuse", "abc", "pqr"}, expectedConfig: &cfg.Config{ FileSystem: cfg.FileSystemConfig{ - FuseMaxPagesLimit: int64(cfg.DefaultFuseMaxPagesLimit()), + FuseMaxPagesLimit: int64(cfg.BucketTypeRapid.DefaultFuseMaxPagesLimit()), DirMode: 0755, FileMode: 0644, FuseOptions: []string{}, @@ -1336,7 +1336,7 @@ func TestArgsParsing_FileSystemFlags(t *testing.T) { args: []string{"gcsfuse", "--kernel-params-file=/tmp/params", "abc", "pqr"}, expectedConfig: &cfg.Config{ FileSystem: cfg.FileSystemConfig{ - FuseMaxPagesLimit: int64(cfg.DefaultFuseMaxPagesLimit()), + FuseMaxPagesLimit: int64(cfg.BucketTypeRapid.DefaultFuseMaxPagesLimit()), DirMode: 0755, FileMode: 0644, FuseOptions: []string{}, @@ -1354,7 +1354,7 @@ func TestArgsParsing_FileSystemFlags(t *testing.T) { args: []string{"gcsfuse", "--config-file", createTempConfigFile(t, "file-system:\n kernel-params-file: /tmp/config_params"), "abc", "pqr"}, expectedConfig: &cfg.Config{ FileSystem: cfg.FileSystemConfig{ - FuseMaxPagesLimit: int64(cfg.DefaultFuseMaxPagesLimit()), + FuseMaxPagesLimit: int64(cfg.BucketTypeRapid.DefaultFuseMaxPagesLimit()), DirMode: 0755, FileMode: 0644, FuseOptions: []string{}, diff --git a/tools/config-gen/main.go b/tools/config-gen/main.go index 759abb02c5..4261db46a6 100644 --- a/tools/config-gen/main.go +++ b/tools/config-gen/main.go @@ -24,6 +24,7 @@ import ( "path" "reflect" "slices" + "strings" "text/template" // NOLINT "unicode" @@ -78,7 +79,10 @@ func write(dataObj any, outputFile, templateFile string) (err error) { // Define the custom function map. funcMap := template.FuncMap{ "formatValue": formatValue, - "title": cases.Title(language.English).String, + "title": func(s string) string { + s = cases.Title(language.English).String(s) + return strings.ReplaceAll(s, "-", "") + }, } file := path.Base(templateFile) diff --git a/tools/config-gen/parser.go b/tools/config-gen/parser.go index 9b78f86f3e..2d96bcf563 100644 --- a/tools/config-gen/parser.go +++ b/tools/config-gen/parser.go @@ -127,7 +127,7 @@ func validateParam(param Param) error { // Validate bucket-based optimizations if present. if param.Optimizations != nil { - validBucketTypes := []string{"zonal", "hierarchical", "flat", "pirlo"} + validBucketTypes := []string{"zonal", "hierarchical", "flat", "pirlo", "rapid", "non-rapid"} for _, bto := range param.Optimizations.BucketTypeOptimization { if !slices.Contains(validBucketTypes, bto.BucketType) { return fmt.Errorf("invalid bucket-type %q for flag %s; must be one of: %v", diff --git a/tools/config-gen/parser_test.go b/tools/config-gen/parser_test.go index 562d3abc49..7e266ef0ef 100644 --- a/tools/config-gen/parser_test.go +++ b/tools/config-gen/parser_test.go @@ -203,6 +203,10 @@ params: value: true - bucket-type: pirlo value: true + - bucket-type: rapid + value: true + - bucket-type: non-rapid + value: false - config-path: "file-system.max-read-ahead-kb" flag-name: "max-read-ahead-kb" type: "int" @@ -270,6 +274,8 @@ params: {BucketType: "hierarchical", Value: false}, {BucketType: "flat", Value: true}, {BucketType: "pirlo", Value: true}, + {BucketType: "rapid", Value: true}, + {BucketType: "non-rapid", Value: false}, }, } assert.Equal(t, "file-system.enable-kernel-reader", param.ConfigPath) diff --git a/tools/config-gen/templates/config_test.tpl b/tools/config-gen/templates/config_test.tpl index c5e6a75dae..4ddaf924a5 100644 --- a/tools/config-gen/templates/config_test.tpl +++ b/tools/config-gen/templates/config_test.tpl @@ -205,7 +205,7 @@ func TestApplyOptimizations(t *testing.T) { if tc.name == "user_set" { c.{{$flag.GoPath}} = tc.expectedValue.({{$flag.GoType}}) } else { - c.{{$flag.GoPath}} = {{$flag.DefaultValue}} + c.{{$flag.GoPath}} = {{$flag.GoType}}({{$flag.DefaultValue}}) } v := viper.New() diff --git a/tools/integration_tests/flag_optimizations/zonal_bucket_optimization_test.go b/tools/integration_tests/flag_optimizations/zonal_bucket_optimization_test.go index 2d27258964..134923a712 100644 --- a/tools/integration_tests/flag_optimizations/zonal_bucket_optimization_test.go +++ b/tools/integration_tests/flag_optimizations/zonal_bucket_optimization_test.go @@ -157,8 +157,8 @@ func (s *KernelReaderParamsSuite) TestKernelParamVerification() { require.NoError(s.T(), err) optimizedReadAhead := "16384" - optimizedMaxBackground := fmt.Sprintf("%d", cfg.DefaultMaxBackground()) - optimizedCongestion := fmt.Sprintf("%d", cfg.DefaultCongestionThreshold()) + optimizedMaxBackground := fmt.Sprintf("%d", cfg.BucketTypeRapid.DefaultMaxBackground()) + optimizedCongestion := fmt.Sprintf("%d", cfg.BucketTypeRapid.DefaultCongestionThreshold()) s.verifyKernelParam(readAheadPath, s.expectedReadAhead, optimizedReadAhead) s.verifyKernelParam(maxBackgroundPath, s.expectedMaxBackground, optimizedMaxBackground)