nspcc-dev
diff --git a/‎pkg/local_object_storage/blobstor/common/storage.go
Lines changed: 14 additions & 5 deletions b/‎pkg/local_object_storage/blobstor/common/storage.go
Lines changed: 14 additions & 5 deletions
diff --git a/‎pkg/local_object_storage/blobstor/common/storage_test.go
Lines changed: 1 addition & 1 deletion b/‎pkg/local_object_storage/blobstor/common/storage_test.go
Lines changed: 1 addition & 1 deletion
diff --git a/‎pkg/local_object_storage/blobstor/fstree/bench_test.go
Lines changed: 4 additions & 4 deletions b/‎pkg/local_object_storage/blobstor/fstree/bench_test.go
Lines changed: 4 additions & 4 deletions
diff --git a/‎pkg/local_object_storage/blobstor/fstree/fstree.go
Lines changed: 74 additions & 4 deletions b/‎pkg/local_object_storage/blobstor/fstree/fstree.go
Lines changed: 74 additions & 4 deletions
diff --git a/‎pkg/local_object_storage/blobstor/fstree/getstream_test.go
Lines changed: 2 additions & 2 deletions b/‎pkg/local_object_storage/blobstor/fstree/getstream_test.go
Lines changed: 2 additions & 2 deletions
diff --git a/‎pkg/local_object_storage/blobstor/fstree/head.go
Lines changed: 40 additions & 1 deletion b/‎pkg/local_object_storage/blobstor/fstree/head.go
Lines changed: 40 additions & 1 deletion
diff --git a/‎pkg/local_object_storage/blobstor/fstree/head_bench_test.go
Lines changed: 48 additions & 1 deletion b/‎pkg/local_object_storage/blobstor/fstree/head_bench_test.go
Lines changed: 48 additions & 1 deletion
diff --git a/‎pkg/local_object_storage/blobstor/fstree/head_test.go
Lines changed: 4 additions & 4 deletions b/‎pkg/local_object_storage/blobstor/fstree/head_test.go
Lines changed: 4 additions & 4 deletions
@@ -28,8 +28,17 @@ type Storage interface {
 	GetRange(oid.Address, uint64, uint64) ([]byte, error)
 	Head(oid.Address) (*objectSDK.Object, error)
 	Exists(oid.Address) (bool, error)
-	Put(oid.Address, []byte) error
-	PutBatch(map[oid.Address][]byte) error
+	// Put stores object in the storage. Arguments are:
+	//
+	// - addr oid.Address is the address of the object to store.
+	//
+	// - data []byte is the object payload in a binary format or
+	// full data if header is nil.
+	//
+	// - header []byte is the optional object header in a binary format.
+	// If nil, the header is extracted from the payload.
+	Put(addr oid.Address, data []byte, header []byte) error
+	PutBatch(map[oid.Address][2][]byte) error
 	Delete(oid.Address) error
 	Iterate(func(oid.Address, []byte) error, func(oid.Address, error) error) error
 	IterateAddresses(func(oid.Address) error, bool) error
@@ -69,7 +78,7 @@ func CopyBatched(dst, src Storage, batchSize int) error {
 		return fmt.Errorf("initialize destination sub-storage: %w", err)
 	}
 
-	var objBatch = make(map[oid.Address][]byte)
+	var objBatch = make(map[oid.Address][2][]byte)
 
 	err = src.Iterate(func(addr oid.Address, data []byte) error {
 		exists, err := dst.Exists(addr)
@@ -80,13 +89,13 @@ func CopyBatched(dst, src Storage, batchSize int) error {
 		}
 
 		if batchSize <= 1 {
-			err = dst.Put(addr, data)
+			err = dst.Put(addr, data, nil)
 			if err != nil {
 				return fmt.Errorf("put object %s into destination sub-storage: %w", addr, err)
 			}
 			return nil
 		}
-		objBatch[addr] = data
+		objBatch[addr] = [2][]byte{data}
 		if len(objBatch) == batchSize {
 			err = dst.PutBatch(objBatch)
 			if err != nil {
 
@@ -39,7 +39,7 @@ func testCopy(t *testing.T, copier func(dst, src common.Storage) error) {
 		_, _ = rand.Read(data)
 		mObjs[addr] = data
 
-		err := src.Put(addr, data)
+		err := src.Put(addr, data, nil)
 		require.NoError(t, err)
 	}
 
 
@@ -43,7 +43,7 @@ func testReadOp(b *testing.B, fsTree *fstree.FSTree, read func(address oid.Addre
 		obj := generateTestObject(payloadSize)
 		addr := object.AddressOf(obj)
 
-		require.NoError(b, fsTree.Put(addr, obj.Marshal()))
+		require.NoError(b, fsTree.Put(addr, obj.Payload(), obj.CutPayload().Marshal()))
 		b.ReportAllocs()
 		b.ResetTimer()
 		for range b.N {
@@ -57,11 +57,11 @@ func testReadOp(b *testing.B, fsTree *fstree.FSTree, read func(address oid.Addre
 	b.Run(name+"_combined", func(b *testing.B) {
 		const numObjects = 10
 
-		objMap := make(map[oid.Address][]byte, numObjects)
+		objMap := make(map[oid.Address][2][]byte, numObjects)
 		addrs := make([]oid.Address, numObjects)
 		for i := range numObjects {
 			o := generateTestObject(payloadSize)
-			objMap[object.AddressOf(o)] = o.Marshal()
+			objMap[object.AddressOf(o)] = [2][]byte{o.Payload(), o.CutPayload().Marshal()}
 			addrs[i] = object.AddressOf(o)
 		}
 		require.NoError(b, fsTree.PutBatch(objMap))
@@ -85,7 +85,7 @@ func testReadOp(b *testing.B, fsTree *fstree.FSTree, read func(address oid.Addre
 		}
 		require.NoError(b, compressConfig.Init())
 		fsTree.SetCompressor(compressConfig)
-		require.NoError(b, fsTree.Put(addr, obj.Marshal()))
+		require.NoError(b, fsTree.Put(addr, obj.Payload(), obj.CutPayload().Marshal()))
 
 		b.ReportAllocs()
 		b.ResetTimer()
 
@@ -22,6 +22,7 @@ import (
 	objectSDK "github.com/nspcc-dev/neofs-sdk-go/object"
 	oid "github.com/nspcc-dev/neofs-sdk-go/object/id"
 	"go.uber.org/zap"
+	"google.golang.org/protobuf/encoding/protowire"
 )
 
 // FSTree represents an object storage as a filesystem tree.
@@ -90,6 +91,26 @@ const (
 	// combinedDataOff is the offset from the start of the combined prefix to object data.
 	// It's also the length of the prefix in total.
 	combinedDataOff = combinedLengthOff + combinedLenSize
+
+	// streamPrefix is the prefix for streamed objects. It is used to distinguish
+	// streamed objects from regular ones.
+	streamPrefix = 0x7e
+
+	// streamLenHeaderOff is the offset from the start of the stream prefix to
+	// the length of the header data.
+	streamLenHeaderOff = 2
+
+	// streamLenSize is sizeof(uint32), length of a serialized 32-bit BE integer
+	// that represents the length of the header or payload data.
+	streamLenSize = 4
+
+	// streamLenDataOff is the offset from the start of the stream prefix to
+	// the length of the data.
+	streamLenDataOff = streamLenHeaderOff + streamLenSize
+
+	// streamDataOff is the offset from the start of the stream prefix to the
+	// start of the data. It is used to read the data after the header.
+	streamDataOff = streamLenDataOff + streamLenSize
 )
 
 var _ common.Storage = (*FSTree)(nil)
@@ -329,7 +350,7 @@ func (t *FSTree) getPath(addr oid.Address) (string, error) {
 }
 
 // Put puts an object in the storage.
-func (t *FSTree) Put(addr oid.Address, data []byte) error {
+func (t *FSTree) Put(addr oid.Address, data []byte, header []byte) error {
 	if t.readOnly {
 		return common.ErrReadOnly
 	}
@@ -339,7 +360,7 @@ func (t *FSTree) Put(addr oid.Address, data []byte) error {
 	if err := util.MkdirAllX(filepath.Dir(p), t.Permissions); err != nil {
 		return fmt.Errorf("mkdirall for %q: %w", p, err)
 	}
-	data = t.Compress(data)
+	data = t.joinHeaderAndPayload(header, data)
 
 	err := t.writer.writeData(addr.Object(), p, data)
 	if err != nil {
@@ -349,7 +370,7 @@ func (t *FSTree) Put(addr oid.Address, data []byte) error {
 }
 
 // PutBatch puts a batch of objects in the storage.
-func (t *FSTree) PutBatch(objs map[oid.Address][]byte) error {
+func (t *FSTree) PutBatch(objs map[oid.Address][2][]byte) error {
 	if t.readOnly {
 		return common.ErrReadOnly
 	}
@@ -363,7 +384,7 @@ func (t *FSTree) PutBatch(objs map[oid.Address][]byte) error {
 		writeDataUnits = append(writeDataUnits, writeDataUnit{
 			id:   addr.Object(),
 			path: p,
-			data: t.Compress(data),
+			data: t.joinHeaderAndPayload(data[1], data[0]),
 		})
 	}
 
@@ -375,6 +396,28 @@ func (t *FSTree) PutBatch(objs map[oid.Address][]byte) error {
 	return nil
 }
 
+// joinHeaderAndPayload combines header and payload into a single byte slice.
+func (t *FSTree) joinHeaderAndPayload(header, payload []byte) []byte {
+	if header == nil {
+		return payload
+	}
+
+	hLen := len(header)
+	payload = t.Compress(payload)
+	pLen := len(payload)
+
+	data := make([]byte, hLen+pLen+streamDataOff)
+	data[0] = streamPrefix
+	data[1] = 0 // version 0
+	binary.BigEndian.PutUint32(data[streamLenHeaderOff:], uint32(hLen))
+	binary.BigEndian.PutUint32(data[streamLenDataOff:], uint32(pLen))
+
+	copy(data[streamDataOff:], header)
+	copy(data[streamDataOff+hLen:], payload)
+
+	return data
+}
+
 // Get returns an object from the storage by address.
 func (t *FSTree) Get(addr oid.Address) (*objectSDK.Object, error) {
 	data, err := t.getObjBytes(addr)
@@ -433,6 +476,16 @@ func parseCombinedPrefix(p []byte) ([]byte, uint32) {
 		binary.BigEndian.Uint32(p[combinedLengthOff:combinedDataOff])
 }
 
+// parseStreamPrefix checks the given byte slice for stream prefix and returns
+// the length of the header and data if so (0, 0 otherwise).
+func parseStreamPrefix(p []byte) (uint32, uint32) {
+	if p[0] != streamPrefix || p[1] != 0 { // Only version 0 is supported now.
+		return 0, 0
+	}
+	return binary.BigEndian.Uint32(p[streamLenHeaderOff:streamLenDataOff]),
+		binary.BigEndian.Uint32(p[streamLenDataOff:])
+}
+
 func (t *FSTree) extractCombinedObject(id oid.ID, f *os.File) ([]byte, error) {
 	var (
 		comBuf     [combinedDataOff]byte
@@ -485,6 +538,23 @@ func (t *FSTree) readFullObject(f io.Reader, initial []byte, size int64) ([]byte
 		return nil, fmt.Errorf("read: %w", err)
 	}
 	data = data[:len(initial)+n]
+	hLen, _ := parseStreamPrefix(data)
+	if hLen > 0 {
+		data = data[streamDataOff:]
+		payload, err := t.Decompress(data[hLen:])
+		if err != nil {
+			return nil, fmt.Errorf("decompress payload: %w", err)
+		}
+		pLen := len(payload)
+		payloadNum := protowire.Number(4)
+		n := protowire.SizeTag(payloadNum) + protowire.SizeVarint(uint64(pLen))
+		buf := make([]byte, int(hLen)+pLen+n)
+		copy(buf[:hLen], data)
+		off := binary.PutUvarint(buf[hLen:], protowire.EncodeTag(payloadNum, protowire.BytesType)) + int(hLen)
+		off += binary.PutUvarint(buf[off:], uint64(pLen))
+		copy(buf[off:], payload)
+		data = buf
+	}
 
 	return t.Decompress(data)
 }
 
@@ -41,7 +41,7 @@ func TestGetStream(t *testing.T) {
 		obj.SetID(addr.Object())
 		obj.SetPayload(payload)
 
-		require.NoError(t, tree.Put(addr, obj.Marshal()))
+		require.NoError(t, tree.Put(addr, obj.Marshal(), nil))
 
 		retrievedObj, reader, err := tree.GetStream(addr)
 		require.NoError(t, err)
@@ -108,7 +108,7 @@ func TestGetStreamAfterErrors(t *testing.T) {
 		payload := []byte("test payload")
 		obj.SetPayload(payload)
 
-		require.NoError(t, tree.Put(addr, obj.Marshal()))
+		require.NoError(t, tree.Put(addr, obj.Marshal(), nil))
 
 		objPath := tree.treePath(addr)
 
 
@@ -124,8 +124,47 @@ func (t *FSTree) extractHeaderAndStream(id oid.ID, f *os.File) (*objectSDK.Objec
 
 // readHeaderAndPayload reads an object header from the file and returns reader for payload.
 // This function takes ownership of the io.ReadCloser and will close it if it does not return it.
-func (t *FSTree) readHeaderAndPayload(f io.ReadCloser, initial []byte) (*objectSDK.Object, io.ReadSeekCloser, error) {
+func (t *FSTree) readHeaderAndPayload(f io.ReadSeekCloser, initial []byte) (*objectSDK.Object, io.ReadSeekCloser, error) {
 	var err error
+	var hLen, dLen uint32
+	if len(initial) >= streamDataOff {
+		hLen, dLen = parseStreamPrefix(initial)
+	} else {
+		var p []byte
+		copy(p[:], initial)
+		_, err := io.ReadFull(f, p[len(initial):])
+		if err != nil && !errors.Is(err, io.EOF) && !errors.Is(err, io.ErrUnexpectedEOF) {
+			return nil, f, fmt.Errorf("read stream prefix: %w", err)
+		}
+		hLen, dLen = parseStreamPrefix(p)
+		if hLen == 0 {
+			initial = p[:]
+		}
+	}
+	if hLen > 0 {
+		initial = initial[streamDataOff:]
+		var header []byte
+		if len(initial) < int(hLen) {
+			header = make([]byte, hLen)
+			copy(header, initial)
+			_, err = io.ReadFull(f, header[len(initial):])
+			if err != nil {
+				return nil, nil, fmt.Errorf("read stream header: %w", err)
+			}
+			initial = header
+		}
+		header = initial[:hLen]
+		var obj objectSDK.Object
+		err = obj.Unmarshal(header)
+		if err != nil {
+			return nil, nil, fmt.Errorf("unmarshal object: %w", err)
+		}
+		return &obj, &payloadReader{
+			Reader: io.LimitReader(io.MultiReader(bytes.NewReader(initial[hLen:]), f), int64(dLen)),
+			close:  f.Close,
+		}, nil
+	}
+
 	if len(initial) < objectSDK.MaxHeaderLen {
 		_ = f.Close()
 		initial, err = t.Decompress(initial)
 
@@ -71,7 +71,7 @@ func runHeadVsGetBenchmark(b *testing.B, payloadSize int, compressed bool) {
 	obj := generateTestObject(payloadSize)
 	addr := object.AddressOf(obj)
 
-	require.NoError(b, fsTree.Put(addr, obj.Marshal()))
+	require.NoError(b, fsTree.Put(addr, obj.Payload(), obj.CutPayload().Marshal()))
 
 	b.Run("Head"+suffix, func(b *testing.B) {
 		b.ResetTimer()
@@ -96,6 +96,53 @@ func runHeadVsGetBenchmark(b *testing.B, payloadSize int, compressed bool) {
 	})
 }
 
+func BenchmarkFSTree_NewStructureHead(b *testing.B) {
+	for _, size := range payloadSizes {
+		b.Run(generateSizeLabel(size), func(b *testing.B) {
+			for _, tc := range []struct {
+				name       string
+				compressed bool
+				nilHeader  bool
+			}{
+				{"WithHeader", false, false},
+				{"WithoutHeader", false, true},
+				{"WithHeader_Compressed", true, false},
+				{"WithoutHeader_Compressed", true, true},
+			} {
+				b.Run(tc.name, func(b *testing.B) {
+					fsTree := fstree.New(fstree.WithPath(b.TempDir()))
+					if tc.compressed {
+						compressConfig := &compression.Config{Enabled: true}
+						require.NoError(b, compressConfig.Init())
+						fsTree.SetCompressor(compressConfig)
+					}
+					require.NoError(b, fsTree.Open(false))
+					require.NoError(b, fsTree.Init())
+
+					obj := generateTestObject(size)
+					addr := object.AddressOf(obj)
+
+					if tc.nilHeader {
+						payload := obj.Marshal()
+						require.NoError(b, fsTree.Put(addr, payload, nil))
+					} else {
+						header := obj.CutPayload().Marshal()
+						payload := obj.Payload()
+						require.NoError(b, fsTree.Put(addr, payload, header))
+					}
+
+					b.ReportAllocs()
+					b.ResetTimer()
+					for range b.N {
+						_, err := fsTree.Head(addr)
+						require.NoError(b, err)
+					}
+				})
+			}
+		})
+	}
+}
+
 func generateTestObject(payloadSize int) *objectSDK.Object {
 	obj := objecttest.Object()
 	if payloadSize > 0 {
 
@@ -23,7 +23,7 @@ func TestHeadStorage(t *testing.T) {
 		addAttribute(obj, "test-key1", "test-value1")
 		addAttribute(obj, "test-key2", "test-value2")
 
-		err := fsTree.Put(object.AddressOf(obj), obj.Marshal())
+		err := fsTree.Put(object.AddressOf(obj), obj.Payload(), obj.CutPayload().Marshal())
 		require.NoError(t, err)
 
 		res, err := fsTree.Head(object.AddressOf(obj))
@@ -42,15 +42,15 @@ func TestHeadStorage(t *testing.T) {
 	testCombinedObjects := func(t *testing.T, fsTree *fstree.FSTree, size int) {
 		const numObjects = 100
 
-		objMap := make(map[oid.Address][]byte, numObjects)
+		objMap := make(map[oid.Address][2][]byte, numObjects)
 		objects := make([]*objectSDK.Object, numObjects)
 		for i := range numObjects {
 			obj := generateTestObject(size)
 			obj.SetAttributes()
 			addAttribute(obj, fmt.Sprintf("key-%d", i), fmt.Sprintf("value-%d", i))
 
 			objects[i] = obj
-			objMap[object.AddressOf(obj)] = obj.Marshal()
+			objMap[object.AddressOf(obj)] = [2][]byte{obj.Payload(), obj.CutPayload().Marshal()}
 		}
 
 		require.NoError(t, fsTree.PutBatch(objMap))
@@ -75,7 +75,7 @@ func TestHeadStorage(t *testing.T) {
 			addAttribute(obj, fmt.Sprintf("key%d", i), fmt.Sprintf("value%d", i))
 		}
 
-		err := fsTree.Put(object.AddressOf(obj), obj.Marshal())
+		err := fsTree.Put(object.AddressOf(obj), obj.Payload(), obj.CutPayload().Marshal())
 		require.NoError(t, err)
 
 		res, err := fsTree.Head(object.AddressOf(obj))
Original file line number	Diff line number	Diff line change
`@@ -39,7 +39,7 @@ func testCopy(t *testing.T, copier func(dst, src common.Storage) error) {`
`39`	`39`	`_, _ = rand.Read(data)`
`40`	`40`	`mObjs[addr] = data`
`41`	`41`
`42`		`- err := src.Put(addr, data)`
	`42`	`+ err := src.Put(addr, data, nil)`
`43`	`43`	`require.NoError(t, err)`
`44`	`44`	`}`
`45`	`45`