mirrors/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2026-05-26 03:30:19 +00:00
Code Issues Packages Projects Releases Wiki Activity

mount,filer: re-assert POSIX locks via keepalive (ownership migration + restart) (#9668)

Browse Source 
* mount: renew POSIX lock leases via keepalive

The mount tracks the inode keys it holds locks on and a background loop
renews its session lease (KEEP_ALIVE) with each key's owner filer every
5s, within the filer's 15s TTL. A live mount is never reaped; a dead one
stops renewing and owners reclaim its locks. Tracking is a superset:
holds are added on grant and dropped only on owner release, so a still
held lock is never under-renewed.

* mount,filer: re-assert held POSIX locks via keepalive

The owner filer holds POSIX advisory locks as in-memory soft state, so a key's
owner change (ring rebalance) or an owner restart lost or stranded them: the new
or restarted owner was blind to existing holders and would double-grant.

Make the keepalive carry the mount's held lock ranges per key. The mount mirrors
its own granted locks (posixOwn), and each tick re-asserts them to the key's
current owner, which rebuilds that session's locks from the assertion — self
-healing after a takeover or restart. The owner arbitrates re-asserted locks
against other sessions so it never double-grants; a lock that lost a migration
race is reported, not forced. A bare keepalive (no ranges) still just renews.
This commit is contained in:
Chris Lu
2026-05-25 01:02:45 -07:00
committed by GitHub
parent c97b69f8a4
commit f8caaa4464
8 changed files with 398 additions and 123 deletions
Download Patch File Download Diff File Expand all files Collapse all files

63
weed/filer/posixlock/manager.go
View File

@@ -87,6 +87,69 @@ func (m *Manager) TryLock(key string, lk Range) (Range, bool) {
return Range{}, true
}
// Track records a lock the server already granted, without arbitration. A mount
// uses it to mirror its own held locks so it can re-assert them to the inode's
// current owner filer after an ownership change or owner restart.
func (m *Manager) Track(key string, lk Range) {
m.mu.Lock()
defer m.mu.Unlock()
s, ok := m.byKey[key]
if !ok {
s = &Set{}
m.byKey[key] = s
}
s.Grant(lk)
m.index(lk.Sid, key)
}
// Snapshot returns a copy of the held locks per key. A mount calls it to drive
// re-assertion keepalives; the filer never does.
func (m *Manager) Snapshot() map[string][]Range {
m.mu.Lock()
defer m.mu.Unlock()
out := make(map[string][]Range, len(m.byKey))
for key, s := range m.byKey {
out[key] = append([]Range(nil), s.locks...)
}
return out
}
// Reassert rebuilds session sid's locks on key from the client's authoritative
// list, renewing the lease. It replaces sid's existing locks on the key, then
// re-acquires each asserted lock — arbitrating against other sessions so it never
// double-grants. Locks that lost to another session in a migration window are
// returned as conflicts. The owner filer calls this on a re-assertion keepalive.
func (m *Manager) Reassert(key string, sid uint64, locks []Range) (conflicts []Range) {
m.mu.Lock()
defer m.mu.Unlock()
m.lastSeen[sid] = time.Now()
s := m.byKey[key]
if s == nil {
if len(locks) == 0 {
return nil
}
s = &Set{}
m.byKey[key] = s
}
s.ReleaseSession(sid)
for _, lk := range locks {
lk.Sid = sid
if c, granted := s.Acquire(lk); !granted {
conflicts = append(conflicts, c)
}
}
if setHasSession(s, sid) {
m.index(sid, key)
} else {
m.deindex(sid, key)
}
if s.Empty() {
delete(m.byKey, key)
}
return conflicts
}
// 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()

6
weed/filer/posixlock/posixlock.go
View File

@@ -82,6 +82,12 @@ func (s *Set) Acquire(lk Range) (Range, bool) {
return Range{}, true
}
// Grant inserts lk without a conflict check. It is for a client mirroring locks
// the server already granted (so they are conflict-free), not for arbitration.
func (s *Set) Grant(lk Range) {
s.insert(lk)
}
// 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).

82
weed/filer/posixlock/reassert_test.go Normal file
View File

@@ -0,0 +1,82 @@
package posixlock
import (
"reflect"
"testing"
"time"
)
// A mount's tracked locks round-trip through Snapshot back to a fresh owner via
// Reassert — the owner-restart / ring-change recovery path.
func TestReassertRebuildsOnFreshOwner(t *testing.T) {
const sid = uint64(7)
// Client mirror: two granted locks on one key, one on another.
client := NewManager()
client.Track("a", Range{Start: 0, End: 99, Type: Write, Sid: sid, Owner: 1})
client.Track("a", Range{Start: 200, End: 299, Type: Read, Sid: sid, Owner: 2})
client.Track("b", Range{Start: 0, End: maxEnd, Type: Write, Sid: sid, Owner: 1, IsFlock: true})
// Fresh owner (post-restart / new ring owner) knows nothing.
owner := NewManager()
for key, locks := range client.Snapshot() {
if c := owner.Reassert(key, sid, locks); c != nil {
t.Fatalf("unexpected conflict reasserting %s: %+v", key, c)
}
}
// The owner now reports the same conflicts a foreign session would hit.
if _, granted := owner.TryLock("a", Range{Start: 50, End: 60, Type: Write, Sid: 99, Owner: 1}); granted {
t.Fatal("owner should block a foreign write after rebuild")
}
if _, granted := owner.TryLock("b", Range{Start: 0, End: 0, Type: Read, Sid: 99, Owner: 1, IsFlock: true}); granted {
t.Fatal("owner should block a foreign flock read after rebuild")
}
}
// Re-asserting every tick is idempotent: the owner's view is unchanged.
func TestReassertIdempotent(t *testing.T) {
const sid = uint64(1)
m := NewManager()
m.TryLock("k", Range{Start: 0, End: 99, Type: Write, Sid: sid, Owner: 1})
before := append([]Range(nil), m.byKey["k"].locks...)
m.Reassert("k", sid, before)
m.Reassert("k", sid, before)
if !reflect.DeepEqual(m.byKey["k"].locks, before) {
t.Fatalf("reassert not idempotent:\n got %+v\nwant %+v", m.byKey["k"].locks, before)
}
}
// A lock another session grabbed in the migration window is reported as a
// conflict and not double-granted.
func TestReassertReportsConflict(t *testing.T) {
const mine, other = uint64(1), uint64(2)
m := NewManager()
// Another mount took the lock on this (new) owner during the gap.
m.TryLock("k", Range{Start: 0, End: 99, Type: Write, Sid: other, Owner: 1})
conflicts := m.Reassert("k", mine, []Range{{Start: 0, End: 99, Type: Write, Sid: mine, Owner: 1}})
if len(conflicts) != 1 {
t.Fatalf("expected 1 conflict, got %d: %+v", len(conflicts), conflicts)
}
// The other session keeps the lock; mine was not installed.
if got := len(m.byKey["k"].locks); got != 1 {
t.Fatalf("expected only the incumbent lock, got %d", got)
}
}
// Reassert renews the lease, so a re-asserting mount is not reaped.
func TestReassertRenewsLease(t *testing.T) {
const sid = uint64(1)
m := NewManager()
m.Renew(sid)
m.Reassert("k", sid, []Range{{Start: 0, End: 9, Type: Write, Sid: sid, Owner: 1}})
if reaped := m.ReapExpired(time.Hour); len(reaped) != 0 {
t.Fatalf("freshly re-asserted session should not be reaped: %v", reaped)
}
}
const maxEnd = ^uint64(0)

10
weed/mount/weedfs.go
View File

@@ -15,6 +15,7 @@ import (
"github.com/seaweedfs/seaweedfs/weed/cluster"
"github.com/seaweedfs/seaweedfs/weed/filer"
"github.com/seaweedfs/seaweedfs/weed/filer/posixlock"
"github.com/seaweedfs/seaweedfs/weed/glog"
"github.com/seaweedfs/seaweedfs/weed/mount/meta_cache"
"github.com/seaweedfs/seaweedfs/weed/mount/page_writer"
@@ -140,8 +141,9 @@ 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
posixSid uint64 // this mount's session id, for routed-lock owner identity
posixHint *posixLockHint // local fcntl-lock hint for routed mode
posixOwn *posixlock.Manager // mirror of locks this mount holds, re-asserted via keepalive
rdmaClient *RDMAMountClient
peerRegistrar *PeerRegistrar
peerDirectory *PeerDirectory
@@ -248,6 +250,7 @@ func NewSeaweedFileSystem(option *Option) *WFS {
posixLocks: NewPosixLockTable(),
posixSid: randomPosixSid(),
posixHint: newPosixLockHint(),
posixOwn: posixlock.NewManager(),
refreshingDirs: make(map[util.FullPath]struct{}),
atimeMap: make(map[uint64]time.Time, 8192),
openMtimeCache: make(map[uint64][2]int64, 8192),
@@ -511,6 +514,9 @@ func (wfs *WFS) StartBackgroundTasks() error {
go wfs.loopFlushDirtyMetadata()
go wfs.loopEvictIdleDirCache()
go wfs.loopProactiveFlush()
if wfs.crossMountLocks() {
go wfs.loopRenewPosixLeases()
}
return nil
}

87
weed/mount/weedfs_posix_lock_routed.go
View File

@@ -36,8 +36,11 @@ const (
// 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.
// close indefinitely. It also bounds each keepalive RPC.
posixLockReleaseTimeout = 5 * time.Second
// posixKeepaliveInterval renews the session lease well within the filer's
// posixLockSessionTTL (15s) so a live mount is never reaped.
posixKeepaliveInterval = 5 * time.Second
)
// posixLockHint records, per inode, which owners this mount has taken fcntl locks
@@ -207,9 +210,7 @@ func (wfs *WFS) routedSetLk(cancel <-chan struct{}, in *fuse.LkIn) fuse.Status {
if !resp.GetGranted() {
return fuse.EAGAIN
}
if !lk.IsFlock {
wfs.posixHint.add(in.NodeId, in.Owner)
}
wfs.recordPosixGrant(key, in.NodeId, lk)
return fuse.OK
}
@@ -231,13 +232,27 @@ func (wfs *WFS) routedSetLkw(cancel <-chan struct{}, in *fuse.LkIn) fuse.Status
}
return resp.GetGranted(), nil
})
if status == fuse.OK && !lk.IsFlock {
wfs.posixHint.add(in.NodeId, in.Owner)
if status == fuse.OK {
wfs.recordPosixGrant(key, in.NodeId, lk)
}
return status
}
// recordPosixGrant notes a newly granted lock: the flush hint (fcntl only) and
// the own-lock mirror, which the keepalive re-asserts to the key's owner so the
// lease is renewed and the owner can rebuild after a takeover or restart.
func (wfs *WFS) recordPosixGrant(key string, inode uint64, lk posixlock.Range) {
if !lk.IsFlock {
wfs.posixHint.add(inode, lk.Owner)
}
wfs.posixOwn.Track(key, lk)
}
func (wfs *WFS) routedUnlock(key string, lk posixlock.Range) fuse.Status {
// Drop the range from the mirror first so a keepalive re-assertion can't
// resurrect it; if the RPC below fails, the next re-assertion reconciles the
// owner to this (released) state anyway.
wfs.posixOwn.Unlock(key, lk)
// 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.
@@ -291,6 +306,7 @@ func (wfs *WFS) routedReleasePosixOwner(inode, owner uint64) {
if !ok {
return
}
wfs.posixOwn.ReleasePosixOwner(key, wfs.posixSid, owner)
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 {
@@ -308,6 +324,7 @@ func (wfs *WFS) routedReleaseFlockOwner(inode, owner uint64) {
if !ok {
return
}
wfs.posixOwn.ReleaseFlockOwner(key, wfs.posixSid, owner)
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 {
@@ -349,3 +366,61 @@ func (wfs *WFS) hasPosixOwner(inode, owner uint64) bool {
}
return wfs.posixLocks.HasPosixOwner(inode, owner)
}
// callPosixReassert sends the mount's held locks on key to the key's current
// owner filer as a KEEP_ALIVE, which renews the lease and lets the owner rebuild
// its in-memory state after a takeover or restart.
func (wfs *WFS) callPosixReassert(ctx context.Context, key string, locks []posixlock.Range) error {
pbLocks := make([]*filer_pb.PosixLockRange, 0, len(locks))
for _, l := range locks {
pbLocks = append(pbLocks, &filer_pb.PosixLockRange{
Start: l.Start, End: l.End, Type: l.Type,
Sid: l.Sid, Owner: l.Owner, Pid: l.Pid, IsFlock: l.IsFlock,
})
}
return wfs.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
_, e := client.PosixLock(ctx, &filer_pb.PosixLockRequest{
Key: key,
Op: filer_pb.PosixLockOp_KEEP_ALIVE,
Lock: &filer_pb.PosixLockRange{Sid: wfs.posixSid},
Locks: pbLocks,
})
return e
})
}
// posixKeepaliveConcurrency bounds the parallel keepalive RPCs per tick. A mount
// holding locks on many inodes must renew every lease well within the filer TTL;
// dispatching them concurrently keeps the round trip from scaling with lock count.
const posixKeepaliveConcurrency = 32
// loopRenewPosixLeases re-asserts this mount's held locks to the owner filer of
// every key, which renews the session lease and rebuilds the owner's state after
// a ring change or owner restart. A dead mount stops re-asserting and the owners'
// sweepers reclaim its locks after the TTL.
func (wfs *WFS) loopRenewPosixLeases() {
ticker := time.NewTicker(posixKeepaliveInterval)
defer ticker.Stop()
for range ticker.C {
held := wfs.posixOwn.Snapshot()
var wg sync.WaitGroup
sem := make(chan struct{}, posixKeepaliveConcurrency)
for key, locks := range held {
wg.Add(1)
sem <- struct{}{}
go func() {
defer wg.Done()
defer func() { <-sem }()
// Bound each re-assertion so a stuck filer can't block wg.Wait and
// stall the whole tick, which would let other keys' leases expire
// and get reaped.
ctx, cancel := context.WithTimeout(context.Background(), posixLockReleaseTimeout)
defer cancel()
if err := wfs.callPosixReassert(ctx, key, locks); err != nil {
glog.V(2).Infof("posix reassert %s: %v", key, err)
}
}()
}
wg.Wait()
}
}

4
weed/pb/filer.proto
View File

@@ -388,6 +388,10 @@ message PosixLockRequest {
bool is_moved = 2;
PosixLockOp op = 3;
PosixLockRange lock = 4;
// locks carries the full set a mount holds on key for a KEEP_ALIVE
// re-assertion, so the current owner filer can rebuild its in-memory state
// after an ownership change or restart. lock.sid identifies the session.
repeated PosixLockRange locks = 5;
}
enum PosixLockOp {

241
weed/pb/filer_pb/filer.pb.go
View File

@@ -1983,11 +1983,15 @@ func (x *PosixLockRange) GetIsFlock() bool {
// owner and to index the table. A non-owner filer forwards the request one hop;
// is_moved bounds it so a stale ring view cannot loop.
type PosixLockRequest struct {
state protoimpl.MessageState `protogen:"open.v1"`
Key string `protobuf:"bytes,1,opt,name=key,proto3" json:"key,omitempty"`
IsMoved bool `protobuf:"varint,2,opt,name=is_moved,json=isMoved,proto3" json:"is_moved,omitempty"`
Op PosixLockOp `protobuf:"varint,3,opt,name=op,proto3,enum=filer_pb.PosixLockOp" json:"op,omitempty"`
Lock *PosixLockRange `protobuf:"bytes,4,opt,name=lock,proto3" json:"lock,omitempty"`
state protoimpl.MessageState `protogen:"open.v1"`
Key string `protobuf:"bytes,1,opt,name=key,proto3" json:"key,omitempty"`
IsMoved bool `protobuf:"varint,2,opt,name=is_moved,json=isMoved,proto3" json:"is_moved,omitempty"`
Op PosixLockOp `protobuf:"varint,3,opt,name=op,proto3,enum=filer_pb.PosixLockOp" json:"op,omitempty"`
Lock *PosixLockRange `protobuf:"bytes,4,opt,name=lock,proto3" json:"lock,omitempty"`
// locks carries the full set a mount holds on key for a KEEP_ALIVE
// re-assertion, so the current owner filer can rebuild its in-memory state
// after an ownership change or restart. lock.sid identifies the session.
Locks []*PosixLockRange `protobuf:"bytes,5,rep,name=locks,proto3" json:"locks,omitempty"`
unknownFields protoimpl.UnknownFields
sizeCache protoimpl.SizeCache
}
@@ -2050,6 +2054,13 @@ func (x *PosixLockRequest) GetLock() *PosixLockRange {
return nil
}
func (x *PosixLockRequest) GetLocks() []*PosixLockRange {
if x != nil {
return x.Locks
}
return nil
}
type PosixLockResponse struct {
state protoimpl.MessageState `protogen:"open.v1"`
Granted bool `protobuf:"varint,1,opt,name=granted,proto3" json:"granted,omitempty"` // for TRY_LOCK: whether the lock was granted
@@ -6775,12 +6786,13 @@ const file_filer_proto_rawDesc = "" +
"\x03sid\x18\x04 \x01(\x04R\x03sid\x12\x14\n" +
"\x05owner\x18\x05 \x01(\x04R\x05owner\x12\x10\n" +
"\x03pid\x18\x06 \x01(\rR\x03pid\x12\x19\n" +
"\bis_flock\x18\a \x01(\bR\aisFlock\"\x94\x01\n" +
"\bis_flock\x18\a \x01(\bR\aisFlock\"\xc4\x01\n" +
"\x10PosixLockRequest\x12\x10\n" +
"\x03key\x18\x01 \x01(\tR\x03key\x12\x19\n" +
"\bis_moved\x18\x02 \x01(\bR\aisMoved\x12%\n" +
"\x02op\x18\x03 \x01(\x0e2\x15.filer_pb.PosixLockOpR\x02op\x12,\n" +
"\x04lock\x18\x04 \x01(\v2\x18.filer_pb.PosixLockRangeR\x04lock\"\x86\x01\n" +
"\x04lock\x18\x04 \x01(\v2\x18.filer_pb.PosixLockRangeR\x04lock\x12.\n" +
"\x05locks\x18\x05 \x03(\v2\x18.filer_pb.PosixLockRangeR\x05locks\"\x86\x01\n" +
"\x11PosixLockResponse\x12\x18\n" +
"\agranted\x18\x01 \x01(\bR\agranted\x12!\n" +
"\fhas_conflict\x18\x02 \x01(\bR\vhasConflict\x124\n" +
@@ -7349,113 +7361,114 @@ var file_filer_proto_depIdxs = []int32{
1, // 23: filer_pb.ObjectTransactionResponse.error_code:type_name -> filer_pb.FilerError
2, // 24: filer_pb.PosixLockRequest.op:type_name -> filer_pb.PosixLockOp
23, // 25: filer_pb.PosixLockRequest.lock:type_name -> filer_pb.PosixLockRange
23, // 26: filer_pb.PosixLockResponse.conflict:type_name -> filer_pb.PosixLockRange
21, // 27: filer_pb.ObjectTransactionBatchRequest.transactions:type_name -> filer_pb.ObjectTransactionRequest
22, // 28: filer_pb.ObjectTransactionBatchResponse.responses:type_name -> filer_pb.ObjectTransactionResponse
59, // 29: filer_pb.CreateEntryResponse.metadata_event:type_name -> filer_pb.SubscribeMetadataResponse
1, // 30: filer_pb.CreateEntryResponse.error_code:type_name -> filer_pb.FilerError
10, // 31: filer_pb.UpdateEntryRequest.entry:type_name -> filer_pb.Entry
97, // 32: filer_pb.UpdateEntryRequest.expected_extended:type_name -> filer_pb.UpdateEntryRequest.ExpectedExtendedEntry
59, // 33: filer_pb.UpdateEntryResponse.metadata_event:type_name -> filer_pb.SubscribeMetadataResponse
13, // 34: filer_pb.AppendToEntryRequest.chunks:type_name -> filer_pb.FileChunk
59, // 35: filer_pb.DeleteEntryResponse.metadata_event:type_name -> filer_pb.SubscribeMetadataResponse
12, // 36: filer_pb.StreamRenameEntryResponse.event_notification:type_name -> filer_pb.EventNotification
45, // 37: filer_pb.AssignVolumeResponse.location:type_name -> filer_pb.Location
45, // 38: filer_pb.Locations.locations:type_name -> filer_pb.Location
98, // 39: filer_pb.LookupVolumeResponse.locations_map:type_name -> filer_pb.LookupVolumeResponse.LocationsMapEntry
47, // 40: filer_pb.CollectionListResponse.collections:type_name -> filer_pb.Collection
12, // 41: filer_pb.SubscribeMetadataResponse.event_notification:type_name -> filer_pb.EventNotification
59, // 42: filer_pb.SubscribeMetadataResponse.events:type_name -> filer_pb.SubscribeMetadataResponse
60, // 43: filer_pb.SubscribeMetadataResponse.log_file_refs:type_name -> filer_pb.LogFileChunkRef
13, // 44: filer_pb.LogFileChunkRef.chunks:type_name -> filer_pb.FileChunk
10, // 45: filer_pb.TraverseBfsMetadataResponse.entry:type_name -> filer_pb.Entry
99, // 46: filer_pb.LocateBrokerResponse.resources:type_name -> filer_pb.LocateBrokerResponse.Resource
100, // 47: filer_pb.FilerConf.locations:type_name -> filer_pb.FilerConf.PathConf
10, // 48: filer_pb.CacheRemoteObjectToLocalClusterResponse.entry:type_name -> filer_pb.Entry
59, // 49: filer_pb.CacheRemoteObjectToLocalClusterResponse.metadata_event:type_name -> filer_pb.SubscribeMetadataResponse
81, // 50: filer_pb.TransferLocksRequest.locks:type_name -> filer_pb.Lock
17, // 51: filer_pb.StreamMutateEntryRequest.create_request:type_name -> filer_pb.CreateEntryRequest
29, // 52: filer_pb.StreamMutateEntryRequest.update_request:type_name -> filer_pb.UpdateEntryRequest
35, // 53: filer_pb.StreamMutateEntryRequest.delete_request:type_name -> filer_pb.DeleteEntryRequest
39, // 54: filer_pb.StreamMutateEntryRequest.rename_request:type_name -> filer_pb.StreamRenameEntryRequest
28, // 55: filer_pb.StreamMutateEntryResponse.create_response:type_name -> filer_pb.CreateEntryResponse
30, // 56: filer_pb.StreamMutateEntryResponse.update_response:type_name -> filer_pb.UpdateEntryResponse
36, // 57: filer_pb.StreamMutateEntryResponse.delete_response:type_name -> filer_pb.DeleteEntryResponse
40, // 58: filer_pb.StreamMutateEntryResponse.rename_response:type_name -> filer_pb.StreamRenameEntryResponse
92, // 59: filer_pb.MountListResponse.mounts:type_name -> filer_pb.MountInfo
3, // 60: filer_pb.WriteCondition.Clause.kind:type_name -> filer_pb.WriteCondition.Kind
44, // 61: filer_pb.LookupVolumeResponse.LocationsMapEntry.value:type_name -> filer_pb.Locations
5, // 62: filer_pb.SeaweedFiler.LookupDirectoryEntry:input_type -> filer_pb.LookupDirectoryEntryRequest
7, // 63: filer_pb.SeaweedFiler.ListEntries:input_type -> filer_pb.ListEntriesRequest
17, // 64: filer_pb.SeaweedFiler.CreateEntry:input_type -> filer_pb.CreateEntryRequest
29, // 65: filer_pb.SeaweedFiler.UpdateEntry:input_type -> filer_pb.UpdateEntryRequest
31, // 66: filer_pb.SeaweedFiler.TouchAccessTime:input_type -> filer_pb.TouchAccessTimeRequest
33, // 67: filer_pb.SeaweedFiler.AppendToEntry:input_type -> filer_pb.AppendToEntryRequest
35, // 68: filer_pb.SeaweedFiler.DeleteEntry:input_type -> filer_pb.DeleteEntryRequest
21, // 69: filer_pb.SeaweedFiler.ObjectTransaction:input_type -> filer_pb.ObjectTransactionRequest
26, // 70: filer_pb.SeaweedFiler.ObjectTransactionBatch:input_type -> filer_pb.ObjectTransactionBatchRequest
24, // 71: filer_pb.SeaweedFiler.PosixLock:input_type -> filer_pb.PosixLockRequest
37, // 72: filer_pb.SeaweedFiler.AtomicRenameEntry:input_type -> filer_pb.AtomicRenameEntryRequest
39, // 73: filer_pb.SeaweedFiler.StreamRenameEntry:input_type -> filer_pb.StreamRenameEntryRequest
86, // 74: filer_pb.SeaweedFiler.StreamMutateEntry:input_type -> filer_pb.StreamMutateEntryRequest
41, // 75: filer_pb.SeaweedFiler.AssignVolume:input_type -> filer_pb.AssignVolumeRequest
43, // 76: filer_pb.SeaweedFiler.LookupVolume:input_type -> filer_pb.LookupVolumeRequest
48, // 77: filer_pb.SeaweedFiler.CollectionList:input_type -> filer_pb.CollectionListRequest
50, // 78: filer_pb.SeaweedFiler.DeleteCollection:input_type -> filer_pb.DeleteCollectionRequest
52, // 79: filer_pb.SeaweedFiler.Statistics:input_type -> filer_pb.StatisticsRequest
54, // 80: filer_pb.SeaweedFiler.Ping:input_type -> filer_pb.PingRequest
56, // 81: filer_pb.SeaweedFiler.GetFilerConfiguration:input_type -> filer_pb.GetFilerConfigurationRequest
61, // 82: filer_pb.SeaweedFiler.TraverseBfsMetadata:input_type -> filer_pb.TraverseBfsMetadataRequest
58, // 83: filer_pb.SeaweedFiler.SubscribeMetadata:input_type -> filer_pb.SubscribeMetadataRequest
58, // 84: filer_pb.SeaweedFiler.SubscribeLocalMetadata:input_type -> filer_pb.SubscribeMetadataRequest
68, // 85: filer_pb.SeaweedFiler.KvGet:input_type -> filer_pb.KvGetRequest
70, // 86: filer_pb.SeaweedFiler.KvPut:input_type -> filer_pb.KvPutRequest
73, // 87: filer_pb.SeaweedFiler.CacheRemoteObjectToLocalCluster:input_type -> filer_pb.CacheRemoteObjectToLocalClusterRequest
75, // 88: filer_pb.SeaweedFiler.DistributedLock:input_type -> filer_pb.LockRequest
77, // 89: filer_pb.SeaweedFiler.DistributedUnlock:input_type -> filer_pb.UnlockRequest
79, // 90: filer_pb.SeaweedFiler.FindLockOwner:input_type -> filer_pb.FindLockOwnerRequest
82, // 91: filer_pb.SeaweedFiler.TransferLocks:input_type -> filer_pb.TransferLocksRequest
84, // 92: filer_pb.SeaweedFiler.ReplicateLock:input_type -> filer_pb.ReplicateLockRequest
88, // 93: filer_pb.SeaweedFiler.MountRegister:input_type -> filer_pb.MountRegisterRequest
90, // 94: filer_pb.SeaweedFiler.MountList:input_type -> filer_pb.MountListRequest
6, // 95: filer_pb.SeaweedFiler.LookupDirectoryEntry:output_type -> filer_pb.LookupDirectoryEntryResponse
8, // 96: filer_pb.SeaweedFiler.ListEntries:output_type -> filer_pb.ListEntriesResponse
28, // 97: filer_pb.SeaweedFiler.CreateEntry:output_type -> filer_pb.CreateEntryResponse
30, // 98: filer_pb.SeaweedFiler.UpdateEntry:output_type -> filer_pb.UpdateEntryResponse
32, // 99: filer_pb.SeaweedFiler.TouchAccessTime:output_type -> filer_pb.TouchAccessTimeResponse
34, // 100: filer_pb.SeaweedFiler.AppendToEntry:output_type -> filer_pb.AppendToEntryResponse
36, // 101: filer_pb.SeaweedFiler.DeleteEntry:output_type -> filer_pb.DeleteEntryResponse
22, // 102: filer_pb.SeaweedFiler.ObjectTransaction:output_type -> filer_pb.ObjectTransactionResponse
27, // 103: filer_pb.SeaweedFiler.ObjectTransactionBatch:output_type -> filer_pb.ObjectTransactionBatchResponse
25, // 104: filer_pb.SeaweedFiler.PosixLock:output_type -> filer_pb.PosixLockResponse
38, // 105: filer_pb.SeaweedFiler.AtomicRenameEntry:output_type -> filer_pb.AtomicRenameEntryResponse
40, // 106: filer_pb.SeaweedFiler.StreamRenameEntry:output_type -> filer_pb.StreamRenameEntryResponse
87, // 107: filer_pb.SeaweedFiler.StreamMutateEntry:output_type -> filer_pb.StreamMutateEntryResponse
42, // 108: filer_pb.SeaweedFiler.AssignVolume:output_type -> filer_pb.AssignVolumeResponse
46, // 109: filer_pb.SeaweedFiler.LookupVolume:output_type -> filer_pb.LookupVolumeResponse
49, // 110: filer_pb.SeaweedFiler.CollectionList:output_type -> filer_pb.CollectionListResponse
51, // 111: filer_pb.SeaweedFiler.DeleteCollection:output_type -> filer_pb.DeleteCollectionResponse
53, // 112: filer_pb.SeaweedFiler.Statistics:output_type -> filer_pb.StatisticsResponse
55, // 113: filer_pb.SeaweedFiler.Ping:output_type -> filer_pb.PingResponse
57, // 114: filer_pb.SeaweedFiler.GetFilerConfiguration:output_type -> filer_pb.GetFilerConfigurationResponse
62, // 115: filer_pb.SeaweedFiler.TraverseBfsMetadata:output_type -> filer_pb.TraverseBfsMetadataResponse
59, // 116: filer_pb.SeaweedFiler.SubscribeMetadata:output_type -> filer_pb.SubscribeMetadataResponse
59, // 117: filer_pb.SeaweedFiler.SubscribeLocalMetadata:output_type -> filer_pb.SubscribeMetadataResponse
69, // 118: filer_pb.SeaweedFiler.KvGet:output_type -> filer_pb.KvGetResponse
71, // 119: filer_pb.SeaweedFiler.KvPut:output_type -> filer_pb.KvPutResponse
74, // 120: filer_pb.SeaweedFiler.CacheRemoteObjectToLocalCluster:output_type -> filer_pb.CacheRemoteObjectToLocalClusterResponse
76, // 121: filer_pb.SeaweedFiler.DistributedLock:output_type -> filer_pb.LockResponse
78, // 122: filer_pb.SeaweedFiler.DistributedUnlock:output_type -> filer_pb.UnlockResponse
80, // 123: filer_pb.SeaweedFiler.FindLockOwner:output_type -> filer_pb.FindLockOwnerResponse
83, // 124: filer_pb.SeaweedFiler.TransferLocks:output_type -> filer_pb.TransferLocksResponse
85, // 125: filer_pb.SeaweedFiler.ReplicateLock:output_type -> filer_pb.ReplicateLockResponse
89, // 126: filer_pb.SeaweedFiler.MountRegister:output_type -> filer_pb.MountRegisterResponse
91, // 127: filer_pb.SeaweedFiler.MountList:output_type -> filer_pb.MountListResponse
95, // [95:128] is the sub-list for method output_type
62, // [62:95] is the sub-list for method input_type
62, // [62:62] is the sub-list for extension type_name
62, // [62:62] is the sub-list for extension extendee
0, // [0:62] is the sub-list for field type_name
23, // 26: filer_pb.PosixLockRequest.locks:type_name -> filer_pb.PosixLockRange
23, // 27: filer_pb.PosixLockResponse.conflict:type_name -> filer_pb.PosixLockRange
21, // 28: filer_pb.ObjectTransactionBatchRequest.transactions:type_name -> filer_pb.ObjectTransactionRequest
22, // 29: filer_pb.ObjectTransactionBatchResponse.responses:type_name -> filer_pb.ObjectTransactionResponse
59, // 30: filer_pb.CreateEntryResponse.metadata_event:type_name -> filer_pb.SubscribeMetadataResponse
1, // 31: filer_pb.CreateEntryResponse.error_code:type_name -> filer_pb.FilerError
10, // 32: filer_pb.UpdateEntryRequest.entry:type_name -> filer_pb.Entry
97, // 33: filer_pb.UpdateEntryRequest.expected_extended:type_name -> filer_pb.UpdateEntryRequest.ExpectedExtendedEntry
59, // 34: filer_pb.UpdateEntryResponse.metadata_event:type_name -> filer_pb.SubscribeMetadataResponse
13, // 35: filer_pb.AppendToEntryRequest.chunks:type_name -> filer_pb.FileChunk
59, // 36: filer_pb.DeleteEntryResponse.metadata_event:type_name -> filer_pb.SubscribeMetadataResponse
12, // 37: filer_pb.StreamRenameEntryResponse.event_notification:type_name -> filer_pb.EventNotification
45, // 38: filer_pb.AssignVolumeResponse.location:type_name -> filer_pb.Location
45, // 39: filer_pb.Locations.locations:type_name -> filer_pb.Location
98, // 40: filer_pb.LookupVolumeResponse.locations_map:type_name -> filer_pb.LookupVolumeResponse.LocationsMapEntry
47, // 41: filer_pb.CollectionListResponse.collections:type_name -> filer_pb.Collection
12, // 42: filer_pb.SubscribeMetadataResponse.event_notification:type_name -> filer_pb.EventNotification
59, // 43: filer_pb.SubscribeMetadataResponse.events:type_name -> filer_pb.SubscribeMetadataResponse
60, // 44: filer_pb.SubscribeMetadataResponse.log_file_refs:type_name -> filer_pb.LogFileChunkRef
13, // 45: filer_pb.LogFileChunkRef.chunks:type_name -> filer_pb.FileChunk
10, // 46: filer_pb.TraverseBfsMetadataResponse.entry:type_name -> filer_pb.Entry
99, // 47: filer_pb.LocateBrokerResponse.resources:type_name -> filer_pb.LocateBrokerResponse.Resource
100, // 48: filer_pb.FilerConf.locations:type_name -> filer_pb.FilerConf.PathConf
10, // 49: filer_pb.CacheRemoteObjectToLocalClusterResponse.entry:type_name -> filer_pb.Entry
59, // 50: filer_pb.CacheRemoteObjectToLocalClusterResponse.metadata_event:type_name -> filer_pb.SubscribeMetadataResponse
81, // 51: filer_pb.TransferLocksRequest.locks:type_name -> filer_pb.Lock
17, // 52: filer_pb.StreamMutateEntryRequest.create_request:type_name -> filer_pb.CreateEntryRequest
29, // 53: filer_pb.StreamMutateEntryRequest.update_request:type_name -> filer_pb.UpdateEntryRequest
35, // 54: filer_pb.StreamMutateEntryRequest.delete_request:type_name -> filer_pb.DeleteEntryRequest
39, // 55: filer_pb.StreamMutateEntryRequest.rename_request:type_name -> filer_pb.StreamRenameEntryRequest
28, // 56: filer_pb.StreamMutateEntryResponse.create_response:type_name -> filer_pb.CreateEntryResponse
30, // 57: filer_pb.StreamMutateEntryResponse.update_response:type_name -> filer_pb.UpdateEntryResponse
36, // 58: filer_pb.StreamMutateEntryResponse.delete_response:type_name -> filer_pb.DeleteEntryResponse
40, // 59: filer_pb.StreamMutateEntryResponse.rename_response:type_name -> filer_pb.StreamRenameEntryResponse
92, // 60: filer_pb.MountListResponse.mounts:type_name -> filer_pb.MountInfo
3, // 61: filer_pb.WriteCondition.Clause.kind:type_name -> filer_pb.WriteCondition.Kind
44, // 62: filer_pb.LookupVolumeResponse.LocationsMapEntry.value:type_name -> filer_pb.Locations
5, // 63: filer_pb.SeaweedFiler.LookupDirectoryEntry:input_type -> filer_pb.LookupDirectoryEntryRequest
7, // 64: filer_pb.SeaweedFiler.ListEntries:input_type -> filer_pb.ListEntriesRequest
17, // 65: filer_pb.SeaweedFiler.CreateEntry:input_type -> filer_pb.CreateEntryRequest
29, // 66: filer_pb.SeaweedFiler.UpdateEntry:input_type -> filer_pb.UpdateEntryRequest
31, // 67: filer_pb.SeaweedFiler.TouchAccessTime:input_type -> filer_pb.TouchAccessTimeRequest
33, // 68: filer_pb.SeaweedFiler.AppendToEntry:input_type -> filer_pb.AppendToEntryRequest
35, // 69: filer_pb.SeaweedFiler.DeleteEntry:input_type -> filer_pb.DeleteEntryRequest
21, // 70: filer_pb.SeaweedFiler.ObjectTransaction:input_type -> filer_pb.ObjectTransactionRequest
26, // 71: filer_pb.SeaweedFiler.ObjectTransactionBatch:input_type -> filer_pb.ObjectTransactionBatchRequest
24, // 72: filer_pb.SeaweedFiler.PosixLock:input_type -> filer_pb.PosixLockRequest
37, // 73: filer_pb.SeaweedFiler.AtomicRenameEntry:input_type -> filer_pb.AtomicRenameEntryRequest
39, // 74: filer_pb.SeaweedFiler.StreamRenameEntry:input_type -> filer_pb.StreamRenameEntryRequest
86, // 75: filer_pb.SeaweedFiler.StreamMutateEntry:input_type -> filer_pb.StreamMutateEntryRequest
41, // 76: filer_pb.SeaweedFiler.AssignVolume:input_type -> filer_pb.AssignVolumeRequest
43, // 77: filer_pb.SeaweedFiler.LookupVolume:input_type -> filer_pb.LookupVolumeRequest
48, // 78: filer_pb.SeaweedFiler.CollectionList:input_type -> filer_pb.CollectionListRequest
50, // 79: filer_pb.SeaweedFiler.DeleteCollection:input_type -> filer_pb.DeleteCollectionRequest
52, // 80: filer_pb.SeaweedFiler.Statistics:input_type -> filer_pb.StatisticsRequest
54, // 81: filer_pb.SeaweedFiler.Ping:input_type -> filer_pb.PingRequest
56, // 82: filer_pb.SeaweedFiler.GetFilerConfiguration:input_type -> filer_pb.GetFilerConfigurationRequest
61, // 83: filer_pb.SeaweedFiler.TraverseBfsMetadata:input_type -> filer_pb.TraverseBfsMetadataRequest
58, // 84: filer_pb.SeaweedFiler.SubscribeMetadata:input_type -> filer_pb.SubscribeMetadataRequest
58, // 85: filer_pb.SeaweedFiler.SubscribeLocalMetadata:input_type -> filer_pb.SubscribeMetadataRequest
68, // 86: filer_pb.SeaweedFiler.KvGet:input_type -> filer_pb.KvGetRequest
70, // 87: filer_pb.SeaweedFiler.KvPut:input_type -> filer_pb.KvPutRequest
73, // 88: filer_pb.SeaweedFiler.CacheRemoteObjectToLocalCluster:input_type -> filer_pb.CacheRemoteObjectToLocalClusterRequest
75, // 89: filer_pb.SeaweedFiler.DistributedLock:input_type -> filer_pb.LockRequest
77, // 90: filer_pb.SeaweedFiler.DistributedUnlock:input_type -> filer_pb.UnlockRequest
79, // 91: filer_pb.SeaweedFiler.FindLockOwner:input_type -> filer_pb.FindLockOwnerRequest
82, // 92: filer_pb.SeaweedFiler.TransferLocks:input_type -> filer_pb.TransferLocksRequest
84, // 93: filer_pb.SeaweedFiler.ReplicateLock:input_type -> filer_pb.ReplicateLockRequest
88, // 94: filer_pb.SeaweedFiler.MountRegister:input_type -> filer_pb.MountRegisterRequest
90, // 95: filer_pb.SeaweedFiler.MountList:input_type -> filer_pb.MountListRequest
6, // 96: filer_pb.SeaweedFiler.LookupDirectoryEntry:output_type -> filer_pb.LookupDirectoryEntryResponse
8, // 97: filer_pb.SeaweedFiler.ListEntries:output_type -> filer_pb.ListEntriesResponse
28, // 98: filer_pb.SeaweedFiler.CreateEntry:output_type -> filer_pb.CreateEntryResponse
30, // 99: filer_pb.SeaweedFiler.UpdateEntry:output_type -> filer_pb.UpdateEntryResponse
32, // 100: filer_pb.SeaweedFiler.TouchAccessTime:output_type -> filer_pb.TouchAccessTimeResponse
34, // 101: filer_pb.SeaweedFiler.AppendToEntry:output_type -> filer_pb.AppendToEntryResponse
36, // 102: filer_pb.SeaweedFiler.DeleteEntry:output_type -> filer_pb.DeleteEntryResponse
22, // 103: filer_pb.SeaweedFiler.ObjectTransaction:output_type -> filer_pb.ObjectTransactionResponse
27, // 104: filer_pb.SeaweedFiler.ObjectTransactionBatch:output_type -> filer_pb.ObjectTransactionBatchResponse
25, // 105: filer_pb.SeaweedFiler.PosixLock:output_type -> filer_pb.PosixLockResponse
38, // 106: filer_pb.SeaweedFiler.AtomicRenameEntry:output_type -> filer_pb.AtomicRenameEntryResponse
40, // 107: filer_pb.SeaweedFiler.StreamRenameEntry:output_type -> filer_pb.StreamRenameEntryResponse
87, // 108: filer_pb.SeaweedFiler.StreamMutateEntry:output_type -> filer_pb.StreamMutateEntryResponse
42, // 109: filer_pb.SeaweedFiler.AssignVolume:output_type -> filer_pb.AssignVolumeResponse
46, // 110: filer_pb.SeaweedFiler.LookupVolume:output_type -> filer_pb.LookupVolumeResponse
49, // 111: filer_pb.SeaweedFiler.CollectionList:output_type -> filer_pb.CollectionListResponse
51, // 112: filer_pb.SeaweedFiler.DeleteCollection:output_type -> filer_pb.DeleteCollectionResponse
53, // 113: filer_pb.SeaweedFiler.Statistics:output_type -> filer_pb.StatisticsResponse
55, // 114: filer_pb.SeaweedFiler.Ping:output_type -> filer_pb.PingResponse
57, // 115: filer_pb.SeaweedFiler.GetFilerConfiguration:output_type -> filer_pb.GetFilerConfigurationResponse
62, // 116: filer_pb.SeaweedFiler.TraverseBfsMetadata:output_type -> filer_pb.TraverseBfsMetadataResponse
59, // 117: filer_pb.SeaweedFiler.SubscribeMetadata:output_type -> filer_pb.SubscribeMetadataResponse
59, // 118: filer_pb.SeaweedFiler.SubscribeLocalMetadata:output_type -> filer_pb.SubscribeMetadataResponse
69, // 119: filer_pb.SeaweedFiler.KvGet:output_type -> filer_pb.KvGetResponse
71, // 120: filer_pb.SeaweedFiler.KvPut:output_type -> filer_pb.KvPutResponse
74, // 121: filer_pb.SeaweedFiler.CacheRemoteObjectToLocalCluster:output_type -> filer_pb.CacheRemoteObjectToLocalClusterResponse
76, // 122: filer_pb.SeaweedFiler.DistributedLock:output_type -> filer_pb.LockResponse
78, // 123: filer_pb.SeaweedFiler.DistributedUnlock:output_type -> filer_pb.UnlockResponse
80, // 124: filer_pb.SeaweedFiler.FindLockOwner:output_type -> filer_pb.FindLockOwnerResponse
83, // 125: filer_pb.SeaweedFiler.TransferLocks:output_type -> filer_pb.TransferLocksResponse
85, // 126: filer_pb.SeaweedFiler.ReplicateLock:output_type -> filer_pb.ReplicateLockResponse
89, // 127: filer_pb.SeaweedFiler.MountRegister:output_type -> filer_pb.MountRegisterResponse
91, // 128: filer_pb.SeaweedFiler.MountList:output_type -> filer_pb.MountListResponse
96, // [96:129] is the sub-list for method output_type
63, // [63:96] is the sub-list for method input_type
63, // [63:63] is the sub-list for extension type_name
63, // [63:63] is the sub-list for extension extendee
0, // [0:63] is the sub-list for field type_name
}
func init() { file_filer_proto_init() }

28
weed/server/filer_grpc_server_posix_lock.go
View File

@@ -61,6 +61,7 @@ func (fs *FilerServer) PosixLock(ctx context.Context, req *filer_pb.PosixLockReq
IsMoved: true,
Op: req.Op,
Lock: req.Lock,
Locks: req.Locks,
}
glog.V(4).InfofCtx(ctx, "PosixLock %s op=%v: forwarding to owner %s", req.Key, req.Op, owner)
var resp *filer_pb.PosixLockResponse
@@ -107,13 +108,38 @@ func (fs *FilerServer) PosixLock(ctx context.Context, req *filer_pb.PosixLockReq
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)
// A re-assertion carries the mount's held locks on this key so the owner
// can rebuild its in-memory state after an ownership change or restart; a
// bare keepalive just renews the lease.
if len(req.Locks) > 0 {
held := make([]posixlock.Range, 0, len(req.Locks))
for _, l := range req.Locks {
held = append(held, posixRangeFromPb(l))
}
if conflicts := fs.posixLocks.Reassert(req.Key, lk.Sid, held); len(conflicts) > 0 {
glog.Warningf("posix reassert %s sid %d: %d lock(s) lost to another session: %+v", req.Key, lk.Sid, len(conflicts), conflicts)
}
} else {
fs.posixLocks.Renew(lk.Sid)
}
default:
return &filer_pb.PosixLockResponse{}, fmt.Errorf("unknown posix lock op %v", req.Op)
}
return resp, nil
}
func posixRangeFromPb(l *filer_pb.PosixLockRange) posixlock.Range {
return posixlock.Range{
Start: l.GetStart(),
End: l.GetEnd(),
Type: l.GetType(),
Sid: l.GetSid(),
Owner: l.GetOwner(),
Pid: l.GetPid(),
IsFlock: l.GetIsFlock(),
}
}
func posixRangeToPb(r posixlock.Range) *filer_pb.PosixLockRange {
return &filer_pb.PosixLockRange{
Start: r.Start,