fix(diskcache): synchronize concurrent rotation state

diskcache/concurrency_test.go

@@ -0,0 +1,76 @@
+// Unless explicitly stated otherwise all files in this repository are licensed
+// under the MIT License.
+// This product includes software developed at Guance Cloud (https://www.guance.com/).
+// Copyright 2021-present Guance, Inc.
+
+package diskcache
+
+import (
+	"errors"
+	"fmt"
+	"runtime"
+	T "testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestPutGetConcurrentRotation(t *T.T) {
+	const records = 256
+
+	c, err := Open(
+		WithPath(t.TempDir()),
+		WithBatchSize(1),
+		WithNoPos(true),
+		WithNoSync(true),
+	)
+	require.NoError(t, err)
+	t.Cleanup(func() {
+		require.NoError(t, c.Close())
+	})
+
+	start := make(chan struct{})
+	results := make(chan error, 2)
+
+	go func() {
+		<-start
+		for i := range records {
+			if err := c.Put([]byte{byte(i)}); err != nil {
+				results <- fmt.Errorf("put record %d: %w", i, err)
+				return
+			}
+		}
+		results <- nil
+	}()
+
+	go func() {
+		<-start
+		deadline := time.Now().Add(2 * time.Second)
+		for next := 0; next < records; {
+			err := c.Get(func(got []byte) error {
+				if len(got) != 1 || got[0] != byte(next) {
+					return fmt.Errorf("record %d: got %v", next, got)
+				}
+				next++
+				return nil
+			})
+			if err == nil {
+				continue
+			}
+			if !errors.Is(err, ErrNoData) {
+				results <- fmt.Errorf("get record %d: %w", next, err)
+				return
+			}
+			if time.Now().After(deadline) {
+				results <- fmt.Errorf("get record %d: timed out", next)
+				return
+			}
+			runtime.Gosched()
+		}
+		results <- nil
+	}()
+
+	close(start)
+	require.NoError(t, <-results)
+	require.NoError(t, <-results)
+}

diskcache/get.go

@@ -93,19 +93,20 @@ func (c *DiskCache) doGet(buf []byte, fn Fn, bfn BufFunc) error {
 	}()
 
 	// wakeup sleeping write file, rotate it for succession reading!
-	if time.Since(c.wfdLastWrite) > c.wakeup && c.curBatchSize > 0 {
-		wakeupVec.WithLabelValues(c.path).Inc()
-
-		if err = func() error {
-			c.wlock.Lock()
-			defer c.wlock.Unlock()
-
-			return c.rotate()
-		}(); err != nil {
+	// A held write lock means the write file is active, not sleeping.
+	if c.wlock.TryLock() {
+		idleTime := time.Since(c.wfdLastWrite)
+		batchSize := c.curBatchSize
+		if idleTime > c.wakeup && batchSize > 0 {
+			wakeupVec.WithLabelValues(c.path).Inc()
+			err = c.rotate()
+		}
+		c.wlock.Unlock()
+		if err != nil {
 			return NewCacheError(OpGet, err, "failed_to_wakeup_sleeping_write_file").
 				WithPath(c.path).
 				WithDetails(fmt.Sprintf("idle_time=%v, batch_size=%d",
-					time.Since(c.wfdLastWrite), c.curBatchSize))
+					idleTime, batchSize))
 		}
 	}
 

diskcache/rotate.go

@@ -28,6 +28,9 @@ func (c *DiskCache) Rotate() error {
 		return NewCacheError(OpRotate, ErrClosed, "cache_closed").WithPath(c.path)
 	}
 
+	c.wlock.Lock()
+	defer c.wlock.Unlock()
+
 	return c.rotate()
 }