mirrors/dae
1
0
Fork 0
mirror of https://github.com/daeuniverse/dae.git synced 2025-07-04 07:17:55 +07:00
Code Issues Packages Projects Releases Wiki Activity

feat: add min moving average latencies policy

Browse Source
This commit is contained in:
mzz2017
2023-02-19 01:49:36 +08:00
parent 500248ea5c
commit 156769bf55
6 changed files with 31 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
component/outbound/dialer/alive_dialer_set.go
View File

@ -114,10 +114,14 @@ func (a *AliveDialerSet) NotifyLatencyChange(dialer *Dialer, alive bool) {
switch a.selectionPolicy {
case consts.DialerSelectionPolicy_MinLastLatency:
latency, hasLatency = dialer.MustGetLatencies10(a.CheckTyp).LastLatency()
latency, hasLatency = dialer.mustGetCollection(a.CheckTyp).Latencies10.LastLatency()
minPolicy = true
case consts.DialerSelectionPolicy_MinAverage10Latencies:
latency, hasLatency = dialer.MustGetLatencies10(a.CheckTyp).AvgLatency()
latency, hasLatency = dialer.mustGetCollection(a.CheckTyp).Latencies10.AvgLatency()
minPolicy = true
case consts.DialerSelectionPolicy_MinMovingAverageLatencies:
latency = dialer.mustGetCollection(a.CheckTyp).MovingAverage
hasLatency = latency > 0
minPolicy = true
}

19
component/outbound/dialer/connectivity_check.go
View File

@ -49,6 +49,7 @@ type collection struct {
// AliveDialerSetSet uses reference counting.
AliveDialerSetSet AliveDialerSetSet
Latencies10 *LatenciesN
MovingAverage time.Duration
Alive bool
}
@ -93,10 +94,6 @@ func (d *Dialer) mustGetCollection(typ *NetworkType) *collection {
panic("invalid param")
}
func (d *Dialer) MustGetLatencies10(typ *NetworkType) *LatenciesN {
return d.mustGetCollection(typ).Latencies10
}
func (d *Dialer) MustGetAlive(typ *NetworkType) bool {
return d.mustGetCollection(typ).Alive
}
@ -447,16 +444,18 @@ func (d *Dialer) Check(timeout time.Duration,
if ok, err = opts.CheckFunc(ctx, opts.networkType); ok && err == nil {
// No error.
latency := time.Since(start)
latencies10 := d.mustGetCollection(opts.networkType).Latencies10
latencies10.AppendLatency(latency)
avg, _ := latencies10.AvgLatency()
collection.Latencies10.AppendLatency(latency)
avg, _ := collection.Latencies10.AvgLatency()
collection.MovingAverage = (collection.MovingAverage + latency) / 2
collection.Alive = true
d.Log.WithFields(logrus.Fields{
"network": opts.networkType.String(),
"node": d.name,
"last": latency.Truncate(time.Millisecond),
"avg_10": avg.Truncate(time.Millisecond),
"mov_avg": collection.MovingAverage.Truncate(time.Millisecond),
}).Debugln("Connectivity Check")
collection.Alive = true
} else {
// Append timeout if there is any error or unexpected status code.
if err != nil {
@ -472,8 +471,8 @@ func (d *Dialer) Check(timeout time.Duration,
"err": err.Error(),
}).Debugln("Connectivity Check Failed")
}
latencies10 := collection.Latencies10
latencies10.AppendLatency(timeout)
collection.Latencies10.AppendLatency(timeout)
collection.MovingAverage = (collection.MovingAverage + timeout) / 2
collection.Alive = false
}
// Inform DialerGroups to update state.

7
component/outbound/dialer_group.go
View File

@ -41,7 +41,8 @@ func NewDialerGroup(option *dialer.GlobalOption, name string, dialers []*dialer.
switch p.Policy {
case consts.DialerSelectionPolicy_Random,
consts.DialerSelectionPolicy_MinLastLatency,
consts.DialerSelectionPolicy_MinAverage10Latencies:
consts.DialerSelectionPolicy_MinAverage10Latencies,
consts.DialerSelectionPolicy_MinMovingAverageLatencies:
// Need to know the alive state or latency.
networkType := &dialer.NetworkType{
@ -212,7 +213,9 @@ func (g *DialerGroup) Select(networkType *dialer.NetworkType) (d *dialer.Dialer,
}
return g.Dialers[g.selectionPolicy.FixedIndex], 0, nil
case consts.DialerSelectionPolicy_MinLastLatency, consts.DialerSelectionPolicy_MinAverage10Latencies:
case consts.DialerSelectionPolicy_MinLastLatency,
consts.DialerSelectionPolicy_MinAverage10Latencies,
consts.DialerSelectionPolicy_MinMovingAverageLatencies:
d, latency := a.GetMinLatency()
if d == nil {
// No alive dialer.

3
component/outbound/dialer_selection_policy.go
View File

@ -25,7 +25,8 @@ func NewDialerSelectionPolicyFromGroupParam(param *config.GroupParam) (policy *D
switch consts.DialerSelectionPolicy(val) {
case consts.DialerSelectionPolicy_Random,
consts.DialerSelectionPolicy_MinAverage10Latencies,
consts.DialerSelectionPolicy_MinLastLatency:
consts.DialerSelectionPolicy_MinLastLatency,
consts.DialerSelectionPolicy_MinMovingAverageLatencies:
return &DialerSelectionPolicy{
Policy: consts.DialerSelectionPolicy(val),
}, nil