Store all created cluster nodes in allNodes

This commit is contained in:
Vladimir Mihailenco 2018-02-27 14:35:03 +02:00
parent e5648e8c7c
commit 9df09066e2
1 changed files with 41 additions and 43 deletions

View File

@ -205,8 +205,8 @@ type clusterNodes struct {
mu sync.RWMutex mu sync.RWMutex
allAddrs []string allAddrs []string
addrs []string allNodes map[string]*clusterNode
nodes map[string]*clusterNode clusterAddrs []string
closed bool closed bool
nodeCreateGroup singleflight.Group nodeCreateGroup singleflight.Group
@ -219,7 +219,7 @@ func newClusterNodes(opt *ClusterOptions) *clusterNodes {
opt: opt, opt: opt,
allAddrs: opt.Addrs, allAddrs: opt.Addrs,
nodes: make(map[string]*clusterNode), allNodes: make(map[string]*clusterNode),
} }
} }
@ -233,14 +233,14 @@ func (c *clusterNodes) Close() error {
c.closed = true c.closed = true
var firstErr error var firstErr error
for _, node := range c.nodes { for _, node := range c.allNodes {
if err := node.Client.Close(); err != nil && firstErr == nil { if err := node.Client.Close(); err != nil && firstErr == nil {
firstErr = err firstErr = err
} }
} }
c.addrs = nil c.allNodes = nil
c.nodes = nil c.clusterAddrs = nil
return firstErr return firstErr
} }
@ -250,8 +250,8 @@ func (c *clusterNodes) Addrs() ([]string, error) {
c.mu.RLock() c.mu.RLock()
closed := c.closed closed := c.closed
if !closed { if !closed {
if len(c.addrs) > 0 { if len(c.clusterAddrs) > 0 {
addrs = c.addrs addrs = c.clusterAddrs
} else { } else {
addrs = c.allAddrs addrs = c.allAddrs
} }
@ -276,25 +276,20 @@ func (c *clusterNodes) NextGeneration() uint32 {
func (c *clusterNodes) GC(generation uint32) { func (c *clusterNodes) GC(generation uint32) {
var collected []*clusterNode var collected []*clusterNode
c.mu.Lock() c.mu.Lock()
for i := 0; i < len(c.addrs); { for addr, node := range c.allNodes {
addr := c.addrs[i]
node := c.nodes[addr]
if node.Generation() >= generation { if node.Generation() >= generation {
i++
continue continue
} }
c.addrs = append(c.addrs[:i], c.addrs[i+1:]...) c.clusterAddrs = remove(c.clusterAddrs, addr)
delete(c.nodes, addr) delete(c.allNodes, addr)
collected = append(collected, node) collected = append(collected, node)
} }
c.mu.Unlock() c.mu.Unlock()
time.AfterFunc(time.Minute, func() {
for _, node := range collected { for _, node := range collected {
_ = node.Client.Close() _ = node.Client.Close()
} }
})
} }
func (c *clusterNodes) All() ([]*clusterNode, error) { func (c *clusterNodes) All() ([]*clusterNode, error) {
@ -305,23 +300,28 @@ func (c *clusterNodes) All() ([]*clusterNode, error) {
return nil, pool.ErrClosed return nil, pool.ErrClosed
} }
nodes := make([]*clusterNode, 0, len(c.nodes)) cp := make([]*clusterNode, 0, len(c.allNodes))
for _, node := range c.nodes { for _, node := range c.allNodes {
nodes = append(nodes, node) cp = append(cp, node)
} }
return nodes, nil return cp, nil
} }
func (c *clusterNodes) GetOrCreate(addr string) (*clusterNode, error) { func (c *clusterNodes) GetOrCreate(addr string) (*clusterNode, error) {
var node *clusterNode var node *clusterNode
var ok bool var err error
c.mu.RLock() c.mu.RLock()
if !c.closed { if c.closed {
node, ok = c.nodes[addr] err = pool.ErrClosed
} else {
node = c.allNodes[addr]
} }
c.mu.RUnlock() c.mu.RUnlock()
if ok { if err != nil {
return nil, err
}
if node != nil {
return node, nil return node, nil
} }
@ -329,9 +329,6 @@ func (c *clusterNodes) GetOrCreate(addr string) (*clusterNode, error) {
node := newClusterNode(c.opt, addr) node := newClusterNode(c.opt, addr)
return node, node.Test() return node, node.Test()
}) })
if err != nil {
return nil, err
}
c.mu.Lock() c.mu.Lock()
defer c.mu.Unlock() defer c.mu.Unlock()
@ -340,18 +337,20 @@ func (c *clusterNodes) GetOrCreate(addr string) (*clusterNode, error) {
return nil, pool.ErrClosed return nil, pool.ErrClosed
} }
node, ok = c.nodes[addr] node, ok := c.allNodes[addr]
if ok { if ok {
_ = v.(*clusterNode).Close() _ = v.(*clusterNode).Close()
return node, nil return node, err
} }
node = v.(*clusterNode) node = v.(*clusterNode)
c.allAddrs = appendIfNotExists(c.allAddrs, addr) c.allAddrs = appendIfNotExists(c.allAddrs, addr)
c.addrs = append(c.addrs, addr) if err == nil {
c.nodes[addr] = node c.clusterAddrs = append(c.clusterAddrs, addr)
}
c.allNodes[addr] = node
return node, nil return node, err
} }
func (c *clusterNodes) Random() (*clusterNode, error) { func (c *clusterNodes) Random() (*clusterNode, error) {
@ -679,11 +678,8 @@ func (c *ClusterClient) WrapProcess(
} }
func (c *ClusterClient) Process(cmd Cmder) error { func (c *ClusterClient) Process(cmd Cmder) error {
if c.process != nil {
return c.process(cmd) return c.process(cmd)
} }
return c.defaultProcess(cmd)
}
func (c *ClusterClient) defaultProcess(cmd Cmder) error { func (c *ClusterClient) defaultProcess(cmd Cmder) error {
_, node, err := c.cmdSlotAndNode(cmd) _, node, err := c.cmdSlotAndNode(cmd)
@ -918,7 +914,9 @@ func (c *ClusterClient) reloadState() bool {
state, err := c.loadState() state, err := c.loadState()
if err == nil { if err == nil {
c._state.Store(state) c._state.Store(state)
time.AfterFunc(time.Minute, func() {
c.nodes.GC(state.generation) c.nodes.GC(state.generation)
})
return true return true
} }