func (c *Consistent) GetNeighborsEx(name string, n int, roles proto.ServerRoles) ([]proto.Node, error) {
	c.RLock()
	defer c.RUnlock()
	c.cacheLock.RLock()
	defer c.cacheLock.RUnlock()
	if len(c.circle) == 0 {
		return nil, ErrEmptyCircle
	}
	count := len(c.sortedHashes) / c.NumberOfReplicas
	if count < n {
		n = count
	}
	var (
		key   = hashKey(name)
		i     = c.search(key)
		start = i
		res   = make([]proto.Node, 0, n)
		elem  = *c.circle[c.sortedHashes[i]]
	)
	var noFilter = roles == nil || len(roles) == 0
	if noFilter || roles.Contains(elem.Role) {
		res = append(res, elem)
	}
	if noFilter && len(res) == n {
		return res, nil
	}
	for i = start + 1; i != start; i++ {
		if i >= len(c.sortedHashes) {
			i = 0
		}
		elem = *c.circle[c.sortedHashes[i]]
		if noFilter || roles.Contains(elem.Role) {
			if !sliceContainsMember(res, elem) {
				res = append(res, elem)
			}
		}
		if len(res) == n {
			break
		}
	}
	return res, nil
}
GetNeighborsEx returns the N closest distinct nodes to the name input in the circle.
package consistent
import (
	"errors"
	"expvar"
	"sort"
	"strconv"
	"sync"
	mw "github.com/zserge/metric"
	"github.com/SQLess/SQLess/conf"
	"github.com/SQLess/SQLess/crypto/hash"
	"github.com/SQLess/SQLess/proto"
	"github.com/SQLess/SQLess/utils/log"
)
type NodeKeys []proto.NodeKey
func (x NodeKeys) Len() int { return len(x) }
func (x NodeKeys) Less(i, j int) bool { return x[i].Less(&x[j]) }
func (x NodeKeys) Swap(i, j int) { x[i], x[j] = x[j], x[i] }
var (
	ErrEmptyCircle = errors.New("empty circle")
	ErrKeyNotFound = errors.New("node key not found")
)
const (
	mwKeyDHTNodeCount = "service:DHT:NodeCount"
)
func init() {
	expvar.Publish(mwKeyDHTNodeCount, mw.NewGauge("1M1h"))
}
type Consistent struct {
	circle map[proto.NodeKey]*proto.Node
	sortedHashes     NodeKeys
	NumberOfReplicas int
	persist          Persistence
	cacheLock        sync.RWMutex
	sync.RWMutex
}
func InitConsistent(storePath string, persistImpl Persistence, initBP bool) (c *Consistent, err error) {
	var BPNodes []proto.Node
	if initBP {
		if conf.GConf == nil {
			log.Fatal("must call conf.LoadConfig first")
		}
		BPNodes = conf.GConf.SeedBPNodes
	}
	c = &Consistent{
		NumberOfReplicas: 20,
		circle:           make(map[proto.NodeKey]*proto.Node),
		persist:          persistImpl,
	}
	err = c.persist.Init(storePath, BPNodes)
	if err != nil {
		log.WithError(err).Error("init persist BP nodes failed")
		return
	}
	nodes, err := c.persist.GetAllNodeInfo()
	if err != nil {
		log.WithError(err).Error("get all node id failed")
		return
	}
	log.Debugf("c.persist.GetAllNodeInfo: %#v", nodes)
	_, isKMSStorage := c.persist.(*KMSStorage)
	if isKMSStorage {
		for _, n := range nodes[:] {
			c.Add(n)
		}
	} else {
		for _, n := range nodes[:] {
			c.AddCache(n)
		}
	}
	return
}
func (c *Consistent) nodeKey(nodeID proto.NodeID, idx int) string {
	return strconv.Itoa(idx) + string(nodeID)
}
func (c *Consistent) Add(node proto.Node) (err error) {
	log.WithField("node", node).Debug("add node to consistent ring")
	c.Lock()
	defer c.Unlock()
	return c.add(node)
}
func (c *Consistent) Remove(nodeID proto.NodeID) (err error) {
	c.Lock()
	defer c.Unlock()
	err = c.persist.DelNode(nodeID)
	if err != nil {
		log.WithField("node", nodeID).WithError(err).Error("del node failed")
		return
	}
	c.RemoveCache(nodeID)
	return
}
func (c *Consistent) Set(nodes []proto.Node) (err error) {
	c.Lock()
	defer c.Unlock()
	err = c.persist.Reset()
	if err != nil {
		log.WithError(err).Error("reset bucket failed")
		return
	}
	c.ResetCache()
	for _, v := range nodes {
		c.add(v)
	}
	return
}
func (c *Consistent) add(node proto.Node) (err error) {
	err = c.persist.SetNode(&node)
	if err != nil {
		log.WithField("node", node).WithError(err).Error("set node info failed")
		return
	}
	c.AddCache(node)
	return
}
func (c *Consistent) AddCache(node proto.Node) {
	c.cacheLock.Lock()
	defer c.cacheLock.Unlock()
	for i := 0; i < c.NumberOfReplicas; i++ {
		c.circle[hashKey(c.nodeKey(node.ID, i))] = &node
	}
	c.updateSortedHashes()
	expvar.Get(mwKeyDHTNodeCount).(mw.Metric).Add(float64(len(c.circle) / c.NumberOfReplicas))
	return
}
func (c *Consistent) RemoveCache(nodeID proto.NodeID) {
	c.cacheLock.Lock()
	defer c.cacheLock.Unlock()
	for i := 0; i < c.NumberOfReplicas; i++ {
		delete(c.circle, hashKey(c.nodeKey(nodeID, i)))
	}
	c.updateSortedHashes()
}
func (c *Consistent) ResetCache() {
	c.cacheLock.Lock()
	defer c.cacheLock.Unlock()
	c.circle = make(map[proto.NodeKey]*proto.Node)
	c.sortedHashes = NodeKeys{}
}
func (c *Consistent) GetNeighbor(name string) (proto.Node, error) {
	c.RLock()
	defer c.RUnlock()
	c.cacheLock.RLock()
	defer c.cacheLock.RUnlock()
	if len(c.circle) == 0 {
		return proto.Node{}, ErrEmptyCircle
	}
	key := hashKey(name)
	i := c.search(key)
	return *c.circle[c.sortedHashes[i]], nil
}
func (c *Consistent) GetNode(name string) (*proto.Node, error) {
	c.RLock()
	defer c.RUnlock()
	c.cacheLock.RLock()
	defer c.cacheLock.RUnlock()
	n, ok := c.circle[hashKey(c.nodeKey(proto.NodeID(name), 0))]
	if ok {
		return n, nil
	}
	return nil, ErrKeyNotFound
}
func (c *Consistent) search(key proto.NodeKey) (i int) {
	f := func(x int) bool {
		return key.Less(&c.sortedHashes[x])
	}
	i = sort.Search(len(c.sortedHashes), f)
	if i >= len(c.sortedHashes) {
		i = 0
	}
	return
}
func (c *Consistent) GetTwoNeighbors(name string) (proto.Node, proto.Node, error) {
	c.RLock()
	defer c.RUnlock()
	c.cacheLock.RLock()
	defer c.cacheLock.RUnlock()
	if len(c.circle) == 0 {
		return proto.Node{}, proto.Node{}, ErrEmptyCircle
	}
	key := hashKey(name)
	i := c.search(key)
	a := *c.circle[c.sortedHashes[i]]
	if len(c.sortedHashes)/c.NumberOfReplicas == 1 {
		return a, proto.Node{}, nil
	}
	start := i
	var b proto.Node
	for i = start + 1; i != start; i++ {
		if i >= len(c.sortedHashes) {
			i = 0
		}
		b = *c.circle[c.sortedHashes[i]]
		if b.ID != a.ID {
			break
		}
	}
	return a, b, nil
}