x/time/rate: Limiter allows more configured with multiple goroutines

[ianzhang1988]

Go version

go version go1.18.5 linux/amd64

Output of go env in your module/workspace:

GO111MODULE="on"
GOARCH="amd64"
GOBIN=""
GOCACHE="/root/.cache/go-build"
GOENV="/root/.config/go/env"
GOEXE=""
GOEXPERIMENT=""
GOFLAGS=""
GOHOSTARCH="amd64"
GOHOSTOS="linux"
GOINSECURE=""
GOMODCACHE="/root/.gvm/pkgsets/go1.18.5/global/pkg/mod"
GONOPROXY=""
GONOSUMDB=""
GOOS="linux"
GOPATH="/root/.gvm/pkgsets/go1.18.5/global"
GOPRIVATE=""
GOPROXY="https://goproxy.cn,direct"
GOROOT="/root/.gvm/gos/go1.18.5"
GOSUMDB="sum.golang.org"
GOTMPDIR=""
GOTOOLDIR="/root/.gvm/gos/go1.18.5/pkg/tool/linux_amd64"
GOVCS=""
GOVERSION="go1.18.5"
GCCGO="gccgo"
GOAMD64="v1"
AR="ar"
CC="gcc"
CXX="g++"
CGO_ENABLED="0"
GOMOD="/data/zhangyang/goproject/go.mod"
GOWORK=""
CGO_CFLAGS="-g -O2"
CGO_CPPFLAGS=""
CGO_CXXFLAGS="-g -O2"
CGO_FFLAGS="-g -O2"
CGO_LDFLAGS="-g -O2"
PKG_CONFIG="pkg-config"
GOGCCFLAGS="-fPIC -m64 -fmessage-length=0 -fdebug-prefix-map=/tmp/go-build3572669853=/tmp/go-build -gno-record-gcc-switches"

What did you do?

Here is my code

package main

import (
	"fmt"
	"time"
	. "golang.org/x/time/rate"
)

func Produce(ch chan uint64) {
	var counter uint64 = 0
	for {
		ch <- counter
		counter += 1
	}
}

func ConsumWithLimitDelay(ch chan uint64, chout chan uint64, lim *Limiter) {
	for {
		n := lim.Reserve()
		if !n.OK() {
			continue
		}

		time.Sleep(n.Delay())
		chout <- <-ch
	}
}

func Count(ch chan uint64) {
	var counter uint64 = 0

	lastTime := time.Now()

	for {
		_ = <-ch
		counter += 1
		du := time.Since(lastTime)
		if du > 1*time.Second {
			fmt.Printf("%d %v\n", counter, du)
			counter = 0
			lastTime = lastTime.Add(du)
		}
	}
}

func main() {
	ch := make(chan uint64, 1000)
	ch2 := make(chan uint64, 100)
	go Produce(ch)
	lim := NewLimiter(200000.0, 1000)
	for i := 0; i < 100; i++ {
		go ConsumWithLimitDelay(ch, ch2, lim)
	}
	Count(ch2)
}

limit is set to 200k, but actually is 300k on my machine.

here is what i think is going wrong in rate.go:
sometimes lim.advance(t) would return a time t which is in the past, then update to lim.last and causing problem.
(see // !!! comment in code below)

func (lim *Limiter) reserveN(t time.Time, n int, maxFutureReserve time.Duration) Reservation {
	lim.mu.Lock()
	defer lim.mu.Unlock()

	if lim.limit == Inf {
		return Reservation{
			ok:        true,
			lim:       lim,
			tokens:    n,
			timeToAct: t,
		}
	} else if lim.limit == 0 {
		var ok bool
		if lim.burst >= n {
			ok = true
			lim.burst -= n
		}
		return Reservation{
			ok:        ok,
			lim:       lim,
			tokens:    lim.burst,
			timeToAct: t,
		}
	}

        // !!! with this block of code would set rate.go work correctly
	// if t.Before(lim.last) {
	// 	return Reservation{
	// 		ok:    false,
	// 		lim:   lim,
	// 		limit: lim.limit,
	// 	}
	// }

        // !!! t could be in the past in multiple goroutine
	t, tokens := lim.advance(t)

	// Calculate the remaining number of tokens resulting from the request.
	tokens -= float64(n)

	// Calculate the wait duration
	var waitDuration time.Duration
	if tokens < 0 {
		waitDuration = lim.limit.durationFromTokens(-tokens)
	}

	// Decide result
	ok := n <= lim.burst && waitDuration <= maxFutureReserve

	// Prepare reservation
	r := Reservation{
		ok:    ok,
		lim:   lim,
		limit: lim.limit,
	}
	if ok {
		r.tokens = n
		r.timeToAct = t.Add(waitDuration)

                // !!! some time here would update last to a past time, causing the problem
		// Update state
		lim.last = t
		lim.tokens = tokens
		lim.lastEvent = r.timeToAct
	}

	return r
}

func (lim *Limiter) advance(t time.Time) (newT time.Time, newTokens float64) {
	last := lim.last
	if t.Before(last) {
		last = t
	}

	// Calculate the new number of tokens, due to time that passed.
	elapsed := t.Sub(last)
	delta := lim.limit.tokensFromDuration(elapsed)
	tokens := lim.tokens + delta
	if burst := float64(lim.burst); tokens > burst {
		tokens = burst
	}
	return t, tokens
}

What did you see happen?

limiter can't limit properly in multiple goroutine

What did you expect to see?

limiter limit properly in multiple goroutine

[seankhliao]

cc @Sajmani maybe?

it does appear that by passing a time before a lock is obtained to reserveN that drift can be an issue (so setting t = time.Now() in reserveN results in an accurate limit).

related #23145 ?

[ianzhang1988]

Is there any update on this?

[ianzhang1988]

I think the fix is simple as below.

func (lim *Limiter) advance(t time.Time) (newT time.Time, newTokens float64) {
	last := lim.last
	if t.Before(last) {
		// last = t
		t = last // here
	}

	// Calculate the new number of tokens, due to time that passed.
	elapsed := t.Sub(last)
	delta := lim.limit.tokensFromDuration(elapsed)
	tokens := lim.tokens + delta
	if burst := float64(lim.burst); tokens > burst {
		tokens = burst
	}
	return t, tokens
}

if t is in the past, advance should not produce any token. In the sense that token for this t is already produced in last call of advance.