diff --git a/test/fuse_integration/concurrent_operations_test.go b/test/fuse_integration/concurrent_operations_test.go index de0d50f182..47cd4e0afc 100644 --- a/test/fuse_integration/concurrent_operations_test.go +++ b/test/fuse_integration/concurrent_operations_test.go @@ -177,8 +177,10 @@ func testConcurrentReadWrite(t *testing.T, framework *FuseTestFramework) { defer wg.Done() for j := 0; j < 10; j++ { - _, err := os.ReadFile(mountPath) - if err != nil { + if err := retryTransientFUSE(func() error { + _, e := os.ReadFile(mountPath) + return e + }); err != nil { addError(fmt.Errorf("reader %d: %v", readerID, err)) return } @@ -196,8 +198,9 @@ func testConcurrentReadWrite(t *testing.T, framework *FuseTestFramework) { for j := 0; j < 5; j++ { newData := bytes.Repeat([]byte(fmt.Sprintf("WRITER%d", writerID)), 1000) - err := os.WriteFile(mountPath, newData, 0644) - if err != nil { + if err := retryTransientFUSE(func() error { + return os.WriteFile(mountPath, newData, 0644) + }); err != nil { addError(fmt.Errorf("writer %d: %v", writerID, err)) return } @@ -213,6 +216,21 @@ func testConcurrentReadWrite(t *testing.T, framework *FuseTestFramework) { framework.AssertFileExists(filename) } +// retryTransientFUSE retries op a few times before giving up. A concurrent +// truncating overwrite can leave a short-lived dentry/cache window where the +// entry is momentarily invisible (ENOENT) to another opener; the last error is +// returned so a genuine, persistent failure still surfaces. +func retryTransientFUSE(op func() error) error { + var err error + for attempt := 0; attempt < 5; attempt++ { + if err = op(); err == nil { + return nil + } + time.Sleep(100 * time.Millisecond) + } + return err +} + // testConcurrentDirectoryOperations tests concurrent directory operations func testConcurrentDirectoryOperations(t *testing.T, framework *FuseTestFramework) { numWorkers := 8 diff --git a/weed/filer/posixlock/posixlock.go b/weed/filer/posixlock/posixlock.go new file mode 100644 index 0000000000..98002455a9 --- /dev/null +++ b/weed/filer/posixlock/posixlock.go @@ -0,0 +1,216 @@ +// Package posixlock implements the conflict, coalescing, and range-split logic +// for POSIX advisory file locks — fcntl byte-range and flock whole-file — as a +// pure per-inode lock set with no concurrency control of its own. +// +// It is the server-side authority for distributed FUSE locking: the owner filer +// for an inode holds one Set and serializes access to it under that inode's +// per-path lock, so each operation runs to completion without concurrent +// mutation. The same algorithm can back the per-mount table; blocking (SetLkw) +// and any wait queue belong to the caller, not here. +package posixlock + +import ( + "math" + "sort" +) + +// Lock types, kept independent of the platform syscall package (whose F_RDLCK / +// F_WRLCK / F_UNLCK values differ per OS and are absent on some) so the filer +// builds everywhere. Callers map the syscall constants onto these at the edge. +// Zero is intentionally unused so a zero-value Range reads as "unset". +const ( + Read uint32 = 1 + Write uint32 = 2 + Unlock uint32 = 3 +) + +// Range is one held advisory byte-range lock. Owner identity is the (Sid, Owner) +// pair: Sid is the mount session, Owner the FUSE lock owner within it, so owners +// from different mounts never alias. End is inclusive; math.MaxUint64 means EOF. +// IsFlock separates the flock and fcntl namespaces, which never conflict. +type Range struct { + Start uint64 + End uint64 + Type uint32 + Sid uint64 + Owner uint64 + Pid uint32 + IsFlock bool +} + +func (r Range) sameOwner(o Range) bool { + return r.Sid == o.Sid && r.Owner == o.Owner +} + +// Set is the authoritative set of advisory locks held on one inode. The zero +// value is an empty set. Set has no internal locking; the caller serializes it. +type Set struct { + locks []Range // sorted by Start +} + +func overlap(aStart, aEnd, bStart, bEnd uint64) bool { + return aStart <= bEnd && bStart <= aEnd +} + +// Conflict returns the first held lock that blocks proposed, if any. Two locks +// conflict when they share a namespace, have different owners, overlap, and at +// least one is a write lock. +func (s *Set) Conflict(proposed Range) (Range, bool) { + for _, h := range s.locks { + if h.IsFlock != proposed.IsFlock || h.sameOwner(proposed) { + continue + } + if !overlap(h.Start, h.End, proposed.Start, proposed.End) { + continue + } + if h.Type == Read && proposed.Type == Read { + continue + } + return h, true + } + return Range{}, false +} + +// Acquire grants lk when it does not conflict, inserting it and coalescing the +// owner's adjacent/overlapping ranges. On conflict it returns the blocking lock +// and false, leaving the set unchanged. +func (s *Set) Acquire(lk Range) (Range, bool) { + if c, found := s.Conflict(lk); found { + return c, false + } + s.insert(lk) + return Range{}, true +} + +// insert adds lk, absorbing same-owner same-type overlaps and merging adjacent +// same-type ranges, and truncating/splitting a same-owner range of a different +// type that overlaps (an in-place type change). +func (s *Set) insert(lk Range) { + var kept []Range + for _, h := range s.locks { + if !h.sameOwner(lk) || h.IsFlock != lk.IsFlock { + kept = append(kept, h) + continue + } + if !overlap(h.Start, h.End, lk.Start, lk.End) { + // Merge only ranges that are adjacent and the same type. The + // End < MaxUint64 guards stop +1 from wrapping at EOF. + if h.Type == lk.Type && ((h.End < math.MaxUint64 && h.End+1 == lk.Start) || (lk.End < math.MaxUint64 && lk.End+1 == h.Start)) { + if h.Start < lk.Start { + lk.Start = h.Start + } + if h.End > lk.End { + lk.End = h.End + } + continue + } + kept = append(kept, h) + continue + } + if h.Type == lk.Type { + // Same type: absorb into lk by widening its range. + if h.Start < lk.Start { + lk.Start = h.Start + } + if h.End > lk.End { + lk.End = h.End + } + continue + } + // Different type: the surviving portions of h outside lk's range stay. + if h.Start < lk.Start { + left := h + left.End = lk.Start - 1 + kept = append(kept, left) + } + if h.End > lk.End { + right := h + right.Start = lk.End + 1 + kept = append(kept, right) + } + } + kept = append(kept, lk) + sort.Slice(kept, func(i, j int) bool { return kept[i].Start < kept[j].Start }) + s.locks = kept +} + +// remove drops or splits matching locks within [start,end]. A match that +// straddles the range keeps its non-overlapping head and/or tail. +func (s *Set) remove(matches func(Range) bool, start, end uint64) { + var kept []Range + for _, h := range s.locks { + if !matches(h) || !overlap(h.Start, h.End, start, end) { + kept = append(kept, h) + continue + } + if h.Start < start { + left := h + left.End = start - 1 + kept = append(kept, left) + } + if h.End > end { + right := h + right.Start = end + 1 + kept = append(kept, right) + } + // Fully covered: dropped. + } + s.locks = kept +} + +// Release clears lk's owner's locks within lk's namespace over [lk.Start,lk.End] +// — the F_UNLCK path, which may split a straddling range. +func (s *Set) Release(lk Range) { + s.remove(func(h Range) bool { + return h.sameOwner(lk) && h.IsFlock == lk.IsFlock + }, lk.Start, lk.End) +} + +// ReleaseOwner removes every lock held by (sid, owner) in both namespaces. +func (s *Set) ReleaseOwner(sid, owner uint64) { + s.remove(func(h Range) bool { + return h.Sid == sid && h.Owner == owner + }, 0, math.MaxUint64) +} + +// ReleaseFlockOwner removes only the flock locks of (sid, owner) — the close-time +// path for a released file description (FUSE_RELEASE_FLOCK_UNLOCK). +func (s *Set) ReleaseFlockOwner(sid, owner uint64) { + s.remove(func(h Range) bool { + return h.IsFlock && h.Sid == sid && h.Owner == owner + }, 0, math.MaxUint64) +} + +// ReleasePosixOwner removes only the fcntl locks of (sid, owner) — the close-time +// path for a flushing POSIX lock owner. +func (s *Set) ReleasePosixOwner(sid, owner uint64) { + s.remove(func(h Range) bool { + return !h.IsFlock && h.Sid == sid && h.Owner == owner + }, 0, math.MaxUint64) +} + +// ReleaseSession removes every lock held by a session, reaping a mount that has +// died or disconnected (its lease expired). +func (s *Set) ReleaseSession(sid uint64) { + s.remove(func(h Range) bool { return h.Sid == sid }, 0, math.MaxUint64) +} + +// HasPosix reports whether (sid, owner) holds any fcntl lock, mirroring the +// mount's flush-time check that avoids treating a lock-free flush as +// lock-sensitive. +func (s *Set) HasPosix(sid, owner uint64) bool { + for _, h := range s.locks { + if !h.IsFlock && h.Sid == sid && h.Owner == owner { + return true + } + } + return false +} + +// Locks returns the held locks, sorted by Start. The slice aliases internal +// state; the caller must not mutate it. +func (s *Set) Locks() []Range { return s.locks } + +// Empty reports whether no locks are held, so the caller can drop the inode's +// entry from its table. +func (s *Set) Empty() bool { return len(s.locks) == 0 } diff --git a/weed/filer/posixlock/posixlock_test.go b/weed/filer/posixlock/posixlock_test.go new file mode 100644 index 0000000000..3cbb998435 --- /dev/null +++ b/weed/filer/posixlock/posixlock_test.go @@ -0,0 +1,292 @@ +package posixlock + +import ( + "math" + "testing" +) + +// acquire is a test helper asserting the lock is granted. +func mustAcquire(t *testing.T, s *Set, lk Range) { + t.Helper() + if _, granted := s.Acquire(lk); !granted { + t.Fatalf("expected lock granted: %+v", lk) + } +} + +func TestNonOverlappingLocksFromDifferentOwners(t *testing.T) { + s := &Set{} + mustAcquire(t, s, Range{Start: 0, End: 49, Type: Write, Owner: 1, Pid: 10}) + mustAcquire(t, s, Range{Start: 50, End: 99, Type: Write, Owner: 2, Pid: 20}) +} + +func TestOverlappingReadLocksFromDifferentOwners(t *testing.T) { + s := &Set{} + mustAcquire(t, s, Range{Start: 0, End: 99, Type: Read, Owner: 1, Pid: 10}) + mustAcquire(t, s, Range{Start: 50, End: 149, Type: Read, Owner: 2, Pid: 20}) +} + +func TestOverlappingWriteReadConflict(t *testing.T) { + s := &Set{} + mustAcquire(t, s, Range{Start: 0, End: 99, Type: Write, Owner: 1, Pid: 10}) + if _, granted := s.Acquire(Range{Start: 50, End: 149, Type: Read, Owner: 2, Pid: 20}); granted { + t.Fatal("expected conflict") + } +} + +func TestOverlappingWriteWriteConflict(t *testing.T) { + s := &Set{} + mustAcquire(t, s, Range{Start: 0, End: 99, Type: Write, Owner: 1, Pid: 10}) + if _, granted := s.Acquire(Range{Start: 50, End: 149, Type: Write, Owner: 2, Pid: 20}); granted { + t.Fatal("expected conflict") + } +} + +func TestSameOwnerUpgradeReadToWrite(t *testing.T) { + s := &Set{} + mustAcquire(t, s, Range{Start: 0, End: 99, Type: Read, Owner: 1, Pid: 10}) + mustAcquire(t, s, Range{Start: 0, End: 99, Type: Write, Owner: 1, Pid: 10}) + + c, found := s.Conflict(Range{Start: 0, End: 99, Type: Write, Owner: 2, Pid: 20}) + if !found || c.Type != Write { + t.Fatalf("expected conflicting write lock after upgrade, got %+v found=%v", c, found) + } +} + +func TestSameOwnerDowngradeWriteToRead(t *testing.T) { + s := &Set{} + mustAcquire(t, s, Range{Start: 0, End: 99, Type: Write, Owner: 1, Pid: 10}) + mustAcquire(t, s, Range{Start: 0, End: 99, Type: Read, Owner: 1, Pid: 10}) + // Another owner can now take a shared read lock. + mustAcquire(t, s, Range{Start: 0, End: 99, Type: Read, Owner: 2, Pid: 20}) +} + +func TestLockCoalescing(t *testing.T) { + s := &Set{} + mustAcquire(t, s, Range{Start: 0, End: 9, Type: Write, Owner: 1, Pid: 10}) + mustAcquire(t, s, Range{Start: 10, End: 19, Type: Write, Owner: 1, Pid: 10}) + + if len(s.locks) != 1 { + t.Fatalf("expected 1 coalesced lock, got %d: %+v", len(s.locks), s.locks) + } + if s.locks[0].Start != 0 || s.locks[0].End != 19 { + t.Errorf("expected coalesced [0,19], got [%d,%d]", s.locks[0].Start, s.locks[0].End) + } +} + +func TestLockSplitting(t *testing.T) { + s := &Set{} + mustAcquire(t, s, Range{Start: 0, End: 99, Type: Write, Owner: 1, Pid: 10}) + s.Release(Range{Start: 40, End: 59, Type: Unlock, Owner: 1, Pid: 10}) + + if len(s.locks) != 2 { + t.Fatalf("expected 2 locks after split, got %d: %+v", len(s.locks), s.locks) + } + if s.locks[0].Start != 0 || s.locks[0].End != 39 { + t.Errorf("expected left [0,39], got [%d,%d]", s.locks[0].Start, s.locks[0].End) + } + if s.locks[1].Start != 60 || s.locks[1].End != 99 { + t.Errorf("expected right [60,99], got [%d,%d]", s.locks[1].Start, s.locks[1].End) + } +} + +func TestConflictReportsHolder(t *testing.T) { + s := &Set{} + mustAcquire(t, s, Range{Start: 10, End: 50, Type: Write, Owner: 1, Pid: 10}) + + c, found := s.Conflict(Range{Start: 30, End: 70, Type: Read, Owner: 2, Pid: 20}) + if !found { + t.Fatal("expected a conflict") + } + if c.Type != Write || c.Pid != 10 || c.Start != 10 || c.End != 50 { + t.Fatalf("unexpected conflict report: %+v", c) + } +} + +func TestConflictNoneForSharedReads(t *testing.T) { + s := &Set{} + mustAcquire(t, s, Range{Start: 10, End: 50, Type: Read, Owner: 1, Pid: 10}) + if _, found := s.Conflict(Range{Start: 30, End: 70, Type: Read, Owner: 2, Pid: 20}); found { + t.Fatal("two read locks should not conflict") + } +} + +func TestConflictSameOwnerNone(t *testing.T) { + s := &Set{} + mustAcquire(t, s, Range{Start: 0, End: 99, Type: Write, Owner: 1, Pid: 10}) + if _, found := s.Conflict(Range{Start: 0, End: 99, Type: Write, Owner: 1, Pid: 10}); found { + t.Fatal("an owner should not conflict with itself") + } +} + +func TestReleaseOwner(t *testing.T) { + s := &Set{} + mustAcquire(t, s, Range{Start: 0, End: 49, Type: Write, Owner: 1, Pid: 10}) + mustAcquire(t, s, Range{Start: 50, End: 99, Type: Write, Owner: 1, Pid: 10}) + mustAcquire(t, s, Range{Start: 200, End: 299, Type: Read, Owner: 2, Pid: 20}) + + s.ReleaseOwner(0, 1) + + if _, found := s.Conflict(Range{Start: 0, End: 99, Type: Write, Owner: 3, Pid: 30}); found { + t.Fatal("owner 1's locks should be gone") + } + c, found := s.Conflict(Range{Start: 200, End: 299, Type: Write, Owner: 3, Pid: 30}) + if !found || c.Type != Read { + t.Fatalf("owner 2's read lock should remain, got %+v found=%v", c, found) + } +} + +func TestFlockAndFcntlDoNotConflict(t *testing.T) { + s := &Set{} + mustAcquire(t, s, Range{Start: 0, End: 99, Type: Write, Owner: 1, Pid: 10}) + mustAcquire(t, s, Range{Start: 0, End: math.MaxUint64, Type: Write, Owner: 2, Pid: 20, IsFlock: true}) +} + +func TestReleasePosixOwnerKeepsFlock(t *testing.T) { + s := &Set{} + mustAcquire(t, s, Range{Start: 0, End: math.MaxUint64, Type: Write, Owner: 1, Pid: 10, IsFlock: true}) + s.ReleasePosixOwner(0, 1) + if _, found := s.Conflict(Range{Start: 0, End: math.MaxUint64, Type: Write, Owner: 2, Pid: 20, IsFlock: true}); !found { + t.Fatal("flock lock should remain after ReleasePosixOwner") + } +} + +func TestReleaseFlockOwnerKeepsPosix(t *testing.T) { + s := &Set{} + mustAcquire(t, s, Range{Start: 0, End: 99, Type: Write, Owner: 1, Pid: 10}) + mustAcquire(t, s, Range{Start: 0, End: math.MaxUint64, Type: Write, Owner: 2, Pid: 10, IsFlock: true}) + + s.ReleaseFlockOwner(0, 2) + + if _, found := s.Conflict(Range{Start: 0, End: 99, Type: Write, Owner: 3, Pid: 30}); !found { + t.Fatal("fcntl lock should remain after ReleaseFlockOwner") + } + if _, found := s.Conflict(Range{Start: 0, End: math.MaxUint64, Type: Write, Owner: 4, Pid: 40, IsFlock: true}); found { + t.Fatal("flock lock should be gone after ReleaseFlockOwner") + } +} + +func TestHasPosixIgnoresMissingOwnerAndFlock(t *testing.T) { + s := &Set{} + if s.HasPosix(0, 1) { + t.Fatal("empty set should report no posix owner") + } + mustAcquire(t, s, Range{Start: 0, End: math.MaxUint64, Type: Write, Owner: 1, Pid: 10, IsFlock: true}) + if s.HasPosix(0, 1) { + t.Fatal("a flock owner is not a posix owner") + } + mustAcquire(t, s, Range{Start: 0, End: 99, Type: Write, Owner: 2, Pid: 20}) + if !s.HasPosix(0, 2) { + t.Fatal("posix owner should be reported") + } +} + +func TestWholeFileLock(t *testing.T) { + s := &Set{} + mustAcquire(t, s, Range{Start: 0, End: math.MaxUint64, Type: Write, Owner: 1, Pid: 10}) + if _, granted := s.Acquire(Range{Start: 0, End: math.MaxUint64, Type: Write, Owner: 2, Pid: 20}); granted { + t.Fatal("whole-file lock should block another owner") + } + if _, granted := s.Acquire(Range{Start: 100, End: 200, Type: Read, Owner: 2, Pid: 20}); granted { + t.Fatal("partial overlap with whole-file lock should conflict") + } +} + +func TestReleaseNoExistingLocks(t *testing.T) { + s := &Set{} + s.Release(Range{Start: 0, End: 99, Type: Unlock, Owner: 1, Pid: 10}) + if !s.Empty() { + t.Fatal("releasing on an empty set should be a no-op") + } +} + +func TestSameOwnerReplaceDifferentType(t *testing.T) { + s := &Set{} + mustAcquire(t, s, Range{Start: 0, End: 99, Type: Write, Owner: 1, Pid: 10}) + mustAcquire(t, s, Range{Start: 30, End: 60, Type: Read, Owner: 1, Pid: 10}) + + if len(s.locks) != 3 { + t.Fatalf("expected 3 locks after partial type change, got %d: %+v", len(s.locks), s.locks) + } + if s.locks[0].Type != Write || s.locks[0].Start != 0 || s.locks[0].End != 29 { + t.Errorf("expected write [0,29], got %+v", s.locks[0]) + } + if s.locks[1].Type != Read || s.locks[1].Start != 30 || s.locks[1].End != 60 { + t.Errorf("expected read [30,60], got %+v", s.locks[1]) + } + if s.locks[2].Type != Write || s.locks[2].Start != 61 || s.locks[2].End != 99 { + t.Errorf("expected write [61,99], got %+v", s.locks[2]) + } +} + +func TestNonAdjacentRangesNotCoalesced(t *testing.T) { + s := &Set{} + mustAcquire(t, s, Range{Start: 5, End: math.MaxUint64, Type: Write, Owner: 1, Pid: 10}) + mustAcquire(t, s, Range{Start: 0, End: 2, Type: Write, Owner: 1, Pid: 10}) + + if len(s.locks) != 2 { + t.Fatalf("gap [3,4] should prevent coalescing, got %d: %+v", len(s.locks), s.locks) + } + if s.locks[0].Start != 0 || s.locks[0].End != 2 { + t.Errorf("expected [0,2], got [%d,%d]", s.locks[0].Start, s.locks[0].End) + } + if s.locks[1].Start != 5 || s.locks[1].End != math.MaxUint64 { + t.Errorf("expected [5,MaxUint64], got [%d,%d]", s.locks[1].Start, s.locks[1].End) + } +} + +func TestAdjacencyNoOverflowAtMaxUint64(t *testing.T) { + s := &Set{} + mustAcquire(t, s, Range{Start: 100, End: math.MaxUint64, Type: Write, Owner: 1, Pid: 10}) + mustAcquire(t, s, Range{Start: 0, End: 0, Type: Write, Owner: 1, Pid: 10}) + + if len(s.locks) != 2 { + t.Fatalf("MaxUint64+1 must not wrap and falsely merge, got %d: %+v", len(s.locks), s.locks) + } +} + +// Sessions are part of owner identity: the same FUSE Owner number on two +// different mounts (Sid) is two distinct owners and must contend. +func TestTwoSessionsSameOwnerDoNotAlias(t *testing.T) { + s := &Set{} + mustAcquire(t, s, Range{Start: 0, End: 99, Type: Write, Sid: 1, Owner: 5, Pid: 10}) + + if _, granted := s.Acquire(Range{Start: 50, End: 149, Type: Write, Sid: 2, Owner: 5, Pid: 20}); granted { + t.Fatal("same Owner number on a different session must conflict, not alias") + } + // A read on session 2 against a session-1 read is fine (shared). + s2 := &Set{} + mustAcquire(t, s2, Range{Start: 0, End: 99, Type: Read, Sid: 1, Owner: 5}) + mustAcquire(t, s2, Range{Start: 0, End: 99, Type: Read, Sid: 2, Owner: 5}) +} + +// Reaping a dead mount drops only that session's locks. +func TestReleaseSessionReapsOnlyThatSession(t *testing.T) { + s := &Set{} + mustAcquire(t, s, Range{Start: 0, End: 49, Type: Write, Sid: 1, Owner: 1, Pid: 10}) + mustAcquire(t, s, Range{Start: 0, End: math.MaxUint64, Type: Write, Sid: 1, Owner: 2, Pid: 11, IsFlock: true}) + mustAcquire(t, s, Range{Start: 50, End: 99, Type: Write, Sid: 2, Owner: 1, Pid: 20}) + + s.ReleaseSession(1) + + for _, h := range s.locks { + if h.Sid == 1 { + t.Fatalf("session 1 lock survived reaping: %+v", h) + } + } + c, found := s.Conflict(Range{Start: 50, End: 99, Type: Write, Sid: 3, Owner: 9}) + if !found || c.Sid != 2 { + t.Fatalf("session 2's lock should remain, got %+v found=%v", c, found) + } +} + +func TestEmptyAfterReleasingAll(t *testing.T) { + s := &Set{} + mustAcquire(t, s, Range{Start: 0, End: 99, Type: Write, Owner: 1, Pid: 10}) + if s.Empty() { + t.Fatal("set should not be empty with a held lock") + } + s.Release(Range{Start: 0, End: 99, Type: Unlock, Owner: 1, Pid: 10}) + if !s.Empty() { + t.Fatal("set should be empty after releasing the only lock") + } +}