Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Revert "Implement ItemBucketRateLimiter" #88846

Merged
merged 1 commit into from Mar 18, 2020
Merged

Revert "Implement ItemBucketRateLimiter" #88846

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
1 change: 0 additions & 1 deletion staging/src/k8s.io/client-go/util/workqueue/BUILD
View file
Open in desktop
Expand Up @@ -20,7 +20,6 @@ go_test(
deps = [
"//staging/src/k8s.io/apimachinery/pkg/util/clock:go_default_library",
"//staging/src/k8s.io/apimachinery/pkg/util/wait:go_default_library",
"//vendor/golang.org/x/time/rate:go_default_library",
],
)

Expand Down
48 changes: 0 additions & 48 deletions staging/src/k8s.io/client-go/util/workqueue/default_rate_limiters.go
View file
Open in desktop
Expand Up @@ -62,54 +62,6 @@ func (r *BucketRateLimiter) NumRequeues(item interface{}) int {
func (r *BucketRateLimiter) Forget(item interface{}) {
}

// ItemBucketRateLimiter implements a workqueue ratelimiter API using standard rate.Limiter.
// Each key is using a separate limiter.
type ItemBucketRateLimiter struct {
r rate.Limit
burst int

limitersLock sync.Mutex
limiters map[interface{}]*rate.Limiter
}

var _ RateLimiter = &ItemBucketRateLimiter{}

// NewItemBucketRateLimiter creates new ItemBucketRateLimiter instance.
func NewItemBucketRateLimiter(r rate.Limit, burst int) *ItemBucketRateLimiter {
return &ItemBucketRateLimiter{
r: r,
burst: burst,
limiters: make(map[interface{}]*rate.Limiter),
}
}

// When returns a time.Duration which we need to wait before item is processed.
func (r *ItemBucketRateLimiter) When(item interface{}) time.Duration {
r.limitersLock.Lock()
defer r.limitersLock.Unlock()

limiter, ok := r.limiters[item]
if !ok {
limiter = rate.NewLimiter(r.r, r.burst)
r.limiters[item] = limiter
}

return limiter.Reserve().Delay()
}

// NumRequeues returns always 0 (doesn't apply to ItemBucketRateLimiter).
func (r *ItemBucketRateLimiter) NumRequeues(item interface{}) int {
return 0
}

// Forget removes item from the internal state.
func (r *ItemBucketRateLimiter) Forget(item interface{}) {
r.limitersLock.Lock()
defer r.limitersLock.Unlock()

delete(r.limiters, item)
}

// ItemExponentialFailureRateLimiter does a simple baseDelay*2^<num-failures> limit
// dealing with max failures and expiration are up to the caller
type ItemExponentialFailureRateLimiter struct {
Expand Down
29 changes: 0 additions & 29 deletions staging/src/k8s.io/client-go/util/workqueue/default_rate_limiters_test.go
View file
Open in desktop
Expand Up @@ -19,8 +19,6 @@ package workqueue
import (
"testing"
"time"

"golang.org/x/time/rate"
)

func TestItemExponentialFailureRateLimiter(t *testing.T) {
Expand Down Expand Up @@ -98,33 +96,6 @@ func TestItemExponentialFailureRateLimiterOverFlow(t *testing.T) {

}

func TestItemBucketRateLimiter(t *testing.T) {
limiter := NewItemBucketRateLimiter(rate.Every(100*time.Millisecond), 1)

// Use initial burst.
if got := limiter.When("one"); got != 0 {
t.Errorf("limiter.When(two) = %v; want 0", got)
}
for i := 0; i < 1000; i++ {
limiter.When("one")
}
// limiter.When should be at this point = 1000 * rate.Limit.
// We set the threshold 1s below this value to avoid race conditions.
if got, want := limiter.When("one"), 990*100*time.Millisecond; got < want {
t.Errorf("limiter.When(one) = %v; want at least %v", got, want)
}

if got := limiter.When("two"); got != 0 {
t.Errorf("limiter.When(two) = %v; want 0", got)
}

limiter.Forget("one")
// Use new budget.
if got := limiter.When("one"); got != 0 {
t.Errorf("limiter.When(two) = %v; want 0", got)
}
}

func TestItemFastSlowRateLimiter(t *testing.T) {
limiter := NewItemFastSlowRateLimiter(5*time.Millisecond, 10*time.Second, 3)

Expand Down