replicator: Copy object from the local storage without decoding

Signed-off-by: Leonard Lyubich <[email protected]>
nspcc-dev · Nov 23, 2023 · 3905d55 · 3905d55
1 parent 42cf34f
commit 3905d55
Show file tree

Hide file tree

Showing 23 changed files with 1,671 additions and 51 deletions.
diff --git a/cmd/blobovnicza-to-peapod/blobovniczatree/get.go b/cmd/blobovnicza-to-peapod/blobovniczatree/get.go
@@ -2,13 +2,15 @@ package blobovniczatree
 
 import (
 	"fmt"
+	"io"
 	"path/filepath"
 
 	"github.com/nspcc-dev/neofs-node/cmd/blobovnicza-to-peapod/blobovnicza"
 	"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/blobstor/common"
 	"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/util/logicerr"
 	apistatus "github.com/nspcc-dev/neofs-sdk-go/client/status"
 	objectSDK "github.com/nspcc-dev/neofs-sdk-go/object"
+	oid "github.com/nspcc-dev/neofs-sdk-go/object/id"
 	"go.uber.org/zap"
 )
 
@@ -141,3 +143,8 @@ func (b *Blobovniczas) getObject(blz *blobovnicza.Blobovnicza, prm blobovnicza.G
 
 	return common.GetRes{Object: obj, RawData: data}, nil
 }
+
+func (b *Blobovniczas) OpenObjectStream(objAddr oid.Address) (io.ReadSeekCloser, error) {
+	// FIXME
+	panic("OpenObjectStream must not be called on Blobovniczas")
+}
diff --git a/cmd/neofs-node/config/replicator/config.go b/cmd/neofs-node/config/replicator/config.go
@@ -31,3 +31,9 @@ func PutTimeout(c *config.Config) time.Duration {
 func PoolSize(c *config.Config) int {
 	return int(config.IntSafe(c.Sub(subsection), "pool_size"))
 }
+
+// UseSingleMessageBuffer returns value of the boolean
+// 'use_single_message_buffer' config parameter from 'replicator' section.
+func UseSingleMessageBuffer(c *config.Config) bool {
+	return config.BoolSafe(c.Sub(subsection), "use_single_message_buffer")
+}
diff --git a/cmd/neofs-node/config/replicator/config_test.go b/cmd/neofs-node/config/replicator/config_test.go
@@ -16,13 +16,15 @@ func TestReplicatorSection(t *testing.T) {
 
 		require.Equal(t, replicatorconfig.PutTimeoutDefault, replicatorconfig.PutTimeout(empty))
 		require.Equal(t, 0, replicatorconfig.PoolSize(empty))
+		require.Equal(t, false, replicatorconfig.UseSingleMessageBuffer(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))
+		require.Equal(t, true, replicatorconfig.UseSingleMessageBuffer(c))
 	}
 
 	configtest.ForEachFileType(path, fileConfigTest)

diff --git a/cmd/neofs-node/object.go b/cmd/neofs-node/object.go
@@ -187,6 +187,7 @@ func initObjectService(c *cfg) {
 		replicator.WithRemoteSender(
 			putsvc.NewRemoteSender(keyStorage, (*coreClientConstructor)(clientConstructor)),
 		),
+		replicator.CopyBinaryObjectWithSingleBuffer(replicatorconfig.UseSingleMessageBuffer(c.appCfg)),
 	)
 
 	c.policer = policer.New(

diff --git a/config/example/node.env b/config/example/node.env
@@ -84,6 +84,7 @@ NEOFS_POLICER_MAX_WORKERS=21
 # Replicator section
 NEOFS_REPLICATOR_PUT_TIMEOUT=15s
 NEOFS_REPLICATOR_POOL_SIZE=10
+NEOFS_REPLICATOR_USE_SINGLE_MESSAGE_BUFFER=true
 
 # Object service section
 NEOFS_OBJECT_DELETE_TOMBSTONE_LIFETIME=10

diff --git a/config/example/node.json b/config/example/node.json
@@ -118,7 +118,8 @@
   },
   "replicator": {
     "pool_size": 10,
-    "put_timeout": "15s"
+    "put_timeout": "15s",
+    "use_single_message_buffer": true
   },
   "object": {
     "delete": {

diff --git a/config/example/node.yaml b/config/example/node.yaml
@@ -107,6 +107,8 @@ policer:
 replicator:
   put_timeout: 15s  # timeout for the Replicator PUT remote operation (defaults to 1m)
   pool_size: 10     # maximum amount of concurrent replications
+  use_single_message_buffer: true # EXPERIMENTAL: whether single message buffer should be allocated
+    # when packing binary object into messages for replication to the particular remote storage node
 
 object:
   delete:

diff --git a/go.mod b/go.mod
@@ -17,9 +17,9 @@ require (
 	github.com/nspcc-dev/hrw/v2 v2.0.0-20231115095647-bf62f4ad0a43
 	github.com/nspcc-dev/locode-db v0.4.1-0.20231120112844-64febbb63ce3
 	github.com/nspcc-dev/neo-go v0.103.1
-	github.com/nspcc-dev/neofs-api-go/v2 v2.14.0
+	github.com/nspcc-dev/neofs-api-go/v2 v2.14.1-0.20231123192948-3118b92f4a4a
 	github.com/nspcc-dev/neofs-contract v0.18.0
-	github.com/nspcc-dev/neofs-sdk-go v1.0.0-rc.11.0.20231113180740-3b24af0410c0
+	github.com/nspcc-dev/neofs-sdk-go v1.0.0-rc.11.0.20231123193402-a60e899f7b32
 	github.com/nspcc-dev/tzhash v1.7.1
 	github.com/olekukonko/tablewriter v0.0.5
 	github.com/panjf2000/ants/v2 v2.8.2
@@ -50,7 +50,7 @@ require (
 	github.com/decred/dcrd/dcrec/secp256k1/v4 v4.0.1 // indirect
 	github.com/fsnotify/fsnotify v1.6.0 // indirect
 	github.com/golang/protobuf v1.5.3 // indirect
-	github.com/golang/snappy v0.0.3 // indirect
+	github.com/golang/snappy v0.0.4 // indirect
 	github.com/google/go-querystring v1.1.0 // indirect
 	github.com/gorilla/websocket v1.4.2 // indirect
 	github.com/hashicorp/golang-lru v1.0.2 // indirect
@@ -74,6 +74,7 @@ require (
 	github.com/nats-io/nuid v1.0.1 // indirect
 	github.com/nspcc-dev/dbft v0.0.0-20230515113611-25db6ba61d5c // indirect
 	github.com/nspcc-dev/go-ordered-json v0.0.0-20220111165707-25110be27d22 // indirect
+	github.com/nspcc-dev/hrw v1.0.9 // indirect
 	github.com/nspcc-dev/neo-go/pkg/interop v0.0.0-20231020160724-c3955f87d1b5 // indirect
 	github.com/nspcc-dev/neofs-crypto v0.4.0 // indirect
 	github.com/nspcc-dev/rfc6979 v0.2.0 // indirect
@@ -110,3 +111,5 @@ retract (
 	v1.22.1 // Contains retraction only.
 	v1.22.0 // Published accidentally.
 )
+
+replace google.golang.org/grpc => github.com/cthulhu-rider/grpc-go v0.0.0-20231123095204-ced09a8c28ff