Mutexmap

concurrency/atomicmap.go

@@ -0,0 +1,121 @@
+/*
+Copyright 2024 The Dapr Authors
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+    http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package concurrency
+
+import (
+	"sync"
+
+	"golang.org/x/exp/constraints"
+)
+
+type AtomicValue[T constraints.Integer] struct {
+	lock  sync.RWMutex
+	value T
+}
+
+func (a *AtomicValue[T]) Load() T {
+	a.lock.RLock()
+	defer a.lock.RUnlock()
+	return a.value
+}
+
+func (a *AtomicValue[T]) Store(v T) {
+	a.lock.Lock()
+	defer a.lock.Unlock()
+	a.value = v
+}
+
+func (a *AtomicValue[T]) Add(v T) T {
+	a.lock.Lock()
+	defer a.lock.Unlock()
+	a.value += v
+	return a.value
+}
+
+type AtomicMap[K comparable, T constraints.Integer] struct {
+	lock  sync.RWMutex
+	items map[K]*AtomicValue[T]
+}
+
+func NewAtomicMapStringInt64() *AtomicMap[string, int64] {
+	return &AtomicMap[string, int64]{
+		items: make(map[string]*AtomicValue[int64]),
+	}
+}
+
+func NewAtomicMapStringInt32() *AtomicMap[string, int32] {
+	return &AtomicMap[string, int32]{
+		items: make(map[string]*AtomicValue[int32]),
+	}
+}
+
+func NewAtomicMapStringUint64() *AtomicMap[string, uint64] {
+	return &AtomicMap[string, uint64]{
+		items: make(map[string]*AtomicValue[uint64]),
+	}
+}
+
+func NewAtomicMapStringUint32() *AtomicMap[string, uint32] {
+	return &AtomicMap[string, uint32]{
+		items: make(map[string]*AtomicValue[uint32]),
+	}
+}
+
+func (a *AtomicMap[K, T]) Get(key K) (*AtomicValue[T], bool) {
+	a.lock.RLock()
+	defer a.lock.RUnlock()
+
+	item, ok := a.items[key]
+	if !ok {
+		return nil, false
+	}
+	return item, true
+}
+
+func (a *AtomicMap[K, T]) GetOrCreate(key K, createT T) *AtomicValue[T] {
+	a.lock.RLock()
+	item, ok := a.items[key]
+	a.lock.RUnlock()
+	if !ok {
+		a.lock.Lock()
+		// Double-check the key exists to avoid race condition
+		item, ok = a.items[key]
+		if !ok {
+			item = &AtomicValue[T]{value: createT}
+			a.items[key] = item
+		}
+		a.lock.Unlock()
+	}
+	return item
+}
+
+func (a *AtomicMap[K, T]) Delete(key K) {
+	a.lock.Lock()
+	delete(a.items, key)
+	a.lock.Unlock()
+}
+
+func (a *AtomicMap[K, T]) ForEach(fn func(key K, value *AtomicValue[T])) {
+	a.lock.RLock()
+	defer a.lock.RUnlock()
+	for k, v := range a.items {
+		fn(k, v)
+	}
+}
+
+func (a *AtomicMap[K, T]) Clear() {
+	a.lock.Lock()
+	defer a.lock.Unlock()
+	a.items = make(map[K]*AtomicValue[T])
+}

concurrency/atomicmap_test.go

@@ -0,0 +1,78 @@
+/*
+Copyright 2024 The Dapr Authors
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+    http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package concurrency
+
+import (
+	"sync"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestAtomicMapInt32_New_Get_Delete(t *testing.T) {
+	m := NewAtomicMapStringInt32()
+	require.NotNil(t, m)
+	require.NotNil(t, m.items)
+	require.Empty(t, m.items)
+
+	t.Run("basic operations", func(t *testing.T) {
+		key := "key1"
+		value := int32(10)
+
+		// Initially, the key should not exist
+		_, ok := m.Get(key)
+		require.False(t, ok)
+
+		// Add a value and check it
+		m.GetOrCreate(key, 0).Store(value)
+		result, ok := m.Get(key)
+		require.True(t, ok)
+		assert.Equal(t, value, result.Load())
+
+		// Delete the key and check it no longer exists
+		m.Delete(key)
+		_, ok = m.Get(key)
+		require.False(t, ok)
+	})
+
+	t.Run("concurrent access multiple keys", func(t *testing.T) {
+		var wg sync.WaitGroup
+		keys := []string{"key1", "key2", "key3"}
+		iterations := 100
+
+		wg.Add(len(keys) * 2)
+		for _, key := range keys {
+			go func(k string) {
+				defer wg.Done()
+				for i := 0; i < iterations; i++ {
+					m.GetOrCreate(k, 0).Add(1)
+				}
+			}(key)
+			go func(k string) {
+				defer wg.Done()
+				for i := 0; i < iterations; i++ {
+					m.GetOrCreate(k, 0).Add(-1)
+				}
+			}(key)
+		}
+		wg.Wait()
+
+		for _, key := range keys {
+			val, ok := m.Get(key)
+			require.True(t, ok)
+			require.Equal(t, int32(0), val.Load())
+		}
+	})
+}

concurrency/mutexmap.go

@@ -0,0 +1,91 @@
+/*
+Copyright 2024 The Dapr Authors
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+    http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package concurrency
+
+import (
+	"sync"
+)
+
+type MutexMap[T comparable] struct {
+	lock  sync.RWMutex
+	items map[T]*sync.RWMutex
+}
+
+func NewMutexMapString() *MutexMap[string] {
+	return &MutexMap[string]{
+		items: make(map[string]*sync.RWMutex),
+	}
+}
+
+func (a *MutexMap[T]) Lock(key T) {
+	a.lock.RLock()
+	mutex, ok := a.items[key]
+	a.lock.RUnlock()
+	if !ok {
+		a.lock.Lock()
+		mutex, ok = a.items[key]
+		if !ok {
+			mutex = &sync.RWMutex{}
+			a.items[key] = mutex
+		}
+		a.lock.Unlock()
+	}
+	mutex.Lock()
+}
+
+func (a *MutexMap[T]) Unlock(key T) {
+	a.lock.RLock()
+	mutex, ok := a.items[key]
+	a.lock.RUnlock()
+	if ok {
+		mutex.Unlock()
+	}
+}
+
+func (a *MutexMap[T]) RLock(key T) {
+	a.lock.RLock()
+	mutex, ok := a.items[key]
+	a.lock.RUnlock()
+	if !ok {
+		a.lock.Lock()
+		mutex, ok = a.items[key]
+		if !ok {
+			mutex = &sync.RWMutex{}
+			a.items[key] = mutex
+		}
+		a.lock.Unlock()
+	}
+	mutex.Lock()
+}
+
+func (a *MutexMap[T]) RUnlock(key T) {
+	a.lock.RLock()
+	mutex, ok := a.items[key]
+	a.lock.RUnlock()
+	if ok {
+		mutex.Unlock()
+	}
+}
+
+func (a *MutexMap[T]) Delete(key T) {
+	a.lock.Lock()
+	delete(a.items, key)
+	a.lock.Unlock()
+}
+
+func (a *MutexMap[T]) Clear() {
+	a.lock.Lock()
+	a.items = make(map[T]*sync.RWMutex)
+	a.lock.Unlock()
+}

concurrency/mutexmap_test.go

@@ -0,0 +1,104 @@
+/*
+Copyright 2024 The Dapr Authors
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+    http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package concurrency
+
+import (
+	"sync"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestNewMutexMap_Add_Delete(t *testing.T) {
+	mm := NewMutexMapString()
+
+	t.Run("New mutex map", func(t *testing.T) {
+		require.NotNil(t, mm)
+		require.NotNil(t, mm.items)
+		require.Empty(t, mm.items)
+	})
+
+	t.Run("Lock and unlock mutex", func(t *testing.T) {
+		mm.Lock("key1")
+		_, ok := mm.items["key1"]
+		require.True(t, ok)
+		mm.Unlock("key1")
+	})
+
+	t.Run("Concurrently lock and unlock mutexes", func(t *testing.T) {
+		var counter int
+		var wg sync.WaitGroup
+
+		numGoroutines := 10
+		wg.Add(numGoroutines)
+
+		// Concurrently lock and unlock for each key
+		for i := 0; i < numGoroutines; i++ {
+			go func() {
+				defer wg.Done()
+				mm.Lock("key1")
+				counter++
+				mm.Unlock("key1")
+			}()
+		}
+		wg.Wait()
+
+		require.Equal(t, 10, counter)
+	})
+
+	t.Run("RLock and RUnlock mutex", func(t *testing.T) {
+		mm.RLock("key1")
+		_, ok := mm.items["key1"]
+		require.True(t, ok)
+		mm.RUnlock("key1")
+	})
+
+	t.Run("Concurrently RLock and RUnlock mutexes", func(t *testing.T) {
+		var counter int
+		var wg sync.WaitGroup
+
+		numGoroutines := 10
+		wg.Add(numGoroutines)
+
+		// Concurrently RLock and RUnlock for each key
+		for i := 0; i < numGoroutines; i++ {
+			go func() {
+				defer wg.Done()
+				mm.RLock("key1")
+				counter++
+				mm.RUnlock("key1")
+			}()
+		}
+		wg.Wait()
+
+		require.Equal(t, 10, counter)
+	})
+
+	t.Run("Delete mutex", func(t *testing.T) {
+		mm.Lock("key1")
+		mm.Unlock("key1")
+		mm.Delete("key1")
+		_, ok := mm.items["key1"]
+		require.False(t, ok)
+	})
+
+	t.Run("Clear all mutexes", func(t *testing.T) {
+		mm.Lock("key1")
+		mm.Unlock("key1")
+		mm.Lock("key2")
+		mm.Unlock("key2")
+		mm.Clear()
+		require.Empty(t, mm.items)
+	})
+}