[#1296] writecache: Add count limit

Signed-off-by: Dmitrii Stepanov <d.stepanov@yadro.com>

cmd/frostfs-node/config.go

@@ -152,6 +152,7 @@ type shardCfg struct {
 		maxObjSize       uint64
 		flushWorkerCount int
 		sizeLimit        uint64
+		countLimit       uint64
 		noSync           bool
 	}
 
@@ -275,6 +276,7 @@ func (a *applicationConfiguration) setShardWriteCacheConfig(newConfig *shardCfg,
 		wc.smallObjectSize = writeCacheCfg.SmallObjectSize()
 		wc.flushWorkerCount = writeCacheCfg.WorkerCount()
 		wc.sizeLimit = writeCacheCfg.SizeLimit()
+		wc.countLimit = writeCacheCfg.CountLimit()
 		wc.noSync = writeCacheCfg.NoSync()
 	}
 }
@@ -867,6 +869,7 @@ func (c *cfg) getWriteCacheOpts(shCfg shardCfg) []writecache.Option {
 			writecache.WithSmallObjectSize(wcRead.smallObjectSize),
 			writecache.WithFlushWorkersCount(wcRead.flushWorkerCount),
 			writecache.WithMaxCacheSize(wcRead.sizeLimit),
+			writecache.WithMaxCacheCount(wcRead.countLimit),
 			writecache.WithNoSync(wcRead.noSync),
 			writecache.WithLogger(c.log),
 		)

cmd/frostfs-node/config/engine/config_test.go

@@ -12,6 +12,7 @@ import (
 	fstreeconfig "git.frostfs.info/TrueCloudLab/frostfs-node/cmd/frostfs-node/config/engine/shard/blobstor/fstree"
 	gcconfig "git.frostfs.info/TrueCloudLab/frostfs-node/cmd/frostfs-node/config/engine/shard/gc"
 	piloramaconfig "git.frostfs.info/TrueCloudLab/frostfs-node/cmd/frostfs-node/config/engine/shard/pilorama"
+	writecacheconfig "git.frostfs.info/TrueCloudLab/frostfs-node/cmd/frostfs-node/config/engine/shard/writecache"
 	configtest "git.frostfs.info/TrueCloudLab/frostfs-node/cmd/frostfs-node/config/test"
 	"git.frostfs.info/TrueCloudLab/frostfs-node/pkg/local_object_storage/shard/mode"
 	"github.com/stretchr/testify/require"
@@ -78,6 +79,7 @@ func TestEngineSection(t *testing.T) {
 				require.EqualValues(t, 134217728, wc.MaxObjectSize())
 				require.EqualValues(t, 30, wc.WorkerCount())
 				require.EqualValues(t, 3221225472, wc.SizeLimit())
+				require.EqualValues(t, 49, wc.CountLimit())
 
 				require.Equal(t, "tmp/0/meta", meta.Path())
 				require.Equal(t, fs.FileMode(0o644), meta.BoltDB().Perm())
@@ -133,6 +135,7 @@ func TestEngineSection(t *testing.T) {
 				require.EqualValues(t, 134217728, wc.MaxObjectSize())
 				require.EqualValues(t, 30, wc.WorkerCount())
 				require.EqualValues(t, 4294967296, wc.SizeLimit())
+				require.EqualValues(t, writecacheconfig.CountLimitDefault, wc.CountLimit())
 
 				require.Equal(t, "tmp/1/meta", meta.Path())
 				require.Equal(t, fs.FileMode(0o644), meta.BoltDB().Perm())

cmd/frostfs-node/config/engine/shard/writecache/config.go

@@ -21,6 +21,9 @@ const (
 
 	// SizeLimitDefault is a default write-cache size limit.
 	SizeLimitDefault = 1 << 30
+
+	// CountLimitDefault is a default write-cache count limit.
+	CountLimitDefault = 0
 )
 
 // From wraps config section into Config.
@@ -115,6 +118,22 @@ func (x *Config) SizeLimit() uint64 {
 	return SizeLimitDefault
 }
 
+// CountLimit returns the value of "max_object_count" config parameter.
+//
+// Returns CountLimitDefault if the value is not a positive number.
+func (x *Config) CountLimit() uint64 {
+	c := config.SizeInBytesSafe(
+		(*config.Config)(x),
+		"max_object_count",
+	)
+
+	if c > 0 {
+		return c
+	}
+
+	return CountLimitDefault
+}
+
 // NoSync returns the value of "no_sync" config parameter.
 //
 // Returns false if the value is not a boolean.

config/example/node.env

@@ -105,6 +105,7 @@ FROSTFS_STORAGE_SHARD_0_WRITECACHE_SMALL_OBJECT_SIZE=16384
 FROSTFS_STORAGE_SHARD_0_WRITECACHE_MAX_OBJECT_SIZE=134217728
 FROSTFS_STORAGE_SHARD_0_WRITECACHE_FLUSH_WORKER_COUNT=30
 FROSTFS_STORAGE_SHARD_0_WRITECACHE_CAPACITY=3221225472
+FROSTFS_STORAGE_SHARD_0_WRITECACHE_MAX_OBJECT_COUNT=49
 ### Metabase config
 FROSTFS_STORAGE_SHARD_0_METABASE_PATH=tmp/0/meta
 FROSTFS_STORAGE_SHARD_0_METABASE_PERM=0644

config/example/node.json

@@ -148,7 +148,8 @@
           "small_object_size": 16384,
           "max_object_size": 134217728,
           "flush_worker_count": 30,
-          "capacity": 3221225472
+          "capacity": 3221225472,
+          "max_object_count": 49
         },
         "metabase": {
           "path": "tmp/0/meta",

config/example/node.yaml

@@ -171,6 +171,7 @@ storage:
         no_sync: true
         path: tmp/0/cache  # write-cache root directory
         capacity: 3221225472  # approximate write-cache total size, bytes
+        max_object_count: 49
 
       metabase:
         path: tmp/0/meta  # metabase path

docs/storage-node-configuration.md

@@ -293,9 +293,10 @@ writecache:
 ```
 
 | Parameter            | Type       | Default value | Description                                                                                                                   |
-|----------------------|------------|---------------|----------------------------------------------------------------------------------------------------------------------|
+|----------------------|------------|---------------|-------------------------------------------------------------------------------------------------------------------------------|
 | `path`               | `string`   |               | Path to the metabase file.                                                                                                    |
-| `capacity`           | `size`     | unrestricted  | Approximate maximum size of the writecache. If the writecache is full, objects are written to the blobstor directly. |
+| `capacity`           | `size`     | `1G`          | Approximate maximum size of the writecache. If the writecache is full, objects are written to the blobstor directly.          |
+| `max_object_count`   | `int`      | unrestricted  | Approximate maximum objects count in the writecache. If the writecache is full, objects are written to the blobstor directly. |
 | `small_object_size`  | `size`     | `32K`         | Maximum object size for "small" objects. This objects are stored in a key-value database instead of a file-system.            |
 | `max_object_size`    | `size`     | `64M`         | Maximum object size allowed to be stored in the writecache.                                                                   |
 | `flush_worker_count` | `int`      | `20`          | Amount of background workers that move data from the writecache to the blobstor.                                              |

pkg/local_object_storage/writecache/options.go

@@ -29,6 +29,9 @@ type options struct {
 	// maxCacheSize is the maximum total size of all objects saved in cache (DB + FS).
 	// 1 GiB by default.
 	maxCacheSize uint64
+	// maxCacheCount is the maximum total count of all object saved in cache.
+	// 0 (no limit) by default.
+	maxCacheCount uint64
 	// objCounters contains atomic counters for the number of objects stored in cache.
 	objCounters counters
 	// maxBatchSize is the maximum batch size for the small object database.
@@ -108,6 +111,13 @@ func WithMaxCacheSize(sz uint64) Option {
 	}
 }
 
+// WithMaxCacheCount sets maximum write-cache objects count.
+func WithMaxCacheCount(v uint64) Option {
+	return func(o *options) {
+		o.maxCacheCount = v
+	}
+}
+
 // WithMaxBatchSize sets max batch size for the small object database.
 func WithMaxBatchSize(sz int) Option {
 	return func(o *options) {

pkg/local_object_storage/writecache/put.go

@@ -76,8 +76,7 @@ func (c *cache) Put(ctx context.Context, prm common.PutPrm) (common.PutRes, erro
 // putSmall persists small objects to the write-cache database and
 // pushes the to the flush workers queue.
 func (c *cache) putSmall(obj objectInfo) error {
-	cacheSize := c.estimateCacheSize()
+	if !c.hasEnoughSpaceDB() {
-	if c.maxCacheSize < c.incSizeDB(cacheSize) {
 		return ErrOutOfSpace
 	}
 
@@ -107,8 +106,7 @@ func (c *cache) putSmall(obj objectInfo) error {
 
 // putBig writes object to FSTree and pushes it to the flush workers queue.
 func (c *cache) putBig(ctx context.Context, addr string, prm common.PutPrm) error {
-	cacheSz := c.estimateCacheSize()
+	if !c.hasEnoughSpaceFS() {
-	if c.maxCacheSize < c.incSizeFS(cacheSz) {
 		return ErrOutOfSpace
 	}
 

pkg/local_object_storage/writecache/state.go

@@ -9,7 +9,7 @@ import (
 	"go.etcd.io/bbolt"
 )
 
-func (c *cache) estimateCacheSize() uint64 {
+func (c *cache) estimateCacheSize() (uint64, uint64) {
 	dbCount := c.objCounters.DB()
 	fsCount := c.objCounters.FS()
 	if fsCount > 0 {
@@ -19,15 +19,23 @@ func (c *cache) estimateCacheSize() uint64 {
 	fsSize := fsCount * c.maxObjectSize
 	c.metrics.SetEstimateSize(dbSize, fsSize)
 	c.metrics.SetActualCounters(dbCount, fsCount)
-	return dbSize + fsSize
+	return dbCount + fsCount, dbSize + fsSize
 }
 
-func (c *cache) incSizeDB(sz uint64) uint64 {
+func (c *cache) hasEnoughSpaceDB() bool {
-	return sz + c.smallObjectSize
+	return c.hasEnoughSpace(c.smallObjectSize)
 }
 
-func (c *cache) incSizeFS(sz uint64) uint64 {
+func (c *cache) hasEnoughSpaceFS() bool {
-	return sz + c.maxObjectSize
+	return c.hasEnoughSpace(c.maxObjectSize)
+}
+
+func (c *cache) hasEnoughSpace(objectSize uint64) bool {
+	count, size := c.estimateCacheSize()
+	if c.maxCacheCount > 0 && count+1 > c.maxCacheCount {
+		return false
+	}
+	return c.maxCacheSize >= size+objectSize
 }
 
 var _ fstree.FileCounter = &counters{}