mirror/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2024-11-23 18:49:17 +08:00
Code Issues Packages Projects Releases Wiki Activity

ExclusiveLocker only create one renew goroutine (#6269)

Browse Source 
Co-authored-by: liguowei <liguowei@xinye.com>
This commit is contained in:
Numblgw 2024-11-22 00:27:12 +08:00 committed by GitHub
parent aebf3952b7
commit e56327e3b0
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 43 additions and 32 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

75
weed/wdclient/exclusive_locks/exclusive_locker.go
View File

@ -19,10 +19,13 @@ const (
type ExclusiveLocker struct { type ExclusiveLocker struct {
token int64 token int64
lockTsNs int64 lockTsNs int64
isLocked bool isLocked atomic.Bool
masterClient *wdclient.MasterClient masterClient *wdclient.MasterClient
lockName string lockName string
message string message string
clientName string
// Each lock has and only has one goroutine
renewGoroutineRunning atomic.Bool
} }
func NewExclusiveLocker(masterClient *wdclient.MasterClient, lockName string) *ExclusiveLocker { func NewExclusiveLocker(masterClient *wdclient.MasterClient, lockName string) *ExclusiveLocker {
@ -33,7 +36,7 @@ func NewExclusiveLocker(masterClient *wdclient.MasterClient, lockName string) *E
} }
func (l *ExclusiveLocker) IsLocked() bool { func (l *ExclusiveLocker) IsLocked() bool {
return l.isLocked return l.isLocked.Load()
} }
func (l *ExclusiveLocker) GetToken() (token int64, lockTsNs int64) { func (l *ExclusiveLocker) GetToken() (token int64, lockTsNs int64) {
@ -45,7 +48,7 @@ func (l *ExclusiveLocker) GetToken() (token int64, lockTsNs int64) {
} }
func (l *ExclusiveLocker) RequestLock(clientName string) { func (l *ExclusiveLocker) RequestLock(clientName string) {
if l.isLocked { if l.isLocked.Load() {
return return
} }
@ -74,43 +77,51 @@ func (l *ExclusiveLocker) RequestLock(clientName string) {
} }
} }
l.isLocked = true l.isLocked.Store(true)
l.clientName = clientName
// start a goroutine to renew the lease // Each lock has and only has one goroutine
go func() { if l.renewGoroutineRunning.CompareAndSwap(false, true) {
ctx2, cancel2 := context.WithCancel(context.Background()) // start a goroutine to renew the lease
defer cancel2() go func() {
ctx2, cancel2 := context.WithCancel(context.Background())
defer cancel2()
for l.isLocked { for {
if err := l.masterClient.WithClient(false, func(client master_pb.SeaweedClient) error { if l.isLocked.Load() {
resp, err := client.LeaseAdminToken(ctx2, &master_pb.LeaseAdminTokenRequest{ if err := l.masterClient.WithClient(false, func(client master_pb.SeaweedClient) error {
PreviousToken: atomic.LoadInt64(&l.token), resp, err := client.LeaseAdminToken(ctx2, &master_pb.LeaseAdminTokenRequest{
PreviousLockTime: atomic.LoadInt64(&l.lockTsNs), PreviousToken: atomic.LoadInt64(&l.token),
LockName: l.lockName, PreviousLockTime: atomic.LoadInt64(&l.lockTsNs),
ClientName: clientName, LockName: l.lockName,
Message: l.message, ClientName: l.clientName,
}) Message: l.message,
if err == nil { })
atomic.StoreInt64(&l.token, resp.Token) if err == nil {
atomic.StoreInt64(&l.lockTsNs, resp.LockTsNs) atomic.StoreInt64(&l.token, resp.Token)
// println("ts", l.lockTsNs, "token", l.token) atomic.StoreInt64(&l.lockTsNs, resp.LockTsNs)
// println("ts", l.lockTsNs, "token", l.token)
}
return err
}); err != nil {
glog.Errorf("failed to renew lock: %v", err)
l.isLocked.Store(false)
return
} else {
time.Sleep(RenewInterval)
}
} else {
time.Sleep(RenewInterval)
} }
return err
}); err != nil {
glog.Errorf("failed to renew lock: %v", err)
l.isLocked = false
return
} else {
time.Sleep(RenewInterval)
} }
}()
} }
}()
} }
func (l *ExclusiveLocker) ReleaseLock() { func (l *ExclusiveLocker) ReleaseLock() {
l.isLocked = false l.isLocked.Store(false)
l.clientName = ""
ctx, cancel := context.WithCancel(context.Background()) ctx, cancel := context.WithCancel(context.Background())
defer cancel() defer cancel()