add maglev consistent hashing

Signed-off-by: dongjiang <[email protected]>

Author: dongjiang1989

cluster/loadbalance/consistenthashing/loadbalance.go

@@ -21,9 +21,7 @@ import (
 	"encoding/json"
 	"hash/crc32"
 	"regexp"
-)
 
-import (
 	"dubbo.apache.org/dubbo-go/v3/cluster/loadbalance"
 	"dubbo.apache.org/dubbo-go/v3/common/constant"
 	"dubbo.apache.org/dubbo-go/v3/common/extension"

cluster/loadbalance/loadbalance_benchmarks_test.go

@@ -20,14 +20,13 @@ package loadbalance_test
 import (
 	"fmt"
 	"testing"
-)
 
-import (
 	"dubbo.apache.org/dubbo-go/v3/cluster/loadbalance"
 	_ "dubbo.apache.org/dubbo-go/v3/cluster/loadbalance/aliasmethod"
 	_ "dubbo.apache.org/dubbo-go/v3/cluster/loadbalance/consistenthashing"
 	_ "dubbo.apache.org/dubbo-go/v3/cluster/loadbalance/iwrr"
 	_ "dubbo.apache.org/dubbo-go/v3/cluster/loadbalance/leastactive"
+	_ "dubbo.apache.org/dubbo-go/v3/cluster/loadbalance/maglevconsistenthashing"
 	_ "dubbo.apache.org/dubbo-go/v3/cluster/loadbalance/p2c"
 	_ "dubbo.apache.org/dubbo-go/v3/cluster/loadbalance/random"
 	_ "dubbo.apache.org/dubbo-go/v3/cluster/loadbalance/ringhash"
@@ -85,3 +84,7 @@ func BenchmarkRandomLoadbalance(b *testing.B) {
 func BenchmarkAliasMethodLoadbalance(b *testing.B) {
 	Benchloadbalance(b, extension.GetLoadbalance(constant.LoadBalanceKeyAliasMethod))
 }
+
+func BenchmarkMaglevConsistentHashingLoadBalance(b *testing.B) {
+	Benchloadbalance(b, extension.GetLoadbalance(constant.LoadBalanceKeyMaglevConsistentHashing))
+}

cluster/loadbalance/maglevconsistenthashing/doc.go

@@ -0,0 +1,19 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+// Package maglevconsistenthashing implements ConsistentHash load balance strategy.
+package maglevconsistenthashing

cluster/loadbalance/maglevconsistenthashing/loadbalance.go

@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package maglevconsistenthashing
+
+import (
+	"dubbo.apache.org/dubbo-go/v3/cluster/loadbalance"
+	"dubbo.apache.org/dubbo-go/v3/common/constant"
+	"dubbo.apache.org/dubbo-go/v3/common/extension"
+	"dubbo.apache.org/dubbo-go/v3/protocol"
+)
+
+func init() {
+	extension.SetLoadbalance(constant.LoadBalanceKeyMaglevConsistentHashing, newMaglevConsistentHashingLoadBalance)
+}
+
+type maglevConsistentHashingLoadBalance struct{}
+
+// newMaglevConsistentHashingLoadBalance returns a loadbalancer using maglev consistent hashing algorithm.
+func newMaglevConsistentHashingLoadBalance() loadbalance.LoadBalance {
+	return &maglevConsistentHashingLoadBalance{}
+}
+
+// Select gets invoker based on interleaved weighted round robine load balancing strategy
+func (lb *maglevConsistentHashingLoadBalance) Select(invokers []protocol.Invoker, invocation protocol.Invocation) protocol.Invoker {
+	count := len(invokers)
+	if count == 0 {
+		return nil
+	}
+	if count == 1 {
+		return invokers[0]
+	}
+
+	maglev := NewMaglev(invokers)
+
+	methodName := invocation.MethodName()
+	key := invokers[0].GetURL().ServiceKey() + "." + methodName
+
+	return maglev.Pick(key)
+}

cluster/loadbalance/maglevconsistenthashing/loadbalance_test.go

@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package maglevconsistenthashing
+
+import (
+	"fmt"
+	"testing"
+
+	"dubbo.apache.org/dubbo-go/v3/common"
+	"dubbo.apache.org/dubbo-go/v3/common/constant"
+	"dubbo.apache.org/dubbo-go/v3/protocol"
+	"dubbo.apache.org/dubbo-go/v3/protocol/invocation"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestMaglevConshashLoadBalanceSelect(t *testing.T) {
+	loadBalance := newMaglevConsistentHashingLoadBalance()
+
+	var invokers []protocol.Invoker
+
+	url, _ := common.NewURL(fmt.Sprintf("dubbo://%s:%d/org.apache.demo.HelloService",
+		constant.LocalHostValue, constant.DefaultPort))
+	invokers = append(invokers, protocol.NewBaseInvoker(url))
+	i := loadBalance.Select(invokers, &invocation.RPCInvocation{})
+	assert.True(t, i.GetURL().URLEqual(url))
+
+	for i := 1; i < 10; i++ {
+		url, _ := common.NewURL(fmt.Sprintf("dubbo://192.168.1.%v:20000/org.apache.demo.HelloService", i))
+		invokers = append(invokers, protocol.NewBaseInvoker(url))
+	}
+	loadBalance.Select(invokers, &invocation.RPCInvocation{})
+}
+
+func TestMaglevConshashLoadBalance(t *testing.T) {
+	loadBalance := newMaglevConsistentHashingLoadBalance()
+
+	var invokers []protocol.Invoker
+	loop := 10
+	for i := 1; i <= loop; i++ {
+		url, _ := common.NewURL(fmt.Sprintf("dubbo://192.168.1.%v:20000/org.apache.demo.HelloService?weight=%v", i, i))
+		invokers = append(invokers, protocol.NewBaseInvoker(url))
+	}
+
+	loop = (1 + loop) * loop / 2
+	selected := make(map[protocol.Invoker]int)
+
+	for i := 1; i <= loop; i++ {
+		invoker := loadBalance.Select(invokers, &invocation.RPCInvocation{})
+		selected[invoker]++
+	}
+
+	sum := 0
+	for _, value := range selected {
+		sum += value
+	}
+
+	assert.Equal(t, loop, sum)
+}

cluster/loadbalance/maglevconsistenthashing/maglev.go

@@ -0,0 +1,126 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package maglevconsistenthashing
+
+import (
+	"hash/fnv"
+
+	"dubbo.apache.org/dubbo-go/v3/protocol"
+)
+
+type Maglev struct {
+	invokers    []protocol.Invoker // Instance
+	lookupTable []int
+
+	tableSize int // Prime number
+}
+
+func NewMaglev(invokers []protocol.Invoker) *Maglev {
+	// Next invokers Prime number
+	pn := NextPrime(len(invokers))
+	m := &Maglev{
+		lookupTable: make([]int, pn),
+		invokers:    invokers,
+		tableSize:   pn,
+	}
+	m.buildLookupTable()
+	return m
+}
+
+// buildLookupTable 构建 Maglev 查找表
+func (m *Maglev) buildLookupTable() {
+	permutations := m.generatePermutations()
+	next := make([]int, len(m.invokers))
+	inserted := 0
+
+	for i := range m.lookupTable {
+		m.lookupTable[i] = -1
+	}
+
+	for inserted < m.tableSize {
+		for i := range m.invokers {
+			if inserted >= m.tableSize {
+				break
+			}
+			offset := permutations[i][next[i]%m.tableSize]
+			for m.lookupTable[offset] != -1 {
+				next[i]++
+				offset = permutations[i][next[i]%m.tableSize]
+			}
+			m.lookupTable[offset] = i
+			next[i]++
+			inserted++
+		}
+	}
+}
+
+// generatePermutations 生成每个节点的排列
+func (m *Maglev) generatePermutations() [][]int {
+	permutations := make([][]int, len(m.invokers))
+	for i, node := range m.invokers {
+		permutations[i] = m.calculatePermutation(node)
+	}
+	return permutations
+}
+
+// calculatePermutation 计算单个节点的排列
+func (m *Maglev) calculatePermutation(node protocol.Invoker) []int {
+	u := node.GetURL()
+	address := u.Ip + ":" + u.Port
+	offset := hash(address) % uint64(m.tableSize)
+	skip := 1 + (hash(address+"skip") % uint64(m.tableSize-1))
+	permutation := make([]int, m.tableSize)
+	for i := 0; i < m.tableSize; i++ {
+		permutation[i] = int((offset + uint64(i)*skip) % uint64(m.tableSize))
+	}
+	return permutation
+}
+
+func (m *Maglev) Pick(key string) protocol.Invoker {
+	index := int(hash(key) % uint64(m.tableSize))
+	return m.invokers[m.lookupTable[index]]
+}
+
+// hash 简单的哈希函数示例
+func hash(s string) uint64 {
+	h := fnv.New64a()
+	h.Write([]byte(s))
+	return h.Sum64()
+}
+
+// isPrime 函数用于判断一个数是否为素数
+func isPrime(num int) bool {
+	if num < 2 {
+		return false
+	}
+	for i := 2; i*i <= num; i++ {
+		if num%i == 0 {
+			return false
+		}
+	}
+	return true
+}
+
+// NextPrime 函数用于找出比 n 大的第一个素数
+func NextPrime(n int) int {
+	num := n + 1
+	for !isPrime(num) {
+		num++
+	}
+	return num
+}

common/constant/loadbalance.go

@@ -26,4 +26,5 @@ const (
 	LoadXDSRingHash                             = "xdsringhash"
 	LoadBalanceKeyInterleavedWeightedRoundRobin = "interleavedweightedroundrobin"
 	LoadBalanceKeyAliasMethod                   = "aliasmethod"
+	LoadBalanceKeyMaglevConsistentHashing       = "maglevconsistenthashing"
 )