Leastload: Impl missing Tolerance

app/router/strategy_leastload.go

@@ -3,6 +3,7 @@ package router
 import (
 	"context"
 	"math"
+	"slices"
 	"sort"
 	"time"
 
@@ -77,7 +78,7 @@ func (s *LeastLoadStrategy) PickOutbound(candidates []string) string {
 }
 
 func (s *LeastLoadStrategy) pickOutbounds(candidates []string) []*node {
-	qualified := s.getNodes(candidates, time.Duration(s.settings.MaxRTT))
+	qualified := s.getNodes(candidates)
 	selects := s.selectLeastLoad(qualified)
 	return selects
 }
@@ -138,7 +139,7 @@ func (s *LeastLoadStrategy) selectLeastLoad(nodes []*node) []*node {
 	return nodes[:count]
 }
 
-func (s *LeastLoadStrategy) getNodes(candidates []string, maxRTT time.Duration) []*node {
+func (s *LeastLoadStrategy) getNodes(candidates []string) []*node {
 	if s.observer == nil {
 		errors.LogError(s.ctx, "observer is nil")
 		return make([]*node, 0)
@@ -151,12 +152,10 @@ func (s *LeastLoadStrategy) getNodes(candidates []string, maxRTT time.Duration)
 
 	results := observeResult.(*observatory.ObservationResult)
 
-	outboundlist := outboundList(candidates)
-
 	var ret []*node
 
 	for _, v := range results.Status {
-		if v.Alive && (v.Delay < maxRTT.Milliseconds() || maxRTT == 0) && outboundlist.contains(v.OutboundTag) {
+		if s.shouldSelectNode(v, candidates) {
 			record := &node{
 				Tag:              v.OutboundTag,
 				CountAll:         1,
@@ -172,8 +171,8 @@ func (s *LeastLoadStrategy) getNodes(candidates []string, maxRTT time.Duration)
 				record.RTTDeviationCost = time.Duration(s.costs.Apply(v.OutboundTag, float64(v.HealthPing.Deviation)))
 				record.CountAll = int(v.HealthPing.All)
 				record.CountFail = int(v.HealthPing.Fail)
-
 			}
+
 			ret = append(ret, record)
 		}
 	}
@@ -182,6 +181,23 @@ func (s *LeastLoadStrategy) getNodes(candidates []string, maxRTT time.Duration)
 	return ret
 }
 
+func (s *LeastLoadStrategy) shouldSelectNode(v *observatory.OutboundStatus, candidates []string) bool {
+	maxRTT := time.Duration(s.settings.MaxRTT)
+	if !v.Alive {
+		return false
+	}
+	if maxRTT != 0 && v.Delay >= maxRTT.Milliseconds() {
+		return false
+	}
+	if !slices.Contains(candidates, v.OutboundTag) {
+		return false
+	}
+	if v.HealthPing != nil && v.HealthPing.All > 0 && s.settings.Tolerance > 0 && float64(v.HealthPing.Fail)/float64(v.HealthPing.All) > float64(s.settings.Tolerance) {
+		return false
+	}
+	return true
+}
+
 func leastloadSort(nodes []*node) {
 	sort.Slice(nodes, func(i, j int) bool {
 		left := nodes[i]

common/signal/timer.go

@@ -3,7 +3,11 @@ package signal
 import (
 	"context"
 	"sync"
+	"sync/atomic"
 	"time"
+
+	"github.com/xtls/xray-core/common"
+	"github.com/xtls/xray-core/common/task"
 )
 
 type ActivityUpdater interface {
@@ -11,35 +15,45 @@ type ActivityUpdater interface {
 }
 
 type ActivityTimer struct {
-	mu sync.Mutex
+	mu        sync.RWMutex
-	// timer will be nil if this timer is already finished
+	updated   chan struct{}
-	timer     *time.Timer
+	checkTask *task.Periodic
-	timeout   time.Duration
 	onTimeout func()
+	consumed  atomic.Bool
+	once      sync.Once
 }
 
 func (t *ActivityTimer) Update() {
-	// someone already called Update or closing, just return
+	select {
-	if !t.mu.TryLock() {
+	case t.updated <- struct{}{}:
-		return
+	default:
 	}
-	defer t.mu.Unlock()
+}
-	if t.timer != nil {
+
-		t.timer.Reset(t.timeout)
+func (t *ActivityTimer) check() error {
+	select {
+	case <-t.updated:
+	default:
+		t.finish()
 	}
+	return nil
 }
 
 func (t *ActivityTimer) finish() {
+	t.once.Do(func() {
+		t.consumed.Store(true)
 		t.mu.Lock()
 		defer t.mu.Unlock()
-	if t.timer != nil {
+
-		t.timer.Stop()
+		common.CloseIfExists(t.checkTask)
 		t.onTimeout()
-		t.timer = nil
+	})
-	}
 }
 
 func (t *ActivityTimer) SetTimeout(timeout time.Duration) {
+	if t.consumed.Load() {
+		return
+	}
 	if timeout == 0 {
 		t.finish()
 		return
@@ -47,22 +61,25 @@ func (t *ActivityTimer) SetTimeout(timeout time.Duration) {
 
 	t.mu.Lock()
 	defer t.mu.Unlock()
-	if t.timer != nil {
+	// double check, just in case
-		t.timeout = timeout
+	if t.consumed.Load() {
-		t.timer.Reset(timeout)
+		return
 	}
+	newCheckTask := &task.Periodic{
+		Interval: timeout,
+		Execute:  t.check,
+	}
+	common.CloseIfExists(t.checkTask)
+	t.checkTask = newCheckTask
+	t.Update()
+	common.Must(newCheckTask.Start())
 }
 
 func CancelAfterInactivity(ctx context.Context, cancel context.CancelFunc, timeout time.Duration) *ActivityTimer {
-	activityTimer := &ActivityTimer{
+	timer := &ActivityTimer{
-		timeout:   timeout,
+		updated:   make(chan struct{}, 1),
 		onTimeout: cancel,
 	}
-	// strange situation
+	timer.SetTimeout(timeout)
-	if timeout == 0 {
+	return timer
-		cancel()
-		return activityTimer
-	}
-	activityTimer.timer = time.AfterFunc(timeout, activityTimer.finish)
-	return activityTimer
 }