go/kratos
1
0
Fork 0
mirror of https://github.com/go-kratos/kratos.git synced 2025-03-17 21:07:54 +02:00
Code Issues Releases Activity

style:ewma (#2018)

Browse Source 
* Update node.go

* Update node.go

* Update node.go

* Update node.go
This commit is contained in:
songzhibin97 2022-05-21 16:26:22 +08:00 committed by GitHub
parent 92ba6a3209
commit 0c511808ce
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 32 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

65
selector/node/ewma/node.go
View File

@ -9,6 +9,7 @@ import (
"time"
"github.com/go-kratos/kratos/v2/errors"
"github.com/go-kratos/kratos/v2/selector"
)
@ -75,45 +76,44 @@ func (n *Node) load() (load uint64) {
predictInterval := avgLag / 5
if predictInterval < int64(time.Millisecond*5) {
predictInterval = int64(time.Millisecond * 5)
} else if predictInterval > int64(time.Millisecond*200) {
}
if predictInterval > int64(time.Millisecond*200) {
predictInterval = int64(time.Millisecond * 200)
}
if now-lastPredictTs > predictInterval {
if atomic.CompareAndSwapInt64(&n.predictTs, lastPredictTs, now) {
var (
total int64
count int
predict int64
)
n.lk.RLock()
first := n.inflights.Front()
for first != nil {
lag := now - first.Value.(int64)
if lag > avgLag {
count++
total += lag
}
first = first.Next()
if now-lastPredictTs > predictInterval && atomic.CompareAndSwapInt64(&n.predictTs, lastPredictTs, now) {
var (
total int64
count int
predict int64
)
n.lk.RLock()
first := n.inflights.Front()
for first != nil {
lag := now - first.Value.(int64)
if lag > avgLag {
count++
total += lag
}
if count > (n.inflights.Len()/2 + 1) {
predict = total / int64(count)
}
n.lk.RUnlock()
atomic.StoreInt64(&n.predict, predict)
first = first.Next()
}
if count > (n.inflights.Len()/2 + 1) {
predict = total / int64(count)
}
n.lk.RUnlock()
atomic.StoreInt64(&n.predict, predict)
}
if avgLag == 0 {
// penalty is the penalty value when there is no data when the node is just started.
// The default value is 1e9 * 10
load = penalty * uint64(atomic.LoadInt64(&n.inflight))
} else {
predict := atomic.LoadInt64(&n.predict)
if predict > avgLag {
avgLag = predict
}
load = uint64(avgLag) * uint64(atomic.LoadInt64(&n.inflight))
return
}
predict := atomic.LoadInt64(&n.predict)
if predict > avgLag {
avgLag = predict
}
load = uint64(avgLag) * uint64(atomic.LoadInt64(&n.inflight))
return
}
@ -155,11 +155,10 @@ func (n *Node) Pick() selector.DoneFunc {
success := uint64(1000) // error value ,if error set 1
if di.Err != nil {
if n.errHandler != nil {
if n.errHandler(di.Err) {
success = 0
}
} else if errors.Is(context.DeadlineExceeded, di.Err) || errors.Is(context.Canceled, di.Err) ||
if n.errHandler != nil && n.errHandler(di.Err) {
success = 0
}
if errors.Is(context.DeadlineExceeded, di.Err) || errors.Is(context.Canceled, di.Err) ||
errors.IsServiceUnavailable(di.Err) || errors.IsGatewayTimeout(di.Err) {
success = 0
}