Fix test approach for detecting issues

.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

Makefile

@@ -53,4 +53,4 @@ staticcheck: bin/staticcheck
 
 .PHONY: test
 test:
-	go test -race ./...
+	go test -v -race ./...

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
 )

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=

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)
+}

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,
 	}

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

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))
+}