Fix concurrency issues caused by Pipeline cache function (#128)

GuanceCloud · Dec 11, 2024 · ef1c166 · ef1c166
1 parent 2793865
commit ef1c166
Show file tree

Hide file tree

Showing 2 changed files with 32 additions and 0 deletions.
diff --git a/pipeline/ptinput/plcache/cache.go b/pipeline/ptinput/plcache/cache.go
@@ -134,6 +134,9 @@ func (c *Cache) getPosAndCircle(d time.Duration) (pos, circle int) {
 }
 
 func (c *Cache) Get(key string) (any, bool, error) {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+
 	select {
 	case <-c.stopChannel:
 		return nil, false, ErrClosed

diff --git a/pipeline/ptinput/plcache/cache_test.go b/pipeline/ptinput/plcache/cache_test.go
@@ -1,6 +1,7 @@
 package plcache
 
 import (
+	"sync"
 	"testing"
 	"time"
 
@@ -52,3 +53,31 @@ func TestCache_Stop(t *testing.T) {
 	assert.NotNil(t, err1, "err1 is nil")
 	assert.NotNil(t, err2, "err2 is nil")
 }
+
+func TestCache(t *testing.T) {
+	cache, _ := NewCache(time.Second, 10)
+
+	times := 10_0000
+
+	var wg sync.WaitGroup
+	fnSet := func() {
+		for i := 0; i < times; i++ {
+			cache.Set("k", "v", time.Millisecond*10)
+		}
+		wg.Done()
+	}
+
+	fnGet := func() {
+		for i := 0; i < times; i++ {
+			cache.Get("k1")
+		}
+		wg.Done()
+	}
+
+	for i := 0; i < 50; i++ {
+		wg.Add(2)
+		go fnSet()
+		go fnGet()
+	}
+	wg.Wait()
+}