mount: recursively rename locally

2 files changed, 118 insertions, 43 deletions

diff --git a/weed/filesys/dir_rename.go b/weed/filesys/dir_rename.go
index d50c6dab0..27a14d5c6 100644
--- a/weed/filesys/dir_rename.go
+++ b/weed/filesys/dir_rename.go
@@ -2,6 +2,9 @@ package filesys
 
 import (
 	"context"
+	"fmt"
+	"github.com/chrislusf/seaweedfs/weed/filer"
+	"math"
 
 	"github.com/seaweedfs/fuse"
 	"github.com/seaweedfs/fuse/fs"
@@ -37,6 +40,7 @@ func (dir *Dir) Rename(ctx context.Context, req *fuse.RenameRequest, newDirector
 			OldName:      req.OldName,
 			NewDirectory: newDir.FullPath(),
 			NewName:      req.NewName,
+			Signatures:   []int32{dir.wfs.signature},
 		}
 
 		_, err := client.AtomicRenameEntry(ctx, request)
@@ -53,34 +57,12 @@ func (dir *Dir) Rename(ctx context.Context, req *fuse.RenameRequest, newDirector
 		return fuse.EIO
 	}
 
-	// TODO: replicate renaming logic on filer
-	if err := dir.wfs.metaCache.DeleteEntry(context.Background(), oldPath); err != nil {
-		glog.V(0).Infof("dir Rename delete local %s => %s : %v", oldPath, newPath, err)
-		return fuse.EIO
-	}
-	oldEntry.FullPath = newPath
-	if err := dir.wfs.metaCache.InsertEntry(context.Background(), oldEntry); err != nil {
-		glog.V(0).Infof("dir Rename insert local %s => %s : %v", oldPath, newPath, err)
+	err = dir.moveEntry(context.Background(), util.FullPath(dir.FullPath()), oldEntry, util.FullPath(newDir.FullPath()), req.NewName)
+	if err != nil {
+		glog.V(0).Infof("dir local Rename %s => %s : %v", oldPath, newPath, err)
 		return fuse.EIO
 	}
 
-	oldFsNode := NodeWithId(oldPath.AsInode())
-	newFsNode := NodeWithId(newPath.AsInode())
-	dir.wfs.Server.InvalidateInternalNode(oldFsNode, newFsNode, func(internalNode fs.Node) {
-		if file, ok := internalNode.(*File); ok {
-			glog.V(4).Infof("internal file node %s", file.Name)
-			file.Name = req.NewName
-			file.id = uint64(newFsNode)
-			file.dir = newDir
-		}
-		if dir, ok := internalNode.(*Dir); ok {
-			glog.V(4).Infof("internal dir node %s", dir.name)
-			dir.name = req.NewName
-			dir.id = uint64(newFsNode)
-			dir.parent = newDir
-		}
-	})
-
 	// change file handle
 	dir.wfs.handlesLock.Lock()
 	defer dir.wfs.handlesLock.Unlock()
@@ -96,3 +78,98 @@ func (dir *Dir) Rename(ctx context.Context, req *fuse.RenameRequest, newDirector
 
 	return nil
 }
+
+func (dir *Dir) moveEntry(ctx context.Context, oldParent util.FullPath, entry *filer.Entry, newParent util.FullPath, newName string) error {
+
+	oldName := entry.Name()
+
+	if err := dir.moveSelfEntry(ctx, oldParent, entry, newParent, newName, func() error {
+
+		oldFsNode := NodeWithId(oldParent.Child(oldName).AsInode())
+		newFsNode := NodeWithId(newParent.Child(newName).AsInode())
+		newDirNode, found := dir.wfs.Server.FindInternalNode(NodeWithId(newParent.AsInode()))
+		var newDir *Dir
+		if found {
+			newDir = newDirNode.(*Dir)
+		}
+		dir.wfs.Server.InvalidateInternalNode(oldFsNode, newFsNode, func(internalNode fs.Node) {
+			if file, ok := internalNode.(*File); ok {
+				glog.V(4).Infof("internal file node %s", oldParent.Child(oldName))
+				file.Name = newName
+				file.id = uint64(newFsNode)
+				if found {
+					file.dir = newDir
+				}
+			}
+			if dir, ok := internalNode.(*Dir); ok {
+				glog.V(4).Infof("internal dir node %s", oldParent.Child(oldName))
+				dir.name = newName
+				dir.id = uint64(newFsNode)
+				if found {
+					dir.parent = newDir
+				}
+			}
+		})
+
+		if entry.IsDirectory() {
+			if err := dir.moveFolderSubEntries(ctx, oldParent, oldName, newParent, newName); err != nil {
+				return err
+			}
+		}
+		return nil
+	}); err != nil {
+		return fmt.Errorf("fail to move %s => %s: %v", oldParent.Child(oldName), newParent.Child(newName), err)
+	}
+
+	return nil
+}
+
+func (dir *Dir) moveFolderSubEntries(ctx context.Context, oldParent util.FullPath, oldName string, newParent util.FullPath, newName string) error {
+
+	currentDirPath := oldParent.Child(oldName)
+	newDirPath := newParent.Child(newName)
+
+	glog.V(1).Infof("moving folder %s => %s", currentDirPath, newDirPath)
+
+	var moveErr error
+	listErr := dir.wfs.metaCache.ListDirectoryEntries(ctx, currentDirPath, "", false, int64(math.MaxInt32), func(item *filer.Entry) bool {
+		moveErr = dir.moveEntry(ctx, currentDirPath, item, newDirPath, item.Name())
+		if moveErr != nil {
+			return false
+		}
+		return true
+	})
+	if listErr != nil {
+		return listErr
+	}
+	if moveErr != nil {
+		return moveErr
+	}
+
+	return nil
+}
+
+func (dir *Dir) moveSelfEntry(ctx context.Context, oldParent util.FullPath, entry *filer.Entry, newParent util.FullPath, newName string, moveFolderSubEntries func() error) error {
+
+	newPath := newParent.Child(newName)
+	oldPath := oldParent.Child(entry.Name())
+
+	entry.FullPath = newPath
+	if err := dir.wfs.metaCache.InsertEntry(ctx, entry); err != nil {
+		glog.V(0).Infof("dir Rename insert local %s => %s : %v", oldPath, newPath, err)
+		return fuse.EIO
+	}
+
+	if moveFolderSubEntries != nil {
+		if moveChildrenErr := moveFolderSubEntries(); moveChildrenErr != nil {
+			return moveChildrenErr
+		}
+	}
+
+	if err := dir.wfs.metaCache.DeleteEntry(ctx, oldPath); err != nil {
+		glog.V(0).Infof("dir Rename delete local %s => %s : %v", oldPath, newPath, err)
+		return fuse.EIO
+	}
+
+	return nil
+}
diff --git a/weed/filesys/meta_cache/meta_cache.go b/weed/filesys/meta_cache/meta_cache.go
index 3f6391c39..69d1655ee 100644
--- a/weed/filesys/meta_cache/meta_cache.go
+++ b/weed/filesys/meta_cache/meta_cache.go
@@ -3,14 +3,12 @@ package meta_cache
 import (
 	"context"
 	"fmt"
-	"os"
-	"sync"
-
 	"github.com/chrislusf/seaweedfs/weed/filer"
 	"github.com/chrislusf/seaweedfs/weed/filer/leveldb"
 	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/util"
 	"github.com/chrislusf/seaweedfs/weed/util/bounded_tree"
+	"os"
 )
 
 // need to have logic similar to FilerStoreWrapper
@@ -18,7 +16,7 @@ import (
 
 type MetaCache struct {
 	localStore filer.VirtualFilerStore
-	sync.RWMutex
+	// sync.RWMutex
 	visitedBoundary *bounded_tree.BoundedTree
 	uidGidMapper    *UidGidMapper
 	invalidateFunc  func(util.FullPath)
@@ -54,8 +52,8 @@ func openMetaStore(dbFolder string) filer.VirtualFilerStore {
 }
 
 func (mc *MetaCache) InsertEntry(ctx context.Context, entry *filer.Entry) error {
-	mc.Lock()
-	defer mc.Unlock()
+	//mc.Lock()
+	//defer mc.Unlock()
 	return mc.doInsertEntry(ctx, entry)
 }
 
@@ -64,8 +62,8 @@ func (mc *MetaCache) doInsertEntry(ctx context.Context, entry *filer.Entry) erro
 }
 
 func (mc *MetaCache) AtomicUpdateEntryFromFiler(ctx context.Context, oldPath util.FullPath, newEntry *filer.Entry) error {
-	mc.Lock()
-	defer mc.Unlock()
+	//mc.Lock()
+	//defer mc.Unlock()
 
 	oldDir, _ := oldPath.DirAndName()
 	if mc.visitedBoundary.HasVisited(util.FullPath(oldDir)) {
@@ -97,14 +95,14 @@ func (mc *MetaCache) AtomicUpdateEntryFromFiler(ctx context.Context, oldPath uti
 }
 
 func (mc *MetaCache) UpdateEntry(ctx context.Context, entry *filer.Entry) error {
-	mc.Lock()
-	defer mc.Unlock()
+	//mc.Lock()
+	//defer mc.Unlock()
 	return mc.localStore.UpdateEntry(ctx, entry)
 }
 
 func (mc *MetaCache) FindEntry(ctx context.Context, fp util.FullPath) (entry *filer.Entry, err error) {
-	mc.RLock()
-	defer mc.RUnlock()
+	//mc.RLock()
+	//defer mc.RUnlock()
 	entry, err = mc.localStore.FindEntry(ctx, fp)
 	if err != nil {
 		return nil, err
@@ -114,14 +112,14 @@ func (mc *MetaCache) FindEntry(ctx context.Context, fp util.FullPath) (entry *fi
 }
 
 func (mc *MetaCache) DeleteEntry(ctx context.Context, fp util.FullPath) (err error) {
-	mc.Lock()
-	defer mc.Unlock()
+	//mc.Lock()
+	//defer mc.Unlock()
 	return mc.localStore.DeleteEntry(ctx, fp)
 }
 
 func (mc *MetaCache) ListDirectoryEntries(ctx context.Context, dirPath util.FullPath, startFileName string, includeStartFile bool, limit int64, eachEntryFunc filer.ListEachEntryFunc) error {
-	mc.RLock()
-	defer mc.RUnlock()
+	//mc.RLock()
+	//defer mc.RUnlock()
 
 	if !mc.visitedBoundary.HasVisited(dirPath) {
 		return fmt.Errorf("unsynchronized dir: %v", dirPath)
@@ -138,8 +136,8 @@ func (mc *MetaCache) ListDirectoryEntries(ctx context.Context, dirPath util.Full
 }
 
 func (mc *MetaCache) Shutdown() {
-	mc.Lock()
-	defer mc.Unlock()
+	//mc.Lock()
+	//defer mc.Unlock()
 	mc.localStore.Shutdown()
 }