From c97b69f8a4d6cbddf9df58389bd21582add38067 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Mon, 25 May 2026 00:00:59 -0700 Subject: [PATCH] filer: session lease + reaping for POSIX locks (#9666) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * filer: session lease + reaping for POSIX locks A mount renews its session lease by keepalive (new KEEP_ALIVE op); the owner filer records last-seen per session and a background sweeper reaps the locks of leased sessions that stop renewing — a dead or partitioned mount. Only sessions that have renewed are leased, so this is inert until mounts run with -posixLock. * mount: route POSIX advisory locks to the owner filer (-posixLock) (#9665) mount: route POSIX advisory locks to the owner filer under -dlm 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. The mount calls its filer and relies on filer-side forwarding to reach the owner. 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. * mount,filer: bound posix-lock release RPCs and stop the reaper on shutdown The unlock/release RPCs run off the syscall path (close/flush) and used context.Background() with no deadline, so a slow or unreachable filer could hang close() indefinitely; bound them to 5s (they still aren't cancelled by an interrupt). The lease-reaping sweeper now selects on a stop channel that FilerServer.Shutdown closes, instead of looping for the process lifetime. --- other/java/client/src/main/proto/filer.proto | 1 + weed/filer/posixlock/manager.go | 55 +++++++++++++++++-- weed/filer/posixlock/manager_test.go | 32 +++++++++++ weed/pb/filer.proto | 1 + weed/pb/filer_pb/filer.pb.go | 9 ++- weed/server/filer_grpc_server_posix_lock.go | 32 +++++++++++ .../filer_grpc_server_posix_lock_test.go | 16 ++++++ weed/server/filer_server.go | 6 ++ 8 files changed, 144 insertions(+), 8 deletions(-) diff --git a/other/java/client/src/main/proto/filer.proto b/other/java/client/src/main/proto/filer.proto index 8e64b3faa..b2ae87131 100644 --- a/other/java/client/src/main/proto/filer.proto +++ b/other/java/client/src/main/proto/filer.proto @@ -396,6 +396,7 @@ enum PosixLockOp { GET_LK = 2; // report a conflicting lock, if any RELEASE_POSIX_OWNER = 3; // drop the owner's fcntl locks (flush-time) RELEASE_FLOCK_OWNER = 4; // drop the owner's flock locks (release-time) + KEEP_ALIVE = 5; // renew the session's lease on this owner (lock.sid) } message PosixLockResponse { diff --git a/weed/filer/posixlock/manager.go b/weed/filer/posixlock/manager.go index 75bdb111b..297cf2f96 100644 --- a/weed/filer/posixlock/manager.go +++ b/weed/filer/posixlock/manager.go @@ -1,6 +1,9 @@ package posixlock -import "sync" +import ( + "sync" + "time" +) // Manager is the owner filer's in-memory authority for POSIX advisory locks // across inodes. Lock state lives here, not in replicated metadata: it is @@ -13,18 +16,58 @@ import "sync" // "hl:"+hex(HardLinkId) for a hardlinked inode — so all names of one inode share // a Set. The Manager is safe for concurrent use. type Manager struct { - mu sync.Mutex - byKey map[string]*Set // inode key -> held locks - bySid map[uint64]map[string]bool // session -> keys it currently holds locks on + mu sync.Mutex + byKey map[string]*Set // inode key -> held locks + bySid map[uint64]map[string]bool // session -> keys it currently holds locks on + lastSeen map[uint64]time.Time // session -> last keepalive; only renewing sessions are leased } func NewManager() *Manager { return &Manager{ - byKey: make(map[string]*Set), - bySid: make(map[uint64]map[string]bool), + byKey: make(map[string]*Set), + bySid: make(map[uint64]map[string]bool), + lastSeen: make(map[uint64]time.Time), } } +// Renew records a keepalive from a session, placing it under lease management. +// Only sessions that have renewed are subject to ReapExpired, so a session that +// never sends keepalives (e.g. before the mount keepalive exists) is never reaped. +func (m *Manager) Renew(sid uint64) { + m.mu.Lock() + defer m.mu.Unlock() + m.lastSeen[sid] = time.Now() +} + +// ReapExpired releases the locks of every leased session whose last keepalive is +// older than ttl — a dead or partitioned mount. Sessions that never renewed are +// left untouched. Returns the reaped session ids. +func (m *Manager) ReapExpired(ttl time.Duration) []uint64 { + m.mu.Lock() + defer m.mu.Unlock() + cutoff := time.Now().Add(-ttl) + var reaped []uint64 + for sid, seen := range m.lastSeen { + if seen.After(cutoff) { + continue + } + for key := range m.bySid[sid] { + s := m.byKey[key] + if s == nil { + continue + } + s.ReleaseSession(sid) + if s.Empty() { + delete(m.byKey, key) + } + } + delete(m.bySid, sid) + delete(m.lastSeen, sid) + reaped = append(reaped, sid) + } + return reaped +} + // TryLock grants lk on key, or returns the conflicting lock and false. The set // is created on first use and dropped again when it empties. func (m *Manager) TryLock(key string, lk Range) (Range, bool) { diff --git a/weed/filer/posixlock/manager_test.go b/weed/filer/posixlock/manager_test.go index defb928f2..0c0bc8c5a 100644 --- a/weed/filer/posixlock/manager_test.go +++ b/weed/filer/posixlock/manager_test.go @@ -6,6 +6,7 @@ import ( "sync" "sync/atomic" "testing" + "time" ) func TestManagerGrantAndConflict(t *testing.T) { @@ -109,6 +110,37 @@ func TestManagerReleaseSessionReapsAcrossKeys(t *testing.T) { } } +func TestManagerReapsOnlyStaleLeasedSessions(t *testing.T) { + m := NewManager() + // Session 1: holds a lock, leased but stale (renewed long ago). + m.TryLock("a", Range{Start: 0, End: 99, Type: Write, Sid: 1, Owner: 1}) + m.Renew(1) + m.lastSeen[1] = time.Now().Add(-time.Hour) + // Session 2: holds a lock, leased and fresh. + m.TryLock("b", Range{Start: 0, End: 99, Type: Write, Sid: 2, Owner: 1}) + m.Renew(2) + // Session 3: holds a lock but never renewed (no lease) — must not be reaped. + m.TryLock("c", Range{Start: 0, End: 99, Type: Write, Sid: 3, Owner: 1}) + + reaped := m.ReapExpired(30 * time.Second) + + if len(reaped) != 1 || reaped[0] != 1 { + t.Fatalf("only the stale leased session should be reaped, got %v", reaped) + } + if _, ok := m.byKey["a"]; ok { + t.Fatal("stale session's lock should be gone") + } + if _, ok := m.byKey["b"]; !ok { + t.Fatal("fresh session's lock must remain") + } + if _, ok := m.byKey["c"]; !ok { + t.Fatal("never-renewed session must not be reaped") + } + if _, ok := m.lastSeen[1]; ok { + t.Fatal("reaped session's lease entry should be cleared") + } +} + // Mutual exclusion under concurrent whole-file flock churn through the Manager: // at most one owner may believe it holds the exclusive lock at any instant. func TestManagerConcurrentFlockMutualExclusion(t *testing.T) { diff --git a/weed/pb/filer.proto b/weed/pb/filer.proto index 8e64b3faa..b2ae87131 100644 --- a/weed/pb/filer.proto +++ b/weed/pb/filer.proto @@ -396,6 +396,7 @@ enum PosixLockOp { GET_LK = 2; // report a conflicting lock, if any RELEASE_POSIX_OWNER = 3; // drop the owner's fcntl locks (flush-time) RELEASE_FLOCK_OWNER = 4; // drop the owner's flock locks (release-time) + KEEP_ALIVE = 5; // renew the session's lease on this owner (lock.sid) } message PosixLockResponse { diff --git a/weed/pb/filer_pb/filer.pb.go b/weed/pb/filer_pb/filer.pb.go index 02a34fc9e..cde52c5a9 100644 --- a/weed/pb/filer_pb/filer.pb.go +++ b/weed/pb/filer_pb/filer.pb.go @@ -144,6 +144,7 @@ const ( PosixLockOp_GET_LK PosixLockOp = 2 // report a conflicting lock, if any PosixLockOp_RELEASE_POSIX_OWNER PosixLockOp = 3 // drop the owner's fcntl locks (flush-time) PosixLockOp_RELEASE_FLOCK_OWNER PosixLockOp = 4 // drop the owner's flock locks (release-time) + PosixLockOp_KEEP_ALIVE PosixLockOp = 5 // renew the session's lease on this owner (lock.sid) ) // Enum value maps for PosixLockOp. @@ -154,6 +155,7 @@ var ( 2: "GET_LK", 3: "RELEASE_POSIX_OWNER", 4: "RELEASE_FLOCK_OWNER", + 5: "KEEP_ALIVE", } PosixLockOp_value = map[string]int32{ "TRY_LOCK": 0, @@ -161,6 +163,7 @@ var ( "GET_LK": 2, "RELEASE_POSIX_OWNER": 3, "RELEASE_FLOCK_OWNER": 4, + "KEEP_ALIVE": 5, } ) @@ -7153,7 +7156,7 @@ const file_filer_proto_rawDesc = "" + "\x15EXISTING_IS_DIRECTORY\x10\x03\x12\x14\n" + "\x10EXISTING_IS_FILE\x10\x04\x12\x18\n" + "\x14ENTRY_ALREADY_EXISTS\x10\x05\x12\x17\n" + - "\x13PRECONDITION_FAILED\x10\x06*e\n" + + "\x13PRECONDITION_FAILED\x10\x06*u\n" + "\vPosixLockOp\x12\f\n" + "\bTRY_LOCK\x10\x00\x12\n" + "\n" + @@ -7161,7 +7164,9 @@ const file_filer_proto_rawDesc = "" + "\n" + "\x06GET_LK\x10\x02\x12\x17\n" + "\x13RELEASE_POSIX_OWNER\x10\x03\x12\x17\n" + - "\x13RELEASE_FLOCK_OWNER\x10\x042\xbc\x16\n" + + "\x13RELEASE_FLOCK_OWNER\x10\x04\x12\x0e\n" + + "\n" + + "KEEP_ALIVE\x10\x052\xbc\x16\n" + "\fSeaweedFiler\x12g\n" + "\x14LookupDirectoryEntry\x12%.filer_pb.LookupDirectoryEntryRequest\x1a&.filer_pb.LookupDirectoryEntryResponse\"\x00\x12N\n" + "\vListEntries\x12\x1c.filer_pb.ListEntriesRequest\x1a\x1d.filer_pb.ListEntriesResponse\"\x000\x01\x12L\n" + diff --git a/weed/server/filer_grpc_server_posix_lock.go b/weed/server/filer_grpc_server_posix_lock.go index 29d1f4f31..4abe102fc 100644 --- a/weed/server/filer_grpc_server_posix_lock.go +++ b/weed/server/filer_grpc_server_posix_lock.go @@ -3,6 +3,7 @@ package weed_server import ( "context" "fmt" + "time" "github.com/seaweedfs/seaweedfs/weed/filer/posixlock" "github.com/seaweedfs/seaweedfs/weed/glog" @@ -10,6 +11,35 @@ import ( "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" ) +const ( + // posixLockSessionTTL is how long a mount's lease survives without a + // keepalive before its locks are reaped; posixLockSweepInterval is how often + // each filer checks. The mount renews well within the TTL. + posixLockSessionTTL = 15 * time.Second + posixLockSweepInterval = 5 * time.Second +) + +// startPosixLockSweeper periodically reaps the locks of leased sessions (mounts) +// that stopped sending keepalives. Sessions that never renew are never reaped, so +// this is inert until mounts run with -posixLock. +func (fs *FilerServer) startPosixLockSweeper() { + fs.posixLockSweeperStop = make(chan struct{}) + go func() { + ticker := time.NewTicker(posixLockSweepInterval) + defer ticker.Stop() + for { + select { + case <-fs.posixLockSweeperStop: + return + case <-ticker.C: + if reaped := fs.posixLocks.ReapExpired(posixLockSessionTTL); len(reaped) > 0 { + glog.V(2).Infof("posix lock: reaped %d expired session(s): %v", len(reaped), reaped) + } + } + } + }() +} + // PosixLock applies one advisory lock operation against the in-memory lock table // of the inode's owner filer. The owner is resolved from req.Key on the same // ring the gateway and DLM use; a non-owner filer forwards the request one hop @@ -76,6 +106,8 @@ func (fs *FilerServer) PosixLock(ctx context.Context, req *filer_pb.PosixLockReq fs.posixLocks.ReleasePosixOwner(req.Key, lk.Sid, lk.Owner) case filer_pb.PosixLockOp_RELEASE_FLOCK_OWNER: fs.posixLocks.ReleaseFlockOwner(req.Key, lk.Sid, lk.Owner) + case filer_pb.PosixLockOp_KEEP_ALIVE: + fs.posixLocks.Renew(lk.Sid) default: return &filer_pb.PosixLockResponse{}, fmt.Errorf("unknown posix lock op %v", req.Op) } diff --git a/weed/server/filer_grpc_server_posix_lock_test.go b/weed/server/filer_grpc_server_posix_lock_test.go index 98be261be..96423971b 100644 --- a/weed/server/filer_grpc_server_posix_lock_test.go +++ b/weed/server/filer_grpc_server_posix_lock_test.go @@ -88,6 +88,22 @@ func TestPosixLockReleasePosixOwnerKeepsFlock(t *testing.T) { } } +func TestPosixLockKeepAlive(t *testing.T) { + fs := newPosixTestServer() + resp, err := fs.PosixLock(context.Background(), &filer_pb.PosixLockRequest{ + Key: "s3.fuse.lock:/x", Op: filer_pb.PosixLockOp_KEEP_ALIVE, + Lock: pbLock(0, 0, posixlock.Unlock, 7, 0, 0, false), + }) + if err != nil || resp == nil { + t.Fatalf("keep_alive should succeed: err=%v", err) + } + // A renewed session that goes stale is reaped; a never-renewed one is not. + fs.posixLocks.TryLock("s3.fuse.lock:/x", posixlock.Range{Start: 0, End: 9, Type: posixlock.Write, Sid: 7, Owner: 1}) + if reaped := fs.posixLocks.ReapExpired(0); len(reaped) != 1 || reaped[0] != 7 { + t.Fatalf("renewed session 7 should be reapable at ttl=0, got %v", reaped) + } +} + // A request whose key is owned by another filer is forwarded to it; the owner // applies it and the sender does not. The owner's ring points back at the bogus // sender, so without is_moved on the forwarded hop it would re-forward and fail. diff --git a/weed/server/filer_server.go b/weed/server/filer_server.go index f6d73636f..a6c2e1508 100644 --- a/weed/server/filer_server.go +++ b/weed/server/filer_server.go @@ -139,6 +139,8 @@ type FilerServer struct { // here rather than in replicated metadata: it is transient coordination, so // keeping it off the meta-log avoids churn. posixLocks *posixlock.Manager + // posixLockSweeperStop stops the lease-reaping sweeper goroutine on Shutdown. + posixLockSweeperStop chan struct{} } func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption) (fs *FilerServer, err error) { @@ -180,6 +182,7 @@ func NewFilerServer(defaultMux, readonlyMux *http.ServeMux, option *FilerOption) entryLockTable: util.NewLockTable[util.FullPath](), posixLocks: posixlock.NewManager(), } + fs.startPosixLockSweeper() fs.mountPeerRegistry = filer.NewMountPeerRegistry() go fs.runMountPeerRegistrySweeper() fs.listenersCond = sync.NewCond(&fs.listenersLock) @@ -322,6 +325,9 @@ func (fs *FilerServer) checkWithMaster() { // This prevents data corruption when the process receives SIGTERM during active uploads. func (fs *FilerServer) Shutdown() { glog.V(0).Infof("Shutting down filer") + if fs.posixLockSweeperStop != nil { + close(fs.posixLockSweeperStop) + } fs.filer.Shutdown() }