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
|
package util
import (
"fmt"
"github.com/seaweedfs/seaweedfs/weed/glog"
"sync"
"sync/atomic"
)
// LockTable is a table of locks that can be acquired.
// Locks are acquired in order of request.
type LockTable[T comparable] struct {
mu sync.Mutex
locks map[T]*LockEntry
lockIdSeq int64
}
type LockEntry struct {
mu sync.Mutex
waiters []*ActiveLock // ordered waiters that are blocked by exclusive locks
activeLockOwnerCount int32
lockType LockType
cond *sync.Cond
}
type LockType int
const (
SharedLock LockType = iota
ExclusiveLock
)
type ActiveLock struct {
ID int64
isDeleted bool
intention string // for debugging
}
func NewLockTable[T comparable]() *LockTable[T] {
return &LockTable[T]{
locks: make(map[T]*LockEntry),
}
}
func (lt *LockTable[T]) NewActiveLock(intention string) *ActiveLock {
id := atomic.AddInt64(<.lockIdSeq, 1)
l := &ActiveLock{ID: id, intention: intention}
return l
}
func (lt *LockTable[T]) AcquireLock(intention string, key T, lockType LockType) (lock *ActiveLock) {
lt.mu.Lock()
// Get or create the lock entry for the key
entry, exists := lt.locks[key]
if !exists {
entry = &LockEntry{}
entry.cond = sync.NewCond(&entry.mu)
lt.locks[key] = entry
}
lt.mu.Unlock()
lock = lt.NewActiveLock(intention)
// If the lock is held exclusively, wait
entry.mu.Lock()
if len(entry.waiters) > 0 || lockType == ExclusiveLock {
if glog.V(4) {
fmt.Printf("ActiveLock %d %s wait for %+v type=%v with waiters %d active %d.\n", lock.ID, lock.intention, key, lockType, len(entry.waiters), entry.activeLockOwnerCount)
if len(entry.waiters) > 0 {
for _, waiter := range entry.waiters {
fmt.Printf(" %d", waiter.ID)
}
fmt.Printf("\n")
}
}
entry.waiters = append(entry.waiters, lock)
if lockType == ExclusiveLock {
for !lock.isDeleted && ((len(entry.waiters) > 0 && lock.ID != entry.waiters[0].ID) || entry.activeLockOwnerCount > 0) {
entry.cond.Wait()
}
} else {
for !lock.isDeleted && (len(entry.waiters) > 0 && lock.ID != entry.waiters[0].ID) {
entry.cond.Wait()
}
}
// Remove the transaction from the waiters list
if len(entry.waiters) > 0 && lock.ID == entry.waiters[0].ID {
entry.waiters = entry.waiters[1:]
entry.cond.Broadcast()
}
}
entry.activeLockOwnerCount++
// Otherwise, grant the lock
entry.lockType = lockType
if glog.V(4) {
fmt.Printf("ActiveLock %d %s locked %+v type=%v with waiters %d active %d.\n", lock.ID, lock.intention, key, lockType, len(entry.waiters), entry.activeLockOwnerCount)
if len(entry.waiters) > 0 {
for _, waiter := range entry.waiters {
fmt.Printf(" %d", waiter.ID)
}
fmt.Printf("\n")
}
}
entry.mu.Unlock()
return lock
}
func (lt *LockTable[T]) ReleaseLock(key T, lock *ActiveLock) {
lt.mu.Lock()
defer lt.mu.Unlock()
entry, exists := lt.locks[key]
if !exists {
return
}
entry.mu.Lock()
defer entry.mu.Unlock()
// Remove the transaction from the waiters list
for i, waiter := range entry.waiters {
if waiter == lock {
waiter.isDeleted = true
entry.waiters = append(entry.waiters[:i], entry.waiters[i+1:]...)
break
}
}
// If there are no waiters, release the lock
if len(entry.waiters) == 0 {
delete(lt.locks, key)
}
if glog.V(4) {
fmt.Printf("ActiveLock %d %s unlocked %+v type=%v with waiters %d active %d.\n", lock.ID, lock.intention, key, entry.lockType, len(entry.waiters), entry.activeLockOwnerCount)
if len(entry.waiters) > 0 {
for _, waiter := range entry.waiters {
fmt.Printf(" %d", waiter.ID)
}
fmt.Printf("\n")
}
}
entry.activeLockOwnerCount--
// Notify the next waiter
entry.cond.Broadcast()
}
func main() {
}
|
