allow passing reader to read metadata (#3886)

Signed-off-by: Jörn Friedrich Dreyer <[email protected]>

changelog/unreleased/fix-deadlock-by-passing-reader.md

@@ -0,0 +1,5 @@
+Bugfix: decomposedfs no longer deadlocks when cache is disabled
+
+We now pass a Reader for the locked file to lower level functions so metadata can be read without aquiring a new file lock.
+
+https://github.com/cs3org/reva/pull/3886

pkg/storage/utils/decomposedfs/node/node.go

@@ -405,8 +405,8 @@ func (n *Node) Child(ctx context.Context, name string) (*Node, error) {
 	return c, nil
 }
 
-// Parent returns the parent node
-func (n *Node) Parent() (p *Node, err error) {
+// ParentWithReader returns the parent node
+func (n *Node) ParentWithReader(r io.Reader) (p *Node, err error) {
 	if n.ParentID == "" {
 		return nil, fmt.Errorf("decomposedfs: root has no parent")
 	}
@@ -417,6 +417,9 @@ func (n *Node) Parent() (p *Node, err error) {
 		SpaceRoot: n.SpaceRoot,
 	}
 
+	// fill metadata cache using the reader
+	_, _ = p.XattrsWithReader(r)
+
 	// lookup name and parent id in extended attributes
 	p.ParentID, _ = p.XattrString(prefixes.ParentidAttr)
 	p.Name, _ = p.XattrString(prefixes.NameAttr)
@@ -428,6 +431,11 @@ func (n *Node) Parent() (p *Node, err error) {
 	return
 }
 
+// Parent returns the parent node
+func (n *Node) Parent() (p *Node, err error) {
+	return n.ParentWithReader(nil)
+}
+
 // Owner returns the space owner
 func (n *Node) Owner() *userpb.UserId {
 	return n.SpaceRoot.owner

pkg/storage/utils/decomposedfs/node/xattrs.go

@@ -19,6 +19,7 @@
 package node
 
 import (
+	"io"
 	"strconv"
 
 	"github.com/pkg/xattr"
@@ -84,21 +85,34 @@ func (n *Node) RemoveXattr(key string) error {
 	return n.lu.MetadataBackend().Remove(n.InternalPath(), key)
 }
 
-// Xattrs returns the extended attributes of the node. If the attributes have already
+// XattrsWithReader returns the extended attributes of the node. If the attributes have already
 // been cached they are not read from disk again.
-func (n *Node) Xattrs() (Attributes, error) {
+func (n *Node) XattrsWithReader(r io.Reader) (Attributes, error) {
 	if n.xattrsCache != nil {
 		return n.xattrsCache, nil
 	}
 
-	attrs, err := n.lu.MetadataBackend().All(n.InternalPath())
+	var attrs Attributes
+	var err error
+	if r != nil {
+		attrs, err = n.lu.MetadataBackend().AllWithLockedSource(n.InternalPath(), r)
+	} else {
+		attrs, err = n.lu.MetadataBackend().All(n.InternalPath())
+	}
 	if err != nil {
 		return nil, err
 	}
+
 	n.xattrsCache = attrs
 	return n.xattrsCache, nil
 }
 
+// Xattrs returns the extended attributes of the node. If the attributes have already
+// been cached they are not read from disk again.
+func (n *Node) Xattrs() (Attributes, error) {
+	return n.XattrsWithReader(nil)
+}
+
 // Xattr returns an extended attribute of the node. If the attributes have already
 // been cached it is not read from disk again.
 func (n *Node) Xattr(key string) ([]byte, error) {

pkg/storage/utils/decomposedfs/tree/tree.go

@@ -731,7 +731,7 @@ func (t *Tree) Propagate(ctx context.Context, n *node.Node, sizeDiff int64) (err
 			}
 		}()
 
-		if n, err = n.Parent(); err != nil {
+		if n, err = n.ParentWithReader(f); err != nil {
 			return err
 		}