[#69] object: Initial EC implementation

Signed-off-by: Evgenii Stratonikov <e.stratonikov@yadro.com>

netmap/convert.go

@@ -142,6 +142,8 @@ func (r *Replica) ToGRPCMessage() grpc.Message {
 
 		m.SetSelector(r.selector)
 		m.SetCount(r.count)
+		m.EcDataCount = r.ecDataCount
+		m.EcParityCount = r.ecParityCount
 	}
 
 	return m
@@ -155,6 +157,8 @@ func (r *Replica) FromGRPCMessage(m grpc.Message) error {
 
 	r.selector = v.GetSelector()
 	r.count = v.GetCount()
+	r.ecDataCount = v.GetEcDataCount()
+	r.ecParityCount = v.GetEcParityCount()
 
 	return nil
 }

netmap/marshal.go

@@ -19,8 +19,10 @@ const (
 	attributeSelectorField = 4
 	filterSelectorField    = 5
 
-	countReplicaField    = 1
-	selectorReplicaField = 2
+	countReplicaField         = 1
+	selectorReplicaField      = 2
+	ecDataCountReplicaField   = 3
+	ecParityCountReplicaField = 4
 
 	replicasPolicyField  = 1
 	backupPolicyField    = 2
@@ -134,7 +136,9 @@ func (r *Replica) StableMarshal(buf []byte) []byte {
 	var offset int
 
 	offset += protoutil.UInt32Marshal(countReplicaField, buf[offset:], r.count)
-	protoutil.StringMarshal(selectorReplicaField, buf[offset:], r.selector)
+	offset += protoutil.StringMarshal(selectorReplicaField, buf[offset:], r.selector)
+	offset += protoutil.UInt32Marshal(ecDataCountReplicaField, buf[offset:], r.ecDataCount)
+	protoutil.UInt32Marshal(ecParityCountReplicaField, buf[offset:], r.ecParityCount)
 
 	return buf
 }
@@ -146,6 +150,8 @@ func (r *Replica) StableSize() (size int) {
 
 	size += protoutil.UInt32Size(countReplicaField, r.count)
 	size += protoutil.StringSize(selectorReplicaField, r.selector)
+	size += protoutil.UInt32Size(ecDataCountReplicaField, r.ecDataCount)
+	size += protoutil.UInt32Size(ecParityCountReplicaField, r.ecParityCount)
 
 	return size
 }

netmap/types.go

@@ -50,6 +50,9 @@ type Selector struct {
 type Replica struct {
 	count    uint32
 	selector string
+
+	ecDataCount   uint32
+	ecParityCount uint32
 }
 
 type Operation uint32
@@ -259,6 +262,30 @@ func (r *Replica) SetCount(count uint32) {
 	r.count = count
 }
 
+func (r *Replica) GetECDataCount() uint32 {
+	if r != nil {
+		return r.ecDataCount
+	}
+
+	return 0
+}
+
+func (r *Replica) SetECDataCount(count uint32) {
+	r.ecDataCount = count
+}
+
+func (r *Replica) GetECParityCount() uint32 {
+	if r != nil {
+		return r.ecParityCount
+	}
+
+	return 0
+}
+
+func (r *Replica) SetECParityCount(count uint32) {
+	r.ecParityCount = count
+}
+
 func (p *PlacementPolicy) GetUnique() bool {
 	if p != nil {
 		return p.unique

object/convert.go

@@ -262,6 +262,51 @@ func (h *SplitHeader) FromGRPCMessage(m grpc.Message) error {
 	return nil
 }
 
+func (h *ECHeader) ToGRPCMessage() grpc.Message {
+	var m *object.Header_EC
+
+	if h != nil {
+		m = new(object.Header_EC)
+
+		m.Parent = h.Parent.ToGRPCMessage().(*refsGRPC.ObjectID)
+		m.Index = h.Index
+		m.Total = h.Total
+		m.Header = h.Header
+		m.HeaderLength = h.HeaderLength
+	}
+
+	return m
+}
+
+func (h *ECHeader) FromGRPCMessage(m grpc.Message) error {
+	v, ok := m.(*object.Header_EC)
+	if !ok {
+		return message.NewUnexpectedMessageType(m, v)
+	}
+
+	var err error
+
+	par := v.GetParent()
+	if par == nil {
+		h.Parent = nil
+	} else {
+		if h.Parent == nil {
+			h.Parent = new(refs.ObjectID)
+		}
+
+		err = h.Parent.FromGRPCMessage(par)
+		if err != nil {
+			return err
+		}
+	}
+
+	h.Index = v.GetIndex()
+	h.Total = v.GetTotal()
+	h.Header = v.GetHeader()
+	h.HeaderLength = v.GetHeaderLength()
+	return nil
+}
+
 func (h *Header) ToGRPCMessage() grpc.Message {
 	var m *object.Header
 
@@ -275,6 +320,7 @@ func (h *Header) ToGRPCMessage() grpc.Message {
 		m.SetContainerId(h.cid.ToGRPCMessage().(*refsGRPC.ContainerID))
 		m.SetSessionToken(h.sessionToken.ToGRPCMessage().(*sessionGRPC.SessionToken))
 		m.SetSplit(h.split.ToGRPCMessage().(*object.Header_Split))
+		m.Ec = h.ec.ToGRPCMessage().(*object.Header_EC)
 		m.SetAttributes(AttributesToGRPC(h.attr))
 		m.SetPayloadLength(h.payloadLen)
 		m.SetCreationEpoch(h.creatEpoch)
@@ -313,6 +359,9 @@ func (h *Header) FromGRPCMessage(m grpc.Message) error {
 	if err := h.fillSplitHeader(v); err != nil {
 		return err
 	}
+	if err := h.fillECHeader(v); err != nil {
+		return err
+	}
 
 	h.attr, err = AttributesFromGRPC(v.GetAttributes())
 	if err != nil {
@@ -417,6 +466,19 @@ func (h *Header) fillSplitHeader(v *object.Header) error {
 	return h.split.FromGRPCMessage(split)
 }
 
+func (h *Header) fillECHeader(v *object.Header) error {
+	ec := v.GetEc()
+	if ec == nil {
+		h.ec = nil
+		return nil
+	}
+
+	if h.ec == nil {
+		h.ec = new(ECHeader)
+	}
+	return h.ec.FromGRPCMessage(ec)
+}
+
 func (h *HeaderWithSignature) ToGRPCMessage() grpc.Message {
 	var m *object.HeaderWithSignature
 
@@ -591,6 +653,50 @@ func (s *SplitInfo) FromGRPCMessage(m grpc.Message) error {
 	return nil
 }
 
+func (s *ECInfo) ToGRPCMessage() grpc.Message {
+	var m *object.ECInfo
+
+	if s != nil {
+		m = new(object.ECInfo)
+
+		if s.Chunks != nil {
+			chunks := make([]*object.ECInfo_Chunk, len(s.Chunks))
+			for i := range chunks {
+				chunks[i] = &object.ECInfo_Chunk{
+					Id:    s.Chunks[i].ID.ToGRPCMessage().(*refsGRPC.ObjectID),
+					Index: s.Chunks[i].Index,
+					Total: s.Chunks[i].Total,
+				}
+			}
+			m.Chunks = chunks
+		}
+	}
+
+	return m
+}
+
+func (s *ECInfo) FromGRPCMessage(m grpc.Message) error {
+	v, ok := m.(*object.ECInfo)
+	if !ok {
+		return message.NewUnexpectedMessageType(m, v)
+	}
+
+	chunks := v.GetChunks()
+	if chunks == nil {
+		s.Chunks = nil
+	} else {
+		s.Chunks = make([]ECChunk, len(chunks))
+		for i := range chunks {
+			if err := s.Chunks[i].ID.FromGRPCMessage(chunks[i].Id); err != nil {
+				return err
+			}
+			s.Chunks[i].Index = chunks[i].Index
+			s.Chunks[i].Total = chunks[i].Total
+		}
+	}
+	return nil
+}
+
 func (r *GetRequestBody) ToGRPCMessage() grpc.Message {
 	var m *object.GetRequest_Body
 

object/marshal.go

@@ -26,6 +26,12 @@ const (
 	splitHdrChildrenField        = 5
 	splitHdrSplitIDField         = 6
 
+	ecHdrParentField       = 1
+	ecHdrIndexField        = 2
+	ecHdrTotalField        = 3
+	ecHdrHeaderLengthField = 4
+	ecHdrHeaderField       = 5
+
 	hdrVersionField         = 1
 	hdrContainerIDField     = 2
 	hdrOwnerIDField         = 3
@@ -37,6 +43,7 @@ const (
 	hdrSessionTokenField    = 9
 	hdrAttributesField      = 10
 	hdrSplitField           = 11
+	hdrECField              = 12
 
 	hdrWithSigHeaderField    = 1
 	hdrWithSigSignatureField = 2
@@ -229,6 +236,43 @@ func (h *SplitHeader) Unmarshal(data []byte) error {
 	return message.Unmarshal(h, data, new(object.Header_Split))
 }
 
+func (h *ECHeader) StableMarshal(buf []byte) []byte {
+	if h == nil {
+		return []byte{}
+	}
+
+	if buf == nil {
+		buf = make([]byte, h.StableSize())
+	}
+
+	var offset int
+
+	offset += proto.NestedStructureMarshal(ecHdrParentField, buf[offset:], h.Parent)
+	offset += proto.UInt32Marshal(ecHdrIndexField, buf[offset:], h.Index)
+	offset += proto.UInt32Marshal(ecHdrTotalField, buf[offset:], h.Total)
+	offset += proto.UInt32Marshal(ecHdrHeaderLengthField, buf[offset:], h.HeaderLength)
+	proto.BytesMarshal(ecHdrHeaderField, buf[offset:], h.Header)
+	return buf
+}
+
+func (h *ECHeader) StableSize() (size int) {
+	if h == nil {
+		return 0
+	}
+
+	size += proto.NestedStructureSize(ecHdrParentField, h.Parent)
+	size += proto.UInt32Size(ecHdrIndexField, h.Index)
+	size += proto.UInt32Size(ecHdrTotalField, h.Total)
+	size += proto.UInt32Size(ecHdrHeaderLengthField, h.HeaderLength)
+	size += proto.BytesSize(ecHdrHeaderField, h.Header)
+
+	return size
+}
+
+func (h *ECHeader) Unmarshal(data []byte) error {
+	return message.Unmarshal(h, data, new(object.Header_EC))
+}
+
 func (h *Header) StableMarshal(buf []byte) []byte {
 	if h == nil {
 		return []byte{}
@@ -255,6 +299,7 @@ func (h *Header) StableMarshal(buf []byte) []byte {
 	}
 
 	proto.NestedStructureMarshal(hdrSplitField, buf[offset:], h.split)
+	proto.NestedStructureMarshal(hdrECField, buf[offset:], h.ec)
 
 	return buf
 }
@@ -277,6 +322,7 @@ func (h *Header) StableSize() (size int) {
 		size += proto.NestedStructureSize(hdrAttributesField, &h.attr[i])
 	}
 	size += proto.NestedStructureSize(hdrSplitField, h.split)
+	size += proto.NestedStructureSize(hdrECField, h.ec)
 
 	return size
 }

object/message_test.go

@@ -16,6 +16,7 @@ func TestMessageConvert(t *testing.T) {
 		func(empty bool) message.Message { return objecttest.GenerateHeader(empty) },
 		func(empty bool) message.Message { return objecttest.GenerateObject(empty) },
 		func(empty bool) message.Message { return objecttest.GenerateSplitInfo(empty) },
+		func(empty bool) message.Message { return objecttest.GenerateECInfo(empty) },
 		func(empty bool) message.Message { return objecttest.GenerateGetRequestBody(empty) },
 		func(empty bool) message.Message { return objecttest.GenerateGetRequest(empty) },
 		func(empty bool) message.Message { return objecttest.GenerateGetObjectPartInit(empty) },

object/test/generate.go

@@ -137,6 +137,31 @@ func GenerateSplitInfo(empty bool) *object.SplitInfo {
 	return m
 }
 
+func GenerateECInfo(empty bool) *object.ECInfo {
+	m := new(object.ECInfo)
+
+	if !empty {
+		m.Chunks = make([]object.ECChunk, 2)
+		for i := range m.Chunks {
+			m.Chunks[i] = *GenerateECChunk(false)
+		}
+	}
+
+	return m
+}
+
+func GenerateECChunk(empty bool) *object.ECChunk {
+	m := new(object.ECChunk)
+
+	if !empty {
+		m.ID = *refstest.GenerateObjectID(false)
+		m.Index = 4
+		m.Total = 7
+	}
+
+	return m
+}
+
 func GenerateGetRequestBody(empty bool) *object.GetRequestBody {
 	m := new(object.GetRequestBody)
 

object/types.go

@@ -39,6 +39,15 @@ type SplitHeader struct {
 	splitID []byte
 }
 
+type ECHeader struct {
+	Parent       *refs.ObjectID
+	Index        uint32
+	Total        uint32
+	Header       []byte
+	HeaderLength uint32
+	Signature    []byte
+}
+
 type Header struct {
 	version *refs.Version
 
@@ -59,6 +68,8 @@ type Header struct {
 	attr []Attribute
 
 	split *SplitHeader
+
+	ec *ECHeader
 }
 
 type HeaderWithSignature struct {
@@ -88,6 +99,16 @@ type SplitInfo struct {
 	link *refs.ObjectID
 }
 
+type ECChunk struct {
+	ID    refs.ObjectID
+	Index uint32
+	Total uint32
+}
+
+type ECInfo struct {
+	Chunks []ECChunk
+}
+
 type GetRequestBody struct {
 	addr *refs.Address
 
@@ -655,6 +676,17 @@ func (h *Header) SetSplit(v *SplitHeader) {
 	h.split = v
 }
 
+func (h *Header) GetEC() *ECHeader {
+	if h != nil {
+		return h.ec
+	}
+	return nil
+}
+
+func (h *Header) SetEC(v *ECHeader) {
+	h.ec = v
+}
+
 func (h *HeaderWithSignature) GetHeader() *Header {
 	if h != nil {
 		return h.header