[#2003] neofs-node: Allow to configure replicator pool size

Signed-off-by: Evgenii Stratonikov <evgeniy@morphbits.ru>

CHANGELOG.md

@@ -9,6 +9,7 @@ Changelog for NeoFS Node
 - `session` flag support to `neofs-cli object hash` (#2029)
 - Shard can now change mode when encountering background disk errors (#2035)
 - Background workers and object service now use separate client caches (#2048)
+- `replicator.pool_size` config field to tune replicator pool size (#2049)
 
 ### Changed
 - `object lock` command reads CID and OID the same way other commands do (#1971)
@@ -35,6 +36,9 @@ Changelog for NeoFS Node
 ### Updating from v0.34.0
 Pass CID and OID parameters via the `--cid` and `--oid` flags, not as the command arguments.
 
+Replicator pool size can now be fine-tuned with `replicator.pool_size` config field.
+The default value is taken from `object.put.pool_size_remote` as in earlier versions.
+
 ## [0.34.0] - 2022-10-31 - Marado (마라도, 馬羅島)
 
 ### Added

cmd/neofs-node/config.go

@@ -29,6 +29,7 @@ import (
 	metricsconfig "github.com/nspcc-dev/neofs-node/cmd/neofs-node/config/metrics"
 	nodeconfig "github.com/nspcc-dev/neofs-node/cmd/neofs-node/config/node"
 	objectconfig "github.com/nspcc-dev/neofs-node/cmd/neofs-node/config/object"
+	replicatorconfig "github.com/nspcc-dev/neofs-node/cmd/neofs-node/config/replicator"
 	"github.com/nspcc-dev/neofs-node/pkg/core/container"
 	netmapCore "github.com/nspcc-dev/neofs-node/pkg/core/netmap"
 	"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/blobstor"
@@ -488,6 +489,8 @@ type cfgObjectRoutines struct {
 
 	putRemoteCapacity int
 
+	replicatorPoolSize int
+
 	replication *ants.Pool
 }
 
@@ -821,7 +824,12 @@ func initObjectPool(cfg *config.Config) (pool cfgObjectRoutines) {
 	pool.putRemote, err = ants.NewPool(pool.putRemoteCapacity, optNonBlocking)
 	fatalOnErr(err)
 
-	pool.replication, err = ants.NewPool(pool.putRemoteCapacity)
+	pool.replicatorPoolSize = replicatorconfig.PoolSize(cfg)
+	if pool.replicatorPoolSize <= 0 {
+		pool.replicatorPoolSize = pool.putRemoteCapacity
+	}
+
+	pool.replication, err = ants.NewPool(pool.replicatorPoolSize)
 	fatalOnErr(err)
 
 	return pool

cmd/neofs-node/config/replicator/config.go

@@ -25,3 +25,9 @@ func PutTimeout(c *config.Config) time.Duration {
 
 	return PutTimeoutDefault
 }
+
+// PoolSize returns the value of "pool_size" config parameter
+// from "replicator" section.
+func PoolSize(c *config.Config) int {
+	return int(config.IntSafe(c.Sub(subsection), "pool_size"))
+}

cmd/neofs-node/config/replicator/config_test.go

@@ -15,12 +15,14 @@ func TestReplicatorSection(t *testing.T) {
 		empty := configtest.EmptyConfig()
 
 		require.Equal(t, replicatorconfig.PutTimeoutDefault, replicatorconfig.PutTimeout(empty))
+		require.Equal(t, 0, replicatorconfig.PoolSize(empty))
 	})
 
 	const path = "../../../../config/example/node"
 
 	var fileConfigTest = func(c *config.Config) {
 		require.Equal(t, 15*time.Second, replicatorconfig.PutTimeout(c))
+		require.Equal(t, 10, replicatorconfig.PoolSize(c))
 	}
 
 	configtest.ForEachFileType(path, fileConfigTest)

cmd/neofs-node/object.go

@@ -231,7 +231,7 @@ func initObjectService(c *cfg) {
 				)
 			}
 		}),
-		policer.WithMaxCapacity(c.cfgObject.pool.putRemoteCapacity),
+		policer.WithMaxCapacity(c.cfgObject.pool.replicatorPoolSize),
 		policer.WithPool(c.cfgObject.pool.replication),
 		policer.WithNodeLoader(c),
 	)

config/example/node.env

@@ -78,6 +78,7 @@ NEOFS_POLICER_HEAD_TIMEOUT=15s
 
 # Replicator section
 NEOFS_REPLICATOR_PUT_TIMEOUT=15s
+NEOFS_REPLICATOR_POOL_SIZE=10
 
 # Object service section
 NEOFS_OBJECT_PUT_POOL_SIZE_REMOTE=100

config/example/node.json

@@ -121,6 +121,7 @@
     "head_timeout": "15s"
   },
   "replicator": {
+    "pool_size": 10,
     "put_timeout": "15s"
   },
   "object": {

config/example/node.yaml

@@ -101,6 +101,7 @@ policer:
 
 replicator:
   put_timeout: 15s  # timeout for the Replicator PUT remote operation
+  pool_size: 10     # maximum amount of concurrent replications
 
 object:
   put:

docs/storage-node-configuration.md

@@ -404,11 +404,13 @@ Configuration for the Replicator service.
 ```yaml
 replicator:
   put_timeout: 15s
+  pool_size: 10
 ```
 
-| Parameter     | Type       | Default value | Description                                 |
-|---------------|------------|---------------|---------------------------------------------|
-| `put_timeout` | `duration` | `5s`          | Timeout for performing the `PUT` operation. |
+| Parameter     | Type       | Default value                          | Description                                 |
+|---------------|------------|----------------------------------------|---------------------------------------------|
+| `put_timeout` | `duration` | `5s`                                   | Timeout for performing the `PUT` operation. |
+| `pool_size`   | `int`      | Equal to `object.put.pool_size_remote` | Maximum amount of concurrent replications.  |
 
 # `object` section
 Contains pool sizes for object operations with remote nodes.