Initial commit.
This commit is contained in:
@@ -1,67 +1,69 @@
|
|||||||
package keyedmutex
|
package keyedmutex
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"container/list"
|
|
||||||
"sync"
|
"sync"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
type keyedLock struct {
|
||||||
|
lock sync.Mutex
|
||||||
|
count int
|
||||||
|
}
|
||||||
|
|
||||||
type KeyedMutex[K comparable] struct {
|
type KeyedMutex[K comparable] struct {
|
||||||
mu *sync.Mutex
|
mu sync.Mutex
|
||||||
waitList map[K]*list.List
|
byKey map[K]*keyedLock
|
||||||
}
|
}
|
||||||
|
|
||||||
func New[K comparable]() KeyedMutex[K] {
|
func New[K comparable]() *KeyedMutex[K] {
|
||||||
return KeyedMutex[K]{
|
return &KeyedMutex[K]{
|
||||||
mu: new(sync.Mutex),
|
byKey: map[K]*keyedLock{},
|
||||||
waitList: map[K]*list.List{},
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func (m KeyedMutex[K]) Lock(key K) {
|
func (m *KeyedMutex[K]) getLock(key K) *keyedLock {
|
||||||
if ch := m.lock(key); ch != nil {
|
|
||||||
<-ch
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func (m KeyedMutex[K]) lock(key K) chan struct{} {
|
|
||||||
m.mu.Lock()
|
m.mu.Lock()
|
||||||
defer m.mu.Unlock()
|
defer m.mu.Unlock()
|
||||||
|
|
||||||
if waitList, ok := m.waitList[key]; ok {
|
item, ok := m.byKey[key]
|
||||||
ch := make(chan struct{})
|
if !ok {
|
||||||
waitList.PushBack(ch)
|
item = &keyedLock{}
|
||||||
return ch
|
m.byKey[key] = item
|
||||||
|
}
|
||||||
|
item.count++
|
||||||
|
return item
|
||||||
}
|
}
|
||||||
|
|
||||||
m.waitList[key] = list.New()
|
func (m *KeyedMutex[K]) release(key K, unlock bool) {
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func (m KeyedMutex[K]) TryLock(key K) bool {
|
|
||||||
m.mu.Lock()
|
m.mu.Lock()
|
||||||
defer m.mu.Unlock()
|
defer m.mu.Unlock()
|
||||||
|
|
||||||
if _, ok := m.waitList[key]; ok {
|
item, ok := m.byKey[key]
|
||||||
return false
|
|
||||||
}
|
|
||||||
|
|
||||||
m.waitList[key] = list.New()
|
|
||||||
return true
|
|
||||||
}
|
|
||||||
|
|
||||||
func (m KeyedMutex[K]) Unlock(key K) {
|
|
||||||
m.mu.Lock()
|
|
||||||
defer m.mu.Unlock()
|
|
||||||
|
|
||||||
waitList, ok := m.waitList[key]
|
|
||||||
if !ok {
|
if !ok {
|
||||||
panic("unlock of unlocked mutex")
|
panic("unlock of unlocked mutex")
|
||||||
}
|
}
|
||||||
|
|
||||||
if waitList.Len() == 0 {
|
item.count--
|
||||||
delete(m.waitList, key)
|
if unlock {
|
||||||
} else {
|
item.lock.Unlock()
|
||||||
ch := waitList.Remove(waitList.Front()).(chan struct{})
|
}
|
||||||
ch <- struct{}{}
|
|
||||||
|
if item.count == 0 {
|
||||||
|
delete(m.byKey, key)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (m *KeyedMutex[K]) Lock(key K) {
|
||||||
|
m.getLock(key).lock.Lock()
|
||||||
|
}
|
||||||
|
|
||||||
|
func (m *KeyedMutex[K]) TryLock(key K) bool {
|
||||||
|
if ok := m.getLock(key).lock.TryLock(); !ok {
|
||||||
|
m.release(key, false)
|
||||||
|
return false
|
||||||
|
}
|
||||||
|
return true
|
||||||
|
}
|
||||||
|
|
||||||
|
func (m *KeyedMutex[K]) Unlock(key K) {
|
||||||
|
m.release(key, true)
|
||||||
|
}
|
||||||
|
|||||||
@@ -1,81 +1,92 @@
|
|||||||
package keyedmutex
|
package keyedmutex
|
||||||
|
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"sync"
|
"sync"
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
|
||||||
)
|
)
|
||||||
|
|
||||||
func TestKeyedMutex(t *testing.T) {
|
func TestLock_CleansUpMap(t *testing.T) {
|
||||||
checkState := func(t *testing.T, m KeyedMutex[string], keys ...string) {
|
|
||||||
if len(m.waitList) != len(keys) {
|
|
||||||
t.Fatal(m.waitList, keys)
|
|
||||||
}
|
|
||||||
|
|
||||||
for _, key := range keys {
|
|
||||||
if _, ok := m.waitList[key]; !ok {
|
|
||||||
t.Fatal(key)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
m := New[string]()
|
m := New[string]()
|
||||||
checkState(t, m)
|
|
||||||
|
|
||||||
m.Lock("a")
|
m.Lock("a")
|
||||||
checkState(t, m, "a")
|
|
||||||
m.Lock("b")
|
|
||||||
checkState(t, m, "a", "b")
|
|
||||||
m.Lock("c")
|
|
||||||
checkState(t, m, "a", "b", "c")
|
|
||||||
|
|
||||||
if m.TryLock("a") {
|
|
||||||
t.Fatal("a")
|
|
||||||
}
|
|
||||||
if m.TryLock("b") {
|
|
||||||
t.Fatal("b")
|
|
||||||
}
|
|
||||||
if m.TryLock("c") {
|
|
||||||
t.Fatal("c")
|
|
||||||
}
|
|
||||||
|
|
||||||
if !m.TryLock("d") {
|
|
||||||
t.Fatal("d")
|
|
||||||
}
|
|
||||||
|
|
||||||
checkState(t, m, "a", "b", "c", "d")
|
|
||||||
|
|
||||||
if !m.TryLock("e") {
|
|
||||||
t.Fatal("e")
|
|
||||||
}
|
|
||||||
checkState(t, m, "a", "b", "c", "d", "e")
|
|
||||||
|
|
||||||
m.Unlock("c")
|
|
||||||
checkState(t, m, "a", "b", "d", "e")
|
|
||||||
m.Unlock("a")
|
m.Unlock("a")
|
||||||
checkState(t, m, "b", "d", "e")
|
if len(m.byKey) != 0 {
|
||||||
m.Unlock("e")
|
t.Fatalf("expected empty byKey, got %v", m.byKey)
|
||||||
checkState(t, m, "b", "d")
|
}
|
||||||
|
}
|
||||||
|
|
||||||
wg := sync.WaitGroup{}
|
func TestTryLock_SucceedsOnFreeKey(t *testing.T) {
|
||||||
for i := 0; i < 8; i++ {
|
m := New[string]()
|
||||||
|
if !m.TryLock("a") {
|
||||||
|
t.Fatal("expected TryLock to succeed on free key")
|
||||||
|
}
|
||||||
|
m.Unlock("a")
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestTryLock_FailsOnLockedKey(t *testing.T) {
|
||||||
|
m := New[string]()
|
||||||
|
m.Lock("a")
|
||||||
|
if m.TryLock("a") {
|
||||||
|
t.Fatal("expected TryLock to fail on locked key")
|
||||||
|
}
|
||||||
|
m.Unlock("a")
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestTryLock_FailureDoesNotCorruptLock(t *testing.T) {
|
||||||
|
// A failed TryLock must not call Unlock on the key's mutex.
|
||||||
|
// The old bug did this unconditionally, so the Unlock below would
|
||||||
|
// double-unlock and panic.
|
||||||
|
m := New[string]()
|
||||||
|
m.Lock("a")
|
||||||
|
m.TryLock("a") // must return false and leave the lock intact
|
||||||
|
m.Unlock("a") // must not panic
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestTryLock_FailureDecrementsCount(t *testing.T) {
|
||||||
|
// A failed TryLock must undo its getLock increment so that the
|
||||||
|
// original holder's Unlock cleans up the map entry.
|
||||||
|
m := New[string]()
|
||||||
|
m.Lock("a")
|
||||||
|
m.TryLock("a") // fails; a leaked count would leave a stale map entry
|
||||||
|
m.Unlock("a")
|
||||||
|
if len(m.byKey) != 0 {
|
||||||
|
t.Fatalf("expected empty byKey after unlock, got %v", m.byKey)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestMultipleKeys_AreIndependent(t *testing.T) {
|
||||||
|
m := New[string]()
|
||||||
|
m.Lock("a")
|
||||||
|
if !m.TryLock("b") {
|
||||||
|
t.Fatal("expected TryLock on different key to succeed while 'a' is locked")
|
||||||
|
}
|
||||||
|
m.Unlock("b")
|
||||||
|
m.Unlock("a")
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestConcurrentLock_MutualExclusion(t *testing.T) {
|
||||||
|
m := New[string]()
|
||||||
|
const N = 100
|
||||||
|
var wg sync.WaitGroup
|
||||||
|
var shared int // intentionally non-atomic: race detector catches improper access
|
||||||
|
|
||||||
|
for range N {
|
||||||
wg.Add(1)
|
wg.Add(1)
|
||||||
go func() {
|
go func() {
|
||||||
defer wg.Done()
|
defer wg.Done()
|
||||||
m.Lock("b")
|
m.Lock("a")
|
||||||
m.Unlock("b")
|
shared++
|
||||||
|
m.Unlock("a")
|
||||||
}()
|
}()
|
||||||
}
|
}
|
||||||
|
|
||||||
time.Sleep(100 * time.Millisecond)
|
|
||||||
m.Unlock("b")
|
|
||||||
wg.Wait()
|
wg.Wait()
|
||||||
|
|
||||||
checkState(t, m, "d")
|
if shared != N {
|
||||||
|
t.Fatalf("expected %d, got %d", N, shared)
|
||||||
m.Unlock("d")
|
}
|
||||||
checkState(t, m)
|
if len(m.byKey) != 0 {
|
||||||
|
t.Fatalf("expected empty byKey after all goroutines done, got %v", m.byKey)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestKeyedMutex_unlockUnlocked(t *testing.T) {
|
func TestKeyedMutex_unlockUnlocked(t *testing.T) {
|
||||||
@@ -93,31 +104,8 @@ func BenchmarkUncontendedMutex(b *testing.B) {
|
|||||||
m := New[string]()
|
m := New[string]()
|
||||||
key := "xyz"
|
key := "xyz"
|
||||||
|
|
||||||
for i := 0; i < b.N; i++ {
|
for b.Loop() {
|
||||||
m.Lock(key)
|
m.Lock(key)
|
||||||
m.Unlock(key)
|
m.Unlock(key)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func BenchmarkContendedMutex(b *testing.B) {
|
|
||||||
m := New[string]()
|
|
||||||
key := "xyz"
|
|
||||||
|
|
||||||
m.Lock(key)
|
|
||||||
|
|
||||||
wg := sync.WaitGroup{}
|
|
||||||
for i := 0; i < b.N; i++ {
|
|
||||||
wg.Add(1)
|
|
||||||
go func() {
|
|
||||||
defer wg.Done()
|
|
||||||
m.Lock(key)
|
|
||||||
m.Unlock(key)
|
|
||||||
}()
|
|
||||||
}
|
|
||||||
|
|
||||||
time.Sleep(time.Second)
|
|
||||||
|
|
||||||
b.ResetTimer()
|
|
||||||
m.Unlock(key)
|
|
||||||
wg.Wait()
|
|
||||||
}
|
|
||||||
|
|||||||
Reference in New Issue
Block a user