Skip to content

Commit 2f7d6cb

Browse files
Aias00Aias00
authored
Merge branch 'develop' into worktree-fix-exchange-client-init-race
2 parents 3dfa7fb + 93fe03a commit 2f7d6cb

37 files changed

Lines changed: 1116 additions & 367 deletions

‎cluster/loadbalance/aliasmethod/alias_method.go‎

Lines changed: 24 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -20,13 +20,16 @@ package aliasmethod // weighted random with alias-method algorithm
2020

2121
import (
2222
"math/rand"
23+
"time"
2324
)
2425

2526
import (
2627
"dubbo.apache.org/dubbo-go/v3/cluster/loadbalance"
2728
"dubbo.apache.org/dubbo-go/v3/protocol/base"
2829
)
2930

31+
const smallInvokerThreshold = 32
32+
3033
type aliasMethodPicker struct {
3134
invokers []base.Invoker // Instance
3235

@@ -46,18 +49,34 @@ func NewAliasMethodPicker(invokers []base.Invoker, invocation base.Invocation) *
4649
// Alias Method: https://en.wikipedia.org/wiki/Alias_method
4750
func (am *aliasMethodPicker) init(invocation base.Invocation) {
4851
n := len(am.invokers)
49-
weights := make([]int64, n)
5052
am.alias = make([]int, n)
5153
am.prob = make([]float64, n)
5254

5355
totalWeight := int64(0)
5456

55-
scaledProb := make([]float64, n)
56-
small := make([]int, 0, n)
57-
large := make([]int, 0, n)
57+
var (
58+
weightStack [smallInvokerThreshold]int64
59+
scaledProbStack [smallInvokerThreshold]float64
60+
smallStack [smallInvokerThreshold]int
61+
largeStack [smallInvokerThreshold]int
62+
)
63+
weights := weightStack[:]
64+
scaledProb := scaledProbStack[:]
65+
small := smallStack[:0]
66+
large := largeStack[:0]
67+
if n > smallInvokerThreshold {
68+
weights = make([]int64, n)
69+
scaledProb = make([]float64, n)
70+
small = make([]int, 0, n)
71+
large = make([]int, 0, n)
72+
} else {
73+
weights = weights[:n]
74+
scaledProb = scaledProb[:n]
75+
}
5876

77+
now := time.Now().Unix()
5978
for i, invoker := range am.invokers {
60-
weight := loadbalance.GetWeight(invoker, invocation)
79+
weight := loadbalance.GetWeightAt(invoker, invocation, now)
6180
weights[i] = weight
6281
totalWeight += weight
6382
}

‎cluster/loadbalance/consistenthashing/selector.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -85,7 +85,7 @@ func (c *selector) toKey(args []any) string {
8585
var sb strings.Builder
8686
for i := range c.argumentIndex {
8787
if i >= 0 && i < len(args) {
88-
_, _ = fmt.Fprint(&sb, args[i].(string))
88+
_, _ = fmt.Fprint(&sb, args[i])
8989
}
9090
}
9191
return sb.String()

‎cluster/loadbalance/iwrr/iwrr.go‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@ package iwrr
2020
import (
2121
"math/rand"
2222
"sync"
23+
"time"
2324
)
2425

2526
import (
@@ -85,9 +86,10 @@ func NewInterleavedweightedRoundRobin(invokers []base.Invoker, invocation base.I
8586
size := uint64(len(invokers))
8687
offset := rand.Uint64() % size //NOSONAR
8788
step := int64(0)
89+
now := time.Now().Unix()
8890
for idx := uint64(0); idx < size; idx++ {
8991
invoker := invokers[(idx+offset)%size]
90-
weight := loadbalance.GetWeight(invoker, invocation)
92+
weight := loadbalance.GetWeightAt(invoker, invocation, now)
9193
step = gcdInt(step, weight)
9294
iwrrp.current.push(&iwrrEntry{
9395
invoker: invoker,

‎cluster/loadbalance/leastactive/loadbalance.go‎

Lines changed: 24 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@ package leastactive
1919

2020
import (
2121
"math/rand"
22+
"time"
2223
)
2324

2425
import (
@@ -31,6 +32,9 @@ import (
3132
const (
3233
// Key is used to set the load balance extension
3334
Key = "leastactive"
35+
36+
minStackInvokerCount = 8
37+
maxStackInvokerCount = 32
3438
)
3539

3640
func init() {
@@ -57,21 +61,33 @@ func (lb *leastActiveLoadBalance) Select(invokers []base.Invoker, invocation bas
5761
}
5862

5963
var (
60-
leastActive int32 = -1 // The least active value of all invokers
61-
totalWeight int64 // The number of invokers having the same least active value (LEAST_ACTIVE)
62-
firstWeight int64 // Initial value, used for comparison
63-
leastCount int // The number of invokers having the same least active value (LEAST_ACTIVE)
64-
leastIndexes = make([]int, count) // The index of invokers having the same least active value (LEAST_ACTIVE)
65-
sameWeight = true // Every invoker has the same weight value?
66-
weights = make([]int64, count) // The weight of every invokers
64+
leastActive int32 = -1 // The least active value of all invokers
65+
totalWeight int64 // Sum of weights of invokers having the same least active value (LEAST_ACTIVE)
66+
firstWeight int64 // Initial value, used for comparison
67+
leastCount int // The number of invokers having the same least active value (LEAST_ACTIVE)
68+
leastIndexes []int // The index of invokers having the same least active value (LEAST_ACTIVE)
69+
sameWeight = true // Every invoker has the same weight value?
70+
weights []int64 // The weight of every invokers
71+
)
72+
var (
73+
leastIndexStack [maxStackInvokerCount]int
74+
weightStack [maxStackInvokerCount]int64
6775
)
76+
if count >= minStackInvokerCount && count <= maxStackInvokerCount {
77+
leastIndexes = leastIndexStack[:count]
78+
weights = weightStack[:count]
79+
} else {
80+
leastIndexes = make([]int, count)
81+
weights = make([]int64, count)
82+
}
6883

84+
now := time.Now().Unix()
6985
for i := 0; i < count; i++ {
7086
invoker := invokers[i]
7187
// Active number
7288
active := base.GetMethodStatus(invoker.GetURL(), invocation.MethodName()).GetActive()
7389
// current weight (maybe in warmUp)
74-
afterWarmup := loadbalance.GetWeight(invoker, invocation)
90+
afterWarmup := loadbalance.GetWeightAt(invoker, invocation, now)
7591
// save for later use
7692
weights[i] = afterWarmup
7793
// There are smaller active services

‎cluster/loadbalance/loadbalance_benchmarks_test.go‎

Lines changed: 70 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@ package loadbalance_test
2020
import (
2121
"fmt"
2222
"testing"
23+
"time"
2324
)
2425

2526
import (
@@ -47,6 +48,19 @@ func Generate() []base.Invoker {
4748
return invokers
4849
}
4950

51+
func generateInvokers(count int, weighted bool) []base.Invoker {
52+
invokers := make([]base.Invoker, 0, count)
53+
for i := 1; i <= count; i++ {
54+
rawURL := fmt.Sprintf("dubbo://192.168.1.%v:20000/org.apache.demo.HelloService", i)
55+
if weighted {
56+
rawURL = fmt.Sprintf("%s?weight=%d", rawURL, i)
57+
}
58+
url, _ := common.NewURL(rawURL)
59+
invokers = append(invokers, base.NewBaseInvoker(url))
60+
}
61+
return invokers
62+
}
63+
5064
func Benchloadbalance(b *testing.B, lb loadbalance.LoadBalance) {
5165
b.Helper()
5266
invokers := Generate()
@@ -84,3 +98,59 @@ func BenchmarkRandomLoadbalance(b *testing.B) {
8498
func BenchmarkAliasMethodLoadbalance(b *testing.B) {
8599
Benchloadbalance(b, extension.GetLoadbalance(constant.LoadBalanceKeyAliasMethod))
86100
}
101+
102+
func BenchmarkGetWeight(b *testing.B) {
103+
invokers := Generate()
104+
inv := &invocation.RPCInvocation{}
105+
b.ReportAllocs()
106+
b.ResetTimer()
107+
for i := 0; i < b.N; i++ {
108+
loadbalance.GetWeight(invokers[i%len(invokers)], inv)
109+
}
110+
}
111+
112+
func BenchmarkGetWeightAt(b *testing.B) {
113+
invokers := Generate()
114+
inv := &invocation.RPCInvocation{}
115+
now := time.Now().Unix()
116+
b.ReportAllocs()
117+
b.ResetTimer()
118+
for i := 0; i < b.N; i++ {
119+
loadbalance.GetWeightAt(invokers[i%len(invokers)], inv, now)
120+
}
121+
}
122+
123+
func benchmarkLoadBalanceSmallMedium(b *testing.B, lb loadbalance.LoadBalance) {
124+
b.Helper()
125+
for _, count := range []int{2, 4, 8, 16, 32, 33} {
126+
for _, weighted := range []bool{false, true} {
127+
name := fmt.Sprintf("invokers=%d", count)
128+
if weighted {
129+
name += "/weighted"
130+
} else {
131+
name += "/uniform"
132+
}
133+
b.Run(name, func(b *testing.B) {
134+
invokers := generateInvokers(count, weighted)
135+
rpcInvocation := &invocation.RPCInvocation{}
136+
b.ReportAllocs()
137+
b.ResetTimer()
138+
for i := 0; i < b.N; i++ {
139+
lb.Select(invokers, rpcInvocation)
140+
}
141+
})
142+
}
143+
}
144+
}
145+
146+
func BenchmarkRandomLoadbalanceSmallMedium(b *testing.B) {
147+
benchmarkLoadBalanceSmallMedium(b, extension.GetLoadbalance(constant.LoadBalanceKeyRandom))
148+
}
149+
150+
func BenchmarkLeastactiveLoadbalanceSmallMedium(b *testing.B) {
151+
benchmarkLoadBalanceSmallMedium(b, extension.GetLoadbalance(constant.LoadBalanceKeyLeastActive))
152+
}
153+
154+
func BenchmarkAliasMethodLoadbalanceSmallMedium(b *testing.B) {
155+
benchmarkLoadBalanceSmallMedium(b, extension.GetLoadbalance(constant.LoadBalanceKeyAliasMethod))
156+
}

‎cluster/loadbalance/random/loadbalance.go‎

Lines changed: 18 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@ package random
1919

2020
import (
2121
"math/rand"
22+
"time"
2223
)
2324

2425
import (
@@ -52,12 +53,27 @@ func (lb *randomLoadBalance) Select(invokers []base.Invoker, invocation base.Inv
5253
// Every invoker has the same weight?
5354
sameWeight := true
5455
// the maxWeight of every invokers, the minWeight = 0 or the maxWeight of the last invoker
55-
weights := make([]int64, length)
56+
// Use stack buffers for common small invoker lists to avoid the temporary weights slice allocation.
57+
var weights []int64
58+
switch {
59+
case length <= 8:
60+
var weightStack [8]int64
61+
weights = weightStack[:length:length]
62+
case length <= 16:
63+
var weightStack [16]int64
64+
weights = weightStack[:length:length]
65+
case length <= 32:
66+
var weightStack [32]int64
67+
weights = weightStack[:length:length]
68+
default:
69+
weights = make([]int64, length)
70+
}
5671
// The sum of weights
5772
var totalWeight int64 = 0
5873

74+
now := time.Now().Unix()
5975
for i := 0; i < length; i++ {
60-
weight := loadbalance.GetWeight(invokers[i], invocation)
76+
weight := loadbalance.GetWeightAt(invokers[i], invocation, now)
6177
//Sum
6278
totalWeight += weight
6379
// save for later use

‎cluster/loadbalance/roundrobin/loadbalance.go‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -74,12 +74,13 @@ func (lb *rrLoadBalance) Select(invokers []base.Invoker, invocation base.Invocat
7474
totalWeight = int64(0)
7575
maxCurrentWeight = int64(math.MinInt64)
7676
now = time.Now()
77+
nowUnix = now.Unix()
7778
selectedInvoker base.Invoker
7879
selectedWeightRobin *weightedRoundRobin
7980
)
8081

8182
for _, invoker := range invokers {
82-
weight := max(loadbalance.GetWeight(invoker, invocation), 0)
83+
weight := max(loadbalance.GetWeightAt(invoker, invocation, nowUnix), 0)
8384

8485
identifier := invoker.GetURL().Key()
8586
wr := &weightedRoundRobin{weight: weight}

‎cluster/loadbalance/util.go‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -28,7 +28,14 @@ import (
2828

2929
// GetWeight returns the weight for the load‑balancing strategy.
3030
func GetWeight(invoker base.Invoker, invocation base.Invocation) int64 {
31+
return GetWeightAt(invoker, invocation, time.Now().Unix())
32+
}
3133

34+
// GetWeightAt returns the weight for the load-balancing strategy using the
35+
// provided Unix timestamp. Callers that loop over many invokers should compute
36+
// now once and pass it here to avoid repeated time.Now() calls and to keep the
37+
// warmup calculation consistent within a single selection.
38+
func GetWeightAt(invoker base.Invoker, invocation base.Invocation, now int64) int64 {
3239
url := invoker.GetURL()
3340

3441
// Method‑level or registry‑level weight taken from URL parameters — highest priority.
@@ -46,7 +53,6 @@ func GetWeight(invoker base.Invoker, invocation base.Invocation) int64 {
4653

4754
// Warm‑up adjustment (same logic as before).
4855
if weight > 0 {
49-
now := time.Now().Unix()
5056
ts := url.GetParamInt(constant.RemoteTimestampKey, now)
5157
if uptime := now - ts; uptime > 0 {
5258
warm := url.GetParamInt(constant.WarmupKey, constant.DefaultWarmup)

‎cluster/router/script/router.go‎

Lines changed: 14 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -70,9 +70,22 @@ func (s *ScriptRouter) Process(event *config_center.ConfigChangeEvent) {
7070
s.mu.Lock()
7171
defer s.mu.Unlock()
7272

73+
if event.ConfigType == remoting.EventTypeDel {
74+
in, _ := ins.GetInstances(s.scriptType)
75+
76+
if in != nil && s.enabled {
77+
in.Destroy(s.rawScript)
78+
}
79+
s.enabled = false
80+
s.rawScript = ""
81+
s.scriptType = ""
82+
return
83+
}
84+
7385
rawConf, ok := event.Value.(string)
7486
if !ok {
75-
panic(ok)
87+
logger.Errorf("[Router][Script] route config value must be string, actualType=%T", event.Value)
88+
return
7689
}
7790
cfg, err := parseRoute(rawConf)
7891
if err != nil {
@@ -128,15 +141,6 @@ func (s *ScriptRouter) Process(event *config_center.ConfigChangeEvent) {
128141
}
129142
}
130143

131-
case remoting.EventTypeDel:
132-
in, _ := ins.GetInstances(s.scriptType)
133-
134-
if in != nil && s.enabled {
135-
in.Destroy(s.rawScript)
136-
}
137-
s.enabled = false
138-
s.rawScript = ""
139-
s.scriptType = ""
140144
}
141145
}
142146

‎cluster/router/script/router_test.go‎

Lines changed: 30 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -227,6 +227,36 @@ script: |
227227
}
228228
}
229229

230+
func TestScriptRouterProcessSkipsNonStringConfig(t *testing.T) {
231+
s := &ScriptRouter{
232+
enabled: true,
233+
scriptType: "javascript",
234+
rawScript: "old script",
235+
}
236+
237+
assert.NotPanics(t, func() {
238+
s.Process(&config_center.ConfigChangeEvent{Key: "", Value: 123, ConfigType: remoting.EventTypeUpdate})
239+
})
240+
assert.True(t, s.enabled)
241+
assert.Equal(t, "javascript", s.scriptType)
242+
assert.Equal(t, "old script", s.rawScript)
243+
}
244+
245+
func TestScriptRouterProcessDelSkipsConfigBody(t *testing.T) {
246+
s := &ScriptRouter{
247+
enabled: true,
248+
scriptType: "javascript",
249+
rawScript: "old script",
250+
}
251+
252+
assert.NotPanics(t, func() {
253+
s.Process(&config_center.ConfigChangeEvent{Key: "", Value: nil, ConfigType: remoting.EventTypeDel})
254+
})
255+
assert.False(t, s.enabled)
256+
assert.Empty(t, s.scriptType)
257+
assert.Empty(t, s.rawScript)
258+
}
259+
230260
func checkInvokersSame(invokers []base.Invoker, otherInvokers []base.Invoker) bool {
231261
k := map[string]struct{}{}
232262
for _, invoker := range otherInvokers {

0 commit comments

Comments
 (0)