Disable write back cache when streaming writes are enabled

cfg/rationalize.go

@@ -73,6 +73,8 @@ func resolveStreamingWriteConfig(w *WriteConfig) {
 		w.CreateEmptyFile = false
 	}
 
+	w.BlockSizeMb *= util.MiB
+
 	if w.GlobalMaxBlocks == -1 {
 		w.GlobalMaxBlocks = math.MaxInt64
 	}

cfg/rationalize_test.go

@@ -315,6 +315,7 @@ func TestRationalize_WriteConfig(t *testing.T) {
 		config                   *Config
 		expectedCreateEmptyFile  bool
 		expectedMaxBlocksPerFile int64
+		expectedBlockSizeMB      int64
 	}{
 		{
 			name: "valid_config_streaming_writes_enabled",
@@ -329,12 +330,13 @@ func TestRationalize_WriteConfig(t *testing.T) {
 			},
 			expectedCreateEmptyFile:  false,
 			expectedMaxBlocksPerFile: math.MaxInt64,
+			expectedBlockSizeMB:      10 * 1024 * 1024,
 		},
 		{
 			name: "valid_config_global_max_blocks_less_than_blocks_per_file",
 			config: &Config{
 				Write: WriteConfig{
-					BlockSizeMb:                       10,
+					BlockSizeMb:                       5,
 					CreateEmptyFile:                   true,
 					ExperimentalEnableStreamingWrites: true,
 					GlobalMaxBlocks:                   10,
@@ -343,12 +345,13 @@ func TestRationalize_WriteConfig(t *testing.T) {
 			},
 			expectedCreateEmptyFile:  false,
 			expectedMaxBlocksPerFile: 10,
+			expectedBlockSizeMB:      5 * 1024 * 1024,
 		},
 		{
 			name: "valid_config_global_max_blocks_more_than_blocks_per_file",
 			config: &Config{
 				Write: WriteConfig{
-					BlockSizeMb:                       10,
+					BlockSizeMb:                       64,
 					CreateEmptyFile:                   true,
 					ExperimentalEnableStreamingWrites: true,
 					GlobalMaxBlocks:                   20,
@@ -357,6 +360,7 @@ func TestRationalize_WriteConfig(t *testing.T) {
 			},
 			expectedCreateEmptyFile:  false,
 			expectedMaxBlocksPerFile: 10,
+			expectedBlockSizeMB:      64 * 1024 * 1024,
 		},
 	}
 
@@ -367,6 +371,7 @@ func TestRationalize_WriteConfig(t *testing.T) {
 			if assert.NoError(t, actualErr) {
 				assert.Equal(t, tc.expectedCreateEmptyFile, tc.config.Write.CreateEmptyFile)
 				assert.Equal(t, tc.expectedMaxBlocksPerFile, tc.config.Write.MaxBlocksPerFile)
+				assert.Equal(t, tc.expectedBlockSizeMB, tc.config.Write.BlockSizeMb)
 			}
 		})
 	}

cfg/validate.go

@@ -17,8 +17,9 @@ package cfg
 import (
 	"errors"
 	"fmt"
-
 	"math"
+
+	"github.com/googlecloudplatform/gcsfuse/v2/internal/util"
 )
 
 const (
@@ -155,8 +156,8 @@ func isValidWriteStreamingConfig(wc *WriteConfig) error {
 		return nil
 	}
 
-	if wc.BlockSizeMb <= 0 {
-		return fmt.Errorf("invalid value of write-block-size-mb; can't be less than 1")
+	if wc.BlockSizeMb <= 0 || wc.BlockSizeMb > util.MaxMiBsInInt64 {
+		return fmt.Errorf("invalid value of write-block-size-mb; can't be less than 1 or more than %d", util.MaxMiBsInInt64)
 	}
 	if !(wc.MaxBlocksPerFile == -1 || wc.MaxBlocksPerFile >= 2) {
 		return fmt.Errorf("invalid value of write-max-blocks-per-file: %d; should be >=2 or -1 (for infinite)", wc.MaxBlocksPerFile)

cfg/validate_test.go

@@ -18,6 +18,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/googlecloudplatform/gcsfuse/v2/internal/util"
 	"github.com/stretchr/testify/assert"
 )
 
@@ -451,6 +452,13 @@ func Test_isValidWriteStreamingConfig_ErrorScenarios(t *testing.T) {
 			GlobalMaxBlocks:                   -1,
 			MaxBlocksPerFile:                  -1,
 		}},
+		{"very_large_block_size", WriteConfig{
+			BlockSizeMb:                       util.MaxMiBsInInt64 + 1,
+			CreateEmptyFile:                   false,
+			ExperimentalEnableStreamingWrites: true,
+			GlobalMaxBlocks:                   -1,
+			MaxBlocksPerFile:                  -1,
+		}},
 		{"negative_block_size", WriteConfig{
 			BlockSizeMb:                       -1,
 			CreateEmptyFile:                   false,

cmd/config_validation_test.go

@@ -24,6 +24,7 @@ import (
 	"time"
 
 	"github.com/googlecloudplatform/gcsfuse/v2/cfg"
+	"github.com/googlecloudplatform/gcsfuse/v2/internal/util"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -190,7 +191,7 @@ func TestValidateConfigFile_WriteConfig(t *testing.T) {
 			expectedConfig: &cfg.Config{
 				Write: cfg.WriteConfig{
 					CreateEmptyFile:                   false,
-					BlockSizeMb:                       64,
+					BlockSizeMb:                       64 * util.MiB,
 					ExperimentalEnableStreamingWrites: false,
 					GlobalMaxBlocks:                   math.MaxInt64,
 					MaxBlocksPerFile:                  math.MaxInt64},
@@ -202,7 +203,7 @@ func TestValidateConfigFile_WriteConfig(t *testing.T) {
 			expectedConfig: &cfg.Config{
 				Write: cfg.WriteConfig{
 					CreateEmptyFile:                   false, // changed due to enabled streaming writes.
-					BlockSizeMb:                       10,
+					BlockSizeMb:                       10 * util.MiB,
 					ExperimentalEnableStreamingWrites: true,
 					GlobalMaxBlocks:                   20,
 					MaxBlocksPerFile:                  2,

cmd/mount.go

@@ -168,6 +168,8 @@ func getFuseMountConfig(fsName string, newConfig *cfg.Config) *fuse.MountConfig
 		// access two files under same directory parallely, then the lookups also
 		// happen parallely.
 		EnableParallelDirOps: !(newConfig.FileSystem.DisableParallelDirops),
+		// We disable write-back cache when streaming writes are enabled.
+		DisableWritebackCaching: newConfig.Write.ExperimentalEnableStreamingWrites,
 	}
 
 	mountCfg.ErrorLogger = logger.NewLegacyLogger(logger.LevelError, "fuse: ")

cmd/root_test.go

@@ -24,6 +24,7 @@ import (
 	"time"
 
 	"github.com/googlecloudplatform/gcsfuse/v2/cfg"
+	"github.com/googlecloudplatform/gcsfuse/v2/internal/util"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -207,7 +208,7 @@ func TestArgsParsing_WriteConfigFlags(t *testing.T) {
 			args:                          []string{"gcsfuse", "--create-empty-file=true", "abc", "pqr"},
 			expectedCreateEmptyFile:       true,
 			expectedEnableStreamingWrites: false,
-			expectedWriteBlockSizeMB:      64,
+			expectedWriteBlockSizeMB:      64 * util.MiB,
 			expectedWriteGlobalMaxBlocks:  math.MaxInt64,
 			expectedWriteMaxBlocksPerFile: math.MaxInt64,
 		},
@@ -216,7 +217,7 @@ func TestArgsParsing_WriteConfigFlags(t *testing.T) {
 			args:                          []string{"gcsfuse", "--create-empty-file=false", "abc", "pqr"},
 			expectedCreateEmptyFile:       false,
 			expectedEnableStreamingWrites: false,
-			expectedWriteBlockSizeMB:      64,
+			expectedWriteBlockSizeMB:      64 * util.MiB,
 			expectedWriteGlobalMaxBlocks:  math.MaxInt64,
 			expectedWriteMaxBlocksPerFile: math.MaxInt64,
 		},
@@ -225,7 +226,7 @@ func TestArgsParsing_WriteConfigFlags(t *testing.T) {
 			args:                          []string{"gcsfuse", "abc", "pqr"},
 			expectedCreateEmptyFile:       false,
 			expectedEnableStreamingWrites: false,
-			expectedWriteBlockSizeMB:      64,
+			expectedWriteBlockSizeMB:      64 * util.MiB,
 			expectedWriteGlobalMaxBlocks:  math.MaxInt64,
 			expectedWriteMaxBlocksPerFile: math.MaxInt64,
 		},
@@ -234,7 +235,7 @@ func TestArgsParsing_WriteConfigFlags(t *testing.T) {
 			args:                          []string{"gcsfuse", "--experimental-enable-streaming-writes", "abc", "pqr"},
 			expectedCreateEmptyFile:       false,
 			expectedEnableStreamingWrites: true,
-			expectedWriteBlockSizeMB:      64,
+			expectedWriteBlockSizeMB:      64 * util.MiB,
 			expectedWriteGlobalMaxBlocks:  math.MaxInt64,
 			expectedWriteMaxBlocksPerFile: math.MaxInt64,
 		},
@@ -243,7 +244,7 @@ func TestArgsParsing_WriteConfigFlags(t *testing.T) {
 			args:                          []string{"gcsfuse", "--experimental-enable-streaming-writes=false", "abc", "pqr"},
 			expectedCreateEmptyFile:       false,
 			expectedEnableStreamingWrites: false,
-			expectedWriteBlockSizeMB:      64,
+			expectedWriteBlockSizeMB:      64 * util.MiB,
 			expectedWriteGlobalMaxBlocks:  math.MaxInt64,
 			expectedWriteMaxBlocksPerFile: math.MaxInt64,
 		},
@@ -252,7 +253,7 @@ func TestArgsParsing_WriteConfigFlags(t *testing.T) {
 			args:                          []string{"gcsfuse", "--experimental-enable-streaming-writes", "--write-block-size-mb=10", "abc", "pqr"},
 			expectedCreateEmptyFile:       false,
 			expectedEnableStreamingWrites: true,
-			expectedWriteBlockSizeMB:      10,
+			expectedWriteBlockSizeMB:      10 * util.MiB,
 			expectedWriteGlobalMaxBlocks:  math.MaxInt64,
 			expectedWriteMaxBlocksPerFile: math.MaxInt64,
 		},
@@ -261,7 +262,7 @@ func TestArgsParsing_WriteConfigFlags(t *testing.T) {
 			args:                          []string{"gcsfuse", "--experimental-enable-streaming-writes", "--write-global-max-blocks=10", "abc", "pqr"},
 			expectedCreateEmptyFile:       false,
 			expectedEnableStreamingWrites: true,
-			expectedWriteBlockSizeMB:      64,
+			expectedWriteBlockSizeMB:      64 * util.MiB,
 			expectedWriteGlobalMaxBlocks:  10,
 			expectedWriteMaxBlocksPerFile: math.MaxInt64,
 		},
@@ -270,7 +271,7 @@ func TestArgsParsing_WriteConfigFlags(t *testing.T) {
 			args:                          []string{"gcsfuse", "--experimental-enable-streaming-writes", "--write-max-blocks-per-file=10", "abc", "pqr"},
 			expectedCreateEmptyFile:       false,
 			expectedEnableStreamingWrites: true,
-			expectedWriteBlockSizeMB:      64,
+			expectedWriteBlockSizeMB:      64 * util.MiB,
 			expectedWriteGlobalMaxBlocks:  math.MaxInt64,
 			expectedWriteMaxBlocksPerFile: 10,
 		},

internal/util/util.go

@@ -34,6 +34,8 @@ const (
 	Parallel   = "Parallel"
 
 	MaxMiBsInUint64 uint64 = math.MaxUint64 >> 20
+	MaxMiBsInInt64  int64  = math.MaxInt64 >> 20
+	MiB                    = 1024 * 1024
 
 	// HeapSizeToRssConversionFactor is a constant factor
 	// which we multiply to the calculated heap-size