adding concurrency test

Signed-off-by: Soule BA <[email protected]>

cache/cache.go

@@ -130,7 +130,7 @@ func (c *Cache[T]) Add(object T) error {
 }
 
 // Update adds an item to the cache, replacing any existing item.
-// If the cache is full, Set will return an error.
+// If the cache is full, it will return an error.
 func (c *Cache[T]) Update(object T) error {
 	key, err := c.keyFunc(object)
 	if err != nil {
@@ -197,6 +197,7 @@ func (c *Cache[T]) Get(object T) (item T, exists bool, err error) {
 	return item, true, nil
 }
 
+// GetByKey returns the object for the given key.
 func (c *Cache[T]) GetByKey(key string) (T, bool, error) {
 	var res T
 	items, found, err := c.get(key)

cache/cache_test.go

@@ -17,6 +17,9 @@ limitations under the License.
 package cache
 
 import (
+	"fmt"
+	"math/rand"
+	"sync"
 	"testing"
 	"time"
 
@@ -119,7 +122,7 @@ func TestCache(t *testing.T) {
 	t.Run("Add expiring keys", func(t *testing.T) {
 		g := NewWithT(t)
 		// new cache with a cleanup interval of 1 second
-		cache, err := New[IdentifiableObject](2, IdentifiableObjectKeyFunc,
+		cache, err := New(2, IdentifiableObjectKeyFunc,
 			WithCleanupInterval[IdentifiableObject](1*time.Second),
 			WithMetricsRegisterer[IdentifiableObject](prometheus.NewPedanticRegistry()))
 		g.Expect(err).ToNot(HaveOccurred())
@@ -164,3 +167,71 @@ func TestCache(t *testing.T) {
 		g.Expect(item.Object).To(BeNil())
 	})
 }
+
+func TestCache_Concurrent(t *testing.T) {
+	const (
+		concurrency = 500
+		keysNum     = 10
+	)
+	g := NewWithT(t)
+	// create a cache that can hold 10 items and have no cleanup
+	cache, err := New(10, IdentifiableObjectKeyFunc,
+		WithCleanupInterval[IdentifiableObject](1*time.Second),
+		WithMetricsRegisterer[IdentifiableObject](prometheus.NewPedanticRegistry()))
+	g.Expect(err).ToNot(HaveOccurred())
+
+	objmap := createObjectMap(keysNum)
+
+	wg := sync.WaitGroup{}
+	run := make(chan bool)
+
+	// simulate concurrent read and write
+	for i := 0; i < concurrency; i++ {
+		key := rand.Intn(keysNum)
+		wg.Add(2)
+		go func() {
+			defer wg.Done()
+			_ = cache.Add(objmap[key])
+		}()
+		go func() {
+			defer wg.Done()
+			<-run
+			_, _, _ = cache.Get(objmap[key])
+		}()
+	}
+	close(run)
+	wg.Wait()
+
+	keys := cache.ListKeys()
+	g.Expect(len(keys)).To(Equal(len(objmap)))
+
+	for _, obj := range objmap {
+		val, found, err := cache.Get(obj)
+		g.Expect(err).ToNot(HaveOccurred())
+		g.Expect(found).To(BeTrue(), "object %s not found", obj.Name)
+		g.Expect(val).To(Equal(obj))
+	}
+}
+
+func createObjectMap(num int) map[int]IdentifiableObject {
+	objMap := make(map[int]IdentifiableObject)
+	for i := 0; i < num; i++ {
+		obj := IdentifiableObject{
+			ObjMetadata: object.ObjMetadata{
+				Namespace: "test-ns",
+				Name:      fmt.Sprintf("test-%d", i),
+				GroupKind: schema.GroupKind{
+					Group: "test-group",
+					Kind:  "TestObject",
+				},
+			},
+			Object: struct {
+				token string
+			}{
+				token: "test-token",
+			},
+		}
+		objMap[i] = obj
+	}
+	return objMap
+}

cache/go.mod

@@ -39,6 +39,7 @@ require (
 	github.com/prometheus/client_model v0.5.0 // indirect
 	github.com/prometheus/common v0.48.0 // indirect
 	github.com/prometheus/procfs v0.12.0 // indirect
+	github.com/stretchr/objx v0.5.2 // indirect
 	github.com/xlab/treeprint v1.2.0 // indirect
 	go.starlark.net v0.0.0-20230525235612-a134d8f9ddca // indirect
 	golang.org/x/net v0.24.0 // indirect

cache/go.sum

@@ -155,8 +155,9 @@ github.com/spf13/pflag v1.0.5 h1:iy+VFUOCP1a+8yFto/drg2CJ5u0yRoB7fZw3DKv/JXA=
 github.com/spf13/pflag v1.0.5/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
-github.com/stretchr/objx v0.5.0 h1:1zr/of2m5FGMsad5YfcqgdqdWrIhu+EBEJRhR1U7z/c=
 github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo=
+github.com/stretchr/objx v0.5.2 h1:xuMeJ0Sdp5ZMRXx/aWO6RZxdr3beISkG5/G/aIRr3pY=
+github.com/stretchr/objx v0.5.2/go.mod h1:FRsXN1f5AsAjCGJKqEizvkpNtU+EGNCLh3NxZ/8L+MA=
 github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
 github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=

cache/lru_test.go

@@ -17,6 +17,8 @@ limitations under the License.
 package cache
 
 import (
+	"math/rand"
+	"sync"
 	"testing"
 
 	"github.com/fluxcd/cli-utils/pkg/object"
@@ -354,3 +356,47 @@ func Test_LRU_Delete(t *testing.T) {
 	g.Expect(err).ToNot(HaveOccurred())
 	g.Expect(cache.ListKeys()).To(BeEmpty())
 }
+
+func TestLRU_Concurrent(t *testing.T) {
+	const (
+		concurrency = 500
+		keysNum     = 10
+	)
+	g := NewWithT(t)
+	// create a cache that can hold 10 items and have no cleanup
+	cache, err := NewLRU(10, IdentifiableObjectKeyFunc,
+		WithMetricsRegisterer[IdentifiableObject](prometheus.NewPedanticRegistry()))
+	g.Expect(err).ToNot(HaveOccurred())
+
+	objmap := createObjectMap(keysNum)
+
+	wg := sync.WaitGroup{}
+	run := make(chan bool)
+
+	// simulate concurrent read and write
+	for i := 0; i < concurrency; i++ {
+		key := rand.Intn(keysNum)
+		wg.Add(2)
+		go func() {
+			defer wg.Done()
+			_ = cache.Add(objmap[key])
+		}()
+		go func() {
+			defer wg.Done()
+			<-run
+			_, _, _ = cache.Get(objmap[key])
+		}()
+	}
+	close(run)
+	wg.Wait()
+
+	keys := cache.ListKeys()
+	g.Expect(len(keys)).To(Equal(len(objmap)))
+
+	for _, obj := range objmap {
+		val, found, err := cache.Get(obj)
+		g.Expect(err).ToNot(HaveOccurred())
+		g.Expect(found).To(BeTrue(), "object %s not found", obj.Name)
+		g.Expect(val).To(Equal(obj))
+	}
+}