Fix/pubsub ping mutex (#1878)

* Fix PubSub.Ping to hold the lock

* Fix PubSub.Ping to hold the lock

* add write cmd data-race test

Signed-off-by: monkey92t <golang@88.com>

Co-authored-by: monkey92t <golang@88.com>

pubsub.go

@@ -252,13 +252,16 @@ func (c *PubSub) Ping(ctx context.Context, payload ...string) error {
 	}
 	cmd := NewCmd(ctx, args...)
 
-	cn, err := c.connWithLock(ctx)
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	cn, err := c.conn(ctx, nil)
 	if err != nil {
 		return err
 	}
 
 	err = c.writeCmd(ctx, cn, cmd)
-	c.releaseConnWithLock(ctx, cn, err, false)
+	c.releaseConn(ctx, cn, err, false)
 	return err
 }
 
@@ -361,6 +364,8 @@ func (c *PubSub) ReceiveTimeout(ctx context.Context, timeout time.Duration) (int
 		c.cmd = NewCmd(ctx)
 	}
 
+	// Don't hold the lock to allow subscriptions and pings.
+
 	cn, err := c.connWithLock(ctx)
 	if err != nil {
 		return nil, err
@@ -371,6 +376,7 @@ func (c *PubSub) ReceiveTimeout(ctx context.Context, timeout time.Duration) (int
 	})
 
 	c.releaseConnWithLock(ctx, cn, err, timeout > 0)
+
 	if err != nil {
 		return nil, err
 	}

race_test.go

@@ -373,6 +373,18 @@ var _ = Describe("cluster races", func() {
 		Expect(err).NotTo(HaveOccurred())
 		Expect(val).To(Equal(int64(C * N)))
 	})
+
+	It("write cmd data-race", func() {
+		pubsub := client.Subscribe(ctx)
+		defer pubsub.Close()
+
+		pubsub.Channel(redis.WithChannelHealthCheckInterval(time.Millisecond))
+		for i := 0; i < 100; i++ {
+			key := fmt.Sprintf("channel_%d", i)
+			pubsub.Subscribe(ctx, key)
+			pubsub.Unsubscribe(ctx, key)
+		}
+	})
 })
 
 func bigVal() []byte {