forked from mirror/redis
Ensure slots are initialised. Return non-failing connections to pool
This commit is contained in:
parent
9de2369192
commit
51f0a7b0a7
|
@ -28,11 +28,14 @@ type ClusterClient struct {
|
||||||
func NewClusterClient(opt *ClusterOptions) *ClusterClient {
|
func NewClusterClient(opt *ClusterOptions) *ClusterClient {
|
||||||
client := &ClusterClient{
|
client := &ClusterClient{
|
||||||
addrs: opt.Addrs,
|
addrs: opt.Addrs,
|
||||||
|
slots: make([][]string, hashSlots),
|
||||||
clients: make(map[string]*Client),
|
clients: make(map[string]*Client),
|
||||||
opt: opt,
|
opt: opt,
|
||||||
_reload: 1,
|
_reload: 1,
|
||||||
}
|
}
|
||||||
client.commandable.process = client.process
|
client.commandable.process = client.process
|
||||||
|
client.reloadIfDue()
|
||||||
|
|
||||||
go client.reaper(time.NewTicker(5 * time.Minute))
|
go client.reaper(time.NewTicker(5 * time.Minute))
|
||||||
return client
|
return client
|
||||||
}
|
}
|
||||||
|
@ -176,14 +179,15 @@ func (c *ClusterClient) resetClients() (err error) {
|
||||||
func (c *ClusterClient) setSlots(slots []ClusterSlotInfo) {
|
func (c *ClusterClient) setSlots(slots []ClusterSlotInfo) {
|
||||||
c.slotsMx.Lock()
|
c.slotsMx.Lock()
|
||||||
|
|
||||||
c.slots = make([][]string, hashSlots)
|
|
||||||
c.resetClients()
|
c.resetClients()
|
||||||
|
|
||||||
seen := make(map[string]struct{})
|
seen := make(map[string]struct{})
|
||||||
for _, addr := range c.addrs {
|
for _, addr := range c.addrs {
|
||||||
seen[addr] = struct{}{}
|
seen[addr] = struct{}{}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
for i := 0; i < hashSlots; i++ {
|
||||||
|
c.slots[i] = c.slots[i][:0]
|
||||||
|
}
|
||||||
for _, info := range slots {
|
for _, info := range slots {
|
||||||
for slot := info.Start; slot <= info.End; slot++ {
|
for slot := info.Start; slot <= info.End; slot++ {
|
||||||
c.slots[slot] = info.Addrs
|
c.slots[slot] = info.Addrs
|
||||||
|
|
|
@ -48,7 +48,8 @@ var _ = Describe("ClusterClient", func() {
|
||||||
|
|
||||||
It("should initialize", func() {
|
It("should initialize", func() {
|
||||||
Expect(subject.addrs).To(HaveLen(3))
|
Expect(subject.addrs).To(HaveLen(3))
|
||||||
Expect(subject._reload).To(Equal(uint32(1)))
|
Expect(subject.slots).To(HaveLen(16384))
|
||||||
|
Expect(subject._reload).To(Equal(uint32(0)))
|
||||||
})
|
})
|
||||||
|
|
||||||
It("should update slots cache", func() {
|
It("should update slots cache", func() {
|
||||||
|
@ -74,6 +75,16 @@ var _ = Describe("ClusterClient", func() {
|
||||||
}))
|
}))
|
||||||
})
|
})
|
||||||
|
|
||||||
|
It("should close", func() {
|
||||||
|
populate()
|
||||||
|
Expect(subject.Close()).NotTo(HaveOccurred())
|
||||||
|
Expect(subject.clients).To(BeEmpty())
|
||||||
|
Expect(subject.slots[0]).To(BeEmpty())
|
||||||
|
Expect(subject.slots[8191]).To(BeEmpty())
|
||||||
|
Expect(subject.slots[8192]).To(BeEmpty())
|
||||||
|
Expect(subject.slots[16383]).To(BeEmpty())
|
||||||
|
})
|
||||||
|
|
||||||
It("should check if reload is due", func() {
|
It("should check if reload is due", func() {
|
||||||
subject._reload = 0
|
subject._reload = 0
|
||||||
Expect(subject._reload).To(Equal(uint32(0)))
|
Expect(subject._reload).To(Equal(uint32(0)))
|
||||||
|
|
5
redis.go
5
redis.go
|
@ -56,8 +56,9 @@ func (c *baseClient) initConn(cn *conn) error {
|
||||||
func (c *baseClient) freeConn(cn *conn, ei error) error {
|
func (c *baseClient) freeConn(cn *conn, ei error) error {
|
||||||
if cn.rd.Buffered() > 0 {
|
if cn.rd.Buffered() > 0 {
|
||||||
return c.connPool.Remove(cn)
|
return c.connPool.Remove(cn)
|
||||||
}
|
} else if ei == nil {
|
||||||
if _, ok := ei.(redisError); ok {
|
return c.connPool.Put(cn)
|
||||||
|
} else if _, ok := ei.(redisError); ok {
|
||||||
return c.connPool.Put(cn)
|
return c.connPool.Put(cn)
|
||||||
}
|
}
|
||||||
return c.connPool.Remove(cn)
|
return c.connPool.Remove(cn)
|
||||||
|
|
Loading…
Reference in New Issue