mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-09-19 14:34:15 +00:00
filer: in-memory POSIX lock authority (Manager) (#9663)
Concurrent multi-inode authority over the per-inode Set: a Set per opaque inode key (path, or hl:<HardLinkId>) plus a session->keys index so a dead mount's locks reap in O(locks held). Lock state stays in memory like the distributed lock manager's, off the replicated meta-log. TryLock/Unlock/ GetLk/ReleasePosixOwner/ReleaseFlockOwner/ReleaseSession; empty sets and stale index entries are pruned on release.
This commit is contained in:
Chris Lu
@@ -0,0 +1,153 @@
|
||||
package posixlock
|
||||
|
||||
import "sync"
|
||||
|
||||
// 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
|
||||
// transient coordination, so keeping it out of the meta-log avoids churn and
|
||||
// does not pollute what subscribers see (the distributed lock manager holds its
|
||||
// locks the same way). A `Set` per inode key, plus a session index so a dead
|
||||
// mount's locks are reaped in O(locks held) rather than by scanning every inode.
|
||||
//
|
||||
// key is an opaque inode identity supplied by the caller — the file's path, or
|
||||
// "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
|
||||
}
|
||||
|
||||
func NewManager() *Manager {
|
||||
return &Manager{
|
||||
byKey: make(map[string]*Set),
|
||||
bySid: make(map[uint64]map[string]bool),
|
||||
}
|
||||
}
|
||||
|
||||
// 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) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
s, ok := m.byKey[key]
|
||||
if !ok {
|
||||
s = &Set{}
|
||||
}
|
||||
if c, granted := s.Acquire(lk); !granted {
|
||||
return c, false
|
||||
}
|
||||
if !ok {
|
||||
m.byKey[key] = s
|
||||
}
|
||||
m.index(lk.Sid, key)
|
||||
return Range{}, true
|
||||
}
|
||||
|
||||
// Unlock releases lk's owner's locks within its namespace over lk's range.
|
||||
func (m *Manager) Unlock(key string, lk Range) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
s := m.byKey[key]
|
||||
if s == nil {
|
||||
return
|
||||
}
|
||||
s.Release(lk)
|
||||
m.afterRelease(key, s, lk.Sid)
|
||||
}
|
||||
|
||||
// GetLk reports the lock that would block proposed on key, if any.
|
||||
func (m *Manager) GetLk(key string, proposed Range) (Range, bool) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
s := m.byKey[key]
|
||||
if s == nil {
|
||||
return Range{}, false
|
||||
}
|
||||
return s.Conflict(proposed)
|
||||
}
|
||||
|
||||
// ReleasePosixOwner drops (sid, owner)'s fcntl locks on key — the flush-time path.
|
||||
func (m *Manager) ReleasePosixOwner(key string, sid, owner uint64) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
s := m.byKey[key]
|
||||
if s == nil {
|
||||
return
|
||||
}
|
||||
s.ReleasePosixOwner(sid, owner)
|
||||
m.afterRelease(key, s, sid)
|
||||
}
|
||||
|
||||
// ReleaseFlockOwner drops (sid, owner)'s flock locks on key — the release-time path.
|
||||
func (m *Manager) ReleaseFlockOwner(key string, sid, owner uint64) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
s := m.byKey[key]
|
||||
if s == nil {
|
||||
return
|
||||
}
|
||||
s.ReleaseFlockOwner(sid, owner)
|
||||
m.afterRelease(key, s, sid)
|
||||
}
|
||||
|
||||
// ReleaseSession drops every lock held by a session across all inodes it touched,
|
||||
// reaping a mount whose lease expired. O(locks held by the session).
|
||||
func (m *Manager) ReleaseSession(sid uint64) {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
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)
|
||||
}
|
||||
|
||||
// afterRelease prunes the session index when sid no longer holds any lock on key,
|
||||
// and drops the set when it empties. Only sid's presence can have changed, since
|
||||
// a release only removes sid's locks.
|
||||
func (m *Manager) afterRelease(key string, s *Set, sid uint64) {
|
||||
if s.Empty() {
|
||||
delete(m.byKey, key)
|
||||
m.deindex(sid, key)
|
||||
return
|
||||
}
|
||||
if !setHasSession(s, sid) {
|
||||
m.deindex(sid, key)
|
||||
}
|
||||
}
|
||||
|
||||
func (m *Manager) index(sid uint64, key string) {
|
||||
keys := m.bySid[sid]
|
||||
if keys == nil {
|
||||
keys = make(map[string]bool)
|
||||
m.bySid[sid] = keys
|
||||
}
|
||||
keys[key] = true
|
||||
}
|
||||
|
||||
func (m *Manager) deindex(sid uint64, key string) {
|
||||
keys := m.bySid[sid]
|
||||
if keys == nil {
|
||||
return
|
||||
}
|
||||
delete(keys, key)
|
||||
if len(keys) == 0 {
|
||||
delete(m.bySid, sid)
|
||||
}
|
||||
}
|
||||
|
||||
func setHasSession(s *Set, sid uint64) bool {
|
||||
for _, l := range s.Locks() {
|
||||
if l.Sid == sid {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
@@ -0,0 +1,162 @@
|
||||
package posixlock
|
||||
|
||||
import (
|
||||
"math"
|
||||
"runtime"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestManagerGrantAndConflict(t *testing.T) {
|
||||
m := NewManager()
|
||||
if _, granted := m.TryLock("a", Range{Start: 0, End: 99, Type: Write, Sid: 1, Owner: 1}); !granted {
|
||||
t.Fatal("first lock should be granted")
|
||||
}
|
||||
if c, granted := m.TryLock("a", Range{Start: 50, End: 149, Type: Write, Sid: 2, Owner: 1}); granted {
|
||||
t.Fatalf("overlapping lock from another session should conflict, got grant; conflict=%+v", c)
|
||||
}
|
||||
// A different key is independent.
|
||||
if _, granted := m.TryLock("b", Range{Start: 0, End: 99, Type: Write, Sid: 2, Owner: 1}); !granted {
|
||||
t.Fatal("lock on a different key should be granted")
|
||||
}
|
||||
}
|
||||
|
||||
func TestManagerUnlockCleansEmptyKeyAndIndex(t *testing.T) {
|
||||
m := NewManager()
|
||||
lk := Range{Start: 0, End: 99, Type: Write, Sid: 1, Owner: 1}
|
||||
m.TryLock("a", lk)
|
||||
|
||||
if !m.bySid[1]["a"] {
|
||||
t.Fatal("session index should record the held key")
|
||||
}
|
||||
m.Unlock("a", Range{Start: 0, End: 99, Type: Unlock, Sid: 1, Owner: 1})
|
||||
|
||||
if _, ok := m.byKey["a"]; ok {
|
||||
t.Fatal("empty set should be dropped from byKey")
|
||||
}
|
||||
if _, ok := m.bySid[1]; ok {
|
||||
t.Fatal("session index should be pruned when it holds nothing")
|
||||
}
|
||||
}
|
||||
|
||||
func TestManagerPartialUnlockKeepsIndex(t *testing.T) {
|
||||
m := NewManager()
|
||||
m.TryLock("a", Range{Start: 0, End: 49, Type: Write, Sid: 1, Owner: 1})
|
||||
m.TryLock("a", Range{Start: 100, End: 149, Type: Write, Sid: 1, Owner: 1})
|
||||
// Release one of the two ranges; the session still holds the other.
|
||||
m.Unlock("a", Range{Start: 0, End: 49, Type: Unlock, Sid: 1, Owner: 1})
|
||||
|
||||
if !m.bySid[1]["a"] {
|
||||
t.Fatal("session still holds a lock on the key; index must remain")
|
||||
}
|
||||
if _, ok := m.byKey["a"]; !ok {
|
||||
t.Fatal("key should remain while a lock is held")
|
||||
}
|
||||
}
|
||||
|
||||
func TestManagerGetLk(t *testing.T) {
|
||||
m := NewManager()
|
||||
m.TryLock("a", Range{Start: 10, End: 50, Type: Write, Sid: 1, Owner: 1, Pid: 7})
|
||||
c, found := m.GetLk("a", Range{Start: 30, End: 70, Type: Read, Sid: 2, Owner: 1})
|
||||
if !found || c.Pid != 7 {
|
||||
t.Fatalf("expected conflict from pid 7, got %+v found=%v", c, found)
|
||||
}
|
||||
if _, found := m.GetLk("missing", Range{Start: 0, End: 1, Type: Write, Sid: 9, Owner: 9}); found {
|
||||
t.Fatal("missing key should report no conflict")
|
||||
}
|
||||
}
|
||||
|
||||
func TestManagerReleasePosixOwnerKeepsFlockAndIndex(t *testing.T) {
|
||||
m := NewManager()
|
||||
m.TryLock("a", Range{Start: 0, End: 99, Type: Write, Sid: 1, Owner: 1})
|
||||
m.TryLock("a", Range{Start: 0, End: math.MaxUint64, Type: Write, Sid: 1, Owner: 1, IsFlock: true})
|
||||
|
||||
m.ReleasePosixOwner("a", 1, 1)
|
||||
|
||||
// flock lock for the same session remains, so the index must remain too.
|
||||
if !m.bySid[1]["a"] {
|
||||
t.Fatal("session still holds the flock lock; index must remain")
|
||||
}
|
||||
if _, found := m.GetLk("a", Range{Start: 0, End: 10, Type: Write, Sid: 2, Owner: 2, IsFlock: true}); !found {
|
||||
t.Fatal("flock lock should survive ReleasePosixOwner")
|
||||
}
|
||||
if _, found := m.GetLk("a", Range{Start: 0, End: 10, Type: Write, Sid: 2, Owner: 2}); found {
|
||||
t.Fatal("fcntl lock should be gone after ReleasePosixOwner")
|
||||
}
|
||||
}
|
||||
|
||||
func TestManagerReleaseSessionReapsAcrossKeys(t *testing.T) {
|
||||
m := NewManager()
|
||||
m.TryLock("a", Range{Start: 0, End: 99, Type: Write, Sid: 1, Owner: 1})
|
||||
m.TryLock("b", Range{Start: 0, End: 99, Type: Write, Sid: 1, Owner: 2})
|
||||
m.TryLock("b", Range{Start: 200, End: 299, Type: Write, Sid: 2, Owner: 1})
|
||||
|
||||
m.ReleaseSession(1)
|
||||
|
||||
if _, ok := m.bySid[1]; ok {
|
||||
t.Fatal("reaped session should be gone from the index")
|
||||
}
|
||||
if _, ok := m.byKey["a"]; ok {
|
||||
t.Fatal("key a held only session 1's lock and should be dropped")
|
||||
}
|
||||
// Session 2's lock on b survives.
|
||||
if _, found := m.GetLk("b", Range{Start: 200, End: 299, Type: Write, Sid: 9, Owner: 9}); !found {
|
||||
t.Fatal("session 2's lock on b should remain after reaping session 1")
|
||||
}
|
||||
if !m.bySid[2]["b"] {
|
||||
t.Fatal("session 2 index entry should remain")
|
||||
}
|
||||
}
|
||||
|
||||
// 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) {
|
||||
m := NewManager()
|
||||
const (
|
||||
key = "inode"
|
||||
workers = 16
|
||||
iters = 400
|
||||
)
|
||||
var (
|
||||
wg sync.WaitGroup
|
||||
holder atomic.Int64
|
||||
overlap atomic.Int32
|
||||
)
|
||||
for w := 0; w < workers; w++ {
|
||||
wg.Add(1)
|
||||
go func(id int) {
|
||||
defer wg.Done()
|
||||
lk := Range{Start: 0, End: math.MaxUint64, Type: Write, Sid: uint64(id + 1), Owner: 1, IsFlock: true}
|
||||
unlock := lk
|
||||
unlock.Type = Unlock
|
||||
token := int64(id + 1)
|
||||
for i := 0; i < iters; i++ {
|
||||
for {
|
||||
if _, granted := m.TryLock(key, lk); granted {
|
||||
break
|
||||
}
|
||||
runtime.Gosched()
|
||||
}
|
||||
if prev := holder.Swap(token); prev != 0 {
|
||||
overlap.Add(1)
|
||||
}
|
||||
runtime.Gosched()
|
||||
if !holder.CompareAndSwap(token, 0) {
|
||||
overlap.Add(1)
|
||||
}
|
||||
m.Unlock(key, unlock)
|
||||
}
|
||||
}(w)
|
||||
}
|
||||
wg.Wait()
|
||||
if n := overlap.Load(); n != 0 {
|
||||
t.Fatalf("mutual exclusion violated %d times", n)
|
||||
}
|
||||
if len(m.byKey) != 0 {
|
||||
t.Fatalf("all locks released; byKey should be empty, got %d", len(m.byKey))
|
||||
}
|
||||
if len(m.bySid) != 0 {
|
||||
t.Fatalf("all locks released; bySid should be empty, got %d", len(m.bySid))
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user