feat(cluster): add more fields to vegas capeva

filter/adaptivesvc/capeva/capacity_evaluator.go

@@ -2,10 +2,10 @@ package capeva
 
 type CapacityEvaluator interface {
 	// EstimatedCapacity is estimated capacity, which reflects the maximum requests handled by the provider.
-	EstimatedCapacity() int64
+	EstimatedCapacity() uint64
 
 	// ActualCapacity is actual requests on the provider.
-	ActualCapacity() int64
+	ActualCapacity() uint64
 
 	// NewCapacityUpdater returns a capacity updater
 	NewCapacityUpdater() CapacityUpdater

filter/adaptivesvc/capeva/utils.go

@@ -0,0 +1,35 @@
+package capeva
+
+import "go.uber.org/atomic"
+
+func after(crt, next *atomic.Uint64) bool {
+	return crt.Load() > next.Load()
+}
+
+// setValueIfLess sets newValue to v if newValue is less than v
+func setValueIfLess(v *atomic.Uint64, newValue uint64) {
+	vuint64 := v.Load()
+	for vuint64 > newValue {
+		if !v.CAS(vuint64, newValue) {
+			vuint64 = v.Load()
+		}
+	}
+}
+
+func slowStart(est, thresh *atomic.Uint64) {
+	var estValue, threshValue, newEst uint64
+
+	for {
+		estValue = est.Load()
+		threshValue = thresh.Load()
+
+		newEst = estValue * 2
+		if threshValue < newEst {
+			newEst = threshValue
+		}
+
+		if est.CAS(estValue, newEst) {
+			break
+		}
+	}
+}

filter/adaptivesvc/capeva/vegas.go

@@ -2,49 +2,72 @@ package capeva
 
 import (
 	"go.uber.org/atomic"
+	"math"
 )
 
 const (
-	defaultRoundSize uint64 = 100
+	defaultThreshold uint64 = 100
+
+	defaultRoundSize uint64 = 10
+
+	defaultAlpha uint64 = 2
+	defaultBeta  uint64 = 4
+	defaultGamma uint64 = 1
 )
 
 // Vegas is a capacity evaluator using Vegas congestion avoid.
 // RTT is not exactly same as TCP's,
 // in this case, RTT means the time that the provider perform a method.
 type Vegas struct {
-	Estimated, Actual *atomic.Uint64
+	Estimated, Actual, Threshold *atomic.Uint64
+
+	// TODO(justxuewei): load values from config
+	Alpha, Beta, Gamma uint64
 
 	// RoundSize specifies the size of the round, which reflects on the speed of updating estimation.
 	// The MinRTT and CntRTT will be reset in the next round.
+	// The smaller RoundSize is, the faster reevaluating estimated capacity is.
 	RoundSize uint64
 
-	Seq          *atomic.Uint64
-	NextRoundSeq *atomic.Uint64
+	Seq                *atomic.Uint64
+	NextRoundLeftBound *atomic.Uint64
 
 	BaseRTT *atomic.Uint64
 	MinRTT  *atomic.Uint64
-	CntRTT  *atomic.Uint64
+	CntRTT  *atomic.Uint64 // not used so far
 }
 
 func NewVegas() *Vegas {
+	estimated := atomic.NewUint64(1)
+	threshold := atomic.NewUint64(defaultThreshold)
+
+	minRTT := atomic.NewUint64(math.MaxUint64)
+
 	return &Vegas{
-		Estimated:    &atomic.Uint64{},
-		Actual:       &atomic.Uint64{},
-		RoundSize:    defaultRoundSize,
-		Seq:          &atomic.Uint64{},
-		NextRoundSeq: &atomic.Uint64{},
-		BaseRTT:      &atomic.Uint64{},
-		MinRTT:       &atomic.Uint64{},
-		CntRTT:       &atomic.Uint64{},
+		Estimated: estimated,
+		Actual:    &atomic.Uint64{},
+		Threshold: threshold,
+
+		Alpha:     defaultAlpha,
+		Beta:      defaultBeta,
+		Gamma:     defaultGamma,
+		RoundSize: defaultRoundSize,
+
+		Seq:                &atomic.Uint64{},
+		NextRoundLeftBound: &atomic.Uint64{},
+
+		BaseRTT: &atomic.Uint64{},
+		MinRTT:  minRTT,
+		CntRTT:  &atomic.Uint64{},
 	}
 }
 
-func (v *Vegas) EstimatedCapacity() int64 {
-	panic("implement me")
+func (v *Vegas) EstimatedCapacity() uint64 {
+	return v.Estimated.Load()
 }
 
-func (v *Vegas) ActualCapacity() int64 {
-	panic("implement me")
+func (v *Vegas) ActualCapacity() uint64 {
+	return v.Actual.Load()
 }
 
 func (v *Vegas) NewCapacityUpdater() CapacityUpdater {

filter/adaptivesvc/capeva/vegas_updater.go

@@ -1,6 +1,10 @@
 package capeva
 
-import "time"
+import (
+	"go.uber.org/atomic"
+	"math"
+	"time"
+)
 
 type vegasUpdater struct {
 	eva *Vegas
@@ -15,28 +19,52 @@ func newVegasUpdater(eva *Vegas) CapacityUpdater {
 	}
 }
 
-func (v *vegasUpdater) Succeed() {
-	v.updateAfterReturn()
+func (u *vegasUpdater) Succeed() {
+	u.updateAfterReturn()
 }
 
-func (v *vegasUpdater) Failed() {
-	v.updateAfterReturn()
+func (u *vegasUpdater) Failed() {
+	u.updateAfterReturn()
 }
 
-func (v *vegasUpdater) updateAfterReturn() {
-	v.eva.Actual.Add(-1)
-	v.updateRTTs()
+func (u *vegasUpdater) updateAfterReturn() {
+	u.eva.Actual.Add(-1)
+	u.updateRTTs()
+
+	u.reevaEstCap()
 }
 
-func (v *vegasUpdater) updateRTTs() {
+func (u *vegasUpdater) updateRTTs() {
 	// update BaseRTT
-	curRTT := uint64(time.Now().Sub(v.startedTime))
-	oldBaseRTT := v.eva.BaseRTT.Load()
-	for oldBaseRTT > curRTT {
-		if !v.eva.BaseRTT.CAS(oldBaseRTT, curRTT) {
-			oldBaseRTT = v.eva.BaseRTT.Load()
-		}
-	}
+	curRTT := uint64(time.Now().Sub(u.startedTime))
+	setValueIfLess(u.eva.BaseRTT, curRTT)
 	// update MinRTT
+	setValueIfLess(u.eva.MinRTT, curRTT)
 	// update CntRTT
+	u.eva.CntRTT.Add(1)
+}
+
+// reevaEstCap reevaluates estimated capacity if the round
+func (u *vegasUpdater) reevaEstCap() {
+	if after(u.eva.Seq, u.eva.NextRoundLeftBound) {
+		// update next round
+		u.eva.NextRoundLeftBound.Add(u.eva.RoundSize)
+
+		rtt := u.eva.MinRTT.Load()
+		baseRTT := u.eva.BaseRTT.Load()
+
+		thresh := u.eva.Threshold.Load()
+		estCap := u.eva.EstimatedCapacity()
+
+		targetEstCap := estCap * baseRTT / rtt
+		diff := estCap * (rtt - baseRTT) / baseRTT
+
+		if diff > u.eva.Gamma && estCap <= thresh {
+
+		}
+
+		// reset MinRTT & CntRTT
+		u.eva.MinRTT.Store(math.MaxUint64)
+		u.eva.CntRTT.Store(0)
+	}
 }