Hi,

The current token bucket implementation is great. Nonetheless we need some underlying data to implement more powerful functionalities.

For example, if we want to implement a function CanAccept which only checks available tokens without taking it; implement a function that exports rate limiter's metrics to better understand current flow and bottleneck on our services. Both functions will benefit if we can get the underlying data like avail and capacity, including the already public Rate().

Let me know if there are questions. Highly appreciate it if anyone can help review this issue. Thanks!

This adds an interface that is used internally to allow a caller to intercept calls to time.Now() and time.Sleep(), and do better testing.

Fixes #17

@howbazaar this is about as simple as it gets. Unfortunately the factory functions are a bit tedious. Could be changed to "WithDeps" (or something like that) and have a Deps struct, but for now that is going to hold exactly 1 field. This code is simple enough that it probably won't accumulate more. Your call.

I noticed that inside the function adjustavailableTokens()

func (tb *Bucket) adjustavailableTokens(tick int64) {
	if tb.availableTokens >= tb.capacity {
		return
	}
	tb.availableTokens += (tick - tb.latestTick) * tb.quantum
	if tb.availableTokens > tb.capacity {
		tb.availableTokens = tb.capacity
	}
	tb.latestTick = tick
	return
}

the tb.latestTick is not updated if tb.availableTokens >= tb.capacity。

That makes the description of the variable latestTick holds the latest tick for which we know the number of tokens in the bucket. not so accurate, IMO.

And I wrote a snippet of code which can produce surprising results:

func main() {
	bucket := ratelimit.NewBucketWithQuantum(time.Second*1, 100, 20)
	fmt.Printf("Avail=%d\n", bucket.Available())

	fmt.Printf("%v\n", time.Now())
	fmt.Printf("Pause wait for 5s\n")
	time.Sleep(time.Second * 5)
	fmt.Printf("%v\n", time.Now())
	fmt.Printf("Avail=%d\n", bucket.Available())

	fmt.Printf("Request token up to 100\n")
	cnt := bucket.TakeAvailable(100)
	fmt.Printf("Token taken=%d\n", cnt)

        // It will surprise you.
	fmt.Printf("Avail=%d\n", bucket.Available())
}

Output

Avail=100                                                                                          
2019-09-26 01:12:47.9410106 +0800 CST m=+0.003992001                                                Pause wait for 5s                                                                                   
2019-09-26 01:12:52.9614404 +0800 CST m=+5.024421801                                                Avail=100                                                                                           
Request token up to 100                                                                             
Token taken=100                                                                                     
Avail=100             

That is, after taken all tokens out of the bucket, the bucket is still full.

Is this by design or just an implementation bug?

func (tb *Bucket) take(now time.Time, count int64, maxWait time.Duration) (time.Duration, bool) {
	...
	avail := tb.availableTokens - count
	if avail >= 0 {
		tb.availableTokens = avail
		return 0, true
	}
	...
	tb.availableTokens = avail
	return waitTime, true
}

If not enough tokens are available, no tokens should be removed from the bucket. (see https://en.wikipedia.org/wiki/Token_bucket#Algorithm) Otherwise, the availableTokens would be less than zero and would never be enough if many threads are trying to take token from the bucket circularly.

please see this unit test:

type mockClock struct {
	mutex sync.RWMutex
	realStartTime time.Time
	startTime time.Time
}

func newMockClock() *mockClock {
	now := time.Now()
	return &mockClock{
		realStartTime: now,
		startTime: now,
	}
}

func (mc *mockClock) Now() time.Time {
	mc.mutex.RLock()
	defer mc.mutex.RUnlock()
	return mc.startTime.Add(time.Now().Sub(mc.realStartTime))
}

func (mc *mockClock) Sleep(d time.Duration) {
	time.Sleep(d)
}

func (mc *mockClock) RollBack(d time.Duration) {
	mc.mutex.Lock()
	defer mc.mutex.Unlock()
	mc.startTime = mc.startTime.Add(-d)
}

func TestSystemClockRollBack(t *testing.T) {
	mc := newMockClock()
	bucket := NewBucketWithRateAndClock(10, 1, mc)

	if bucket.TakeAvailable(1) != 1 {
		t.Fail()
	}
	time.Sleep(time.Second / 10 + 1)
	if bucket.TakeAvailable(1) != 1 {
		t.Fail()
	}
	time.Sleep(time.Second / 10 + 1)
	mc.RollBack(time.Hour * 8)
	bucket.TakeAvailable(1) 
	time.Sleep(time.Second / 10 + 1)
	if bucket.TakeAvailable(1) != 1 {
		t.Fail()
	}
	time.Sleep(time.Second / 10 + 1)
	if bucket.TakeAvailable(1) != 1 {
		t.Fail()
	}
}

We use ratelimit in Kubernetes. I am working on something that uses it, and I am trying to write a test that doesn't actually take a long time to run. As part of that I inject a fake clock interface. It works really well until it hits the ratelimiter logic.

ratelimit uses time.Now() and time.Sleep() internally. Would you be opposed to something like NewBucketWithClock(..., clock Clock) where Clock was an interface that wrapped time.Now() and similar functions? Users who don't care will ignore it and get the default (which just calls the real time.Now()) and users who do care could provide a mocked clock. If that is OK, I can try to work up a quick patch.

Or is there a better way to test through this? I really don't want my test doing a bunch of 100ms sleeps to try to prove that it behaves. I will have a number of cases to test and that adds up fast.

I've been playing around with a few potential changes to this package for potential speed improvements and found that the act of defering the mutex unlocks is a significant slow down to this package. Once these were removed and appropriate adjustments made in the code it appears to have resulted in an 3 times increase in parallel calls to Wait().

Before:

go test -bench . -benchmem -cpu 1,2,4,8

PASS
BenchmarkWait           10000000               152 ns/op               0 B/op          0 allocs/op
BenchmarkWait-2         10000000               155 ns/op               0 B/op          0 allocs/op
BenchmarkWait-4         10000000               153 ns/op               0 B/op          0 allocs/op
BenchmarkWait-8         10000000               152 ns/op               0 B/op          0 allocs/op
BenchmarkWaitParallel   10000000               160 ns/op               0 B/op          0 allocs/op
BenchmarkWaitParallel-2 10000000               164 ns/op               0 B/op          0 allocs/op
BenchmarkWaitParallel-4 10000000               178 ns/op               0 B/op          0 allocs/op
BenchmarkWaitParallel-8 10000000               185 ns/op               0 B/op          0 allocs/op

After:

go test -bench . -benchmem -cpu 1,2,4,8

PASS
BenchmarkWait           30000000                56.1 ns/op             0 B/op          0 allocs/op
BenchmarkWait-2         30000000                56.0 ns/op             0 B/op          0 allocs/op
BenchmarkWait-4         30000000                55.8 ns/op             0 B/op          0 allocs/op
BenchmarkWait-8         30000000                55.8 ns/op             0 B/op          0 allocs/op
BenchmarkWaitParallel   30000000                56.9 ns/op             0 B/op          0 allocs/op
BenchmarkWaitParallel-2 30000000                59.3 ns/op             0 B/op          0 allocs/op
BenchmarkWaitParallel-4 20000000                68.1 ns/op             0 B/op          0 allocs/op
BenchmarkWaitParallel-8 20000000                70.8 ns/op             0 B/op          0 allocs/op

I'm new in github. Found your package small and more understandable then others.

But little confused with tb.avail in Bucket.abjust:

tb.avail += (currentTick - tb.availTick) * tb.quantum

why used just currentTick , but not currentTick * tb.capacity.

For example, we have

t0----------------t1---------------t2 dt = t1-t0 = t2-t1 currentTick = int64(t2.Sub(t0) / dt) = 2. But in such interval [t0,t2] we must have 2 * tb.capacity tokens in the bucket, not just 2.

Once adjustavailableTokens() is called, the number of tokens should be updated in accordance with the tick even if the buket is full.

Semantically, extra tokens are discarded though we, in fact, simply returns from the function.

For example the code:

func test1st() {
	bucket := ratelimit.NewBucket(time.Minute, 1000)
	bucket.TakeAvailable(2000)
	fmt.Println("test#1: left", bucket.Available())
}

func test2nd() {
	bucket := ratelimit.NewBucket(time.Minute, 1000)
	bucket.Take(2000)
	fmt.Println("test#2: left", bucket.Available())
}

func main() {
	test1st()
	test2nd()
}

Print the output:

test#1: left 0
test#2: left -1000

So the question is: How to make it to discard if the available tokens are insufficient.

Hi,

in https://github.com/restic/restic/issues/1760, a of restic reported that apparently after roughly 24 hours, restic forgets the upload rate limit and uploads without any limit (they used 500KiB). We're using ratelimit.Reader() with a bucket created as follows:

https://github.com/restic/restic/blob/abdd59ea1b7dc9ee0be1806da711c630fe2d8fab/internal/limiter/static_limiter.go#L24

Is there anything we're doing wrong? Is this maybe a bug in the ratelimit.Reader? do you have any idea what may cause this issue?

I'm at a loss what's going on here...

error detail like this: connectex: A connection attempt failed because the connected party did not properly respond after a period of time, or established connection failed because connected host has failed to respond.

Hi @manadart @mitechie @rogpeppe,

I guys you are the maintainers, thanks for doing that. I wonder if this lib is still being maintained?

If yes, could you please cut a new release? People need the tagged release for easy to handle the go module thing. Yeah, before cutting a new release, it'd be better to enable the go module.

PS: I can help with it if you are busy and I just want to know if this is still being maintained.

In the process of using ratelimit,there's a lock in the source code . can you replace it with atomic?

func (tb *Bucket) TakeAvailable(count int64) int64 { tb.mu.Lock() defer tb.mu.Unlock() return tb.takeAvailable(tb.clock.Now(), count) }

when i use go mod to install this package, it shows up

cannot load github.com/juju/ratelimit: github.com/juju/[email protected]: invalid pseudo-version: tag (v1.0.1) found on revision 59fac5042749 is already canonical, so should not be replaced with a pseudo-version derived from that tag

looks wrong tag are build?

Since this project is a common utils, and used by many other open source projects as a vendor dependency, LGPLv3 with static-linking exception may be controversial in some situation for Go language libraries.

Many open source libraries in Go in licensed in MIT/BSD/Apache, so is it possible to change the license for the project to more permissive license (e.g. MIT/BSD/Apache) or dual license (e.g. LGPL + MIT) ?

I found similar discuss in https://github.com/davidmoreno/onion/issues/56

@kingsamchen @nammn

Thanks!

We import package ratelimit in Kubernetes and met a scenario, not sure whether the issue should be filed here. The scenario is that when time happened moves backwards for any reason, func take that calculate the time to sleep would get a huge number here:

	endTick := currentTick + (-avail+tb.quantum-1)/tb.quantum
	endTime := tb.startTime.Add(time.Duration(endTick) * tb.fillInterval)
	waitTime := endTime.Sub(now)
	if waitTime > maxWait {
		return 0, false
	}

An available way for the issue might be adding a protection before calculating waitTime, say a comparision between now and startTime.

/cc @thockin