mount: route POSIX advisory locks to the owner filer under -dlm (#9669)

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

@@ -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")

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),

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))
 }

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,

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
 }

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)
+}

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")
+	}
+}