From f8caaa44645ba8b4ce456e615eeef44e7f12f0f3 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Mon, 25 May 2026 01:02:45 -0700 Subject: [PATCH] mount,filer: re-assert POSIX locks via keepalive (ownership migration + restart) (#9668) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * 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. --- weed/filer/posixlock/manager.go | 63 +++++ weed/filer/posixlock/posixlock.go | 6 + weed/filer/posixlock/reassert_test.go | 82 +++++++ weed/mount/weedfs.go | 10 +- weed/mount/weedfs_posix_lock_routed.go | 87 ++++++- weed/pb/filer.proto | 4 + weed/pb/filer_pb/filer.pb.go | 241 +++++++++++--------- weed/server/filer_grpc_server_posix_lock.go | 28 ++- 8 files changed, 398 insertions(+), 123 deletions(-) create mode 100644 weed/filer/posixlock/reassert_test.go diff --git a/weed/filer/posixlock/manager.go b/weed/filer/posixlock/manager.go index 297cf2f96..8fb56ce53 100644 --- a/weed/filer/posixlock/manager.go +++ b/weed/filer/posixlock/manager.go @@ -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() diff --git a/weed/filer/posixlock/posixlock.go b/weed/filer/posixlock/posixlock.go index 98002455a..0fdac08a7 100644 --- a/weed/filer/posixlock/posixlock.go +++ b/weed/filer/posixlock/posixlock.go @@ -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). diff --git a/weed/filer/posixlock/reassert_test.go b/weed/filer/posixlock/reassert_test.go new file mode 100644 index 000000000..948c47b0d --- /dev/null +++ b/weed/filer/posixlock/reassert_test.go @@ -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) diff --git a/weed/mount/weedfs.go b/weed/mount/weedfs.go index 164f70d97..b560735c6 100644 --- a/weed/mount/weedfs.go +++ b/weed/mount/weedfs.go @@ -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 } diff --git a/weed/mount/weedfs_posix_lock_routed.go b/weed/mount/weedfs_posix_lock_routed.go index 279d7dbc2..489d6721f 100644 --- a/weed/mount/weedfs_posix_lock_routed.go +++ b/weed/mount/weedfs_posix_lock_routed.go @@ -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() + } +} diff --git a/weed/pb/filer.proto b/weed/pb/filer.proto index b2ae87131..0bcb44185 100644 --- a/weed/pb/filer.proto +++ b/weed/pb/filer.proto @@ -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 { diff --git a/weed/pb/filer_pb/filer.pb.go b/weed/pb/filer_pb/filer.pb.go index cde52c5a9..77373db6d 100644 --- a/weed/pb/filer_pb/filer.pb.go +++ b/weed/pb/filer_pb/filer.pb.go @@ -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() } diff --git a/weed/server/filer_grpc_server_posix_lock.go b/weed/server/filer_grpc_server_posix_lock.go index 4abe102fc..e01377774 100644 --- a/weed/server/filer_grpc_server_posix_lock.go +++ b/weed/server/filer_grpc_server_posix_lock.go @@ -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,