diff --git a/error.go b/error.go index 8c811966..8013de44 100644 --- a/error.go +++ b/error.go @@ -22,6 +22,12 @@ var ErrPoolExhausted = pool.ErrPoolExhausted // ErrPoolTimeout timed out waiting to get a connection from the connection pool. var ErrPoolTimeout = pool.ErrPoolTimeout +// ErrCrossSlot is returned when keys are used in the same Redis command and +// the keys are not in the same hash slot. This error is returned by Redis +// Cluster and will be returned by the client when TxPipeline or TxPipelined +// is used on a ClusterClient with keys in different slots. +var ErrCrossSlot = proto.RedisError("CROSSSLOT Keys in request don't hash to the same slot") + // HasErrorPrefix checks if the err is a Redis error and the message contains a prefix. func HasErrorPrefix(err error, prefix string) bool { var rErr Error diff --git a/internal/pool/pool.go b/internal/pool/pool.go index e7d951e2..3ee3dea6 100644 --- a/internal/pool/pool.go +++ b/internal/pool/pool.go @@ -325,6 +325,7 @@ func (p *ConnPool) waitTurn(ctx context.Context) error { start := time.Now() timer := timers.Get().(*time.Timer) + defer timers.Put(timer) timer.Reset(p.cfg.PoolTimeout) select { @@ -332,7 +333,6 @@ func (p *ConnPool) waitTurn(ctx context.Context) error { if !timer.Stop() { <-timer.C } - timers.Put(timer) return ctx.Err() case p.queue <- struct{}{}: p.waitDurationNs.Add(time.Since(start).Nanoseconds()) @@ -340,10 +340,8 @@ func (p *ConnPool) waitTurn(ctx context.Context) error { if !timer.Stop() { <-timer.C } - timers.Put(timer) return nil case <-timer.C: - timers.Put(timer) atomic.AddUint32(&p.stats.Timeouts, 1) return ErrPoolTimeout } diff --git a/osscluster.go b/osscluster.go index a68f7eab..55017d8b 100644 --- a/osscluster.go +++ b/osscluster.go @@ -340,6 +340,7 @@ type clusterNode struct { latency uint32 // atomic generation uint32 // atomic failing uint32 // atomic + loaded uint32 // atomic // last time the latency measurement was performed for the node, stored in nanoseconds // from epoch @@ -406,6 +407,7 @@ func (n *clusterNode) Latency() time.Duration { func (n *clusterNode) MarkAsFailing() { atomic.StoreUint32(&n.failing, uint32(time.Now().Unix())) + atomic.StoreUint32(&n.loaded, 0) } func (n *clusterNode) Failing() bool { @@ -449,11 +451,21 @@ func (n *clusterNode) SetLastLatencyMeasurement(t time.Time) { } func (n *clusterNode) Loading() bool { + loaded := atomic.LoadUint32(&n.loaded) + if loaded == 1 { + return false + } + + // check if the node is loading ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) defer cancel() err := n.Client.Ping(ctx).Err() - return err != nil && isLoadingError(err) + loading := err != nil && isLoadingError(err) + if !loading { + atomic.StoreUint32(&n.loaded, 1) + } + return loading } //------------------------------------------------------------------------------ @@ -1497,6 +1509,10 @@ func (c *ClusterClient) processTxPipeline(ctx context.Context, cmds []Cmder) err // Trim multi .. exec. cmds = cmds[1 : len(cmds)-1] + if len(cmds) == 0 { + return nil + } + state, err := c.state.Get(ctx) if err != nil { setCmdsErr(cmds, err) @@ -1504,6 +1520,12 @@ func (c *ClusterClient) processTxPipeline(ctx context.Context, cmds []Cmder) err } cmdsMap := c.mapCmdsBySlot(cmds) + // TxPipeline does not support cross slot transaction. + if len(cmdsMap) > 1 { + setCmdsErr(cmds, ErrCrossSlot) + return ErrCrossSlot + } + for slot, cmds := range cmdsMap { node, err := state.slotMasterNode(slot) if err != nil { diff --git a/osscluster_test.go b/osscluster_test.go index 99341183..10023218 100644 --- a/osscluster_test.go +++ b/osscluster_test.go @@ -462,8 +462,7 @@ var _ = Describe("ClusterClient", func() { Describe("pipelining", func() { var pipe *redis.Pipeline - assertPipeline := func() { - keys := []string{"A", "B", "C", "D", "E", "F", "G"} + assertPipeline := func(keys []string) { It("follows redirects", func() { if !failover { @@ -482,13 +481,12 @@ var _ = Describe("ClusterClient", func() { Expect(err).NotTo(HaveOccurred()) Expect(cmds).To(HaveLen(14)) - _ = client.ForEachShard(ctx, func(ctx context.Context, node *redis.Client) error { - defer GinkgoRecover() - Eventually(func() int64 { - return node.DBSize(ctx).Val() - }, 30*time.Second).ShouldNot(BeZero()) - return nil - }) + // Check that all keys are set. + for _, key := range keys { + Eventually(func() string { + return client.Get(ctx, key).Val() + }, 30*time.Second).Should(Equal(key + "_value")) + } if !failover { for _, key := range keys { @@ -517,14 +515,14 @@ var _ = Describe("ClusterClient", func() { }) It("works with missing keys", func() { - pipe.Set(ctx, "A", "A_value", 0) - pipe.Set(ctx, "C", "C_value", 0) + pipe.Set(ctx, "A{s}", "A_value", 0) + pipe.Set(ctx, "C{s}", "C_value", 0) _, err := pipe.Exec(ctx) Expect(err).NotTo(HaveOccurred()) - a := pipe.Get(ctx, "A") - b := pipe.Get(ctx, "B") - c := pipe.Get(ctx, "C") + a := pipe.Get(ctx, "A{s}") + b := pipe.Get(ctx, "B{s}") + c := pipe.Get(ctx, "C{s}") cmds, err := pipe.Exec(ctx) Expect(err).To(Equal(redis.Nil)) Expect(cmds).To(HaveLen(3)) @@ -547,7 +545,8 @@ var _ = Describe("ClusterClient", func() { AfterEach(func() {}) - assertPipeline() + keys := []string{"A", "B", "C", "D", "E", "F", "G"} + assertPipeline(keys) It("doesn't fail node with context.Canceled error", func() { ctx, cancel := context.WithCancel(context.Background()) @@ -590,7 +589,25 @@ var _ = Describe("ClusterClient", func() { AfterEach(func() {}) - assertPipeline() + // TxPipeline doesn't support cross slot commands. + // Use hashtag to force all keys to the same slot. + keys := []string{"A{s}", "B{s}", "C{s}", "D{s}", "E{s}", "F{s}", "G{s}"} + assertPipeline(keys) + + // make sure CrossSlot error is returned + It("returns CrossSlot error", func() { + pipe.Set(ctx, "A{s}", "A_value", 0) + pipe.Set(ctx, "B{t}", "B_value", 0) + Expect(hashtag.Slot("A{s}")).NotTo(Equal(hashtag.Slot("B{t}"))) + _, err := pipe.Exec(ctx) + Expect(err).To(MatchError(redis.ErrCrossSlot)) + }) + + // doesn't fail when no commands are queued + It("returns no error when there are no commands", func() { + _, err := pipe.Exec(ctx) + Expect(err).NotTo(HaveOccurred()) + }) }) })