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
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
|
package lock_manager
import (
"fmt"
"github.com/google/uuid"
"github.com/seaweedfs/seaweedfs/weed/glog"
"sync"
"time"
)
var LockErrorNonEmptyTokenOnNewLock = fmt.Errorf("lock: non-empty token on a new lock")
var LockErrorNonEmptyTokenOnExpiredLock = fmt.Errorf("lock: non-empty token on an expired lock")
var LockErrorTokenMismatch = fmt.Errorf("lock: token mismatch")
var UnlockErrorTokenMismatch = fmt.Errorf("unlock: token mismatch")
var LockNotFound = fmt.Errorf("lock not found")
// LockManager local lock manager, used by distributed lock manager
type LockManager struct {
locks map[string]*Lock
accessLock sync.RWMutex
}
type Lock struct {
Token string
ExpiredAtNs int64
Key string // only used for moving locks
Owner string
}
func NewLockManager() *LockManager {
t := &LockManager{
locks: make(map[string]*Lock),
}
go t.CleanUp()
return t
}
func (lm *LockManager) Lock(path string, expiredAtNs int64, token string, owner string) (lockOwner, renewToken string, err error) {
lm.accessLock.Lock()
defer lm.accessLock.Unlock()
glog.V(4).Infof("lock %s %v %v %v", path, time.Unix(0, expiredAtNs), token, owner)
if oldValue, found := lm.locks[path]; found {
if oldValue.ExpiredAtNs > 0 && oldValue.ExpiredAtNs < time.Now().UnixNano() {
// lock is expired, set to a new lock
if token != "" {
glog.V(4).Infof("lock expired key %s non-empty token %v owner %v ts %s", path, token, owner, time.Unix(0, oldValue.ExpiredAtNs))
err = LockErrorNonEmptyTokenOnExpiredLock
return
} else {
// new lock
renewToken = uuid.New().String()
glog.V(4).Infof("key %s new token %v owner %v", path, renewToken, owner)
lm.locks[path] = &Lock{Token: renewToken, ExpiredAtNs: expiredAtNs, Owner: owner}
return
}
}
// not expired
lockOwner = oldValue.Owner
if oldValue.Token == token {
// token matches, renew the lock
renewToken = uuid.New().String()
glog.V(4).Infof("key %s old token %v owner %v => %v owner %v", path, oldValue.Token, oldValue.Owner, renewToken, owner)
lm.locks[path] = &Lock{Token: renewToken, ExpiredAtNs: expiredAtNs, Owner: owner}
return
} else {
if token == "" {
// new lock
glog.V(4).Infof("key %s locked by %v", path, oldValue.Owner)
err = fmt.Errorf("lock already owned by %v", oldValue.Owner)
return
}
glog.V(4).Infof("key %s expected token %v owner %v received %v from %v", path, oldValue.Token, oldValue.Owner, token, owner)
err = fmt.Errorf("lock: token mismatch")
return
}
} else {
glog.V(4).Infof("key %s no lock owner %v", path, owner)
if token == "" {
// new lock
glog.V(4).Infof("key %s new token %v owner %v", path, token, owner)
renewToken = uuid.New().String()
lm.locks[path] = &Lock{Token: renewToken, ExpiredAtNs: expiredAtNs, Owner: owner}
return
} else {
glog.V(4).Infof("key %s non-empty token %v owner %v", path, token, owner)
err = LockErrorNonEmptyTokenOnNewLock
return
}
}
}
func (lm *LockManager) Unlock(path string, token string) (isUnlocked bool, err error) {
lm.accessLock.Lock()
defer lm.accessLock.Unlock()
if oldValue, found := lm.locks[path]; found {
now := time.Now()
if oldValue.ExpiredAtNs > 0 && oldValue.ExpiredAtNs < now.UnixNano() {
// lock is expired, delete it
isUnlocked = true
glog.V(4).Infof("key %s expired at %v", path, time.Unix(0, oldValue.ExpiredAtNs))
delete(lm.locks, path)
return
}
if oldValue.Token == token {
isUnlocked = true
glog.V(4).Infof("key %s unlocked with %v", path, token)
delete(lm.locks, path)
return
} else {
isUnlocked = false
err = UnlockErrorTokenMismatch
return
}
}
err = LockNotFound
return
}
func (lm *LockManager) CleanUp() {
for {
time.Sleep(1 * time.Minute)
now := time.Now().UnixNano()
lm.accessLock.Lock()
for key, value := range lm.locks {
if value == nil {
continue
}
if now > value.ExpiredAtNs {
glog.V(4).Infof("key %s expired at %v", key, time.Unix(0, value.ExpiredAtNs))
delete(lm.locks, key)
}
}
lm.accessLock.Unlock()
}
}
// SelectLocks takes out locks by key
// if keyFn return true, the lock will be taken out
func (lm *LockManager) SelectLocks(selectFn func(key string) bool) (locks []*Lock) {
lm.accessLock.RLock()
defer lm.accessLock.RUnlock()
now := time.Now().UnixNano()
for key, lock := range lm.locks {
if now > lock.ExpiredAtNs {
glog.V(4).Infof("key %s expired at %v", key, time.Unix(0, lock.ExpiredAtNs))
delete(lm.locks, key)
continue
}
if selectFn(key) {
glog.V(4).Infof("key %s selected and deleted", key)
delete(lm.locks, key)
lock.Key = key
locks = append(locks, lock)
}
}
return
}
// InsertLock inserts a lock unconditionally
func (lm *LockManager) InsertLock(path string, expiredAtNs int64, token string, owner string) {
lm.accessLock.Lock()
defer lm.accessLock.Unlock()
lm.locks[path] = &Lock{Token: token, ExpiredAtNs: expiredAtNs, Owner: owner}
}
func (lm *LockManager) GetLockOwner(key string) (owner string, err error) {
lm.accessLock.RLock()
defer lm.accessLock.RUnlock()
if lock, found := lm.locks[key]; found {
return lock.Owner, nil
}
err = LockNotFound
return
}
|
