services/object: use stream API for Get operation

Network API always had a concept of payload stream, FSTree can also provide it
now after #3431. So use `GetStream` in object service to enable memory-efficient
object retrieval without loading large payloads into memory.

Closes #3439.

Signed-off-by: Andrey Butusov <[email protected]>

Author: End-rey

CHANGELOG.md

@@ -8,6 +8,7 @@ Changelog for NeoFS Node
 ### Fixed
 
 ### Changed
+- Use stream API of FSTree for object service `Get` operation (#3466)
 
 ### Removed
 - `neofs-cli object head --main-only` no-op flag (#3509)

pkg/services/object/get/assemble.go

@@ -66,7 +66,7 @@ func (exec *execCtx) assemble() {
 
 			if ok := exec.overtakePayloadInReverse(children[len(children)-1]); ok {
 				// payload of all children except the last are written, write last payload
-				exec.writeObjectPayload(exec.collectedObject)
+				exec.writeObjectPayload(exec.collectedObject, exec.collectedReader)
 			}
 		}
 	} else if prev != nil {
@@ -77,7 +77,7 @@ func (exec *execCtx) assemble() {
 
 			if ok := exec.overtakePayloadInReverse(*prev); ok {
 				// payload of all children except the last are written, write last payloa
-				exec.writeObjectPayload(exec.collectedObject)
+				exec.writeObjectPayload(exec.collectedObject, exec.collectedReader)
 			}
 		}
 	} else {
@@ -163,12 +163,8 @@ func (exec *execCtx) overtakePayloadDirectly(children []oid.ID, rngs []objectSDK
 			r = &rngs[i]
 		}
 
-		child, ok := exec.getChild(children[i], r, !withRng && checkRight)
-		if !ok {
-			return
-		}
-
-		if ok := exec.writeObjectPayload(child); !ok {
+		retrieved, wrote := exec.copyChild(children[i], r, !withRng && checkRight)
+		if !retrieved || !wrote {
 			return
 		}
 	}

pkg/services/object/get/assembly_v2.go

@@ -51,7 +51,7 @@ func (exec *execCtx) processV2Last(lastID oid.ID) {
 	// see it for comments and optimization suggestions
 	if ok := exec.writeCollectedHeader(); ok {
 		if ok := exec.overtakePayloadInReverse(lastID); ok {
-			exec.writeObjectPayload(exec.collectedObject)
+			exec.writeObjectPayload(exec.collectedObject, exec.collectedReader)
 		}
 	}
 }
@@ -132,13 +132,11 @@ func (exec *execCtx) rangeFromLink(link objectSDK.Link) bool {
 			}
 		}
 
-		part, ok := exec.getChild(child.ObjectID(), rngPerChild, false)
-		if !ok {
+		retrieved, wrote := exec.copyChild(child.ObjectID(), rngPerChild, false)
+		if !retrieved { // failed to fetch child -> abort whole operation
 			return false
 		}
-
-		if !exec.writeObjectPayload(part) {
-			// we have payload, we want to send it but can't so stop here
+		if !wrote { // payload fetch ok but writing failed -> stop further processing
 			return true
 		}
 	}

pkg/services/object/get/exec.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"crypto/ecdsa"
 	"errors"
+	"io"
 
 	clientcore "github.com/nspcc-dev/neofs-node/pkg/core/client"
 	"github.com/nspcc-dev/neofs-node/pkg/core/object"
@@ -14,6 +15,12 @@ import (
 	"go.uber.org/zap"
 )
 
+const (
+	// streamChunkSize is the size of the chunk that is used to read/write
+	// object payload from/to the stream.
+	streamChunkSize = 64 * 1024 // 64 KiB
+)
+
 type statusError struct {
 	status int
 	err    error
@@ -35,6 +42,7 @@ type execCtx struct {
 	log *zap.Logger
 
 	collectedObject *objectSDK.Object
+	collectedReader io.ReadCloser
 
 	curOff uint64
 
@@ -198,6 +206,44 @@ func (exec *execCtx) getChild(id oid.ID, rng *objectSDK.Range, withHdr bool) (*o
 	return child, ok
 }
 
+// copyChild fetches child object payload and streams it directly into current exec writer.
+// Returns if child header was received and if full payload was successfully written.
+func (exec *execCtx) copyChild(id oid.ID, rng *objectSDK.Range, withHdr bool) (bool, bool) {
+	log := exec.log
+	if rng != nil {
+		log = log.With(zap.String("child range", prettyRange(rng)))
+	}
+
+	childWriter := NewDirectChildWriter(exec.prm.objWriter)
+
+	p := exec.prm
+	p.common = p.common.WithLocalOnly(false)
+	p.objWriter = childWriter
+	p.SetRange(rng)
+	p.addr.SetContainer(exec.containerID())
+	p.addr.SetObject(id)
+
+	exec.statusError = exec.svc.get(exec.context(), p.commonPrm, withPayloadRange(rng), withLogger(log))
+
+	obj := childWriter.Object()
+	ok := exec.status == statusOK
+	if ok && withHdr && !exec.isChild(obj) {
+		exec.status = statusUndefined
+		exec.err = errors.New("wrong child header")
+
+		exec.log.Debug("parent address in child object differs")
+	}
+
+	if !ok {
+		if childWriter.HeaderSet() {
+			return true, false
+		}
+		return false, false
+	}
+
+	return true, true
+}
+
 func (exec *execCtx) headChild(id oid.ID) (*objectSDK.Object, bool) {
 	p := exec.prm
 	p.common = p.common.WithLocalOnly(false)
@@ -299,12 +345,18 @@ func (exec *execCtx) writeCollectedHeader() bool {
 	return exec.status == statusOK
 }
 
-func (exec *execCtx) writeObjectPayload(obj *objectSDK.Object) bool {
+func (exec *execCtx) writeObjectPayload(obj *objectSDK.Object, reader io.ReadCloser) bool {
 	if exec.headOnly() {
 		return true
 	}
 
-	err := exec.prm.objWriter.WriteChunk(obj.Payload())
+	var err error
+	if reader != nil {
+		defer func() { _ = reader.Close() }()
+		err = exec.writeFromStream(reader)
+	} else {
+		err = exec.prm.objWriter.WriteChunk(obj.Payload())
+	}
 
 	switch {
 	default:
@@ -322,9 +374,29 @@ func (exec *execCtx) writeObjectPayload(obj *objectSDK.Object) bool {
 	return err == nil
 }
 
+// writeFromStream writes payload from stream.
+func (exec *execCtx) writeFromStream(reader io.ReadCloser) error {
+	buf := make([]byte, streamChunkSize)
+	for {
+		n, err := reader.Read(buf)
+		if n > 0 {
+			if writeErr := exec.prm.objWriter.WriteChunk(buf[:n]); writeErr != nil {
+				return writeErr
+			}
+		}
+
+		if errors.Is(err, io.EOF) {
+			return nil
+		}
+		if err != nil {
+			return err
+		}
+	}
+}
+
 func (exec *execCtx) writeCollectedObject() {
 	if ok := exec.writeCollectedHeader(); ok {
-		exec.writeObjectPayload(exec.collectedObject)
+		exec.writeObjectPayload(exec.collectedObject, exec.collectedReader)
 	}
 }
 

pkg/services/object/get/get_test.go

@@ -1,12 +1,15 @@
 package getsvc
 
 import (
+	"bytes"
 	"context"
 	"crypto/rand"
 	"errors"
 	"fmt"
+	"io"
 	"strconv"
 	"testing"
+	"time"
 
 	"github.com/nspcc-dev/neofs-node/pkg/core/client"
 	netmapcore "github.com/nspcc-dev/neofs-node/pkg/core/netmap"
@@ -113,19 +116,28 @@ func newTestClient() *testClient {
 	}
 }
 
-func (c *testClient) getObject(exec *execCtx, _ client.NodeInfo) (*objectSDK.Object, error) {
+func (c *testClient) getObject(exec *execCtx, _ client.NodeInfo) (*objectSDK.Object, io.ReadCloser, error) {
 	v, ok := c.results[exec.address().EncodeToString()]
 	if !ok {
 		var errNotFound apistatus.ObjectNotFound
 
-		return nil, errNotFound
+		return nil, nil, errNotFound
 	}
 
 	if v.err != nil {
-		return nil, v.err
+		return nil, nil, v.err
 	}
 
-	return cutToRange(v.obj, exec.ctxRange()), nil
+	obj := cutToRange(v.obj, exec.ctxRange())
+
+	if obj != nil && len(obj.Payload()) > 0 {
+		reader := io.NopCloser(bytes.NewReader(obj.Payload()))
+		objWithoutPayload := obj.CutPayload()
+		objWithoutPayload.SetPayloadSize(obj.PayloadSize())
+		return objWithoutPayload, reader, nil
+	}
+
+	return obj, nil, nil
 }
 
 func (c *testClient) addResult(addr oid.Address, obj *objectSDK.Object, err error) {
@@ -135,7 +147,7 @@ func (c *testClient) addResult(addr oid.Address, obj *objectSDK.Object, err erro
 	}{obj: obj, err: err}
 }
 
-func (s *testStorage) get(exec *execCtx) (*objectSDK.Object, error) {
+func (s *testStorage) get(exec *execCtx) (*objectSDK.Object, io.ReadCloser, error) {
 	var (
 		ok    bool
 		obj   *objectSDK.Object
@@ -145,20 +157,29 @@ func (s *testStorage) get(exec *execCtx) (*objectSDK.Object, error) {
 	if _, ok = s.inhumed[sAddr]; ok {
 		var errRemoved apistatus.ObjectAlreadyRemoved
 
-		return nil, errRemoved
+		return nil, nil, errRemoved
 	}
 
 	if info, ok := s.virtual[sAddr]; ok {
-		return nil, objectSDK.NewSplitInfoError(info)
+		return nil, nil, objectSDK.NewSplitInfoError(info)
 	}
 
 	if obj, ok = s.phy[sAddr]; ok {
-		return cutToRange(obj, exec.ctxRange()), nil
+		obj = cutToRange(obj, exec.ctxRange())
+
+		if obj != nil && len(obj.Payload()) > 0 {
+			reader := io.NopCloser(bytes.NewReader(obj.Payload()))
+			objWithoutPayload := obj.CutPayload()
+			objWithoutPayload.SetPayloadSize(obj.PayloadSize())
+			return objWithoutPayload, reader, nil
+		}
+
+		return obj, nil, nil
 	}
 
 	var errNotFound apistatus.ObjectNotFound
 
-	return nil, errNotFound
+	return nil, nil, errNotFound
 }
 
 func cutToRange(o *objectSDK.Object, rng *objectSDK.Range) *objectSDK.Object {
@@ -476,7 +497,8 @@ func generateChain(ln int, cnr cid.ID) ([]*objectSDK.Object, []oid.ID, []byte) {
 }
 
 func TestGetRemoteSmall(t *testing.T) {
-	ctx := context.Background()
+	ctx, cancel := context.WithTimeout(context.Background(), 150*time.Second)
+	t.Cleanup(cancel)
 
 	var cnr container.Container
 	cnr.SetPlacementPolicy(netmaptest.PlacementPolicy())

pkg/services/object/get/local.go

@@ -11,7 +11,7 @@ import (
 func (exec *execCtx) executeLocal() {
 	var err error
 
-	exec.collectedObject, err = exec.svc.localStorage.get(exec)
+	exec.collectedObject, exec.collectedReader, err = exec.svc.localStorage.get(exec)
 
 	var errSplitInfo *objectSDK.SplitInfoError
 

pkg/services/object/get/remote.go

@@ -17,7 +17,7 @@ func (exec *execCtx) processNode(info client.NodeInfo) bool {
 		return true
 	}
 
-	obj, err := remoteClient.getObject(exec, info)
+	obj, reader, err := remoteClient.getObject(exec, info)
 
 	var errSplitInfo *objectSDK.SplitInfoError
 
@@ -33,13 +33,14 @@ func (exec *execCtx) processNode(info client.NodeInfo) bool {
 		exec.status = statusOK
 		exec.err = nil
 
-		// both object and err are nil only if the original
+		// object, reader and err are nil only if the original
 		// request was forwarded to another node and the object
 		// has already been streamed to the requesting party,
 		// or it is a GETRANGEHASH forwarded request whose
 		// response is not an object
-		if obj != nil {
+		if obj != nil || reader != nil {
 			exec.collectedObject = obj
+			exec.collectedReader = reader
 			exec.writeCollectedObject()
 		}
 	case errors.Is(err, apistatus.Error) && !errors.Is(err, apistatus.ErrObjectNotFound):

pkg/services/object/get/service.go

@@ -1,6 +1,8 @@
 package getsvc
 
 import (
+	"io"
+
 	"github.com/nspcc-dev/neofs-node/pkg/core/client"
 	"github.com/nspcc-dev/neofs-node/pkg/local_object_storage/engine"
 	"github.com/nspcc-dev/neofs-node/pkg/services/object/util"
@@ -41,14 +43,14 @@ type Service struct {
 type Option func(*cfg)
 
 type getClient interface {
-	getObject(*execCtx, client.NodeInfo) (*object.Object, error)
+	getObject(*execCtx, client.NodeInfo) (*object.Object, io.ReadCloser, error)
 }
 
 type cfg struct {
 	log *zap.Logger
 
 	localStorage interface {
-		get(*execCtx) (*object.Object, error)
+		get(*execCtx) (*object.Object, io.ReadCloser, error)
 	}
 
 	clientCache interface {