From 1053ca8a2bc59c850d4b4f2407018ad3df40733c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Pawe=C5=82?= Date: Mon, 27 Jun 2022 15:18:15 -0700 Subject: [PATCH 01/10] Fix test approach for detecting issues --- .github/workflows/benchmark.yaml | 1 + Makefile | 2 +- go.mod | 1 - go.sum | 2 - limiter_atomic_int64.go | 90 +++++++++++++++++++++++++ ratelimit.go | 14 +++- ratelimit_bench_test.go | 9 ++- ratelimit_mock_time_test.go | 111 +++++++++++++++++++++++++++++++ ratelimit_test.go | 45 +++++++++++-- tools/go.mod | 1 + 10 files changed, 259 insertions(+), 17 deletions(-) create mode 100644 limiter_atomic_int64.go create mode 100644 ratelimit_mock_time_test.go diff --git a/.github/workflows/benchmark.yaml b/.github/workflows/benchmark.yaml index 42dc477..0a3d5c0 100644 --- a/.github/workflows/benchmark.yaml +++ b/.github/workflows/benchmark.yaml @@ -13,6 +13,7 @@ jobs: - uses: actions/checkout@v3 - uses: actions/setup-go@v3 with: + go-version-file: 'go.mod' check-latest: true cache: true - name: Benchmark diff --git a/Makefile b/Makefile index 7eb6d24..7ff707d 100644 --- a/Makefile +++ b/Makefile @@ -53,4 +53,4 @@ staticcheck: bin/staticcheck .PHONY: test test: - go test -race ./... + go test -v -race ./... diff --git a/go.mod b/go.mod index ce7cfc9..77c1eda 100644 --- a/go.mod +++ b/go.mod @@ -3,7 +3,6 @@ module go.uber.org/ratelimit go 1.18 require ( - github.com/benbjohnson/clock v1.3.0 github.com/stretchr/testify v1.6.1 go.uber.org/atomic v1.7.0 ) diff --git a/go.sum b/go.sum index 471801e..8c610cb 100644 --- a/go.sum +++ b/go.sum @@ -1,5 +1,3 @@ -github.com/benbjohnson/clock v1.3.0 h1:ip6w0uFQkncKQ979AypyG0ER7mqUSBdKLOgAle/AT8A= -github.com/benbjohnson/clock v1.3.0/go.mod h1:J11/hYXuz8f4ySSvYwY0FKfm+ezbsZBKZxNJlLklBHA= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= diff --git a/limiter_atomic_int64.go b/limiter_atomic_int64.go new file mode 100644 index 0000000..6588cd0 --- /dev/null +++ b/limiter_atomic_int64.go @@ -0,0 +1,90 @@ +// Copyright (c) 2022 Uber Technologies, Inc. +// +// Permission is hereby granted, free of charge, to any person obtaining a copy +// of this software and associated documentation files (the "Software"), to deal +// in the Software without restriction, including without limitation the rights +// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +// copies of the Software, and to permit persons to whom the Software is +// furnished to do so, subject to the following conditions: +// +// The above copyright notice and this permission notice shall be included in +// all copies or substantial portions of the Software. +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +// THE SOFTWARE. + +package ratelimit // import "go.uber.org/ratelimit" + +import ( + "time" + + "sync/atomic" +) + +type atomicInt64Limiter struct { + //lint:ignore U1000 Padding is unused but it is crucial to maintain performance + // of this rate limiter in case of collocation with other frequently accessed memory. + prepadding [64]byte // cache line size = 64; created to avoid false sharing. + state int64 // unix nanoseconds of the next permissions issue. + //lint:ignore U1000 like prepadding. + postpadding [56]byte // cache line size - state size = 64 - 8; created to avoid false sharing. + + perRequest time.Duration + maxSlack time.Duration + clock Clock +} + +// newAtomicBased returns a new atomic based limiter. +func newAtomicInt64Based(rate int, opts ...Option) *atomicInt64Limiter { + // TODO consider moving config building to the implementation + // independent code. + config := buildConfig(opts) + perRequest := config.per / time.Duration(rate) + l := &atomicInt64Limiter{ + perRequest: perRequest, + maxSlack: time.Duration(config.slack) * perRequest, + clock: config.clock, + } + atomic.StoreInt64(&l.state, 0) + return l +} + +// Take blocks to ensure that the time spent between multiple +// Take calls is on average time.Second/rate. +func (t *atomicInt64Limiter) Take() time.Time { + var ( + newTimeOfNextPermissionIssue int64 + now int64 + ) + for { + now = t.clock.Now().UnixNano() + timeOfNextPermissionIssue := atomic.LoadInt64(&t.state) + + switch { + case timeOfNextPermissionIssue == 0: + // If this is our first request, then we allow it. + newTimeOfNextPermissionIssue = now + case now-timeOfNextPermissionIssue > int64(t.maxSlack): + // a lot of nanoseconds passed since the last Take call + // we will limit max accumulated time to maxSlack + newTimeOfNextPermissionIssue = now - int64(t.maxSlack) + default: + // calculate the time at which our permission was issued + newTimeOfNextPermissionIssue = timeOfNextPermissionIssue + int64(t.perRequest) + } + + if atomic.CompareAndSwapInt64(&t.state, timeOfNextPermissionIssue, newTimeOfNextPermissionIssue) { + break + } + } + nanosToSleepUntilOurPermissionIsIssued := newTimeOfNextPermissionIssue - now + if nanosToSleepUntilOurPermissionIsIssued > 0 { + t.clock.Sleep(time.Duration(nanosToSleepUntilOurPermissionIsIssued)) + } + return time.Unix(0, newTimeOfNextPermissionIssue) +} diff --git a/ratelimit.go b/ratelimit.go index 7370526..7afce5b 100644 --- a/ratelimit.go +++ b/ratelimit.go @@ -22,8 +22,6 @@ package ratelimit // import "go.uber.org/ratelimit" import ( "time" - - "github.com/benbjohnson/clock" ) // Note: This file is inspired by: @@ -45,6 +43,16 @@ type Clock interface { Sleep(time.Duration) } +type internalClock struct{} + +func (i *internalClock) Now() time.Time { + return time.Now() +} + +func (i *internalClock) Sleep(duration time.Duration) { + time.Sleep(duration) +} + // config configures a limiter. type config struct { clock Clock @@ -60,7 +68,7 @@ func New(rate int, opts ...Option) Limiter { // buildConfig combines defaults with options. func buildConfig(opts []Option) config { c := config{ - clock: clock.New(), + clock: &internalClock{}, slack: 10, per: time.Second, } diff --git a/ratelimit_bench_test.go b/ratelimit_bench_test.go index 60b203b..a1125c0 100644 --- a/ratelimit_bench_test.go +++ b/ratelimit_bench_test.go @@ -14,8 +14,9 @@ func BenchmarkRateLimiter(b *testing.B) { for _, procs := range []int{1, 4, 8, 16} { runtime.GOMAXPROCS(procs) for name, limiter := range map[string]Limiter{ - "atomic": New(b.N * 10000000), - "mutex": newMutexBased(b.N * 10000000), + "atomic": newAtomicBased(b.N * 1000000000000), + "atomic_int64": newAtomicInt64Based(b.N * 1000000000000), + "mutex": newMutexBased(b.N * 1000000000000), } { for ng := 1; ng < 16; ng++ { runner(b, name, procs, ng, limiter, count) @@ -47,7 +48,9 @@ func BenchmarkRateLimiter(b *testing.B) { } func runner(b *testing.B, name string, procs int, ng int, limiter Limiter, count *atomic.Int64) bool { - return b.Run(fmt.Sprintf("type:%s-procs:%d-goroutines:%d", name, procs, ng), func(b *testing.B) { + return b.Run(fmt.Sprintf("type:%s;max_procs:%d;goroutines:%d", name, procs, ng), func(b *testing.B) { + b.ReportAllocs() + var wg sync.WaitGroup trigger := atomic.NewBool(true) n := b.N diff --git a/ratelimit_mock_time_test.go b/ratelimit_mock_time_test.go new file mode 100644 index 0000000..5383cba --- /dev/null +++ b/ratelimit_mock_time_test.go @@ -0,0 +1,111 @@ +package ratelimit + +/** +This fake time implementation is a modification of time mocking +the mechanism used by Ian Lance Taylor in https://github.com/golang/time project +https://github.com/golang/time/commit/579cf78fd858857c0d766e0d63eb2b0ccf29f436 + +Modified parts: + - advanceUnlocked method uses in-place filtering of timers + instead of a full copy on every remove. + Since we have 100s of timers in our tests current linear + the complexity of this operation is OK + If going to have 1000s in the future, we can use heap to store timers. + - advanceUnlocked method yields the processor, after every timer triggering, + allowing other goroutines to run +*/ + +import ( + "runtime" + "sync" + "time" +) + +// testTime is a fake time used for testing. +type testTime struct { + mu sync.Mutex + cur time.Time // current fake time + timers []testTimer // fake timers +} + +// makeTestTime hooks the testTimer into the package. +func makeTestTime() *testTime { + return &testTime{ + cur: time.Now(), + } +} + +// testTimer is a fake timer. +type testTimer struct { + when time.Time + ch chan<- time.Time +} + +// now returns the current fake time. +func (tt *testTime) now() time.Time { + tt.mu.Lock() + defer tt.mu.Unlock() + return tt.cur +} + +// newTimer creates a fake timer. It returns the channel, +// a function to stop the timer (which we don't care about), +// and a function to advance to the next timer. +func (tt *testTime) newTimer(dur time.Duration) (<-chan time.Time, func() bool, func()) { + tt.mu.Lock() + defer tt.mu.Unlock() + ch := make(chan time.Time, 1) + timer := testTimer{ + when: tt.cur.Add(dur), + ch: ch, + } + tt.timers = append(tt.timers, timer) + return ch, func() bool { return true }, tt.advanceToTimer +} + +// advance advances the fake time. +func (tt *testTime) advance(dur time.Duration) { + tt.mu.Lock() + defer tt.mu.Unlock() + tt.advanceUnlocked(dur) +} + +// advanceUnlock advances the fake time, assuming it is already locked. +func (tt *testTime) advanceUnlocked(dur time.Duration) { + tt.cur = tt.cur.Add(dur) + + i := 0 + j := 0 + for i < len(tt.timers) { + if tt.timers[i].when.After(tt.cur) { + if i != j { + tt.timers[j] = tt.timers[i] + } + i++ + j++ + } else { + tt.timers[i].ch <- tt.cur + for i := 0; i < 16; i++ { + runtime.Gosched() + } + i++ + } + } + tt.timers = tt.timers[0:j] +} + +// advanceToTimer advances the time to the next timer. +func (tt *testTime) advanceToTimer() { + tt.mu.Lock() + defer tt.mu.Unlock() + if len(tt.timers) == 0 { + panic("no timer") + } + when := tt.timers[0].when + for _, timer := range tt.timers[1:] { + if timer.when.Before(when) { + when = timer.when + } + } + tt.advanceUnlocked(when.Sub(tt.cur)) +} diff --git a/ratelimit_test.go b/ratelimit_test.go index 7b584b5..30abc43 100644 --- a/ratelimit_test.go +++ b/ratelimit_test.go @@ -7,10 +7,18 @@ import ( "go.uber.org/atomic" - "github.com/benbjohnson/clock" "github.com/stretchr/testify/assert" ) +func (tt *testTime) Now() time.Time { + return tt.now() +} + +func (tt *testTime) Sleep(duration time.Duration) { + timer, _, _ := tt.newTimer(duration) + <-timer +} + type testRunner interface { // createLimiter builds a limiter with given options. createLimiter(int, ...Option) Limiter @@ -27,7 +35,7 @@ type testRunner interface { type runnerImpl struct { t *testing.T - clock *clock.Mock + clock *testTime constructor func(int, ...Option) Limiter count atomic.Int32 // maxDuration is the time we need to move into the future for a test. @@ -54,13 +62,19 @@ func runTest(t *testing.T, fn func(testRunner)) { return newAtomicBased(rate, opts...) }, }, + { + name: "atomic_int64", + constructor: func(rate int, opts ...Option) Limiter { + return newAtomicInt64Based(rate, opts...) + }, + }, } for _, tt := range impls { t.Run(tt.name, func(t *testing.T) { r := runnerImpl{ t: t, - clock: clock.NewMock(), + clock: makeTestTime(), constructor: tt.constructor, doneCh: make(chan struct{}), } @@ -68,7 +82,18 @@ func runTest(t *testing.T, fn func(testRunner)) { defer r.wg.Wait() fn(&r) - r.clock.Add(r.maxDuration) + go func() { + move := func() { + defer func() { + _ = recover() + time.Sleep(10 * time.Millisecond) + }() + r.clock.advanceToTimer() + } + for { + move() + } + }() }) } } @@ -86,6 +111,7 @@ func (r *runnerImpl) startTaking(rls ...Limiter) { for _, rl := range rls { rl.Take() } + r.clock.advance(time.Nanosecond) r.count.Inc() select { case <-r.doneCh: @@ -110,14 +136,14 @@ func (r *runnerImpl) afterFunc(d time.Duration, fn func()) { if d > r.maxDuration { r.maxDuration = d } - + timer, _, _ := r.clock.newTimer(d) r.goWait(func() { select { case <-r.doneCh: return - case <-r.clock.After(d): + case <-timer: + fn() } - fn() }) } @@ -133,6 +159,7 @@ func (r *runnerImpl) goWait(fn func()) { } func TestUnlimited(t *testing.T) { + t.Parallel() now := time.Now() rl := NewUnlimited() for i := 0; i < 1000; i++ { @@ -142,6 +169,7 @@ func TestUnlimited(t *testing.T) { } func TestRateLimiter(t *testing.T) { + t.Parallel() runTest(t, func(r testRunner) { rl := r.createLimiter(100, WithoutSlack) @@ -158,6 +186,7 @@ func TestRateLimiter(t *testing.T) { } func TestDelayedRateLimiter(t *testing.T) { + t.Parallel() runTest(t, func(r testRunner) { slow := r.createLimiter(10, WithoutSlack) fast := r.createLimiter(100, WithoutSlack) @@ -176,6 +205,7 @@ func TestDelayedRateLimiter(t *testing.T) { } func TestPer(t *testing.T) { + t.Parallel() runTest(t, func(r testRunner) { rl := r.createLimiter(7, WithoutSlack, Per(time.Minute)) @@ -189,6 +219,7 @@ func TestPer(t *testing.T) { } func TestSlack(t *testing.T) { + t.Parallel() // To simulate slack, we combine two limiters. // - First, we start a single goroutine with both of them, // during this time the slow limiter will dominate, diff --git a/tools/go.mod b/tools/go.mod index ba45a9b..c15076e 100644 --- a/tools/go.mod +++ b/tools/go.mod @@ -11,6 +11,7 @@ require ( require ( github.com/BurntSushi/toml v1.0.0 // indirect + github.com/storozhukBM/benchart v1.0.0 golang.org/x/exp/typeparams v0.0.0-20220328175248-053ad81199eb // indirect golang.org/x/mod v0.6.0-dev.0.20220106191415-9b9b3d81d5e3 // indirect golang.org/x/sys v0.0.0-20220330033206-e17cdc41300f // indirect From 47a6978800566f8c9fccc66ec46d824f5e249a00 Mon Sep 17 00:00:00 2001 From: bohdanstorozhuk Date: Thu, 14 Jul 2022 00:11:06 +0100 Subject: [PATCH 02/10] Adapt TestInitial to new testTime approach --- ratelimit_test.go | 15 +++++++++++---- 1 file changed, 11 insertions(+), 4 deletions(-) diff --git a/ratelimit_test.go b/ratelimit_test.go index e9c0a0d..adfd22a 100644 --- a/ratelimit_test.go +++ b/ratelimit_test.go @@ -20,6 +20,7 @@ func (tt *testTime) Sleep(duration time.Duration) { } type testRunner interface { + getName() string // createLimiter builds a limiter with given options. createLimiter(int, ...Option) Limiter // startTaking tries to Take() on passed in limiters in a loop/goroutine. @@ -31,12 +32,13 @@ type testRunner interface { // not using clock.AfterFunc because andres-erbsen/clock misses a nap there. afterFunc(d time.Duration, fn func()) // some tests want raw access to the clock. - getClock() *clock.Mock + getClock() *testTime } type runnerImpl struct { t *testing.T + name string clock *testTime constructor func(int, ...Option) Limiter count atomic.Int32 @@ -76,6 +78,7 @@ func runTest(t *testing.T, fn func(testRunner)) { t.Run(tt.name, func(t *testing.T) { r := runnerImpl{ t: t, + name: tt.name, clock: makeTestTime(), constructor: tt.constructor, doneCh: make(chan struct{}), @@ -106,7 +109,11 @@ func (r *runnerImpl) createLimiter(rate int, opts ...Option) Limiter { return r.constructor(rate, opts...) } -func (r *runnerImpl) getClock() *clock.Mock { +func (r *runnerImpl) getName() string { + return r.name +} + +func (r *runnerImpl) getClock() *testTime { return r.clock } @@ -263,7 +270,7 @@ func TestInitial(t *testing.T) { } startWg.Wait() - clk.Add(time.Second) + clk.advance(time.Second) for i := 0; i < 3; i++ { ts := <-results @@ -278,7 +285,7 @@ func TestInitial(t *testing.T) { time.Millisecond * 100, }, have, - "bad timestamps for inital takes", + "bad timestamps for initial takes of "+r.getName(), ) }) }) From 99e7b2c70edce9b643b8ec9c955f1292f1576bab Mon Sep 17 00:00:00 2001 From: bohdanstorozhuk Date: Thu, 14 Jul 2022 00:19:37 +0100 Subject: [PATCH 03/10] Use advanceToTimer approach in TestInitial --- ratelimit_test.go | 16 ++++++++++++++-- 1 file changed, 14 insertions(+), 2 deletions(-) diff --git a/ratelimit_test.go b/ratelimit_test.go index adfd22a..c181ed9 100644 --- a/ratelimit_test.go +++ b/ratelimit_test.go @@ -260,6 +260,20 @@ func TestInitial(t *testing.T) { have []time.Duration startWg sync.WaitGroup ) + + go func() { + move := func() { + defer func() { + _ = recover() + time.Sleep(10 * time.Millisecond) + }() + r.getClock().advanceToTimer() + } + for { + move() + } + }() + startWg.Add(3) for i := 0; i < 3; i++ { @@ -268,9 +282,7 @@ func TestInitial(t *testing.T) { results <- rl.Take() }() } - startWg.Wait() - clk.advance(time.Second) for i := 0; i < 3; i++ { ts := <-results From 86916e306010a42299d465ca23186d0e85ac658e Mon Sep 17 00:00:00 2001 From: bohdanstorozhuk Date: Thu, 14 Jul 2022 00:29:30 +0100 Subject: [PATCH 04/10] Go back to the original approach of removing timers from the list. Fetch exact numbers for goroutines in runtime to make that amount of Gosched calls. --- ratelimit_mock_time_test.go | 22 ++++++++-------------- 1 file changed, 8 insertions(+), 14 deletions(-) diff --git a/ratelimit_mock_time_test.go b/ratelimit_mock_time_test.go index 5383cba..b77b093 100644 --- a/ratelimit_mock_time_test.go +++ b/ratelimit_mock_time_test.go @@ -6,11 +6,6 @@ the mechanism used by Ian Lance Taylor in https://github.com/golang/time project https://github.com/golang/time/commit/579cf78fd858857c0d766e0d63eb2b0ccf29f436 Modified parts: - - advanceUnlocked method uses in-place filtering of timers - instead of a full copy on every remove. - Since we have 100s of timers in our tests current linear - the complexity of this operation is OK - If going to have 1000s in the future, we can use heap to store timers. - advanceUnlocked method yields the processor, after every timer triggering, allowing other goroutines to run */ @@ -72,26 +67,25 @@ func (tt *testTime) advance(dur time.Duration) { // advanceUnlock advances the fake time, assuming it is already locked. func (tt *testTime) advanceUnlocked(dur time.Duration) { - tt.cur = tt.cur.Add(dur) + tt.cur = tt.cur.Add(dur) i := 0 - j := 0 for i < len(tt.timers) { if tt.timers[i].when.After(tt.cur) { - if i != j { - tt.timers[j] = tt.timers[i] - } i++ - j++ } else { tt.timers[i].ch <- tt.cur - for i := 0; i < 16; i++ { + // calculate how many goroutines we currently have in runtime + // and yield the processor, after every timer triggering, + // allowing all other goroutines to run + numOfAllRunningGoroutines := runtime.NumGoroutine() + for i := 0; i < numOfAllRunningGoroutines; i++ { runtime.Gosched() } - i++ + copy(tt.timers[i:], tt.timers[i+1:]) + tt.timers = tt.timers[:len(tt.timers)-1] } } - tt.timers = tt.timers[0:j] } // advanceToTimer advances the time to the next timer. From 260ee118caed8fd5a9aba85a3c8e24ea55a67267 Mon Sep 17 00:00:00 2001 From: bohdanstorozhuk Date: Thu, 14 Jul 2022 01:25:53 +0100 Subject: [PATCH 05/10] Simplify advanceToTimer routine --- ratelimit_mock_time_test.go | 4 ++-- ratelimit_test.go | 18 ++---------------- 2 files changed, 4 insertions(+), 18 deletions(-) diff --git a/ratelimit_mock_time_test.go b/ratelimit_mock_time_test.go index b77b093..05529f6 100644 --- a/ratelimit_mock_time_test.go +++ b/ratelimit_mock_time_test.go @@ -67,7 +67,6 @@ func (tt *testTime) advance(dur time.Duration) { // advanceUnlock advances the fake time, assuming it is already locked. func (tt *testTime) advanceUnlocked(dur time.Duration) { - tt.cur = tt.cur.Add(dur) i := 0 for i < len(tt.timers) { @@ -90,10 +89,11 @@ func (tt *testTime) advanceUnlocked(dur time.Duration) { // advanceToTimer advances the time to the next timer. func (tt *testTime) advanceToTimer() { + defer time.Sleep(2 * time.Millisecond) tt.mu.Lock() defer tt.mu.Unlock() if len(tt.timers) == 0 { - panic("no timer") + return } when := tt.timers[0].when for _, timer := range tt.timers[1:] { diff --git a/ratelimit_test.go b/ratelimit_test.go index c181ed9..2b615ef 100644 --- a/ratelimit_test.go +++ b/ratelimit_test.go @@ -88,15 +88,8 @@ func runTest(t *testing.T, fn func(testRunner)) { fn(&r) go func() { - move := func() { - defer func() { - _ = recover() - time.Sleep(10 * time.Millisecond) - }() - r.clock.advanceToTimer() - } for { - move() + r.clock.advanceToTimer() } }() }) @@ -262,15 +255,8 @@ func TestInitial(t *testing.T) { ) go func() { - move := func() { - defer func() { - _ = recover() - time.Sleep(10 * time.Millisecond) - }() - r.getClock().advanceToTimer() - } for { - move() + r.getClock().advanceToTimer() } }() From 4e9eb9d9fc2d3269dce1f43301e5ec0e68839691 Mon Sep 17 00:00:00 2001 From: bohdanstorozhuk Date: Thu, 14 Jul 2022 01:29:23 +0100 Subject: [PATCH 06/10] Go back to 10 millis between advances --- ratelimit_mock_time_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/ratelimit_mock_time_test.go b/ratelimit_mock_time_test.go index 05529f6..8acb2eb 100644 --- a/ratelimit_mock_time_test.go +++ b/ratelimit_mock_time_test.go @@ -89,7 +89,7 @@ func (tt *testTime) advanceUnlocked(dur time.Duration) { // advanceToTimer advances the time to the next timer. func (tt *testTime) advanceToTimer() { - defer time.Sleep(2 * time.Millisecond) + defer time.Sleep(10 * time.Millisecond) tt.mu.Lock() defer tt.mu.Unlock() if len(tt.timers) == 0 { From 6ca7158d73f3e16c62f56096531a475addf31246 Mon Sep 17 00:00:00 2001 From: bohdanstorozhuk Date: Thu, 14 Jul 2022 01:59:22 +0100 Subject: [PATCH 07/10] Use advanceFor instead of for loop --- ratelimit_mock_time_test.go | 54 +++++++++++++++++++++++-------------- ratelimit_test.go | 19 ++++--------- 2 files changed, 39 insertions(+), 34 deletions(-) diff --git a/ratelimit_mock_time_test.go b/ratelimit_mock_time_test.go index 8acb2eb..eb2abf3 100644 --- a/ratelimit_mock_time_test.go +++ b/ratelimit_mock_time_test.go @@ -46,7 +46,7 @@ func (tt *testTime) now() time.Time { // newTimer creates a fake timer. It returns the channel, // a function to stop the timer (which we don't care about), // and a function to advance to the next timer. -func (tt *testTime) newTimer(dur time.Duration) (<-chan time.Time, func() bool, func()) { +func (tt *testTime) newTimer(dur time.Duration) (<-chan time.Time, func() bool) { tt.mu.Lock() defer tt.mu.Unlock() ch := make(chan time.Time, 1) @@ -55,7 +55,35 @@ func (tt *testTime) newTimer(dur time.Duration) (<-chan time.Time, func() bool, ch: ch, } tt.timers = append(tt.timers, timer) - return ch, func() bool { return true }, tt.advanceToTimer + return ch, func() bool { return true } +} + +func (tt *testTime) advanceFor(dur time.Duration) { + tt.mu.Lock() + defer tt.mu.Unlock() + + targetTime := tt.cur.Add(dur) + for { + if len(tt.timers) == 0 { + tt.cur = targetTime + return + } + when := tt.timers[0].when + for _, timer := range tt.timers[1:] { + if timer.when.Before(when) { + when = timer.when + } + } + if when.After(targetTime) { + tt.cur = targetTime + return + } + if tt.advanceUnlocked(when.Sub(tt.cur)) { + tt.mu.Unlock() + time.Sleep(10 * time.Millisecond) + tt.mu.Lock() + } + } } // advance advances the fake time. @@ -66,7 +94,8 @@ func (tt *testTime) advance(dur time.Duration) { } // advanceUnlock advances the fake time, assuming it is already locked. -func (tt *testTime) advanceUnlocked(dur time.Duration) { +func (tt *testTime) advanceUnlocked(dur time.Duration) bool { + result := false tt.cur = tt.cur.Add(dur) i := 0 for i < len(tt.timers) { @@ -74,6 +103,7 @@ func (tt *testTime) advanceUnlocked(dur time.Duration) { i++ } else { tt.timers[i].ch <- tt.cur + result = true // calculate how many goroutines we currently have in runtime // and yield the processor, after every timer triggering, // allowing all other goroutines to run @@ -85,21 +115,5 @@ func (tt *testTime) advanceUnlocked(dur time.Duration) { tt.timers = tt.timers[:len(tt.timers)-1] } } -} - -// advanceToTimer advances the time to the next timer. -func (tt *testTime) advanceToTimer() { - defer time.Sleep(10 * time.Millisecond) - tt.mu.Lock() - defer tt.mu.Unlock() - if len(tt.timers) == 0 { - return - } - when := tt.timers[0].when - for _, timer := range tt.timers[1:] { - if timer.when.Before(when) { - when = timer.when - } - } - tt.advanceUnlocked(when.Sub(tt.cur)) + return result } diff --git a/ratelimit_test.go b/ratelimit_test.go index 2b615ef..3233c5f 100644 --- a/ratelimit_test.go +++ b/ratelimit_test.go @@ -15,7 +15,7 @@ func (tt *testTime) Now() time.Time { } func (tt *testTime) Sleep(duration time.Duration) { - timer, _, _ := tt.newTimer(duration) + timer, _ := tt.newTimer(duration) <-timer } @@ -87,11 +87,7 @@ func runTest(t *testing.T, fn func(testRunner)) { defer r.wg.Wait() fn(&r) - go func() { - for { - r.clock.advanceToTimer() - } - }() + r.clock.advanceFor(r.maxDuration) }) } } @@ -142,7 +138,7 @@ func (r *runnerImpl) afterFunc(d time.Duration, fn func()) { if d > r.maxDuration { r.maxDuration = d } - timer, _, _ := r.clock.newTimer(d) + timer, _ := r.clock.newTimer(d) r.goWait(func() { select { case <-r.doneCh: @@ -254,14 +250,7 @@ func TestInitial(t *testing.T) { startWg sync.WaitGroup ) - go func() { - for { - r.getClock().advanceToTimer() - } - }() - startWg.Add(3) - for i := 0; i < 3; i++ { go func() { startWg.Done() @@ -270,6 +259,8 @@ func TestInitial(t *testing.T) { } startWg.Wait() + r.getClock().advanceFor(time.Second) + for i := 0; i < 3; i++ { ts := <-results have = append(have, ts.Sub(prev)) From 41e59b71942588f42fabaa88871643242cf304af Mon Sep 17 00:00:00 2001 From: bohdanstorozhuk Date: Thu, 14 Jul 2022 02:40:47 +0100 Subject: [PATCH 08/10] Sort internal timers and use only one function for advance --- ratelimit_mock_time_test.go | 64 ++++++++++++++++--------------------- ratelimit_test.go | 9 ++++-- 2 files changed, 34 insertions(+), 39 deletions(-) diff --git a/ratelimit_mock_time_test.go b/ratelimit_mock_time_test.go index eb2abf3..f283e95 100644 --- a/ratelimit_mock_time_test.go +++ b/ratelimit_mock_time_test.go @@ -12,6 +12,7 @@ Modified parts: import ( "runtime" + "sort" "sync" "time" ) @@ -55,65 +56,56 @@ func (tt *testTime) newTimer(dur time.Duration) (<-chan time.Time, func() bool) ch: ch, } tt.timers = append(tt.timers, timer) + sort.Slice(tt.timers, func(i, j int) bool { + return tt.timers[i].when.Before(tt.timers[j].when) + }) return ch, func() bool { return true } } -func (tt *testTime) advanceFor(dur time.Duration) { +// advance advances the fake time. +func (tt *testTime) advance(dur time.Duration, backoffDuration time.Duration) { tt.mu.Lock() defer tt.mu.Unlock() targetTime := tt.cur.Add(dur) for { - if len(tt.timers) == 0 { + if len(tt.timers) == 0 || tt.timers[0].when.After(targetTime) { tt.cur = targetTime return } - when := tt.timers[0].when - for _, timer := range tt.timers[1:] { - if timer.when.Before(when) { - when = timer.when - } - } - if when.After(targetTime) { - tt.cur = targetTime - return - } - if tt.advanceUnlocked(when.Sub(tt.cur)) { + if tt.advanceUnlocked(tt.timers[0].when.Sub(tt.cur)) && backoffDuration > 0 { + // after every timer triggering, we release our mutex + // and give time for other goroutines to run tt.mu.Unlock() - time.Sleep(10 * time.Millisecond) + time.Sleep(backoffDuration) tt.mu.Lock() } } } -// advance advances the fake time. -func (tt *testTime) advance(dur time.Duration) { - tt.mu.Lock() - defer tt.mu.Unlock() - tt.advanceUnlocked(dur) -} - // advanceUnlock advances the fake time, assuming it is already locked. func (tt *testTime) advanceUnlocked(dur time.Duration) bool { - result := false tt.cur = tt.cur.Add(dur) + if len(tt.timers) == 0 || tt.timers[0].when.After(tt.cur) { + return false + } + i := 0 for i < len(tt.timers) { if tt.timers[i].when.After(tt.cur) { - i++ - } else { - tt.timers[i].ch <- tt.cur - result = true - // calculate how many goroutines we currently have in runtime - // and yield the processor, after every timer triggering, - // allowing all other goroutines to run - numOfAllRunningGoroutines := runtime.NumGoroutine() - for i := 0; i < numOfAllRunningGoroutines; i++ { - runtime.Gosched() - } - copy(tt.timers[i:], tt.timers[i+1:]) - tt.timers = tt.timers[:len(tt.timers)-1] + break + } + tt.timers[i].ch <- tt.cur + i++ + // calculate how many goroutines we currently have in runtime + // and yield the processor, after every timer triggering, + // allowing all other goroutines to run + numOfAllRunningGoroutines := runtime.NumGoroutine() + for j := 0; j < numOfAllRunningGoroutines; j++ { + runtime.Gosched() } } - return result + + tt.timers = tt.timers[i:] + return true } diff --git a/ratelimit_test.go b/ratelimit_test.go index 3233c5f..928e74b 100644 --- a/ratelimit_test.go +++ b/ratelimit_test.go @@ -10,6 +10,8 @@ import ( "github.com/stretchr/testify/assert" ) +const advanceBackoffDuration = 5 * time.Millisecond + func (tt *testTime) Now() time.Time { return tt.now() } @@ -87,7 +89,7 @@ func runTest(t *testing.T, fn func(testRunner)) { defer r.wg.Wait() fn(&r) - r.clock.advanceFor(r.maxDuration) + r.clock.advance(r.maxDuration, advanceBackoffDuration) }) } } @@ -113,7 +115,7 @@ func (r *runnerImpl) startTaking(rls ...Limiter) { for _, rl := range rls { rl.Take() } - r.clock.advance(time.Nanosecond) + r.clock.advance(time.Nanosecond, 0) r.count.Inc() select { case <-r.doneCh: @@ -259,7 +261,7 @@ func TestInitial(t *testing.T) { } startWg.Wait() - r.getClock().advanceFor(time.Second) + r.getClock().advance(time.Second, advanceBackoffDuration) for i := 0; i < 3; i++ { ts := <-results @@ -354,6 +356,7 @@ func TestSlack(t *testing.T) { for _, tt := range tests { t.Run(tt.msg, func(t *testing.T) { + t.Parallel() runTest(t, func(r testRunner) { slow := r.createLimiter(10, WithoutSlack) fast := r.createLimiter(100, tt.opt...) From e2a95a92436d095e17cfe592e18142d6261726c4 Mon Sep 17 00:00:00 2001 From: bohdanstorozhuk Date: Thu, 14 Jul 2022 02:44:51 +0100 Subject: [PATCH 09/10] Add more comments --- ratelimit_mock_time_test.go | 3 +++ 1 file changed, 3 insertions(+) diff --git a/ratelimit_mock_time_test.go b/ratelimit_mock_time_test.go index f283e95..e88f5a2 100644 --- a/ratelimit_mock_time_test.go +++ b/ratelimit_mock_time_test.go @@ -6,6 +6,9 @@ the mechanism used by Ian Lance Taylor in https://github.com/golang/time project https://github.com/golang/time/commit/579cf78fd858857c0d766e0d63eb2b0ccf29f436 Modified parts: + - timers are sorted on every addition, and then we relly of that order, + we could use heap data structure, but sorting is OK for now. + - advance accepts backoffDuration to sleep without lock held after every timer triggering - advanceUnlocked method yields the processor, after every timer triggering, allowing other goroutines to run */ From 31dcb6c3cbfb792836155df0184c4c15da7e197b Mon Sep 17 00:00:00 2001 From: bohdanstorozhuk Date: Thu, 14 Jul 2022 08:21:42 +0100 Subject: [PATCH 10/10] Remove test runner name --- ratelimit_test.go | 9 +-------- 1 file changed, 1 insertion(+), 8 deletions(-) diff --git a/ratelimit_test.go b/ratelimit_test.go index 928e74b..ac18c71 100644 --- a/ratelimit_test.go +++ b/ratelimit_test.go @@ -22,7 +22,6 @@ func (tt *testTime) Sleep(duration time.Duration) { } type testRunner interface { - getName() string // createLimiter builds a limiter with given options. createLimiter(int, ...Option) Limiter // startTaking tries to Take() on passed in limiters in a loop/goroutine. @@ -40,7 +39,6 @@ type testRunner interface { type runnerImpl struct { t *testing.T - name string clock *testTime constructor func(int, ...Option) Limiter count atomic.Int32 @@ -80,7 +78,6 @@ func runTest(t *testing.T, fn func(testRunner)) { t.Run(tt.name, func(t *testing.T) { r := runnerImpl{ t: t, - name: tt.name, clock: makeTestTime(), constructor: tt.constructor, doneCh: make(chan struct{}), @@ -100,10 +97,6 @@ func (r *runnerImpl) createLimiter(rate int, opts ...Option) Limiter { return r.constructor(rate, opts...) } -func (r *runnerImpl) getName() string { - return r.name -} - func (r *runnerImpl) getClock() *testTime { return r.clock } @@ -276,7 +269,7 @@ func TestInitial(t *testing.T) { time.Millisecond * 100, }, have, - "bad timestamps for initial takes of "+r.getName(), + "bad timestamps for initial takes", ) }) })