diff --git a/cfg/config.go b/cfg/config.go index 2edccc1168..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) @@ -1129,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") @@ -1157,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 5f8e3e3241..cbab6ee510 100644 --- a/cfg/config_util.go +++ b/cfg/config_util.go @@ -20,6 +20,8 @@ import ( "runtime" "strings" "time" + + "github.com/spf13/viper" ) const ( @@ -31,22 +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 (bt BucketType) DefaultMaxBackground() int { + if bt.IsNonRapid() { + return 96 + } + return min(max(12, 2*runtime.NumCPU()), maxBackgroundLimit) } -func DefaultMaxBackground() int { - return min(max(12, 2*runtime.NumCPU()), maxBackgroundLimit) +func (bt BucketType) DefaultCongestionThreshold() int { + return (3 * bt.DefaultMaxBackground()) / 4 +} + +func (bt BucketType) DefaultFuseMaxPagesLimit() int { + if bt.IsNonRapid() { + return (16 * 1024 * 1024) / kernelPageSize + } + return (1024 * 1024) / kernelPageSize } -func DefaultCongestionThreshold() int { - // 75 % of DefaultMaxBackground - return (3 * DefaultMaxBackground()) / 4 +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()) } @@ -103,3 +115,60 @@ func GetBucketType(hierarchical, zonal, pirlo bool) BucketType { } return BucketTypeFlat } + +// 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 || bt == BucketTypeNonRapid +} + +// 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 + } + 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) { + if c.DisableAutoconfig || v == nil || input == nil || !input.BucketType.IsNonRapid() { + return + } + + // 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 { + type flagOpt struct { + name string + val int64 + setter func(int64) + } + opts := []flagOpt{ + {"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 { + opt.setter(opt.val) + optimizedFlags[opt.name] = OptimizationResult{ + Optimized: true, + FinalValue: opt.val, + } + } + } + } +} 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 d9b334be91..a911c997d1 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. @@ -161,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/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/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/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 258c5cbfd4..b9d9409044 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 @@ -270,6 +279,9 @@ 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 len(optimizedFlags) > 0 { logger.Info("GCSFuse Config", "Applied optimizations for bucket-type: ", bucketTypeEnum, "Full Config", optimizedFlags) @@ -676,6 +688,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 +3133,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/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 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) + }) + } +} 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)