From 3481f13f54e57af890b31625ea7032049161bc78 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sun, 24 May 2026 23:56:37 -0700 Subject: [PATCH] mount: route POSIX advisory locks to the owner filer under -dlm (#9669) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit With -dlm, GetLk/SetLk/SetLkw and the flush/release cleanup paths go to the inode's owner filer via the PosixLock RPC instead of the local table, so flock/fcntl are honored across mounts. Advisory locking rides the same switch as whole-file write coordination — and is therefore off under writeback cache, which implies single-writer. Keys are the inode identity (HardLinkId else path); SetLkw is client-side polling with the FUSE cancel channel (no server wait queue); a per-mount session id namespaces owners; a local hint avoids a release RPC on every close. Background unlock/release RPCs are bounded so a stuck filer can't hang close(). --- weed/command/mount.go | 5 +- weed/mount/weedfs.go | 10 +- weed/mount/weedfs_file_io.go | 2 +- weed/mount/weedfs_file_lock.go | 9 + weed/mount/weedfs_file_sync.go | 6 +- weed/mount/weedfs_posix_lock_routed.go | 347 ++++++++++++++++++++ weed/mount/weedfs_posix_lock_routed_test.go | 94 ++++++ 7 files changed, 465 insertions(+), 8 deletions(-) create mode 100644 weed/mount/weedfs_posix_lock_routed.go create mode 100644 weed/mount/weedfs_posix_lock_routed_test.go diff --git a/weed/command/mount.go b/weed/command/mount.go index d23cdfc1b..d400deeeb 100644 --- a/weed/command/mount.go +++ b/weed/command/mount.go @@ -61,7 +61,8 @@ type MountOptions struct { dirIdleEvictSec *int - // Distributed lock for cross-mount write coordination + // Distributed locking for cross-mount write coordination and POSIX + // advisory locks (flock/fcntl) distributedLock *bool // POSIX compliance options @@ -152,7 +153,7 @@ func init() { mountReadRetryTime = cmdMount.Flag.Duration("readRetryTime", 6*time.Second, "maximum read retry wait time") // Distributed lock for cross-mount write coordination - mountOptions.distributedLock = cmdMount.Flag.Bool("dlm", false, "enable distributed lock for cross-mount write coordination (only one mount can write a file at a time)") + mountOptions.distributedLock = cmdMount.Flag.Bool("dlm", false, "coordinate writes across mounts (only one mount writes a file at a time) and honor POSIX advisory locks (flock/fcntl) across mounts by routing them to the owner filer") // POSIX compliance options mountOptions.posixDirNlink = cmdMount.Flag.Bool("posix.dirNLink", false, "report POSIX-compliant directory nlink (2 + subdirectory count); costs one directory listing per stat") diff --git a/weed/mount/weedfs.go b/weed/mount/weedfs.go index 6c50ac04a..164f70d97 100644 --- a/weed/mount/weedfs.go +++ b/weed/mount/weedfs.go @@ -97,8 +97,10 @@ type Option struct { // EnableDistributedLock enables DLM-based write coordination across mounts. // When true, opening a file for write acquires a distributed lock that is - // held (with auto-renewal) until the file is closed. Only one mount can - // have a file open for writing at a time. + // held (with auto-renewal) until the file is closed, so only one mount can + // have a file open for writing at a time; POSIX advisory locks (flock/fcntl) + // are also routed to the inode's owner filer so they are honored across + // mounts. Disabled under writeback cache, which implies single-writer. EnableDistributedLock bool // WritebackCache enables async flush on close for improved small file write performance. @@ -138,6 +140,8 @@ type WFS struct { fhLockTable *util.LockTable[FileHandleId] hardLinkLockTable *util.LockTable[string] posixLocks *PosixLockTable + posixSid uint64 // this mount's session id, for routed-lock owner identity + posixHint *posixLockHint // local fcntl-lock hint for routed mode rdmaClient *RDMAMountClient peerRegistrar *PeerRegistrar peerDirectory *PeerDirectory @@ -242,6 +246,8 @@ func NewSeaweedFileSystem(option *Option) *WFS { fhLockTable: util.NewLockTable[FileHandleId](), hardLinkLockTable: util.NewLockTable[string](), posixLocks: NewPosixLockTable(), + posixSid: randomPosixSid(), + posixHint: newPosixLockHint(), refreshingDirs: make(map[util.FullPath]struct{}), atimeMap: make(map[uint64]time.Time, 8192), openMtimeCache: make(map[uint64][2]int64, 8192), diff --git a/weed/mount/weedfs_file_io.go b/weed/mount/weedfs_file_io.go index c4b615676..959d05849 100644 --- a/weed/mount/weedfs_file_io.go +++ b/weed/mount/weedfs_file_io.go @@ -148,7 +148,7 @@ func (wfs *WFS) Release(cancel <-chan struct{}, in *fuse.ReleaseIn) { } } if in.ReleaseFlags&fuse.FUSE_RELEASE_FLOCK_UNLOCK != 0 { - wfs.posixLocks.ReleaseFlockOwner(in.NodeId, in.LockOwner) + wfs.releaseFlockOwner(in.NodeId, in.LockOwner) } wfs.ReleaseHandle(FileHandleId(in.Fh)) } diff --git a/weed/mount/weedfs_file_lock.go b/weed/mount/weedfs_file_lock.go index 08de8b564..0e67d2fd5 100644 --- a/weed/mount/weedfs_file_lock.go +++ b/weed/mount/weedfs_file_lock.go @@ -10,6 +10,9 @@ import ( // If a conflict exists, the conflicting lock is returned in out. // If no conflict, out.Lk.Typ is set to F_UNLCK. func (wfs *WFS) GetLk(cancel <-chan struct{}, in *fuse.LkIn, out *fuse.LkOut) fuse.Status { + if wfs.crossMountLocks() { + return wfs.routedGetLk(cancel, in, out) + } proposed := lockRange{ Start: in.Lk.Start, End: in.Lk.End, @@ -25,6 +28,9 @@ func (wfs *WFS) GetLk(cancel <-chan struct{}, in *fuse.LkIn, out *fuse.LkOut) fu // SetLk sets or clears a POSIX lock (non-blocking). // Returns EAGAIN if the lock conflicts with an existing lock from another owner. func (wfs *WFS) SetLk(cancel <-chan struct{}, in *fuse.LkIn) fuse.Status { + if wfs.crossMountLocks() { + return wfs.routedSetLk(cancel, in) + } lk := lockRange{ Start: in.Lk.Start, End: in.Lk.End, @@ -39,6 +45,9 @@ func (wfs *WFS) SetLk(cancel <-chan struct{}, in *fuse.LkIn) fuse.Status { // SetLkw sets a POSIX lock (blocking). // Waits until the lock can be acquired or the request is cancelled. func (wfs *WFS) SetLkw(cancel <-chan struct{}, in *fuse.LkIn) fuse.Status { + if wfs.crossMountLocks() { + return wfs.routedSetLkw(cancel, in) + } lk := lockRange{ Start: in.Lk.Start, End: in.Lk.End, diff --git a/weed/mount/weedfs_file_sync.go b/weed/mount/weedfs_file_sync.go index c59f0d4d7..2a589c909 100644 --- a/weed/mount/weedfs_file_sync.go +++ b/weed/mount/weedfs_file_sync.go @@ -60,7 +60,7 @@ func (wfs *WFS) Flush(cancel <-chan struct{}, in *fuse.FlushIn) fuse.Status { // If handle is not found, it might have been already released // This is not an error condition for FLUSH if in.LockOwner != 0 { - wfs.posixLocks.ReleasePosixOwner(in.NodeId, in.LockOwner) + wfs.releasePosixOwner(in.NodeId, in.LockOwner) } return fuse.OK } @@ -69,11 +69,11 @@ func (wfs *WFS) Flush(cancel <-chan struct{}, in *fuse.FlushIn) fuse.Status { // did not hold byte-range locks. Only force the synchronous close path when // this owner actually has POSIX locks to release; otherwise writebackCache // would silently degrade to a blocking flush for ordinary close(). - hasPosixLocks := wfs.posixLocks.HasPosixOwner(in.NodeId, in.LockOwner) + hasPosixLocks := wfs.hasPosixOwner(in.NodeId, in.LockOwner) allowAsync := !hasPosixLocks status := wfs.doFlush(fh, in.Uid, in.Gid, allowAsync) if in.LockOwner != 0 { - wfs.posixLocks.ReleasePosixOwner(in.NodeId, in.LockOwner) + wfs.releasePosixOwner(in.NodeId, in.LockOwner) } return status } diff --git a/weed/mount/weedfs_posix_lock_routed.go b/weed/mount/weedfs_posix_lock_routed.go new file mode 100644 index 000000000..4dcfc2ee1 --- /dev/null +++ b/weed/mount/weedfs_posix_lock_routed.go @@ -0,0 +1,347 @@ +package mount + +import ( + "context" + "encoding/binary" + "encoding/hex" + "math/rand/v2" + "sync" + "syscall" + "time" + + "github.com/seaweedfs/go-fuse/v2/fuse" + "github.com/seaweedfs/seaweedfs/weed/filer/posixlock" + "github.com/seaweedfs/seaweedfs/weed/glog" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/seaweedfs/seaweedfs/weed/util" +) + +// randomPosixSid returns a process-stable, cluster-unique-enough session id for +// this mount, used to namespace lock owners so the same FUSE owner value on a +// different mount never aliases, and to scope lease-based reaping. +func randomPosixSid() uint64 { + return binary.BigEndian.Uint64(util.RandomBytes(8)) +} + +// Routed POSIX locking: when -dlm is set, advisory locks are serialized on the +// inode's owner filer instead of in this mount's local table, so locks are +// honored across mounts. The mount calls its filer and relies on filer-side +// forwarding to reach the owner. Blocking (SetLkw) is client-side polling — there +// is no server-side wait queue. + +const ( + posixLockMinBackoff = 5 * time.Millisecond + posixLockMaxBackoff = 200 * time.Millisecond + posixLockKeyPrefix = "s3.fuse.lock:" + // posixLockReleaseTimeout bounds the background unlock/release RPCs. They run + // off the syscall path (close/flush) and must not be cancelled by an + // interrupt, but a deadline keeps a slow or unreachable filer from blocking + // close indefinitely. + posixLockReleaseTimeout = 5 * time.Second +) + +// posixLockHint records, per inode, which owners this mount has taken fcntl locks +// for. Flush consults it to force a synchronous close and route a release without +// an RPC on every close(). It is a superset hint: an extra sync flush or a no-op +// release RPC is harmless, a missed release is not — so it is added on grant and +// cleared on close (which drops the owner's POSIX locks per POSIX semantics). +type posixLockHint struct { + mu sync.Mutex + m map[uint64]map[uint64]struct{} +} + +func newPosixLockHint() *posixLockHint { + return &posixLockHint{m: make(map[uint64]map[uint64]struct{})} +} + +func (h *posixLockHint) add(inode, owner uint64) { + h.mu.Lock() + defer h.mu.Unlock() + owners := h.m[inode] + if owners == nil { + owners = make(map[uint64]struct{}) + h.m[inode] = owners + } + owners[owner] = struct{}{} +} + +func (h *posixLockHint) has(inode, owner uint64) bool { + h.mu.Lock() + defer h.mu.Unlock() + _, ok := h.m[inode][owner] + return ok +} + +func (h *posixLockHint) drop(inode, owner uint64) { + h.mu.Lock() + defer h.mu.Unlock() + if owners := h.m[inode]; owners != nil { + delete(owners, owner) + if len(owners) == 0 { + delete(h.m, inode) + } + } +} + +// posixLockKeyForInode resolves a FUSE inode to its cluster-stable lock identity: +// the HardLinkId for a hardlinked file (so all names share one owner and table), +// else the path. POSIX locks are inode-scoped, so this must never be the FUSE +// NodeId (mount-local) for a multi-named inode. +func (wfs *WFS) posixLockKeyForInode(inode uint64) (string, bool) { + path, status := wfs.inodeToPath.GetPath(inode) + if status != fuse.OK { + return "", false + } + if entry, st := wfs.maybeLoadEntry(path); st == fuse.OK && entry != nil && len(entry.HardLinkId) > 0 { + return posixLockKeyPrefix + "hl:" + hex.EncodeToString(entry.HardLinkId), true + } + return posixLockKeyPrefix + string(path), true +} + +func posixLockTypeToWire(typ uint32) uint32 { + switch typ { + case syscall.F_RDLCK: + return posixlock.Read + case syscall.F_WRLCK: + return posixlock.Write + default: + return posixlock.Unlock + } +} + +func posixLockTypeFromWire(typ uint32) uint32 { + switch typ { + case posixlock.Read: + return syscall.F_RDLCK + case posixlock.Write: + return syscall.F_WRLCK + default: + return syscall.F_UNLCK + } +} + +func (wfs *WFS) posixRangeFromLkIn(in *fuse.LkIn) posixlock.Range { + return posixlock.Range{ + Start: in.Lk.Start, + End: in.Lk.End, + Type: posixLockTypeToWire(in.Lk.Typ), + Sid: wfs.posixSid, + Owner: in.Owner, + Pid: in.Lk.Pid, + IsFlock: in.LkFlags&fuse.FUSE_LK_FLOCK != 0, + } +} + +// posixLockContext derives a context from the FUSE cancel channel so an +// interrupted lock syscall aborts its in-flight RPC. The caller must call the +// returned func to release the watcher goroutine. +func posixLockContext(cancel <-chan struct{}) (context.Context, context.CancelFunc) { + ctx, cancelFn := context.WithCancel(context.Background()) + if cancel != nil { + go func() { + select { + case <-cancel: + cancelFn() + case <-ctx.Done(): + } + }() + } + return ctx, cancelFn +} + +func (wfs *WFS) callPosixLock(ctx context.Context, key string, op filer_pb.PosixLockOp, lk posixlock.Range) (*filer_pb.PosixLockResponse, error) { + var resp *filer_pb.PosixLockResponse + err := wfs.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + var e error + resp, e = client.PosixLock(ctx, &filer_pb.PosixLockRequest{ + Key: key, + Op: op, + Lock: &filer_pb.PosixLockRange{ + Start: lk.Start, End: lk.End, Type: lk.Type, + Sid: lk.Sid, Owner: lk.Owner, Pid: lk.Pid, IsFlock: lk.IsFlock, + }, + }) + return e + }) + return resp, err +} + +func (wfs *WFS) routedGetLk(cancel <-chan struct{}, in *fuse.LkIn, out *fuse.LkOut) fuse.Status { + key, ok := wfs.posixLockKeyForInode(in.NodeId) + if !ok { + return fuse.EINVAL + } + ctx, done := posixLockContext(cancel) + defer done() + resp, err := wfs.callPosixLock(ctx, key, filer_pb.PosixLockOp_GET_LK, wfs.posixRangeFromLkIn(in)) + if err != nil { + glog.Warningf("routed GetLk %s: %v", key, err) + return fuse.EIO + } + if resp.GetHasConflict() { + c := resp.GetConflict() + out.Lk.Start, out.Lk.End, out.Lk.Pid = c.GetStart(), c.GetEnd(), c.GetPid() + out.Lk.Typ = posixLockTypeFromWire(c.GetType()) + } else { + out.Lk.Typ = syscall.F_UNLCK + } + return fuse.OK +} + +func (wfs *WFS) routedSetLk(cancel <-chan struct{}, in *fuse.LkIn) fuse.Status { + key, ok := wfs.posixLockKeyForInode(in.NodeId) + if !ok { + return fuse.EINVAL + } + lk := wfs.posixRangeFromLkIn(in) + if lk.Type == posixlock.Unlock { + return wfs.routedUnlock(key, lk) + } + ctx, done := posixLockContext(cancel) + defer done() + resp, err := wfs.callPosixLock(ctx, key, filer_pb.PosixLockOp_TRY_LOCK, lk) + if err != nil { + glog.Warningf("routed SetLk %s: %v", key, err) + return fuse.EIO + } + if !resp.GetGranted() { + return fuse.EAGAIN + } + if !lk.IsFlock { + wfs.posixHint.add(in.NodeId, in.Owner) + } + return fuse.OK +} + +func (wfs *WFS) routedSetLkw(cancel <-chan struct{}, in *fuse.LkIn) fuse.Status { + key, ok := wfs.posixLockKeyForInode(in.NodeId) + if !ok { + return fuse.EINVAL + } + lk := wfs.posixRangeFromLkIn(in) + if lk.Type == posixlock.Unlock { + return wfs.routedUnlock(key, lk) + } + ctx, done := posixLockContext(cancel) + defer done() + status := posixPollAcquire(cancel, func() (bool, error) { + resp, err := wfs.callPosixLock(ctx, key, filer_pb.PosixLockOp_TRY_LOCK, lk) + if err != nil { + return false, err + } + return resp.GetGranted(), nil + }) + if status == fuse.OK && !lk.IsFlock { + wfs.posixHint.add(in.NodeId, in.Owner) + } + return status +} + +func (wfs *WFS) routedUnlock(key string, lk posixlock.Range) fuse.Status { + // A release must complete even if the syscall was interrupted; cancelling it + // would leak the lock on the owner filer. Bound it so a stuck filer can't + // hang close() forever. + ctx, cancel := context.WithTimeout(context.Background(), posixLockReleaseTimeout) + defer cancel() + if _, err := wfs.callPosixLock(ctx, key, filer_pb.PosixLockOp_UNLOCK, lk); err != nil { + glog.Warningf("routed unlock %s: %v", key, err) + return fuse.EIO + } + return fuse.OK +} + +// posixPollAcquire retries try with capped, jittered backoff until it is granted, +// the cancel channel fires (EINTR), or try errors (EIO). This is the client-side +// stand-in for a server-side wait queue. +func posixPollAcquire(cancel <-chan struct{}, try func() (bool, error)) fuse.Status { + backoff := posixLockMinBackoff + for { + select { + case <-cancel: + return fuse.EINTR + default: + } + granted, err := try() + if err != nil { + return fuse.EIO + } + if granted { + return fuse.OK + } + timer := time.NewTimer(backoff + time.Duration(rand.Int64N(int64(backoff)))) + select { + case <-cancel: + timer.Stop() + return fuse.EINTR + case <-timer.C: + } + if backoff < posixLockMaxBackoff { + if backoff *= 2; backoff > posixLockMaxBackoff { + backoff = posixLockMaxBackoff + } + } + } +} + +func (wfs *WFS) routedReleasePosixOwner(inode, owner uint64) { + if !wfs.posixHint.has(inode, owner) { + return + } + wfs.posixHint.drop(inode, owner) + key, ok := wfs.posixLockKeyForInode(inode) + if !ok { + return + } + ctx, cancel := context.WithTimeout(context.Background(), posixLockReleaseTimeout) + defer cancel() + if _, err := wfs.callPosixLock(ctx, key, filer_pb.PosixLockOp_RELEASE_POSIX_OWNER, posixlock.Range{Sid: wfs.posixSid, Owner: owner}); err != nil { + glog.Warningf("routed release posix owner %s: %v", key, err) + } +} + +func (wfs *WFS) routedReleaseFlockOwner(inode, owner uint64) { + key, ok := wfs.posixLockKeyForInode(inode) + if !ok { + return + } + ctx, cancel := context.WithTimeout(context.Background(), posixLockReleaseTimeout) + defer cancel() + if _, err := wfs.callPosixLock(ctx, key, filer_pb.PosixLockOp_RELEASE_FLOCK_OWNER, posixlock.Range{Sid: wfs.posixSid, Owner: owner}); err != nil { + glog.Warningf("routed release flock owner %s: %v", key, err) + } +} + +// crossMountLocks reports whether advisory locks are routed to the inode's owner +// filer rather than served from the per-mount local table. It tracks -dlm: lock +// coordination rides the same switch as whole-file write coordination, and is +// off under writeback cache (which implies single-writer, so lockClient is nil). +func (wfs *WFS) crossMountLocks() bool { + return wfs.lockClient != nil +} + +// releasePosixOwner / releaseFlockOwner / hasPosixOwner dispatch to the routed +// authority or the local table based on crossMountLocks, keeping the Flush and +// Release call sites flag-agnostic. + +func (wfs *WFS) releasePosixOwner(inode, owner uint64) { + if wfs.crossMountLocks() { + wfs.routedReleasePosixOwner(inode, owner) + return + } + wfs.posixLocks.ReleasePosixOwner(inode, owner) +} + +func (wfs *WFS) releaseFlockOwner(inode, owner uint64) { + if wfs.crossMountLocks() { + wfs.routedReleaseFlockOwner(inode, owner) + return + } + wfs.posixLocks.ReleaseFlockOwner(inode, owner) +} + +func (wfs *WFS) hasPosixOwner(inode, owner uint64) bool { + if wfs.crossMountLocks() { + return owner != 0 && wfs.posixHint.has(inode, owner) + } + return wfs.posixLocks.HasPosixOwner(inode, owner) +} diff --git a/weed/mount/weedfs_posix_lock_routed_test.go b/weed/mount/weedfs_posix_lock_routed_test.go new file mode 100644 index 000000000..cb4d11a0d --- /dev/null +++ b/weed/mount/weedfs_posix_lock_routed_test.go @@ -0,0 +1,94 @@ +package mount + +import ( + "syscall" + "testing" + "time" + + "github.com/seaweedfs/go-fuse/v2/fuse" + "github.com/seaweedfs/seaweedfs/weed/filer/posixlock" +) + +func TestPosixLockTypeMapping(t *testing.T) { + cases := []struct { + sys uint32 + wire uint32 + }{ + {syscall.F_RDLCK, posixlock.Read}, + {syscall.F_WRLCK, posixlock.Write}, + {syscall.F_UNLCK, posixlock.Unlock}, + } + for _, c := range cases { + if got := posixLockTypeToWire(c.sys); got != c.wire { + t.Errorf("toWire(%d) = %d, want %d", c.sys, got, c.wire) + } + if got := posixLockTypeFromWire(c.wire); got != c.sys { + t.Errorf("fromWire(%d) = %d, want %d", c.wire, got, c.sys) + } + } +} + +func TestPosixPollAcquireGrantedImmediately(t *testing.T) { + calls := 0 + st := posixPollAcquire(nil, func() (bool, error) { calls++; return true, nil }) + if st != fuse.OK || calls != 1 { + t.Fatalf("immediate grant: status=%v calls=%d", st, calls) + } +} + +func TestPosixPollAcquireRetriesThenGrants(t *testing.T) { + calls := 0 + st := posixPollAcquire(nil, func() (bool, error) { + calls++ + return calls >= 3, nil + }) + if st != fuse.OK || calls != 3 { + t.Fatalf("retry then grant: status=%v calls=%d", st, calls) + } +} + +func TestPosixPollAcquireError(t *testing.T) { + if st := posixPollAcquire(nil, func() (bool, error) { return false, syscall.EIO }); st != fuse.EIO { + t.Fatalf("error should map to EIO, got %v", st) + } +} + +func TestPosixPollAcquireCancel(t *testing.T) { + cancel := make(chan struct{}) + close(cancel) + done := make(chan fuse.Status, 1) + go func() { + done <- posixPollAcquire(cancel, func() (bool, error) { return false, nil }) + }() + select { + case st := <-done: + if st != fuse.EINTR { + t.Fatalf("cancel should map to EINTR, got %v", st) + } + case <-time.After(2 * time.Second): + t.Fatal("poll did not return on cancel") + } +} + +func TestPosixLockHint(t *testing.T) { + h := newPosixLockHint() + if h.has(1, 2) { + t.Fatal("empty hint should not report a lock") + } + h.add(1, 2) + h.add(1, 3) + if !h.has(1, 2) || !h.has(1, 3) { + t.Fatal("added owners should be reported") + } + h.drop(1, 2) + if h.has(1, 2) { + t.Fatal("dropped owner should be gone") + } + if !h.has(1, 3) { + t.Fatal("sibling owner should remain") + } + h.drop(1, 3) + if _, ok := h.m[1]; ok { + t.Fatal("inode entry should be removed when its last owner drops") + } +}