mount: streaming renaming folders

other/java/client/src/main/proto/filer.proto

@@ -30,6 +30,8 @@ service SeaweedFiler {
 
     rpc AtomicRenameEntry (AtomicRenameEntryRequest) returns (AtomicRenameEntryResponse) {
     }
+    rpc StreamRenameEntry (StreamRenameEntryRequest) returns (stream StreamRenameEntryResponse) {
+    }
 
     rpc AssignVolume (AssignVolumeRequest) returns (AssignVolumeResponse) {
     }
@@ -225,6 +227,18 @@ message AtomicRenameEntryRequest {
 message AtomicRenameEntryResponse {
 }
 
+message StreamRenameEntryRequest {
+    string old_directory = 1;
+    string old_name = 2;
+    string new_directory = 3;
+    string new_name = 4;
+    repeated int32 signatures = 5;
+}
+message StreamRenameEntryResponse {
+    string directory = 1;
+    EventNotification event_notification = 2;
+    int64 ts_ns = 3;
+}
 message AssignVolumeRequest {
     int32 count = 1;
     string collection = 2;

weed/filesys/dir_rename.go

@@ -2,12 +2,10 @@ package filesys
 
 import (
 	"context"
-	"fmt"
 	"github.com/chrislusf/seaweedfs/weed/filer"
-	"math"
-
 	"github.com/seaweedfs/fuse"
 	"github.com/seaweedfs/fuse/fs"
+	"io"
 
 	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
@@ -23,19 +21,12 @@ func (dir *Dir) Rename(ctx context.Context, req *fuse.RenameRequest, newDirector
 
 	glog.V(4).Infof("dir Rename %s => %s", oldPath, newPath)
 
-	// find local old entry
-	oldEntry, err := dir.wfs.metaCache.FindEntry(context.Background(), oldPath)
-	if err != nil {
-		glog.Errorf("dir Rename can not find source %s : %v", oldPath, err)
-		return fuse.ENOENT
-	}
-
 	// update remote filer
-	err = dir.wfs.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
+	err := dir.wfs.WithFilerClient(func(client filer_pb.SeaweedFilerClient) error {
 		ctx, cancel := context.WithCancel(context.Background())
 		defer cancel()
 
-		request := &filer_pb.AtomicRenameEntryRequest{
+		request := &filer_pb.StreamRenameEntryRequest{
 			OldDirectory: dir.FullPath(),
 			OldName:      req.OldName,
 			NewDirectory: newDir.FullPath(),
@@ -43,12 +34,28 @@ func (dir *Dir) Rename(ctx context.Context, req *fuse.RenameRequest, newDirector
 			Signatures:   []int32{dir.wfs.signature},
 		}
 
-		_, err := client.AtomicRenameEntry(ctx, request)
+		stream, err := client.StreamRenameEntry(ctx, request)
 		if err != nil {
 			glog.Errorf("dir AtomicRenameEntry %s => %s : %v", oldPath, newPath, err)
 			return fuse.EXDEV
 		}
 
+		for {
+			resp, recvErr := stream.Recv()
+			if recvErr != nil {
+				if recvErr == io.EOF {
+					break
+				} else {
+					return recvErr
+				}
+			}
+
+			if err = dir.handleRenameResponse(ctx, resp); err != nil {
+				return err
+			}
+
+		}
+
 		return nil
 
 	})
@@ -57,23 +64,25 @@ func (dir *Dir) Rename(ctx context.Context, req *fuse.RenameRequest, newDirector
 		return fuse.EIO
 	}
 
-	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
-	}
-
 	return nil
 }
 
-func (dir *Dir) moveEntry(ctx context.Context, oldParent util.FullPath, entry *filer.Entry, newParent util.FullPath, newName string) error {
 
-	oldName := entry.Name()
+func (dir *Dir) handleRenameResponse(ctx context.Context, resp *filer_pb.StreamRenameEntryResponse) error {
+	// comes from filer StreamRenameEntry, can only be create or delete entry
+
+	if resp.EventNotification.NewEntry != nil {
+		// with new entry, the old entry name also exists. This is the first step to create new entry
+		newEntry := filer.FromPbEntry(resp.EventNotification.NewParentPath, resp.EventNotification.NewEntry)
+		if err := dir.wfs.metaCache.AtomicUpdateEntryFromFiler(ctx, "", newEntry); err != nil {
+			return err
+		}
 
-	oldPath := oldParent.Child(oldName)
-	newPath := newParent.Child(newName)
-	if err := dir.moveSelfEntry(ctx, oldParent, entry, newParent, newName, func() error {
+		oldParent, newParent := util.FullPath(resp.Directory), util.FullPath(resp.EventNotification.NewParentPath)
+		oldName, newName := resp.EventNotification.OldEntry.Name, resp.EventNotification.NewEntry.Name
 
+		oldPath := oldParent.Child(oldName)
+		newPath := newParent.Child(newName)
 		oldFsNode := NodeWithId(oldPath.AsInode())
 		newFsNode := NodeWithId(newPath.AsInode())
 		newDirNode, found := dir.wfs.Server.FindInternalNode(NodeWithId(newParent.AsInode()))
@@ -110,65 +119,13 @@ func (dir *Dir) moveEntry(ctx context.Context, oldParent util.FullPath, entry *f
 		}
 		dir.wfs.handlesLock.Unlock()
 
-		if entry.IsDirectory() {
-			if err := dir.moveFolderSubEntries(ctx, oldParent, oldName, newParent, newName); err != nil {
-				return err
-			}
+	}else if resp.EventNotification.OldEntry != nil {
+		// without new entry, only old entry name exists. This is the second step to delete old entry
+		if err := dir.wfs.metaCache.AtomicUpdateEntryFromFiler(ctx, util.NewFullPath(resp.Directory, resp.EventNotification.OldEntry.Name), nil); err != nil {
+			return err
 		}
-		return nil
-	}); err != nil {
-		return fmt.Errorf("fail to move %s => %s: %v", oldPath, newPath, 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
 }

weed/filesys/meta_cache/meta_cache.go

@@ -2,7 +2,6 @@ package meta_cache
 
 import (
 	"context"
-	"fmt"
 	"github.com/chrislusf/seaweedfs/weed/filer"
 	"github.com/chrislusf/seaweedfs/weed/filer/leveldb"
 	"github.com/chrislusf/seaweedfs/weed/glog"
@@ -122,7 +121,8 @@ func (mc *MetaCache) ListDirectoryEntries(ctx context.Context, dirPath util.Full
 	//defer mc.RUnlock()
 
 	if !mc.visitedBoundary.HasVisited(dirPath) {
-		return fmt.Errorf("unsynchronized dir: %v", dirPath)
+		// if this request comes after renaming, it should be fine
+		glog.Warningf("unsynchronized dir: %v", dirPath)
 	}
 
 	_, err := mc.localStore.ListDirectoryEntries(ctx, dirPath, startFileName, includeStartFile, limit, func(entry *filer.Entry) bool {

weed/pb/filer.proto

@@ -30,6 +30,8 @@ service SeaweedFiler {
 
     rpc AtomicRenameEntry (AtomicRenameEntryRequest) returns (AtomicRenameEntryResponse) {
     }
+    rpc StreamRenameEntry (StreamRenameEntryRequest) returns (stream StreamRenameEntryResponse) {
+    }
 
     rpc AssignVolume (AssignVolumeRequest) returns (AssignVolumeResponse) {
     }
@@ -225,6 +227,18 @@ message AtomicRenameEntryRequest {
 message AtomicRenameEntryResponse {
 }
 
+message StreamRenameEntryRequest {
+    string old_directory = 1;
+    string old_name = 2;
+    string new_directory = 3;
+    string new_name = 4;
+    repeated int32 signatures = 5;
+}
+message StreamRenameEntryResponse {
+    string directory = 1;
+    EventNotification event_notification = 2;
+    int64 ts_ns = 3;
+}
 message AssignVolumeRequest {
     int32 count = 1;
     string collection = 2;

weed/server/filer_grpc_server_rename.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"path/filepath"
+	"time"
 
 	"github.com/chrislusf/seaweedfs/weed/filer"
 	"github.com/chrislusf/seaweedfs/weed/glog"
@@ -33,7 +34,7 @@ func (fs *FilerServer) AtomicRenameEntry(ctx context.Context, req *filer_pb.Atom
 		return nil, fmt.Errorf("%s/%s not found: %v", req.OldDirectory, req.OldName, err)
 	}
 
-	moveErr := fs.moveEntry(ctx, oldParent, oldEntry, newParent, req.NewName, req.Signatures)
+	moveErr := fs.moveEntry(ctx, nil, oldParent, oldEntry, newParent, req.NewName, req.Signatures)
 	if moveErr != nil {
 		fs.filer.RollbackTransaction(ctx)
 		return nil, fmt.Errorf("%s/%s move error: %v", req.OldDirectory, req.OldName, moveErr)
@@ -47,11 +48,49 @@ func (fs *FilerServer) AtomicRenameEntry(ctx context.Context, req *filer_pb.Atom
 	return &filer_pb.AtomicRenameEntryResponse{}, nil
 }
 
-func (fs *FilerServer) moveEntry(ctx context.Context, oldParent util.FullPath, entry *filer.Entry, newParent util.FullPath, newName string, signatures []int32) error {
+func (fs *FilerServer) StreamRenameEntry(req *filer_pb.StreamRenameEntryRequest, stream filer_pb.SeaweedFiler_StreamRenameEntryServer) (err error) {
 
-	if err := fs.moveSelfEntry(ctx, oldParent, entry, newParent, newName, func() error {
+	glog.V(1).Infof("StreamRenameEntry %v", req)
+
+	oldParent := util.FullPath(filepath.ToSlash(req.OldDirectory))
+	newParent := util.FullPath(filepath.ToSlash(req.NewDirectory))
+
+	if err := fs.filer.CanRename(oldParent, newParent); err != nil {
+		return err
+	}
+
+	ctx := context.Background()
+
+	ctx, err = fs.filer.BeginTransaction(ctx)
+	if err != nil {
+		return err
+	}
+
+	oldEntry, err := fs.filer.FindEntry(ctx, oldParent.Child(req.OldName))
+	if err != nil {
+		fs.filer.RollbackTransaction(ctx)
+		return fmt.Errorf("%s/%s not found: %v", req.OldDirectory, req.OldName, err)
+	}
+
+	moveErr := fs.moveEntry(ctx, stream, oldParent, oldEntry, newParent, req.NewName, req.Signatures)
+	if moveErr != nil {
+		fs.filer.RollbackTransaction(ctx)
+		return fmt.Errorf("%s/%s move error: %v", req.OldDirectory, req.OldName, moveErr)
+	} else {
+		if commitError := fs.filer.CommitTransaction(ctx); commitError != nil {
+			fs.filer.RollbackTransaction(ctx)
+			return fmt.Errorf("%s/%s move commit error: %v", req.OldDirectory, req.OldName, commitError)
+		}
+	}
+
+	return nil
+}
+
+func (fs *FilerServer) moveEntry(ctx context.Context, stream filer_pb.SeaweedFiler_StreamRenameEntryServer, oldParent util.FullPath, entry *filer.Entry, newParent util.FullPath, newName string, signatures []int32) error {
+
+	if err := fs.moveSelfEntry(ctx, stream, oldParent, entry, newParent, newName, func() error {
 		if entry.IsDirectory() {
-			if err := fs.moveFolderSubEntries(ctx, oldParent, entry, newParent, newName, signatures); err != nil {
+			if err := fs.moveFolderSubEntries(ctx, stream, oldParent, entry, newParent, newName, signatures); err != nil {
 				return err
 			}
 		}
@@ -63,7 +102,7 @@ func (fs *FilerServer) moveEntry(ctx context.Context, oldParent util.FullPath, e
 	return nil
 }
 
-func (fs *FilerServer) moveFolderSubEntries(ctx context.Context, oldParent util.FullPath, entry *filer.Entry, newParent util.FullPath, newName string, signatures []int32) error {
+func (fs *FilerServer) moveFolderSubEntries(ctx context.Context, stream filer_pb.SeaweedFiler_StreamRenameEntryServer, oldParent util.FullPath, entry *filer.Entry, newParent util.FullPath, newName string, signatures []int32) error {
 
 	currentDirPath := oldParent.Child(entry.Name())
 	newDirPath := newParent.Child(newName)
@@ -84,7 +123,7 @@ func (fs *FilerServer) moveFolderSubEntries(ctx context.Context, oldParent util.
 		for _, item := range entries {
 			lastFileName = item.Name()
 			// println("processing", lastFileName)
-			err := fs.moveEntry(ctx, currentDirPath, item, newDirPath, item.Name(), signatures)
+			err := fs.moveEntry(ctx, stream, currentDirPath, item, newDirPath, item.Name(), signatures)
 			if err != nil {
 				return err
 			}
@@ -96,7 +135,7 @@ func (fs *FilerServer) moveFolderSubEntries(ctx context.Context, oldParent util.
 	return nil
 }
 
-func (fs *FilerServer) moveSelfEntry(ctx context.Context, oldParent util.FullPath, entry *filer.Entry, newParent util.FullPath, newName string, moveFolderSubEntries func() error, signatures []int32) error {
+func (fs *FilerServer) moveSelfEntry(ctx context.Context, stream filer_pb.SeaweedFiler_StreamRenameEntryServer, oldParent util.FullPath, entry *filer.Entry, newParent util.FullPath, newName string, moveFolderSubEntries func() error, signatures []int32) error {
 
 	oldPath, newPath := oldParent.Child(entry.Name()), newParent.Child(newName)
 
@@ -118,6 +157,24 @@ func (fs *FilerServer) moveSelfEntry(ctx context.Context, oldParent util.FullPat
 	if createErr := fs.filer.CreateEntry(ctx, newEntry, false, false, signatures); createErr != nil {
 		return createErr
 	}
+	if stream != nil {
+		if err := stream.Send(&filer_pb.StreamRenameEntryResponse{
+			Directory: string(newParent),
+			EventNotification: &filer_pb.EventNotification{
+				OldEntry:           &filer_pb.Entry{
+					Name: entry.Name(),
+				},
+				NewEntry:           newEntry.ToProtoEntry(),
+				DeleteChunks:       false,
+				NewParentPath:      string(newParent),
+				IsFromOtherCluster: false,
+				Signatures:         nil,
+			},
+			TsNs: time.Now().UnixNano(),
+		}); err != nil {
+			return err
+		}
+	}
 
 	if moveFolderSubEntries != nil {
 		if moveChildrenErr := moveFolderSubEntries(); moveChildrenErr != nil {
@@ -130,6 +187,24 @@ func (fs *FilerServer) moveSelfEntry(ctx context.Context, oldParent util.FullPat
 	if deleteErr != nil {
 		return deleteErr
 	}
+	if stream != nil {
+		if err := stream.Send(&filer_pb.StreamRenameEntryResponse{
+			Directory: string(oldParent),
+			EventNotification: &filer_pb.EventNotification{
+				OldEntry: &filer_pb.Entry{
+					Name: entry.Name(),
+				},
+				NewEntry:           nil,
+				DeleteChunks:       false,
+				NewParentPath:      "",
+				IsFromOtherCluster: false,
+				Signatures:         nil,
+			},
+			TsNs: time.Now().UnixNano(),
+		}); err != nil {
+			return err
+		}
+	}
 
 	return nil