Skip to content

当一致性哈希环上不存在当前server节点时，跳过遍历实例逻辑 #1241

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 2 commits into
base: main
Choose a base branch
from
Open

当一致性哈希环上不存在当前server节点时，跳过遍历实例逻辑 #1241

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
6076992
当一致性哈希环上不存在当前server节点时，跳过遍历实例逻辑
qnnn Sep 7, 2023
a41c257
Merge branch 'polarismesh:main' into main
qnnn Jan 20, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 17 additions & 3 deletions common/hash/ketama.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,8 @@ type continuumPoint struct {

// Continuum consistent hash ring
type Continuum struct {
ring points
ring points
hosts map[string]struct{}
}

type points []continuumPoint
Expand Down Expand Up @@ -70,6 +71,7 @@ func New(buckets map[Bucket]bool) *Continuum {
}

ring := make(points, 0, numBuckets*160)
hosts := make(map[string]struct{}, numBuckets)

var totalWeight uint32
for bucket := range buckets {
Expand All @@ -81,7 +83,10 @@ func New(buckets map[Bucket]bool) *Continuum {

// this is the equivalent of C's promotion rules, but in Go, to maintain exact compatibility with the C library
limit := int(pct * 40.0 * float64(numBuckets))

// 跳过权重为0节点
if limit != 0 {
hosts[bucket.Host] = struct{}{}
}
for k := 0; k < limit; k++ {
/* 40 hashes, 4 numbers per hash = 160 points per bucket */
ss := fmt.Sprintf("%s-%d", bucket.Host, k)
Expand All @@ -100,7 +105,8 @@ func New(buckets map[Bucket]bool) *Continuum {
sort.Sort(ring)

return &Continuum{
ring: ring,
ring: ring,
hosts: hosts,
}
}

Expand All @@ -119,3 +125,11 @@ func (c *Continuum) Hash(h uint) string {

return c.ring[i].bucket.Host
}

func (c *Continuum) ContainsHost(host string) bool {
if len(c.hosts) == 0 {
return false
}
_, ok := c.hosts[host]
return ok
}
4 changes: 2 additions & 2 deletions service/healthcheck/dispatch.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -148,7 +148,7 @@ func (d *Dispatcher) reloadSelfContinuum() bool {
func (d *Dispatcher) reloadManagedClients() {
nextClients := make(map[string]*ClientWithChecker)

if d.continuum != nil {
if d.continuum != nil && d.continuum.ContainsHost(d.svr.localHost) {
d.svr.cacheProvider.RangeHealthCheckClients(func(itemChecker ItemWithChecker, client *model.Client) {
clientId := client.Proto().GetId().GetValue()
host := d.continuum.Hash(itemChecker.GetHashValue())
Expand Down Expand Up @@ -187,7 +187,7 @@ func (d *Dispatcher) reloadManagedClients() {

func (d *Dispatcher) reloadManagedInstances() {
nextInstances := make(map[string]*InstanceWithChecker)
if d.continuum != nil {
if d.continuum != nil && d.continuum.ContainsHost(d.svr.localHost) {
d.svr.cacheProvider.RangeHealthCheckInstances(func(itemChecker ItemWithChecker, instance *model.Instance) {
instanceId := instance.ID()
host := d.continuum.Hash(itemChecker.GetHashValue())
Expand Down