cmd/frostfs-node/config.go

@@ -632,14 +632,8 @@ type cfgAccessPolicyEngine struct {
 type cfgObjectRoutines struct {
 	putRemote *ants.Pool
 
-	putRemoteCapacity int
-
 	putLocal *ants.Pool
 
-	putLocalCapacity int
-
-	replicatorPoolSize int
-
 	replication *ants.Pool
 }
 
@@ -1094,20 +1088,20 @@ func initObjectPool(cfg *config.Config) (pool cfgObjectRoutines) {
 
 	optNonBlocking := ants.WithNonblocking(true)
 
-	pool.putRemoteCapacity = objectconfig.Put(cfg).PoolSizeRemote()
-	pool.putRemote, err = ants.NewPool(pool.putRemoteCapacity, optNonBlocking)
+	putRemoteCapacity := objectconfig.Put(cfg).PoolSizeRemote()
+	pool.putRemote, err = ants.NewPool(putRemoteCapacity, optNonBlocking)
 	fatalOnErr(err)
 
-	pool.putLocalCapacity = objectconfig.Put(cfg).PoolSizeLocal()
-	pool.putLocal, err = ants.NewPool(pool.putLocalCapacity, optNonBlocking)
+	putLocalCapacity := objectconfig.Put(cfg).PoolSizeLocal()
+	pool.putLocal, err = ants.NewPool(putLocalCapacity, optNonBlocking)
 	fatalOnErr(err)
 
-	pool.replicatorPoolSize = replicatorconfig.PoolSize(cfg)
-	if pool.replicatorPoolSize <= 0 {
-		pool.replicatorPoolSize = pool.putRemoteCapacity
+	replicatorPoolSize := replicatorconfig.PoolSize(cfg)
+	if replicatorPoolSize <= 0 {
+		replicatorPoolSize = putRemoteCapacity
 	}
 
-	pool.replication, err = ants.NewPool(pool.replicatorPoolSize)
+	pool.replication, err = ants.NewPool(replicatorPoolSize)
 	fatalOnErr(err)
 
 	return pool
@@ -1241,6 +1235,7 @@ func (c *cfg) reloadConfig(ctx context.Context) {
 		setRuntimeParameters(c)
 		return nil
 	}})
+	components = append(components, dCmp{"pools", c.reloadPools})
 	components = append(components, dCmp{"tracing", func() error {
 		updated, err := tracing.Setup(ctx, *tracingconfig.ToTracingConfig(c.appCfg))
 		if updated {
@@ -1285,6 +1280,28 @@ func (c *cfg) reloadConfig(ctx context.Context) {
 	c.log.Info(logs.FrostFSNodeConfigurationHasBeenReloadedSuccessfully)
 }
 
+func (c *cfg) reloadPools() error {
+	newSize := objectconfig.Put(c.appCfg).PoolSizeLocal()
+	c.reloadPool(c.cfgObject.pool.putLocal, newSize, "object.put.local_pool_size")
+
+	newSize = objectconfig.Put(c.appCfg).PoolSizeRemote()
+	c.reloadPool(c.cfgObject.pool.putRemote, newSize, "object.put.remote_pool_size")
+
+	newSize = replicatorconfig.PoolSize(c.appCfg)
+	c.reloadPool(c.cfgObject.pool.replication, newSize, "replicator.pool_size")
+
+	return nil
+}
+
+func (c *cfg) reloadPool(p *ants.Pool, newSize int, name string) {
+	oldSize := p.Cap()
+	if oldSize != newSize {
+		c.log.Info(logs.FrostFSNodePoolConfigurationUpdate, zap.String("field", name),
+			zap.Int("old", oldSize), zap.Int("new", newSize))
+		p.Tune(newSize)
+	}
+}
+
 func (c *cfg) reloadAppConfig() error {
 	unlock := c.LockAppConfigExclusive()
 	defer unlock()

cmd/frostfs-node/object.go

@@ -263,7 +263,6 @@ func addPolicer(c *cfg, keyStorage *util.KeyStorage, clientConstructor *cache.Cl
 				)
 			}
 		}),
-		policer.WithMaxCapacity(c.cfgObject.pool.replicatorPoolSize),
 		policer.WithPool(c.cfgObject.pool.replication),
 		policer.WithMetrics(c.metricsCollector.PolicerMetrics()),
 	)

internal/logs/logs.go

@@ -451,6 +451,7 @@ const (
 	FrostFSNodeLoggerConfigurationPreparation                               = "logger configuration preparation"
 	FrostFSNodeTracingConfigationUpdated                                    = "tracing configation updated"
 	FrostFSNodeStorageEngineConfigurationUpdate                             = "storage engine configuration update"
+	FrostFSNodePoolConfigurationUpdate                                      = "adjust pool configuration"
 	FrostFSNodeUpdatedConfigurationApplying                                 = "updated configuration applying"
 	FrostFSNodeConfigurationHasBeenReloadedSuccessfully                     = "configuration has been reloaded successfully"
 	FrostFSNodeReadNewlyCreatedContainerAfterTheNotification                = "read newly created container after the notification"

pkg/services/policer/option.go

@@ -64,8 +64,6 @@ type cfg struct {
 
 	taskPool *ants.Pool
 
-	maxCapacity int
-
 	batchSize, cacheSize uint32
 
 	rebalanceFreq, evictDuration, sleepDuration time.Duration
@@ -158,14 +156,6 @@ func WithRedundantCopyCallback(cb RedundantCopyCallback) Option {
 	}
 }
 
-// WithMaxCapacity returns option to set max capacity
-// that can be set to the pool.
-func WithMaxCapacity(capacity int) Option {
-	return func(c *cfg) {
-		c.maxCapacity = capacity
-	}
-}
-
 // WithPool returns option to set pool for
 // policy and replication operations.
 func WithPool(p *ants.Pool) Option {

pkg/services/policer/policer.go

@@ -66,7 +66,7 @@ func New(opts ...Option) *Policer {
 		cfg:   c,
 		cache: cache,
 		objsInWork: &objectsInWork{
-			objs: make(map[oid.Address]struct{}, c.maxCapacity),
+			objs: make(map[oid.Address]struct{}, c.taskPool.Cap()),
 		},
 	}
 }

pkg/services/policer/policer_test.go

@@ -48,16 +48,12 @@ func TestBuryObjectWithoutContainer(t *testing.T) {
 		return nil
 	}
 
-	// Task pool
-	pool, err := ants.NewPool(4)
-	require.NoError(t, err)
-
 	// Policer instance
 	p := New(
 		WithKeySpaceIterator(&sliceKeySpaceIterator{objs: objs}),
 		WithContainerSource(containerSrc),
 		WithBuryFunc(buryFn),
-		WithPool(pool),
+		WithPool(testPool(t)),
 	)
 
 	ctx, cancel := context.WithCancel(context.Background())
@@ -239,6 +235,7 @@ func TestProcessObject(t *testing.T) {
 						gotReplicateTo = append(gotReplicateTo, int(node.PublicKey()[0]))
 					}
 				})),
+				WithPool(testPool(t)),
 			)
 
 			addrWithType := objectcore.AddressWithType{
@@ -276,6 +273,7 @@ func TestProcessObjectError(t *testing.T) {
 	p := New(
 		WithContainerSource(source),
 		WithBuryFunc(buryFn),
+		WithPool(testPool(t)),
 	)
 
 	addrWithType := objectcore.AddressWithType{
@@ -296,9 +294,6 @@ func TestIteratorContract(t *testing.T) {
 		return nil
 	}
 
-	pool, err := ants.NewPool(4)
-	require.NoError(t, err)
-
 	it := &predefinedIterator{
 		scenario: []nextResult{
 			{objs, nil},
@@ -324,7 +319,7 @@ func TestIteratorContract(t *testing.T) {
 		WithKeySpaceIterator(it),
 		WithContainerSource(containerSrc),
 		WithBuryFunc(buryFn),
-		WithPool(pool),
+		WithPool(testPool(t)),
 		func(c *cfg) {
 			c.sleepDuration = time.Millisecond
 		},
@@ -348,6 +343,12 @@ func TestIteratorContract(t *testing.T) {
 	}, it.calls)
 }
 
+func testPool(t *testing.T) *ants.Pool {
+	pool, err := ants.NewPool(4)
+	require.NoError(t, err)
+	return pool
+}
+
 type nextResult struct {
 	objs []objectcore.AddressWithType
 	err  error