Add race test for big vals. Copy connection to avoid race with PubSub.

Makefile

@@ -1,6 +1,6 @@
 all: testdeps
-	go test ./... -test.v -test.cpu=1,2,4
-	go test ./... -test.v -test.short -test.race
+	go test ./... -test.cpu=1,2,4
+	go test ./... -test.short -test.race
 
 testdeps: testdata/redis/src/redis-server
 

cluster.go

@@ -7,6 +7,7 @@ import (
 	"time"
 
 	"gopkg.in/redis.v3/internal/hashtag"
+	"gopkg.in/redis.v3/internal/pool"
 )
 
 // ClusterClient is a Redis Cluster client representing a pool of zero
@@ -80,7 +81,7 @@ func (c *ClusterClient) Close() error {
 	c.clientsMx.Lock()
 
 	if c.closed {
-		return errClosed
+		return pool.ErrClosed
 	}
 	c.closed = true
 	c.resetClients()
@@ -105,7 +106,7 @@ func (c *ClusterClient) getClient(addr string) (*Client, error) {
 	c.clientsMx.Lock()
 	if c.closed {
 		c.clientsMx.Unlock()
-		return nil, errClosed
+		return nil, pool.ErrClosed
 	}
 
 	client, ok = c.clients[addr]

cluster_pipeline.go

@@ -34,15 +34,15 @@ func (pipe *ClusterPipeline) process(cmd Cmder) {
 // Discard resets the pipeline and discards queued commands.
 func (pipe *ClusterPipeline) Discard() error {
 	if pipe.closed {
-		return errClosed
+		return pool.ErrClosed
 	}
 	pipe.cmds = pipe.cmds[:0]
 	return nil
 }
 
 func (pipe *ClusterPipeline) Exec() (cmds []Cmder, retErr error) {
 	if pipe.closed {
-		return nil, errClosed
+		return nil, pool.ErrClosed
 	}
 	if len(pipe.cmds) == 0 {
 		return []Cmder{}, nil

command_test.go

@@ -5,19 +5,22 @@ import (
 	"strconv"
 	"sync"
 	"testing"
+	"time"
 
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
 
 	"gopkg.in/redis.v3"
+	"gopkg.in/redis.v3/internal/pool"
 )
 
 var _ = Describe("Command", func() {
 	var client *redis.Client
 
 	connect := func() *redis.Client {
 		return redis.NewClient(&redis.Options{
-			Addr: redisAddr,
+			Addr:        redisAddr,
+			PoolTimeout: time.Minute,
 		})
 	}
 
@@ -62,19 +65,19 @@ var _ = Describe("Command", func() {
 	})
 
 	It("should handle big vals", func() {
-		val := string(bytes.Repeat([]byte{'*'}, 1<<16))
-		set := client.Set("key", val, 0)
-		Expect(set.Err()).NotTo(HaveOccurred())
-		Expect(set.Val()).To(Equal("OK"))
+		bigVal := string(bytes.Repeat([]byte{'*'}, 1<<16))
+
+		err := client.Set("key", bigVal, 0).Err()
+		Expect(err).NotTo(HaveOccurred())
 
 		// Reconnect to get new connection.
 		Expect(client.Close()).To(BeNil())
 		client = connect()
 
-		get := client.Get("key")
-		Expect(get.Err()).NotTo(HaveOccurred())
-		Expect(len(get.Val())).To(Equal(len(val)))
-		Expect(get.Val()).To(Equal(val))
+		got, err := client.Get("key").Result()
+		Expect(err).NotTo(HaveOccurred())
+		Expect(len(got)).To(Equal(len(bigVal)))
+		Expect(got).To(Equal(bigVal))
 	})
 
 	It("should handle many keys #1", func() {
@@ -140,48 +143,111 @@ var _ = Describe("Command", func() {
 		}
 
 		It("should echo", func() {
-			wg := &sync.WaitGroup{}
-			for i := 0; i < C; i++ {
-				wg.Add(1)
-
-				go func(i int) {
-					defer GinkgoRecover()
-					defer wg.Done()
-
-					for j := 0; j < N; j++ {
-						msg := "echo" + strconv.Itoa(i)
-						echo := client.Echo(msg)
-						Expect(echo.Err()).NotTo(HaveOccurred())
-						Expect(echo.Val()).To(Equal(msg))
-					}
-				}(i)
-			}
-			wg.Wait()
+			perform(C, func() {
+				for i := 0; i < N; i++ {
+					msg := "echo" + strconv.Itoa(i)
+					echo, err := client.Echo(msg).Result()
+					Expect(err).NotTo(HaveOccurred())
+					Expect(echo).To(Equal(msg))
+				}
+			})
 		})
 
 		It("should incr", func() {
 			key := "TestIncrFromGoroutines"
-			wg := &sync.WaitGroup{}
-			for i := 0; i < C; i++ {
-				wg.Add(1)
 
-				go func() {
-					defer GinkgoRecover()
-					defer wg.Done()
-
-					for j := 0; j < N; j++ {
-						err := client.Incr(key).Err()
-						Expect(err).NotTo(HaveOccurred())
-					}
-				}()
-			}
-			wg.Wait()
+			perform(C, func() {
+				for i := 0; i < N; i++ {
+					err := client.Incr(key).Err()
+					Expect(err).NotTo(HaveOccurred())
+				}
+			})
 
 			val, err := client.Get(key).Int64()
 			Expect(err).NotTo(HaveOccurred())
 			Expect(val).To(Equal(int64(C * N)))
 		})
 
+		It("should handle big vals", func() {
+			client2 := connect()
+			defer client2.Close()
+
+			bigVal := string(bytes.Repeat([]byte{'*'}, 1<<16))
+
+			var wg sync.WaitGroup
+			wg.Add(2)
+
+			go func() {
+				defer wg.Done()
+				perform(C, func() {
+					for i := 0; i < N; i++ {
+						got, err := client.Get("key").Result()
+						if err == redis.Nil {
+							continue
+						}
+						Expect(got).To(Equal(bigVal))
+					}
+				})
+			}()
+
+			go func() {
+				defer wg.Done()
+				perform(C, func() {
+					for i := 0; i < N; i++ {
+						err := client2.Set("key", bigVal, 0).Err()
+						Expect(err).NotTo(HaveOccurred())
+					}
+				})
+			}()
+
+			wg.Wait()
+		})
+
+		It("should PubSub", func() {
+			connPool := client.Pool()
+			connPool.(*pool.ConnPool).DialLimiter = nil
+
+			var wg sync.WaitGroup
+			wg.Add(2)
+
+			go func() {
+				defer wg.Done()
+				perform(C, func() {
+					for i := 0; i < N; i++ {
+						pubsub, err := client.Subscribe("mychannel")
+						Expect(err).NotTo(HaveOccurred())
+
+						go func() {
+							defer GinkgoRecover()
+
+							time.Sleep(time.Millisecond)
+							err := pubsub.Close()
+							Expect(err).NotTo(HaveOccurred())
+						}()
+
+						_, err = pubsub.ReceiveMessage()
+						Expect(err.Error()).To(ContainSubstring("closed"))
+					}
+				})
+			}()
+
+			go func() {
+				defer wg.Done()
+				perform(C, func() {
+					for i := 0; i < N; i++ {
+						val := "echo" + strconv.Itoa(i)
+						echo, err := client.Echo(val).Result()
+						Expect(err).NotTo(HaveOccurred())
+						Expect(echo).To(Equal(val))
+					}
+				})
+			}()
+
+			wg.Wait()
+
+			Expect(connPool.Len()).To(Equal(connPool.FreeLen()))
+			Expect(connPool.Len()).To(BeNumerically("<=", 10))
+		})
 	})
 
 })

error.go

@@ -1,15 +1,12 @@
 package redis
 
 import (
-	"errors"
 	"fmt"
 	"io"
 	"net"
 	"strings"
 )
 
-var errClosed = errors.New("redis: client is closed")
-
 // Redis nil reply, .e.g. when key does not exist.
 var Nil = errorf("redis: nil")
 

example_test.go

@@ -220,13 +220,13 @@ func ExampleClient_Watch() {
 }
 
 func ExamplePubSub() {
-	pubsub, err := client.Subscribe("mychannel")
+	pubsub, err := client.Subscribe("mychannel1")
 	if err != nil {
 		panic(err)
 	}
 	defer pubsub.Close()
 
-	err = client.Publish("mychannel", "hello").Err()
+	err = client.Publish("mychannel1", "hello").Err()
 	if err != nil {
 		panic(err)
 	}
@@ -237,17 +237,17 @@ func ExamplePubSub() {
 	}
 
 	fmt.Println(msg.Channel, msg.Payload)
-	// Output: mychannel hello
+	// Output: mychannel1 hello
 }
 
 func ExamplePubSub_Receive() {
-	pubsub, err := client.Subscribe("mychannel")
+	pubsub, err := client.Subscribe("mychannel2")
 	if err != nil {
 		panic(err)
 	}
 	defer pubsub.Close()
 
-	err = client.Publish("mychannel", "hello").Err()
+	err = client.Publish("mychannel2", "hello").Err()
 	if err != nil {
 		panic(err)
 	}
@@ -269,8 +269,8 @@ func ExamplePubSub_Receive() {
 		}
 	}
 
-	// Output: subscribe mychannel
-	// mychannel hello
+	// Output: subscribe mychannel2
+	// mychannel2 hello
 }
 
 func ExampleScript() {