aboutsummaryrefslogtreecommitdiff
path: root/weed
diff options
context:
space:
mode:
Diffstat (limited to 'weed')
-rw-r--r--weed/command/filer_sync.go5
-rw-r--r--weed/filesys/dir_rename.go26
-rw-r--r--weed/filesys/file.go5
-rw-r--r--weed/filesys/wfs.go22
-rw-r--r--weed/pb/filer_pb_tail.go2
-rw-r--r--weed/server/filer_grpc_server_rename.go15
-rw-r--r--weed/util/constants.go2
7 files changed, 62 insertions, 15 deletions
diff --git a/weed/command/filer_sync.go b/weed/command/filer_sync.go
index 326bd1fbe..172be6a9a 100644
--- a/weed/command/filer_sync.go
+++ b/weed/command/filer_sync.go
@@ -169,8 +169,11 @@ func doSubscribeFilerMetaChanges(clientId int32, grpcDialOption grpc.DialOption,
return persistEventFn(resp)
}
+ var lastLogTsNs = time.Now().Nanosecond()
processEventFnWithOffset := pb.AddOffsetFunc(processEventFn, 3*time.Second, func(counter int64, lastTsNs int64) error {
- glog.V(0).Infof("sync %s to %s progressed to %v %0.2f/sec", sourceFiler, targetFiler, time.Unix(0, lastTsNs), float64(counter)/float64(3))
+ now := time.Now().Nanosecond()
+ glog.V(0).Infof("sync %s to %s progressed to %v %0.2f/sec", sourceFiler, targetFiler, time.Unix(0, lastTsNs), float64(counter)/(float64(now-lastLogTsNs)/1e9))
+ lastLogTsNs = now
return setOffset(grpcDialOption, targetFiler, SyncKeyPrefix, sourceFilerSignature, lastTsNs)
})
diff --git a/weed/filesys/dir_rename.go b/weed/filesys/dir_rename.go
index f28c8af92..01a8df175 100644
--- a/weed/filesys/dir_rename.go
+++ b/weed/filesys/dir_rename.go
@@ -3,13 +3,13 @@ package filesys
import (
"context"
"github.com/chrislusf/seaweedfs/weed/filer"
- "github.com/seaweedfs/fuse"
- "github.com/seaweedfs/fuse/fs"
- "io"
-
"github.com/chrislusf/seaweedfs/weed/glog"
"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
"github.com/chrislusf/seaweedfs/weed/util"
+ "github.com/seaweedfs/fuse"
+ "github.com/seaweedfs/fuse/fs"
+ "io"
+ "strings"
)
func (dir *Dir) Rename(ctx context.Context, req *fuse.RenameRequest, newDirectory fs.Node) error {
@@ -37,7 +37,7 @@ func (dir *Dir) Rename(ctx context.Context, req *fuse.RenameRequest, newDirector
stream, err := client.StreamRenameEntry(ctx, request)
if err != nil {
glog.Errorf("dir AtomicRenameEntry %s => %s : %v", oldPath, newPath, err)
- return fuse.EXDEV
+ return fuse.EIO
}
for {
@@ -46,11 +46,19 @@ func (dir *Dir) Rename(ctx context.Context, req *fuse.RenameRequest, newDirector
if recvErr == io.EOF {
break
} else {
+ glog.V(0).Infof("dir Rename %s => %s receive: %v", oldPath, newPath, recvErr)
+ if strings.Contains(recvErr.Error(), "not empty") {
+ return fuse.EEXIST
+ }
+ if strings.Contains(recvErr.Error(), "not directory") {
+ return fuse.ENOTDIR
+ }
return recvErr
}
}
if err = dir.handleRenameResponse(ctx, resp); err != nil {
+ glog.V(0).Infof("dir Rename %s => %s : %v", oldPath, newPath, err)
return err
}
@@ -59,12 +67,8 @@ func (dir *Dir) Rename(ctx context.Context, req *fuse.RenameRequest, newDirector
return nil
})
- if err != nil {
- glog.V(0).Infof("dir Rename %s => %s : %v", oldPath, newPath, err)
- return fuse.EIO
- }
- return nil
+ return err
}
func (dir *Dir) handleRenameResponse(ctx context.Context, resp *filer_pb.StreamRenameEntryResponse) error {
@@ -114,7 +118,9 @@ func (dir *Dir) handleRenameResponse(ctx context.Context, resp *filer_pb.StreamR
if existingHandle, found := dir.wfs.handles[inodeId]; found && existingHandle != nil {
glog.V(4).Infof("opened file handle %s => %s", oldPath, newPath)
delete(dir.wfs.handles, inodeId)
+ existingHandle.handle = newPath.AsInode()
existingHandle.f.entry.Name = newName
+ existingHandle.f.id = newPath.AsInode()
dir.wfs.handles[newPath.AsInode()] = existingHandle
}
dir.wfs.handlesLock.Unlock()
diff --git a/weed/filesys/file.go b/weed/filesys/file.go
index c7652843c..244ed38ea 100644
--- a/weed/filesys/file.go
+++ b/weed/filesys/file.go
@@ -248,11 +248,12 @@ func (file *File) Listxattr(ctx context.Context, req *fuse.ListxattrRequest, res
}
func (file *File) Fsync(ctx context.Context, req *fuse.FsyncRequest) error {
- // fsync works at OS level
+
// write the file chunks to the filerGrpcAddress
glog.V(4).Infof("%s/%s fsync file %+v", file.dir.FullPath(), file.Name, req)
- return nil
+ return file.wfs.Fsync(file, req.Header)
+
}
func (file *File) Forget() {
diff --git a/weed/filesys/wfs.go b/weed/filesys/wfs.go
index 127c160c4..c9a7ac0a1 100644
--- a/weed/filesys/wfs.go
+++ b/weed/filesys/wfs.go
@@ -179,6 +179,28 @@ func (wfs *WFS) AcquireHandle(file *File, uid, gid uint32) (fileHandle *FileHand
return
}
+func (wfs *WFS) Fsync(file *File, header fuse.Header) error {
+
+ inodeId := file.Id()
+
+ wfs.handlesLock.Lock()
+ existingHandle, found := wfs.handles[inodeId]
+ wfs.handlesLock.Unlock()
+
+ if found && existingHandle != nil && existingHandle.f.isOpen > 0 {
+
+ existingHandle.Add(1)
+ defer existingHandle.Done()
+
+ existingHandle.Lock()
+ defer existingHandle.Unlock()
+
+ return existingHandle.doFlush(context.Background(), header)
+ }
+
+ return nil
+}
+
func (wfs *WFS) ReleaseHandle(fullpath util.FullPath, handleId fuse.HandleID) {
wfs.handlesLock.Lock()
defer wfs.handlesLock.Unlock()
diff --git a/weed/pb/filer_pb_tail.go b/weed/pb/filer_pb_tail.go
index 1afddc725..16c7bc8f4 100644
--- a/weed/pb/filer_pb_tail.go
+++ b/weed/pb/filer_pb_tail.go
@@ -84,11 +84,11 @@ func AddOffsetFunc(processEventFn ProcessMetadataFunc, offsetInterval time.Durat
}
counter++
if lastWriteTime.Add(offsetInterval).Before(time.Now()) {
- counter = 0
lastWriteTime = time.Now()
if err := offsetFunc(counter, resp.TsNs); err != nil {
return err
}
+ counter = 0
}
return nil
}
diff --git a/weed/server/filer_grpc_server_rename.go b/weed/server/filer_grpc_server_rename.go
index 62bcccabb..d70df5db7 100644
--- a/weed/server/filer_grpc_server_rename.go
+++ b/weed/server/filer_grpc_server_rename.go
@@ -72,6 +72,21 @@ func (fs *FilerServer) StreamRenameEntry(req *filer_pb.StreamRenameEntryRequest,
return fmt.Errorf("%s/%s not found: %v", req.OldDirectory, req.OldName, err)
}
+ // follow https://pubs.opengroup.org/onlinepubs/000095399/functions/rename.html
+ if oldEntry.IsDirectory() {
+ targetDir := newParent.Child(req.NewName)
+ newEntry, err := fs.filer.FindEntry(ctx, targetDir)
+ if err == nil {
+ if !newEntry.IsDirectory() {
+ fs.filer.RollbackTransaction(ctx)
+ return fmt.Errorf("%s is not directory", targetDir)
+ }
+ if entries, _, _ := fs.filer.ListDirectoryEntries(context.Background(), targetDir, "", false, 1, "", "", ""); len(entries) > 0 {
+ return fmt.Errorf("%s is not empty", targetDir)
+ }
+ }
+ }
+
moveErr := fs.moveEntry(ctx, stream, oldParent, oldEntry, newParent, req.NewName, req.Signatures)
if moveErr != nil {
fs.filer.RollbackTransaction(ctx)
diff --git a/weed/util/constants.go b/weed/util/constants.go
index bd59c327f..c00ea792b 100644
--- a/weed/util/constants.go
+++ b/weed/util/constants.go
@@ -5,7 +5,7 @@ import (
)
var (
- VERSION_NUMBER = fmt.Sprintf("%.02f", 2.84)
+ VERSION_NUMBER = fmt.Sprintf("%.02f", 2.85)
VERSION = sizeLimit + " " + VERSION_NUMBER
COMMIT = ""
)
generated by cgit v1.2.3 (git 2.25.1) at 2025-12-22 08:22:36 +0000