Merge pull request #738 from go-redis/fix/blpop-race-test

Add race test for BLPop

cluster_test.go

@@ -788,8 +788,8 @@ var _ = Describe("ClusterClient timeout", func() {
 	Context("read/write timeout", func() {
 		BeforeEach(func() {
 			opt := redisClusterOptions()
-			opt.ReadTimeout = 200 * time.Millisecond
-			opt.WriteTimeout = 200 * time.Millisecond
+			opt.ReadTimeout = 300 * time.Millisecond
+			opt.WriteTimeout = 300 * time.Millisecond
 			opt.MaxRedirects = 1
 			client = cluster.clusterClient(opt)
 

main_test.go

@@ -142,7 +142,7 @@ func redisRingOptions() *redis.RingOptions {
 	}
 }
 
-func perform(n int, cbs ...func(int)) {
+func performAsync(n int, cbs ...func(int)) *sync.WaitGroup {
 	var wg sync.WaitGroup
 	for _, cb := range cbs {
 		for i := 0; i < n; i++ {
@@ -155,6 +155,11 @@ func perform(n int, cbs ...func(int)) {
 			}(cb, i)
 		}
 	}
+	return &wg
+}
+
+func perform(n int, cbs ...func(int)) {
+	wg := performAsync(n, cbs...)
 	wg.Wait()
 }
 

race_test.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"net"
 	"strconv"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -258,6 +259,30 @@ var _ = Describe("races", func() {
 		Expect(err).NotTo(HaveOccurred())
 		Expect(n).To(Equal(int64(N)))
 	})
+
+	It("should BLPop", func() {
+		var received uint32
+		wg := performAsync(C, func(id int) {
+			for {
+				v, err := client.BLPop(3*time.Second, "list").Result()
+				if err != nil {
+					break
+				}
+				Expect(v).To(Equal([]string{"list", "hello"}))
+				atomic.AddUint32(&received, 1)
+			}
+		})
+
+		perform(C, func(id int) {
+			for i := 0; i < N; i++ {
+				err := client.LPush("list", "hello").Err()
+				Expect(err).NotTo(HaveOccurred())
+			}
+		})
+
+		wg.Wait()
+		Expect(received).To(Equal(uint32(C * N)))
+	})
 })
 
 func bigVal() []byte {