Optimize DynamicRateLimiter to not constantly re-evaluate RPS

While dynamic config lookups are relatively cheap, they're certainly not free and perform several allocations further contributing to GC times. Making matters worse, quotas.RateLimiter has some strange TTL logic such that the result of evaluating the dynamic config value isn't used more than once a minute unless it's lower than the current value.

Delete quotas.RateLimiter in favor of clock.RateLimiter and move the TTL to DynamicRateLimiter. Reduce the TTL to a more reasonable value and only evaluate the function if the time has elapsed.

Remove the logic allowing the rate change to bypass the TTL if its lower than the current rate. This requires evaluating the RPS value constantly. Instead we've shortened the TTL such that we'll reliably pick up changes within a few seconds regardless of the direction of the change.

The main place that the TTL logic seems to be relevant is the Task Dispatch limiter within Matching. Each poller includes an RPS limit and we'd attempt to update the RPS on each request. This is the only place that explicitly provided a TTL to quotas.RateLimiter (60s) rather than relying on the default.

Change the Matching Rate limiter to use DynamicRateLimiter so that it also only updates according to its TTL. This is a change in behavior and will make Matching less responsive to changes specified by user requests. It still complies with the "take most recent value" behavior that is advertised.

Author: natemort

common/clock/ratelimiter.go

@@ -70,6 +70,8 @@ type (
 		SetBurst(newBurst int)
 		// SetLimit sets the Limit value
 		SetLimit(newLimit rate.Limit)
+
+		SetLimitAndBurst(newLimit rate.Limit, newBurst int)
 		// Tokens returns the number of immediately-allowable events when called.
 		// Values >= 1 will lead to Allow returning true.
 		//
@@ -338,6 +340,14 @@ func (r *ratelimiter) SetLimit(newLimit rate.Limit) {
 	r.limiter.SetLimitAt(r.latestNow, newLimit)
 }
 
+func (r *ratelimiter) SetLimitAndBurst(newLimit rate.Limit, newBurst int) {
+	r.mut.Lock()
+	defer r.mut.Unlock()
+	// See SetLimit and SetBurst for more information
+	r.limiter.SetLimitAt(r.latestNow, newLimit)
+	r.limiter.SetBurstAt(r.latestNow, newBurst)
+}
+
 func (r *ratelimiter) Tokens() float64 {
 	now, unlock := r.lockNow()
 	defer unlock()

common/dynamicconfig/quotas/fallbackdynamicratelimiterfactory.go

@@ -49,15 +49,9 @@ type fallbackDynamicRateLimiterFactory struct {
 // GetLimiter returns a new Limiter for the given domain
 func (f fallbackDynamicRateLimiterFactory) GetLimiter(domain string) quotas.Limiter {
 	return quotas.NewDynamicRateLimiter(func() float64 {
-		return limitWithFallback(
-			float64(f.primary(domain)),
-			float64(f.secondary()))
+		if primaryLimit := f.primary(domain); primaryLimit > 0 {
+			return float64(primaryLimit)
+		}
+		return float64(f.secondary())
 	})
 }
-
-func limitWithFallback(primary, secondary float64) float64 {
-	if primary > 0 {
-		return primary
-	}
-	return secondary
-}

common/quotas/dynamicratelimiter.go

@@ -24,48 +24,103 @@ package quotas
 
 import (
 	"context"
+	"math"
+	"sync/atomic"
+	"time"
 
 	"golang.org/x/time/rate"
 
 	"github.com/uber/cadence/common/clock"
 )
 
+const _ttl = time.Second * 5
+const _minBurst = 1
+
 // DynamicRateLimiter implements a dynamic config wrapper around the rate limiter,
 // checks for updates to the dynamic config and updates the rate limiter accordingly
 type DynamicRateLimiter struct {
-	rps RPSFunc
-	rl  *RateLimiter
+	rps            RPSFunc
+	rl             clock.Ratelimiter
+	timeSource     clock.TimeSource
+	ttl            time.Duration
+	lastUpdateTime atomic.Value
+	minBurst       int
+}
+
+type DynamicRateLimiterOpt = func(limiter *DynamicRateLimiter)
+
+func WithTTL(ttl time.Duration) DynamicRateLimiterOpt {
+	return func(limiter *DynamicRateLimiter) {
+		limiter.ttl = ttl
+	}
+}
+
+func WithMinBurst(minBurst int) DynamicRateLimiterOpt {
+	return func(limiter *DynamicRateLimiter) {
+		limiter.minBurst = minBurst
+	}
+}
+
+func WithTimeSource(timeSource clock.TimeSource) DynamicRateLimiterOpt {
+	return func(limiter *DynamicRateLimiter) {
+		limiter.timeSource = timeSource
+	}
 }
 
 // NewDynamicRateLimiter returns a rate limiter which handles dynamic config
-func NewDynamicRateLimiter(rps RPSFunc) *DynamicRateLimiter {
-	initialRps := rps()
-	rl := NewRateLimiter(&initialRps, _defaultRPSTTL, _burstSize)
-	return &DynamicRateLimiter{rps, rl}
+func NewDynamicRateLimiter(rps RPSFunc, opts ...DynamicRateLimiterOpt) Limiter {
+	res := &DynamicRateLimiter{
+		rps:        rps,
+		timeSource: clock.NewRealTimeSource(),
+		ttl:        _ttl,
+		minBurst:   _minBurst,
+	}
+	for _, opt := range opts {
+		opt(res)
+	}
+	res.lastUpdateTime.Store(res.timeSource.Now())
+	res.rl = clock.NewRatelimiter(res.getLimitAndBurst())
+	return res
 }
 
 // Allow immediately returns with true or false indicating if a rate limit
 // token is available or not
 func (d *DynamicRateLimiter) Allow() bool {
-	rps := d.rps()
-	d.rl.UpdateMaxDispatch(&rps)
+	d.maybeRefreshRps()
 	return d.rl.Allow()
 }
 
 // Wait waits up till deadline for a rate limit token
 func (d *DynamicRateLimiter) Wait(ctx context.Context) error {
-	rps := d.rps()
-	d.rl.UpdateMaxDispatch(&rps)
+	d.maybeRefreshRps()
 	return d.rl.Wait(ctx)
 }
 
 // Reserve reserves a rate limit token
 func (d *DynamicRateLimiter) Reserve() clock.Reservation {
-	rps := d.rps()
-	d.rl.UpdateMaxDispatch(&rps)
+	d.maybeRefreshRps()
 	return d.rl.Reserve()
 }
 
 func (d *DynamicRateLimiter) Limit() rate.Limit {
-	return rate.Limit(d.rps())
+	d.maybeRefreshRps()
+	return d.rl.Limit()
+}
+
+func (d *DynamicRateLimiter) maybeRefreshRps() {
+	now := d.timeSource.Now()
+	lastUpdated := d.lastUpdateTime.Load().(time.Time)
+	if now.After(lastUpdated.Add(d.ttl-1)) && d.lastUpdateTime.CompareAndSwap(lastUpdated, now) {
+		d.rl.SetLimitAndBurst(d.getLimitAndBurst())
+	}
+}
+
+func (d *DynamicRateLimiter) getLimitAndBurst() (rate.Limit, int) {
+	rps := d.rps()
+	burst := max(int(math.Ceil(rps)), d.minBurst)
+	// If we have 0 rps we have to zero out the burst to immediately cut off new permits
+	if rps == 0 {
+		burst = 0
+	}
+	return rate.Limit(rps), burst
 }

common/quotas/global/collection/internal/shadowed_test.go

@@ -189,7 +189,7 @@ func TestShadowed(t *testing.T) {
 		})
 	})
 	t.Run("limit", func(t *testing.T) {
-		l := NewShadowedLimiter(&allowlimiter{}, quotas.NewSimpleRateLimiter(t, 0))
+		l := NewShadowedLimiter(&allowlimiter{}, clock.NewRatelimiter(rate.Limit(0), 1))
 		assert.Equal(t, rate.Inf, l.Limit(), "should return the primary limit, not shadowed")
 	})
 }