cmd/mount: add max-writeback-size option to filter blocks for writeback (#6447)

Signed-off-by: jiefenghuang <[email protected]>

Author: jiefenghuang

cmd/flags.go

@@ -196,6 +196,11 @@ func dataCacheFlags() []cli.Flag {
 			Name:  "writeback",
 			Usage: "upload blocks in background",
 		},
+		&cli.StringFlag{
+			Name:  "writeback-threshold-size",
+			Value: "0",
+			Usage: "blocks smaller than this size will be staged, 0 means all staged.",
+		},
 		&cli.StringFlag{
 			Name:  "upload-delay",
 			Value: "0s",

cmd/mount.go

@@ -364,18 +364,19 @@ func getChunkConf(c *cli.Context, format *meta.Format) *chunk.Config {
 		Compress:   format.Compression,
 		HashPrefix: format.HashPrefix,
 
-		GetTimeout:    utils.Duration(c.String("get-timeout")),
-		PutTimeout:    utils.Duration(c.String("put-timeout")),
-		MaxUpload:     c.Int("max-uploads"),
-		MaxStageWrite: c.Int("max-stage-write"),
-		MaxRetries:    c.Int("io-retries"),
-		Writeback:     c.Bool("writeback"),
-		Prefetch:      c.Int("prefetch"),
-		BufferSize:    utils.ParseBytes(c, "buffer-size", 'M'),
-		UploadLimit:   utils.ParseMbps(c, "upload-limit") * 1e6 / 8,
-		DownloadLimit: utils.ParseMbps(c, "download-limit") * 1e6 / 8,
-		UploadDelay:   utils.Duration(c.String("upload-delay")),
-		UploadHours:   c.String("upload-hours"),
+		GetTimeout:             utils.Duration(c.String("get-timeout")),
+		PutTimeout:             utils.Duration(c.String("put-timeout")),
+		MaxUpload:              c.Int("max-uploads"),
+		MaxStageWrite:          c.Int("max-stage-write"),
+		MaxRetries:             c.Int("io-retries"),
+		Writeback:              c.Bool("writeback"),
+		WritebackThresholdSize: int(utils.ParseBytes(c, "writeback-threshold-size", 'B')),
+		Prefetch:               c.Int("prefetch"),
+		BufferSize:             utils.ParseBytes(c, "buffer-size", 'M'),
+		UploadLimit:            utils.ParseMbps(c, "upload-limit") * 1e6 / 8,
+		DownloadLimit:          utils.ParseMbps(c, "download-limit") * 1e6 / 8,
+		UploadDelay:            utils.Duration(c.String("upload-delay")),
+		UploadHours:            c.String("upload-hours"),
 
 		CacheDir:          c.String("cache-dir"),
 		CacheSize:         utils.ParseBytes(c, "cache-size", 'M'),

pkg/chunk/cached_store.go

@@ -454,7 +454,7 @@ func (s *wSlice) upload(indx int) {
 		if off != blen {
 			panic(fmt.Sprintf("block length does not match: %v != %v", off, blen))
 		}
-		if s.writeback {
+		if s.writeback && blen < s.store.conf.WritebackThresholdSize {
 			stagingPath := "unknown"
 			stageFailed := false
 			block.Acquire()
@@ -559,35 +559,36 @@ func (s *wSlice) Abort() {
 
 // Config contains options for cachedStore
 type Config struct {
-	CacheDir          string
-	CacheMode         os.FileMode
-	CacheSize         uint64
-	CacheItems        int64
-	CacheChecksum     string
-	CacheEviction     string
-	CacheScanInterval time.Duration
-	CacheExpire       time.Duration
-	OSCache           bool
-	FreeSpace         float32
-	AutoCreate        bool
-	Compress          string
-	MaxUpload         int
-	MaxStageWrite     int
-	MaxRetries        int
-	UploadLimit       int64 // bytes per second
-	DownloadLimit     int64 // bytes per second
-	Writeback         bool
-	UploadDelay       time.Duration
-	UploadHours       string
-	HashPrefix        bool
-	BlockSize         int
-	GetTimeout        time.Duration
-	PutTimeout        time.Duration
-	CacheFullBlock    bool
-	CacheLargeWrite   bool
-	BufferSize        uint64
-	Readahead         int
-	Prefetch          int
+	CacheDir               string
+	CacheMode              os.FileMode
+	CacheSize              uint64
+	CacheItems             int64
+	CacheChecksum          string
+	CacheEviction          string
+	CacheScanInterval      time.Duration
+	CacheExpire            time.Duration
+	OSCache                bool
+	FreeSpace              float32
+	AutoCreate             bool
+	Compress               string
+	MaxUpload              int
+	MaxStageWrite          int
+	MaxRetries             int
+	UploadLimit            int64 // bytes per second
+	DownloadLimit          int64 // bytes per second
+	Writeback              bool
+	WritebackThresholdSize int
+	UploadDelay            time.Duration
+	UploadHours            string
+	HashPrefix             bool
+	BlockSize              int
+	GetTimeout             time.Duration
+	PutTimeout             time.Duration
+	CacheFullBlock         bool
+	CacheLargeWrite        bool
+	BufferSize             uint64
+	Readahead              int
+	Prefetch               int
 }
 
 func (c *Config) SelfCheck(uuid string) {
@@ -599,20 +600,6 @@ func (c *Config) SelfCheck(uuid string) {
 		}
 		c.CacheDir = "memory"
 	}
-	if !c.Writeback {
-		if c.UploadDelay > 0 || c.UploadHours != "" {
-			logger.Warnf("delayed upload is disabled in non-writeback mode")
-			c.UploadDelay = 0
-			c.UploadHours = ""
-		}
-	}
-	if !c.CacheFullBlock && c.Writeback {
-		logger.Warnf("cache-partial-only is ineffective for stage blocks with writeback enabled")
-	}
-	if _, _, err := c.parseHours(); err != nil {
-		logger.Warnf("invalid value (%s) for upload-hours: %s", c.UploadHours, err)
-		c.UploadHours = ""
-	}
 	if c.MaxUpload <= 0 {
 		logger.Warnf("max-uploads should be greater than 0, set it to 1")
 		c.MaxUpload = 1
@@ -635,6 +622,24 @@ func (c *Config) SelfCheck(uuid string) {
 		logger.Warnf("writeback is not supported in memory cache mode")
 		c.Writeback = false
 	}
+	if c.Writeback {
+		if !c.CacheFullBlock {
+			logger.Warnf("cache-partial-only is ineffective for stage blocks with writeback enabled")
+		}
+		if c.WritebackThresholdSize == 0 {
+			c.WritebackThresholdSize = c.BlockSize + 1
+		}
+	} else {
+		if c.UploadDelay > 0 || c.UploadHours != "" {
+			logger.Warnf("delayed upload is disabled in non-writeback mode")
+			c.UploadDelay = 0
+			c.UploadHours = ""
+		}
+	}
+	if _, _, err := c.parseHours(); err != nil {
+		logger.Warnf("invalid value (%s) for upload-hours: %s", c.UploadHours, err)
+		c.UploadHours = ""
+	}
 	if c.CacheEviction == "" {
 		c.CacheEviction = Eviction2Random
 	} else if c.CacheEviction != Eviction2Random && c.CacheEviction != EvictionNone && c.CacheEviction != EvictionLRU {
@@ -818,7 +823,7 @@ func NewCachedStore(storage object.ObjectStorage, config Config, reg prometheus.
 		store.downLimit = ratelimit.NewBucketWithRate(float64(config.DownloadLimit)*0.85, config.DownloadLimit/10)
 	}
 	store.initMetrics()
-	if store.conf.CacheDir != "memory" && store.conf.Writeback {
+	if store.conf.Writeback {
 		store.startHour, store.endHour, _ = config.parseHours()
 		if store.startHour != store.endHour {
 			logger.Infof("background upload at %d:00 ~ %d:00", store.startHour, store.endHour)
@@ -866,7 +871,7 @@ func NewCachedStore(storage object.ObjectStorage, config Config, reg prometheus.
 		}
 	})
 
-	if store.conf.CacheDir != "memory" && store.conf.Writeback {
+	if store.conf.Writeback {
 		for i := 0; i < store.conf.MaxUpload; i++ {
 			go store.uploader()
 		}

pkg/chunk/cached_store_test.go

@@ -204,6 +204,7 @@ func TestForceUpload(t *testing.T) {
 	config := defaultConf
 	_ = os.RemoveAll(config.CacheDir)
 	config.Writeback = true
+	config.WritebackThresholdSize = config.BlockSize + 1
 	config.UploadDelay = time.Hour
 	config.BlockSize = 4 << 20
 	store := NewCachedStore(blob, config, nil)