Reload slots in background goroutine.

Author: vmihailenco

cluster.go

@@ -1,6 +1,7 @@
 package redis
 
 import (
+	"log"
 	"math/rand"
 	"strings"
 	"sync"
@@ -20,7 +21,8 @@ type ClusterClient struct {
 
 	opt *ClusterOptions
 
-	_reload uint32
+	// Reports where slots reloading is in progress.
+	_reloading uint32
 }
 
 // NewClusterClient initializes a new cluster-aware client using given options.
@@ -31,10 +33,9 @@ func NewClusterClient(opt *ClusterOptions) *ClusterClient {
 		slots:   make([][]string, hashSlots),
 		clients: make(map[string]*Client),
 		opt:     opt,
-		_reload: 1,
 	}
 	client.commandable.process = client.process
-	client.reloadIfDue()
+	client.reloadSlots()
 	go client.reaper(time.NewTicker(5 * time.Minute))
 	return client
 }
@@ -102,8 +103,6 @@ func (c *ClusterClient) randomClient() (client *Client, err error) {
 func (c *ClusterClient) process(cmd Cmder) {
 	var ask bool
 
-	c.reloadIfDue()
-
 	slot := hashSlot(cmd.clusterKey())
 
 	var addr string
@@ -149,7 +148,7 @@ func (c *ClusterClient) process(cmd Cmder) {
 		moved, ask, addr = isMovedError(err)
 		if moved || ask {
 			if moved {
-				c.scheduleReload()
+				c.lazyReloadSlots()
 			}
 			client, err = c.getClient(addr)
 			if err != nil {
@@ -203,29 +202,28 @@ func (c *ClusterClient) setSlots(slots []ClusterSlotInfo) {
 	c.slotsMx.Unlock()
 }
 
-// Closes all connections and reloads slot cache, if due.
-func (c *ClusterClient) reloadIfDue() (err error) {
-	if !atomic.CompareAndSwapUint32(&c._reload, 1, 0) {
-		return
-	}
+func (c *ClusterClient) reloadSlots() {
+	defer atomic.StoreUint32(&c._reloading, 0)
 
 	client, err := c.randomClient()
 	if err != nil {
-		return err
+		log.Printf("redis: randomClient failed: %s", err)
+		return
 	}
 
 	slots, err := client.ClusterSlots().Result()
 	if err != nil {
-		return err
+		log.Printf("redis: ClusterSlots failed: %s", err)
+		return
 	}
 	c.setSlots(slots)
-
-	return nil
 }
 
-// Schedules slots reload on next request.
-func (c *ClusterClient) scheduleReload() {
-	atomic.StoreUint32(&c._reload, 1)
+func (c *ClusterClient) lazyReloadSlots() {
+	if !atomic.CompareAndSwapUint32(&c._reloading, 0, 1) {
+		return
+	}
+	go c.reloadSlots()
 }
 
 // reaper closes idle connections to the cluster.

cluster_client_test.go

@@ -1,23 +1,22 @@
 package redis
 
 import (
+	"fmt"
+
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
 )
 
-// GetSlot returns the cached slot addresses
-func (c *ClusterClient) GetSlot(pos int) []string {
-	c.slotsMx.RLock()
-	defer c.slotsMx.RUnlock()
-
-	return c.slots[pos]
+func (c *ClusterClient) SlotAddrs(slot int) []string {
+	return c.slotAddrs(slot)
 }
 
 // SwapSlot swaps a slot's master/slave address
 // for testing MOVED redirects
 func (c *ClusterClient) SwapSlot(pos int) []string {
 	c.slotsMx.Lock()
 	defer c.slotsMx.Unlock()
+	fmt.Println(pos, c.slots[pos])
 	c.slots[pos][0], c.slots[pos][1] = c.slots[pos][1], c.slots[pos][0]
 	return c.slots[pos]
 }
@@ -49,7 +48,6 @@ var _ = Describe("ClusterClient", func() {
 	It("should initialize", func() {
 		Expect(subject.addrs).To(HaveLen(3))
 		Expect(subject.slots).To(HaveLen(16384))
-		Expect(subject._reload).To(Equal(uint32(0)))
 	})
 
 	It("should update slots cache", func() {
@@ -84,11 +82,4 @@ var _ = Describe("ClusterClient", func() {
 		Expect(subject.slots[8192]).To(BeEmpty())
 		Expect(subject.slots[16383]).To(BeEmpty())
 	})
-
-	It("should check if reload is due", func() {
-		subject._reload = 0
-		Expect(subject._reload).To(Equal(uint32(0)))
-		subject.scheduleReload()
-		Expect(subject._reload).To(Equal(uint32(1)))
-	})
 })

cluster_pipeline.go

@@ -113,7 +113,7 @@ func (c *ClusterPipeline) execClusterCmds(
 			failedCmds[""] = append(failedCmds[""], cmds[i:]...)
 			break
 		} else if moved, ask, addr := isMovedError(err); moved {
-			c.cluster.scheduleReload()
+			c.cluster.lazyReloadSlots()
 			cmd.reset()
 			failedCmds[addr] = append(failedCmds[addr], cmd)
 		} else if ask {

cluster_test.go

@@ -258,25 +258,23 @@ var _ = Describe("Cluster", func() {
 
 		It("should follow redirects", func() {
 			Expect(client.Set("A", "VALUE", 0).Err()).NotTo(HaveOccurred())
-			Expect(redis.HashSlot("A")).To(Equal(6373))
-			Expect(client.SwapSlot(6373)).To(Equal([]string{"127.0.0.1:8224", "127.0.0.1:8221"}))
+
+			slot := redis.HashSlot("A")
+			Expect(client.SwapSlot(slot)).To(Equal([]string{"127.0.0.1:8224", "127.0.0.1:8221"}))
 
 			val, err := client.Get("A").Result()
 			Expect(err).NotTo(HaveOccurred())
 			Expect(val).To(Equal("VALUE"))
-			Expect(client.GetSlot(6373)).To(Equal([]string{"127.0.0.1:8224", "127.0.0.1:8221"}))
+			Expect(client.SlotAddrs(slot)).To(Equal([]string{"127.0.0.1:8224", "127.0.0.1:8221"}))
 
-			val, err = client.Get("A").Result()
-			Expect(err).NotTo(HaveOccurred())
-			Expect(val).To(Equal("VALUE"))
-			Expect(client.GetSlot(6373)).To(Equal([]string{"127.0.0.1:8221", "127.0.0.1:8224"}))
+			Eventually(func() []string {
+				return client.SlotAddrs(slot)
+			}).Should(Equal([]string{"127.0.0.1:8221", "127.0.0.1:8224"}))
 		})
 
 		It("should perform multi-pipelines", func() {
-			// Dummy command to load slots info.
-			Expect(client.Ping().Err()).NotTo(HaveOccurred())
-
 			slot := redis.HashSlot("A")
+			Expect(client.SlotAddrs(slot)).To(Equal([]string{"127.0.0.1:8221", "127.0.0.1:8224"}))
 			Expect(client.SwapSlot(slot)).To(Equal([]string{"127.0.0.1:8224", "127.0.0.1:8221"}))
 
 			pipe := client.Pipeline()