[#165] Don't use recursion in list streaming

Signed-off-by: Denis Kirillov <d.kirillov@yadro.com>
This commit is contained in:
Denis Kirillov 2024-01-21 14:13:37 +03:00
parent 2d7973b3f1
commit 8a30f18ff6

View file

@ -17,6 +17,7 @@ import (
oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id" oid "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/object/id"
"git.frostfs.info/TrueCloudLab/frostfs-sdk-go/user" "git.frostfs.info/TrueCloudLab/frostfs-sdk-go/user"
"go.uber.org/zap" "go.uber.org/zap"
"golang.org/x/exp/maps"
) )
type ( type (
@ -666,33 +667,69 @@ type VersionsByPrefixStreamImpl struct {
ended bool ended bool
latestOnly bool latestOnly bool
currentLatest *data.NodeVersion currentLatest *data.NodeVersion
log *zap.Logger
} }
// Next todo remove recursion func (s *VersionsByPrefixStreamImpl) Next(context.Context) (*data.NodeVersion, error) {
func (s *VersionsByPrefixStreamImpl) Next(ctx context.Context) (*data.NodeVersion, error) {
if s.ended { if s.ended {
return nil, io.EOF return nil, io.EOF
} }
for true {
if s.innerStream == nil { if s.innerStream == nil {
node, err := s.getNodeFromMainStream()
if err != nil {
if errors.Is(err, io.EOF) {
s.ended = true
if s.currentLatest != nil {
return s.currentLatest, nil
}
}
return nil, fmt.Errorf("get node from main stream: %w", err)
}
if err = s.initInnerStream(node); err != nil {
return nil, fmt.Errorf("init inner stream: %w", err)
}
}
nodeVersion, err := s.getNodeVersionFromInnerStream()
if err != nil {
if errors.Is(err, io.EOF) {
s.innerStream = nil
maps.Clear(s.namesMap)
if s.currentLatest != nil && s.currentLatest.ID != s.intermediateRootID {
return s.currentLatest, nil
}
continue
}
return nil, fmt.Errorf("inner stream: %w", err)
}
return nodeVersion, nil
}
panic("unreachable code")
}
func (s *VersionsByPrefixStreamImpl) getNodeFromMainStream() (NodeResponse, error) {
for true {
node, err := s.mainStream.Next() node, err := s.mainStream.Next()
if err != nil { if err != nil {
if errors.Is(err, ErrNodeNotFound) { if errors.Is(err, ErrNodeNotFound) {
return nil, io.EOF return nil, io.EOF
} }
if errors.Is(err, io.EOF) {
s.ended = true
if s.latestOnly && s.currentLatest != nil {
return s.currentLatest, nil
}
}
return nil, fmt.Errorf("main stream next: %w", err) return nil, fmt.Errorf("main stream next: %w", err)
} }
if node.GetNodeID() == s.rootID || !strings.HasPrefix(getFilename(node), s.tailPrefix) { if node.GetNodeID() != s.rootID && strings.HasPrefix(getFilename(node), s.tailPrefix) {
return s.Next(ctx) return node, nil
}
} }
panic("unreachable code")
}
func (s *VersionsByPrefixStreamImpl) initInnerStream(node NodeResponse) (err error) {
if node.GetParentID() == s.rootID { if node.GetParentID() == s.rootID {
s.intermediateRootID = node.GetNodeID() s.intermediateRootID = node.GetNodeID()
} }
@ -700,56 +737,34 @@ func (s *VersionsByPrefixStreamImpl) Next(ctx context.Context) (*data.NodeVersio
if isIntermediate(node) { if isIntermediate(node) {
s.innerStream, err = s.service.GetSubTreeStream(s.ctx, s.bktInfo, versionTree, node.GetNodeID(), maxGetSubTreeDepth) s.innerStream, err = s.service.GetSubTreeStream(s.ctx, s.bktInfo, versionTree, node.GetNodeID(), maxGetSubTreeDepth)
if err != nil { if err != nil {
return nil, fmt.Errorf("get sub tree node from main stream: %w", err) return fmt.Errorf("get sub tree node from main stream: %w", err)
} }
} else { } else {
s.innerStream = &DummySubTreeStream{data: node} s.innerStream = &DummySubTreeStream{data: node}
} }
}
return nil
}
func (s *VersionsByPrefixStreamImpl) getNodeVersionFromInnerStream() (*data.NodeVersion, error) {
for true {
node, err := s.innerStream.Next() node, err := s.innerStream.Next()
if err != nil { if err != nil {
if errors.Is(err, io.EOF) {
s.innerStream = nil
s.namesMap = map[uint64]string{}
if s.latestOnly && s.currentLatest != nil && s.currentLatest.ID != s.intermediateRootID {
return s.currentLatest, nil
}
return s.Next(ctx)
}
return nil, fmt.Errorf("inner stream: %w", err) return nil, fmt.Errorf("inner stream: %w", err)
} }
treeNode, fileName, err := parseTreeNode(node) nodeVersion, skip, err := s.parseNodeResponse(node)
if err != nil { if err != nil {
return s.Next(ctx) return nil, err
} }
if skip {
var parentPrefix string continue
if s.headPrefix != "" { // The root of subTree can also have a parent
parentPrefix = strings.TrimSuffix(s.headPrefix, separator) + separator // To avoid 'foo//bar'
} }
var filepath string
if treeNode.ID != s.intermediateRootID {
if filepath, err = formFilePath(node, fileName, s.namesMap); err != nil {
return nil, fmt.Errorf("invalid node order: %w", err)
}
} else {
filepath = parentPrefix + fileName
s.namesMap[treeNode.ID] = filepath
}
if treeNode.ObjID.Equals(oid.ID{}) { // The node can be intermediate, but we still want to update namesMap
return s.Next(ctx)
}
nodeVersion := newNodeVersionFromTreeNode(filepath, treeNode)
if s.latestOnly { if s.latestOnly {
if s.currentLatest == nil { if s.currentLatest == nil {
s.currentLatest = nodeVersion s.currentLatest = nodeVersion
return s.Next(ctx) continue
} }
if s.currentLatest.FilePath != nodeVersion.FilePath { if s.currentLatest.FilePath != nodeVersion.FilePath {
@ -762,10 +777,42 @@ func (s *VersionsByPrefixStreamImpl) Next(ctx context.Context) (*data.NodeVersio
s.currentLatest = nodeVersion s.currentLatest = nodeVersion
} }
return s.Next(ctx) continue
} }
return nodeVersion, nil return nodeVersion, nil
}
panic("unreachable code")
}
func (s *VersionsByPrefixStreamImpl) parseNodeResponse(node NodeResponse) (res *data.NodeVersion, skip bool, err error) {
trNode, fileName, err := parseTreeNode(node)
if err != nil {
s.log.Debug("parse tree node", zap.Error(err))
return nil, true, nil
}
var parentPrefix string
if s.headPrefix != "" { // The root of subTree can also have a parent
parentPrefix = strings.TrimSuffix(s.headPrefix, separator) + separator // To avoid 'foo//bar'
}
var filepath string
if trNode.ID != s.intermediateRootID {
if filepath, err = formFilePath(node, fileName, s.namesMap); err != nil {
return nil, false, fmt.Errorf("invalid node order: %w", err)
}
} else {
filepath = parentPrefix + fileName
s.namesMap[trNode.ID] = filepath
}
if trNode.ObjID.Equals(oid.ID{}) { // The node can be intermediate, but we still want to update namesMap
return nil, true, nil
}
return newNodeVersionFromTreeNode(filepath, trNode), false, nil
} }
func (c *Tree) InitVersionsByPrefixStream(ctx context.Context, bktInfo *data.BucketInfo, prefix string, latestOnly bool) (data.VersionsStream, error) { func (c *Tree) InitVersionsByPrefixStream(ctx context.Context, bktInfo *data.BucketInfo, prefix string, latestOnly bool) (data.VersionsStream, error) {
@ -787,6 +834,7 @@ func (c *Tree) InitVersionsByPrefixStream(ctx context.Context, bktInfo *data.Buc
headPrefix: strings.TrimSuffix(prefix, tailPrefix), headPrefix: strings.TrimSuffix(prefix, tailPrefix),
tailPrefix: tailPrefix, tailPrefix: tailPrefix,
latestOnly: latestOnly, latestOnly: latestOnly,
log: c.log,
}, nil }, nil
} }