realloc/frostfs-node
1
0
Fork 0
forked from TrueCloudLab/frostfs-node
Code Releases Packages Activity

[#1326] services/tree: Implement GetSubTree RPC

Browse source 
Signed-off-by: Evgenii Stratonikov <evgeniy@nspcc.ru>
This commit is contained in:
Evgenii Stratonikov 2022-04-29 13:06:10 +03:00 • committed by fyrchik
parent 62154da17c
commit aea855e8f3
10 changed files with 431 additions and 255 deletions
Download patch file Download diff file Expand all files Collapse all files

99
pkg/services/tree/service.go
View file

@ -21,6 +21,9 @@ type Service struct {
closeCh chan struct{}
}
// MaxGetSubTreeDepth represents maximum allowed traversal depth in GetSubTree RPC.
const MaxGetSubTreeDepth = 10
var _ TreeServiceServer = (*Service)(nil)
// New creates new tree service instance.
@ -66,7 +69,7 @@ func (s *Service) Add(_ context.Context, req *AddRequest) (*AddResponse, error)
log, err := s.forest.TreeMove(cid, b.GetTreeId(), &pilorama.Move{
Parent: b.GetParentId(),
Child: pilorama.RootID,
Meta: pilorama.Meta{Items: constructMeta(b.GetMeta())},
Meta: pilorama.Meta{Items: protoToMeta(b.GetMeta())},
})
if err != nil {
return nil, err
@ -93,7 +96,7 @@ func (s *Service) AddByPath(_ context.Context, req *AddByPathRequest) (*AddByPat
return nil, err
}
meta := constructMeta(b.GetMeta())
meta := protoToMeta(b.GetMeta())
attr := b.GetPathAttribute()
if len(attr) == 0 {
@ -173,7 +176,7 @@ func (s *Service) Move(_ context.Context, req *MoveRequest) (*MoveResponse, erro
log, err := s.forest.TreeMove(cid, b.GetTreeId(), &pilorama.Move{
Parent: b.GetParentId(),
Child: b.GetNodeId(),
Meta: pilorama.Meta{Items: constructMeta(b.GetMeta())},
Meta: pilorama.Meta{Items: protoToMeta(b.GetMeta())},
})
if err != nil {
return nil, err
@ -211,23 +214,20 @@ func (s *Service) GetNodeByPath(_ context.Context, req *GetNodeByPathRequest) (*
var x GetNodeByPathResponse_Info
x.NodeId = node
x.Timestamp = m.Time
for _, kv := range m.Items {
needAttr := b.AllAttributes
if !needAttr {
if b.AllAttributes {
x.Meta = metaToProto(m.Items)
} else {
for _, kv := range m.Items {
for _, attr := range b.GetAttributes() {
if kv.Key == attr {
needAttr = true
x.Meta = append(x.Meta, &KeyValue{
Key: kv.Key,
Value: kv.Value,
})
break
}
}
}
if needAttr {
x.Meta = append(x.Meta, &KeyValue{
Key: kv.Key,
Value: kv.Value,
})
}
}
info = append(info, &x)
}
@ -239,8 +239,56 @@ func (s *Service) GetNodeByPath(_ context.Context, req *GetNodeByPathRequest) (*
}, nil
}
func (s *Service) GetSubTree(_ context.Context, req *GetSubTreeRequest) (*GetSubTreeResponse, error) {
return nil, errors.New("GetSubTree is unimplemented")
type nodeDepthPair struct {
nodes []uint64
depth uint32
}
func (s *Service) GetSubTree(req *GetSubTreeRequest, srv TreeService_GetSubTreeServer) error {
b := req.GetBody()
if b.GetDepth() > MaxGetSubTreeDepth {
return fmt.Errorf("too big depth: max=%d, got=%d", MaxGetSubTreeDepth, b.GetDepth())
}
var cid cidSDK.ID
if err := cid.Decode(b.GetContainerId()); err != nil {
return err
}
queue := []nodeDepthPair{{[]uint64{b.GetRootId()}, 0}}
for len(queue) != 0 {
for _, nodeID := range queue[0].nodes {
m, err := s.forest.TreeGetMeta(cid, b.GetTreeId(), nodeID)
if err != nil {
return err
}
err = srv.Send(&GetSubTreeResponse{
Body: &GetSubTreeResponse_Body{
NodeId: b.GetRootId(),
ParentId: b.GetRootId(),
Timestamp: m.Time,
Meta: metaToProto(m.Items),
},
})
if err != nil {
return err
}
}
if queue[0].depth < b.GetDepth() {
for _, nodeID := range queue[0].nodes {
children, err := s.forest.TreeGetChildren(cid, b.GetTreeId(), nodeID)
if err != nil {
return err
}
queue = append(queue, nodeDepthPair{children, queue[0].depth + 1})
}
}
queue = queue[1:]
}
return nil
}
// Apply locally applies operation from the remote node to the tree.
@ -284,11 +332,24 @@ loop:
})
}
func constructMeta(arr []*KeyValue) []pilorama.KeyValue {
func protoToMeta(arr []*KeyValue) []pilorama.KeyValue {
meta := make([]pilorama.KeyValue, len(arr))
for i, kv := range arr {
meta[i].Key = kv.Key
meta[i].Value = kv.Value
if kv != nil {
meta[i].Key = kv.Key
meta[i].Value = kv.Value
}
}
return meta
}
func metaToProto(arr []pilorama.KeyValue) []*KeyValue {
meta := make([]*KeyValue, len(arr))
for i, kv := range arr {
meta[i] = &KeyValue{
Key: kv.Key,
Value: kv.Value,
}
}
return meta
}