1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
|
package exclusive_locks
import (
"context"
"sync/atomic"
"time"
"github.com/chrislusf/seaweedfs/weed/glog"
"github.com/chrislusf/seaweedfs/weed/pb/master_pb"
"github.com/chrislusf/seaweedfs/weed/wdclient"
)
const (
RenewInteval = 4 * time.Second
SafeRenewInteval = 3 * time.Second
InitLockInteval = 1 * time.Second
)
type ExclusiveLocker struct {
token int64
lockTsNs int64
isLocking bool
masterClient *wdclient.MasterClient
lockName string
message string
}
func NewExclusiveLocker(masterClient *wdclient.MasterClient, lockName string) *ExclusiveLocker {
return &ExclusiveLocker{
masterClient: masterClient,
lockName: lockName,
}
}
func (l *ExclusiveLocker) IsLocking() bool {
return l.isLocking
}
func (l *ExclusiveLocker) GetToken() (token int64, lockTsNs int64) {
for time.Unix(0, atomic.LoadInt64(&l.lockTsNs)).Add(SafeRenewInteval).Before(time.Now()) {
// wait until now is within the safe lock period, no immediate renewal to change the token
time.Sleep(100 * time.Millisecond)
}
return atomic.LoadInt64(&l.token), atomic.LoadInt64(&l.lockTsNs)
}
func (l *ExclusiveLocker) RequestLock(clientName string) {
if l.isLocking {
return
}
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
// retry to get the lease
for {
if err := l.masterClient.WithClient(false, func(client master_pb.SeaweedClient) error {
resp, err := client.LeaseAdminToken(ctx, &master_pb.LeaseAdminTokenRequest{
PreviousToken: atomic.LoadInt64(&l.token),
PreviousLockTime: atomic.LoadInt64(&l.lockTsNs),
LockName: l.lockName,
ClientName: clientName,
})
if err == nil {
atomic.StoreInt64(&l.token, resp.Token)
atomic.StoreInt64(&l.lockTsNs, resp.LockTsNs)
}
return err
}); err != nil {
println("lock:", err.Error())
time.Sleep(InitLockInteval)
} else {
break
}
}
l.isLocking = true
// start a goroutine to renew the lease
go func() {
ctx2, cancel2 := context.WithCancel(context.Background())
defer cancel2()
for l.isLocking {
if err := l.masterClient.WithClient(false, func(client master_pb.SeaweedClient) error {
resp, err := client.LeaseAdminToken(ctx2, &master_pb.LeaseAdminTokenRequest{
PreviousToken: atomic.LoadInt64(&l.token),
PreviousLockTime: atomic.LoadInt64(&l.lockTsNs),
LockName: l.lockName,
ClientName: clientName,
Message: l.message,
})
if err == nil {
atomic.StoreInt64(&l.token, resp.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)
return
} else {
time.Sleep(RenewInteval)
}
}
}()
}
func (l *ExclusiveLocker) ReleaseLock() {
l.isLocking = false
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
l.masterClient.WithClient(false, func(client master_pb.SeaweedClient) error {
client.ReleaseAdminToken(ctx, &master_pb.ReleaseAdminTokenRequest{
PreviousToken: atomic.LoadInt64(&l.token),
PreviousLockTime: atomic.LoadInt64(&l.lockTsNs),
LockName: l.lockName,
})
return nil
})
atomic.StoreInt64(&l.token, 0)
atomic.StoreInt64(&l.lockTsNs, 0)
}
func (l *ExclusiveLocker) SetMessage(message string) {
l.message = message
}
|
