+++ /dev/null
-All files in this repository are licensed as follows. If you contribute
-to this repository, it is assumed that you license your contribution
-under the same license unless you state otherwise.
-
-All files Copyright (C) 2015 Canonical Ltd. unless otherwise specified in the file.
-
-This software is licensed under the LGPLv3, included below.
-
-As a special exception to the GNU Lesser General Public License version 3
-("LGPL3"), the copyright holders of this Library give you permission to
-convey to a third party a Combined Work that links statically or dynamically
-to this Library without providing any Minimal Corresponding Source or
-Minimal Application Code as set out in 4d or providing the installation
-information set out in section 4e, provided that you comply with the other
-provisions of LGPL3 and provided that you meet, for the Application the
-terms and conditions of the license(s) which apply to the Application.
-
-Except as stated in this special exception, the provisions of LGPL3 will
-continue to comply in full to this Library. If you modify this Library, you
-may apply this exception to your version of this Library, but you are not
-obliged to do so. If you do not wish to do so, delete this exception
-statement from your version. This exception does not (and cannot) modify any
-license terms which apply to the Application, with which you must still
-comply.
-
-
- GNU LESSER GENERAL PUBLIC LICENSE
- Version 3, 29 June 2007
-
- Copyright (C) 2007 Free Software Foundation, Inc. <http://fsf.org/>
- Everyone is permitted to copy and distribute verbatim copies
- of this license document, but changing it is not allowed.
-
-
- This version of the GNU Lesser General Public License incorporates
-the terms and conditions of version 3 of the GNU General Public
-License, supplemented by the additional permissions listed below.
-
- 0. Additional Definitions.
-
- As used herein, "this License" refers to version 3 of the GNU Lesser
-General Public License, and the "GNU GPL" refers to version 3 of the GNU
-General Public License.
-
- "The Library" refers to a covered work governed by this License,
-other than an Application or a Combined Work as defined below.
-
- An "Application" is any work that makes use of an interface provided
-by the Library, but which is not otherwise based on the Library.
-Defining a subclass of a class defined by the Library is deemed a mode
-of using an interface provided by the Library.
-
- A "Combined Work" is a work produced by combining or linking an
-Application with the Library. The particular version of the Library
-with which the Combined Work was made is also called the "Linked
-Version".
-
- The "Minimal Corresponding Source" for a Combined Work means the
-Corresponding Source for the Combined Work, excluding any source code
-for portions of the Combined Work that, considered in isolation, are
-based on the Application, and not on the Linked Version.
-
- The "Corresponding Application Code" for a Combined Work means the
-object code and/or source code for the Application, including any data
-and utility programs needed for reproducing the Combined Work from the
-Application, but excluding the System Libraries of the Combined Work.
-
- 1. Exception to Section 3 of the GNU GPL.
-
- You may convey a covered work under sections 3 and 4 of this License
-without being bound by section 3 of the GNU GPL.
-
- 2. Conveying Modified Versions.
-
- If you modify a copy of the Library, and, in your modifications, a
-facility refers to a function or data to be supplied by an Application
-that uses the facility (other than as an argument passed when the
-facility is invoked), then you may convey a copy of the modified
-version:
-
- a) under this License, provided that you make a good faith effort to
- ensure that, in the event an Application does not supply the
- function or data, the facility still operates, and performs
- whatever part of its purpose remains meaningful, or
-
- b) under the GNU GPL, with none of the additional permissions of
- this License applicable to that copy.
-
- 3. Object Code Incorporating Material from Library Header Files.
-
- The object code form of an Application may incorporate material from
-a header file that is part of the Library. You may convey such object
-code under terms of your choice, provided that, if the incorporated
-material is not limited to numerical parameters, data structure
-layouts and accessors, or small macros, inline functions and templates
-(ten or fewer lines in length), you do both of the following:
-
- a) Give prominent notice with each copy of the object code that the
- Library is used in it and that the Library and its use are
- covered by this License.
-
- b) Accompany the object code with a copy of the GNU GPL and this license
- document.
-
- 4. Combined Works.
-
- You may convey a Combined Work under terms of your choice that,
-taken together, effectively do not restrict modification of the
-portions of the Library contained in the Combined Work and reverse
-engineering for debugging such modifications, if you also do each of
-the following:
-
- a) Give prominent notice with each copy of the Combined Work that
- the Library is used in it and that the Library and its use are
- covered by this License.
-
- b) Accompany the Combined Work with a copy of the GNU GPL and this license
- document.
-
- c) For a Combined Work that displays copyright notices during
- execution, include the copyright notice for the Library among
- these notices, as well as a reference directing the user to the
- copies of the GNU GPL and this license document.
-
- d) Do one of the following:
-
- 0) Convey the Minimal Corresponding Source under the terms of this
- License, and the Corresponding Application Code in a form
- suitable for, and under terms that permit, the user to
- recombine or relink the Application with a modified version of
- the Linked Version to produce a modified Combined Work, in the
- manner specified by section 6 of the GNU GPL for conveying
- Corresponding Source.
-
- 1) Use a suitable shared library mechanism for linking with the
- Library. A suitable mechanism is one that (a) uses at run time
- a copy of the Library already present on the user's computer
- system, and (b) will operate properly with a modified version
- of the Library that is interface-compatible with the Linked
- Version.
-
- e) Provide Installation Information, but only if you would otherwise
- be required to provide such information under section 6 of the
- GNU GPL, and only to the extent that such information is
- necessary to install and execute a modified version of the
- Combined Work produced by recombining or relinking the
- Application with a modified version of the Linked Version. (If
- you use option 4d0, the Installation Information must accompany
- the Minimal Corresponding Source and Corresponding Application
- Code. If you use option 4d1, you must provide the Installation
- Information in the manner specified by section 6 of the GNU GPL
- for conveying Corresponding Source.)
-
- 5. Combined Libraries.
-
- You may place library facilities that are a work based on the
-Library side by side in a single library together with other library
-facilities that are not Applications and are not covered by this
-License, and convey such a combined library under terms of your
-choice, if you do both of the following:
-
- a) Accompany the combined library with a copy of the same work based
- on the Library, uncombined with any other library facilities,
- conveyed under the terms of this License.
-
- b) Give prominent notice with the combined library that part of it
- is a work based on the Library, and explaining where to find the
- accompanying uncombined form of the same work.
-
- 6. Revised Versions of the GNU Lesser General Public License.
-
- The Free Software Foundation may publish revised and/or new versions
-of the GNU Lesser General Public License from time to time. Such new
-versions will be similar in spirit to the present version, but may
-differ in detail to address new problems or concerns.
-
- Each version is given a distinguishing version number. If the
-Library as you received it specifies that a certain numbered version
-of the GNU Lesser General Public License "or any later version"
-applies to it, you have the option of following the terms and
-conditions either of that published version or of any later version
-published by the Free Software Foundation. If the Library as you
-received it does not specify a version number of the GNU Lesser
-General Public License, you may choose any version of the GNU Lesser
-General Public License ever published by the Free Software Foundation.
-
- If the Library as you received it specifies that a proxy can decide
-whether future versions of the GNU Lesser General Public License shall
-apply, that proxy's public statement of acceptance of any version is
-permanent authorization for you to choose that version for the
-Library.
+++ /dev/null
-# ratelimit
---
- import "github.com/juju/ratelimit"
-
-The ratelimit package provides an efficient token bucket implementation. See
-http://en.wikipedia.org/wiki/Token_bucket.
-
-## Usage
-
-#### func Reader
-
-```go
-func Reader(r io.Reader, bucket *Bucket) io.Reader
-```
-Reader returns a reader that is rate limited by the given token bucket. Each
-token in the bucket represents one byte.
-
-#### func Writer
-
-```go
-func Writer(w io.Writer, bucket *Bucket) io.Writer
-```
-Writer returns a writer that is rate limited by the given token bucket. Each
-token in the bucket represents one byte.
-
-#### type Bucket
-
-```go
-type Bucket struct {
-}
-```
-
-Bucket represents a token bucket that fills at a predetermined rate. Methods on
-Bucket may be called concurrently.
-
-#### func NewBucket
-
-```go
-func NewBucket(fillInterval time.Duration, capacity int64) *Bucket
-```
-NewBucket returns a new token bucket that fills at the rate of one token every
-fillInterval, up to the given maximum capacity. Both arguments must be positive.
-The bucket is initially full.
-
-#### func NewBucketWithQuantum
-
-```go
-func NewBucketWithQuantum(fillInterval time.Duration, capacity, quantum int64) *Bucket
-```
-NewBucketWithQuantum is similar to NewBucket, but allows the specification of
-the quantum size - quantum tokens are added every fillInterval.
-
-#### func NewBucketWithRate
-
-```go
-func NewBucketWithRate(rate float64, capacity int64) *Bucket
-```
-NewBucketWithRate returns a token bucket that fills the bucket at the rate of
-rate tokens per second up to the given maximum capacity. Because of limited
-clock resolution, at high rates, the actual rate may be up to 1% different from
-the specified rate.
-
-#### func (*Bucket) Rate
-
-```go
-func (tb *Bucket) Rate() float64
-```
-Rate returns the fill rate of the bucket, in tokens per second.
-
-#### func (*Bucket) Take
-
-```go
-func (tb *Bucket) Take(count int64) time.Duration
-```
-Take takes count tokens from the bucket without blocking. It returns the time
-that the caller should wait until the tokens are actually available.
-
-Note that if the request is irrevocable - there is no way to return tokens to
-the bucket once this method commits us to taking them.
-
-#### func (*Bucket) TakeAvailable
-
-```go
-func (tb *Bucket) TakeAvailable(count int64) int64
-```
-TakeAvailable takes up to count immediately available tokens from the bucket. It
-returns the number of tokens removed, or zero if there are no available tokens.
-It does not block.
-
-#### func (*Bucket) TakeMaxDuration
-
-```go
-func (tb *Bucket) TakeMaxDuration(count int64, maxWait time.Duration) (time.Duration, bool)
-```
-TakeMaxDuration is like Take, except that it will only take tokens from the
-bucket if the wait time for the tokens is no greater than maxWait.
-
-If it would take longer than maxWait for the tokens to become available, it does
-nothing and reports false, otherwise it returns the time that the caller should
-wait until the tokens are actually available, and reports true.
-
-#### func (*Bucket) Wait
-
-```go
-func (tb *Bucket) Wait(count int64)
-```
-Wait takes count tokens from the bucket, waiting until they are available.
-
-#### func (*Bucket) WaitMaxDuration
-
-```go
-func (tb *Bucket) WaitMaxDuration(count int64, maxWait time.Duration) bool
-```
-WaitMaxDuration is like Wait except that it will only take tokens from the
-bucket if it needs to wait for no greater than maxWait. It reports whether any
-tokens have been removed from the bucket If no tokens have been removed, it
-returns immediately.
+++ /dev/null
-// Copyright 2014 Canonical Ltd.
-// Licensed under the LGPLv3 with static-linking exception.
-// See LICENCE file for details.
-
-// The ratelimit package provides an efficient token bucket implementation
-// that can be used to limit the rate of arbitrary things.
-// See http://en.wikipedia.org/wiki/Token_bucket.
-package ratelimit
-
-import (
- "math"
- "strconv"
- "sync"
- "time"
-)
-
-// Bucket represents a token bucket that fills at a predetermined rate.
-// Methods on Bucket may be called concurrently.
-type Bucket struct {
- startTime time.Time
- capacity int64
- quantum int64
- fillInterval time.Duration
-
- // The mutex guards the fields following it.
- mu sync.Mutex
-
- // avail holds the number of available tokens
- // in the bucket, as of availTick ticks from startTime.
- // It will be negative when there are consumers
- // waiting for tokens.
- avail int64
- availTick int64
-}
-
-// NewBucket returns a new token bucket that fills at the
-// rate of one token every fillInterval, up to the given
-// maximum capacity. Both arguments must be
-// positive. The bucket is initially full.
-func NewBucket(fillInterval time.Duration, capacity int64) *Bucket {
- return NewBucketWithQuantum(fillInterval, capacity, 1)
-}
-
-// rateMargin specifes the allowed variance of actual
-// rate from specified rate. 1% seems reasonable.
-const rateMargin = 0.01
-
-// NewBucketWithRate returns a token bucket that fills the bucket
-// at the rate of rate tokens per second up to the given
-// maximum capacity. Because of limited clock resolution,
-// at high rates, the actual rate may be up to 1% different from the
-// specified rate.
-func NewBucketWithRate(rate float64, capacity int64) *Bucket {
- for quantum := int64(1); quantum < 1<<50; quantum = nextQuantum(quantum) {
- fillInterval := time.Duration(1e9 * float64(quantum) / rate)
- if fillInterval <= 0 {
- continue
- }
- tb := NewBucketWithQuantum(fillInterval, capacity, quantum)
- if diff := math.Abs(tb.Rate() - rate); diff/rate <= rateMargin {
- return tb
- }
- }
- panic("cannot find suitable quantum for " + strconv.FormatFloat(rate, 'g', -1, 64))
-}
-
-// nextQuantum returns the next quantum to try after q.
-// We grow the quantum exponentially, but slowly, so we
-// get a good fit in the lower numbers.
-func nextQuantum(q int64) int64 {
- q1 := q * 11 / 10
- if q1 == q {
- q1++
- }
- return q1
-}
-
-// NewBucketWithQuantum is similar to NewBucket, but allows
-// the specification of the quantum size - quantum tokens
-// are added every fillInterval.
-func NewBucketWithQuantum(fillInterval time.Duration, capacity, quantum int64) *Bucket {
- if fillInterval <= 0 {
- panic("token bucket fill interval is not > 0")
- }
- if capacity <= 0 {
- panic("token bucket capacity is not > 0")
- }
- if quantum <= 0 {
- panic("token bucket quantum is not > 0")
- }
- return &Bucket{
- startTime: time.Now(),
- capacity: capacity,
- quantum: quantum,
- avail: capacity,
- fillInterval: fillInterval,
- }
-}
-
-// Wait takes count tokens from the bucket, waiting until they are
-// available.
-func (tb *Bucket) Wait(count int64) {
- if d := tb.Take(count); d > 0 {
- time.Sleep(d)
- }
-}
-
-// WaitMaxDuration is like Wait except that it will
-// only take tokens from the bucket if it needs to wait
-// for no greater than maxWait. It reports whether
-// any tokens have been removed from the bucket
-// If no tokens have been removed, it returns immediately.
-func (tb *Bucket) WaitMaxDuration(count int64, maxWait time.Duration) bool {
- d, ok := tb.TakeMaxDuration(count, maxWait)
- if d > 0 {
- time.Sleep(d)
- }
- return ok
-}
-
-const infinityDuration time.Duration = 0x7fffffffffffffff
-
-// Take takes count tokens from the bucket without blocking. It returns
-// the time that the caller should wait until the tokens are actually
-// available.
-//
-// Note that if the request is irrevocable - there is no way to return
-// tokens to the bucket once this method commits us to taking them.
-func (tb *Bucket) Take(count int64) time.Duration {
- d, _ := tb.take(time.Now(), count, infinityDuration)
- return d
-}
-
-// TakeMaxDuration is like Take, except that
-// it will only take tokens from the bucket if the wait
-// time for the tokens is no greater than maxWait.
-//
-// If it would take longer than maxWait for the tokens
-// to become available, it does nothing and reports false,
-// otherwise it returns the time that the caller should
-// wait until the tokens are actually available, and reports
-// true.
-func (tb *Bucket) TakeMaxDuration(count int64, maxWait time.Duration) (time.Duration, bool) {
- return tb.take(time.Now(), count, maxWait)
-}
-
-// TakeAvailable takes up to count immediately available tokens from the
-// bucket. It returns the number of tokens removed, or zero if there are
-// no available tokens. It does not block.
-func (tb *Bucket) TakeAvailable(count int64) int64 {
- return tb.takeAvailable(time.Now(), count)
-}
-
-// takeAvailable is the internal version of TakeAvailable - it takes the
-// current time as an argument to enable easy testing.
-func (tb *Bucket) takeAvailable(now time.Time, count int64) int64 {
- if count <= 0 {
- return 0
- }
- tb.mu.Lock()
- defer tb.mu.Unlock()
-
- tb.adjust(now)
- if tb.avail <= 0 {
- return 0
- }
- if count > tb.avail {
- count = tb.avail
- }
- tb.avail -= count
- return count
-}
-
-// Available returns the number of available tokens. It will be negative
-// when there are consumers waiting for tokens. Note that if this
-// returns greater than zero, it does not guarantee that calls that take
-// tokens from the buffer will succeed, as the number of available
-// tokens could have changed in the meantime. This method is intended
-// primarily for metrics reporting and debugging.
-func (tb *Bucket) Available() int64 {
- return tb.available(time.Now())
-}
-
-// available is the internal version of available - it takes the current time as
-// an argument to enable easy testing.
-func (tb *Bucket) available(now time.Time) int64 {
- tb.mu.Lock()
- defer tb.mu.Unlock()
- tb.adjust(now)
- return tb.avail
-}
-
-// Capacity returns the capacity that the bucket was created with.
-func (tb *Bucket) Capacity() int64 {
- return tb.capacity
-}
-
-// Rate returns the fill rate of the bucket, in tokens per second.
-func (tb *Bucket) Rate() float64 {
- return 1e9 * float64(tb.quantum) / float64(tb.fillInterval)
-}
-
-// take is the internal version of Take - it takes the current time as
-// an argument to enable easy testing.
-func (tb *Bucket) take(now time.Time, count int64, maxWait time.Duration) (time.Duration, bool) {
- if count <= 0 {
- return 0, true
- }
- tb.mu.Lock()
- defer tb.mu.Unlock()
-
- currentTick := tb.adjust(now)
- avail := tb.avail - count
- if avail >= 0 {
- tb.avail = avail
- return 0, true
- }
- // Round up the missing tokens to the nearest multiple
- // of quantum - the tokens won't be available until
- // that tick.
- 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
- }
- tb.avail = avail
- return waitTime, true
-}
-
-// adjust adjusts the current bucket capacity based on the current time.
-// It returns the current tick.
-func (tb *Bucket) adjust(now time.Time) (currentTick int64) {
- currentTick = int64(now.Sub(tb.startTime) / tb.fillInterval)
-
- if tb.avail >= tb.capacity {
- return
- }
- tb.avail += (currentTick - tb.availTick) * tb.quantum
- if tb.avail > tb.capacity {
- tb.avail = tb.capacity
- }
- tb.availTick = currentTick
- return
-}
+++ /dev/null
-// Copyright 2014 Canonical Ltd.
-// Licensed under the LGPLv3 with static-linking exception.
-// See LICENCE file for details.
-
-package ratelimit
-
-import "io"
-
-type reader struct {
- r io.Reader
- bucket *Bucket
-}
-
-// Reader returns a reader that is rate limited by
-// the given token bucket. Each token in the bucket
-// represents one byte.
-func Reader(r io.Reader, bucket *Bucket) io.Reader {
- return &reader{
- r: r,
- bucket: bucket,
- }
-}
-
-func (r *reader) Read(buf []byte) (int, error) {
- n, err := r.r.Read(buf)
- if n <= 0 {
- return n, err
- }
- r.bucket.Wait(int64(n))
- return n, err
-}
-
-type writer struct {
- w io.Writer
- bucket *Bucket
-}
-
-// Writer returns a reader that is rate limited by
-// the given token bucket. Each token in the bucket
-// represents one byte.
-func Writer(w io.Writer, bucket *Bucket) io.Writer {
- return &writer{
- w: w,
- bucket: bucket,
- }
-}
-
-func (w *writer) Write(buf []byte) (int, error) {
- w.bucket.Wait(int64(len(buf)))
- return w.w.Write(buf)
-}
--- /dev/null
+# This source code refers to The Go Authors for copyright purposes.
+# The master list of authors is in the main Go distribution,
+# visible at http://tip.golang.org/AUTHORS.
--- /dev/null
+# Contributing to Go
+
+Go is an open source project.
+
+It is the work of hundreds of contributors. We appreciate your help!
+
+## Filing issues
+
+When [filing an issue](https://golang.org/issue/new), make sure to answer these five questions:
+
+1. What version of Go are you using (`go version`)?
+2. What operating system and processor architecture are you using?
+3. What did you do?
+4. What did you expect to see?
+5. What did you see instead?
+
+General questions should go to the [golang-nuts mailing list](https://groups.google.com/group/golang-nuts) instead of the issue tracker.
+The gophers there will answer or ask you to file an issue if you've tripped over a bug.
+
+## Contributing code
+
+Please read the [Contribution Guidelines](https://golang.org/doc/contribute.html)
+before sending patches.
+
+Unless otherwise noted, the Go source files are distributed under
+the BSD-style license found in the LICENSE file.
--- /dev/null
+# This source code was written by the Go contributors.
+# The master list of contributors is in the main Go distribution,
+# visible at http://tip.golang.org/CONTRIBUTORS.
--- /dev/null
+Copyright (c) 2009 The Go Authors. All rights reserved.
+
+Redistribution and use in source and binary forms, with or without
+modification, are permitted provided that the following conditions are
+met:
+
+ * Redistributions of source code must retain the above copyright
+notice, this list of conditions and the following disclaimer.
+ * Redistributions in binary form must reproduce the above
+copyright notice, this list of conditions and the following disclaimer
+in the documentation and/or other materials provided with the
+distribution.
+ * Neither the name of Google Inc. nor the names of its
+contributors may be used to endorse or promote products derived from
+this software without specific prior written permission.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
--- /dev/null
+Additional IP Rights Grant (Patents)
+
+"This implementation" means the copyrightable works distributed by
+Google as part of the Go project.
+
+Google hereby grants to You a perpetual, worldwide, non-exclusive,
+no-charge, royalty-free, irrevocable (except as stated in this section)
+patent license to make, have made, use, offer to sell, sell, import,
+transfer and otherwise run, modify and propagate the contents of this
+implementation of Go, where such license applies only to those patent
+claims, both currently owned or controlled by Google and acquired in
+the future, licensable by Google that are necessarily infringed by this
+implementation of Go. This grant does not include claims that would be
+infringed only as a consequence of further modification of this
+implementation. If you or your agent or exclusive licensee institute or
+order or agree to the institution of patent litigation against any
+entity (including a cross-claim or counterclaim in a lawsuit) alleging
+that this implementation of Go or any code incorporated within this
+implementation of Go constitutes direct or contributory patent
+infringement, or inducement of patent infringement, then any patent
+rights granted to you under this License for this implementation of Go
+shall terminate as of the date such litigation is filed.
--- /dev/null
+// Copyright 2015 The Go Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+// Package rate provides a rate limiter.
+package rate
+
+import (
+ "fmt"
+ "math"
+ "sync"
+ "time"
+)
+
+// Limit defines the maximum frequency of some events.
+// Limit is represented as number of events per second.
+// A zero Limit allows no events.
+type Limit float64
+
+// Inf is the infinite rate limit; it allows all events (even if burst is zero).
+const Inf = Limit(math.MaxFloat64)
+
+// Every converts a minimum time interval between events to a Limit.
+func Every(interval time.Duration) Limit {
+ if interval <= 0 {
+ return Inf
+ }
+ return 1 / Limit(interval.Seconds())
+}
+
+// A Limiter controls how frequently events are allowed to happen.
+// It implements a "token bucket" of size b, initially full and refilled
+// at rate r tokens per second.
+// Informally, in any large enough time interval, the Limiter limits the
+// rate to r tokens per second, with a maximum burst size of b events.
+// As a special case, if r == Inf (the infinite rate), b is ignored.
+// See https://en.wikipedia.org/wiki/Token_bucket for more about token buckets.
+//
+// The zero value is a valid Limiter, but it will reject all events.
+// Use NewLimiter to create non-zero Limiters.
+//
+// Limiter has three main methods, Allow, Reserve, and Wait.
+// Most callers should use Wait.
+//
+// Each of the three methods consumes a single token.
+// They differ in their behavior when no token is available.
+// If no token is available, Allow returns false.
+// If no token is available, Reserve returns a reservation for a future token
+// and the amount of time the caller must wait before using it.
+// If no token is available, Wait blocks until one can be obtained
+// or its associated context.Context is canceled.
+//
+// The methods AllowN, ReserveN, and WaitN consume n tokens.
+type Limiter struct {
+ limit Limit
+ burst int
+
+ mu sync.Mutex
+ tokens float64
+ // last is the last time the limiter's tokens field was updated
+ last time.Time
+ // lastEvent is the latest time of a rate-limited event (past or future)
+ lastEvent time.Time
+}
+
+// Limit returns the maximum overall event rate.
+func (lim *Limiter) Limit() Limit {
+ lim.mu.Lock()
+ defer lim.mu.Unlock()
+ return lim.limit
+}
+
+// Burst returns the maximum burst size. Burst is the maximum number of tokens
+// that can be consumed in a single call to Allow, Reserve, or Wait, so higher
+// Burst values allow more events to happen at once.
+// A zero Burst allows no events, unless limit == Inf.
+func (lim *Limiter) Burst() int {
+ return lim.burst
+}
+
+// NewLimiter returns a new Limiter that allows events up to rate r and permits
+// bursts of at most b tokens.
+func NewLimiter(r Limit, b int) *Limiter {
+ return &Limiter{
+ limit: r,
+ burst: b,
+ }
+}
+
+// Allow is shorthand for AllowN(time.Now(), 1).
+func (lim *Limiter) Allow() bool {
+ return lim.AllowN(time.Now(), 1)
+}
+
+// AllowN reports whether n events may happen at time now.
+// Use this method if you intend to drop / skip events that exceed the rate limit.
+// Otherwise use Reserve or Wait.
+func (lim *Limiter) AllowN(now time.Time, n int) bool {
+ return lim.reserveN(now, n, 0).ok
+}
+
+// A Reservation holds information about events that are permitted by a Limiter to happen after a delay.
+// A Reservation may be canceled, which may enable the Limiter to permit additional events.
+type Reservation struct {
+ ok bool
+ lim *Limiter
+ tokens int
+ timeToAct time.Time
+ // This is the Limit at reservation time, it can change later.
+ limit Limit
+}
+
+// OK returns whether the limiter can provide the requested number of tokens
+// within the maximum wait time. If OK is false, Delay returns InfDuration, and
+// Cancel does nothing.
+func (r *Reservation) OK() bool {
+ return r.ok
+}
+
+// Delay is shorthand for DelayFrom(time.Now()).
+func (r *Reservation) Delay() time.Duration {
+ return r.DelayFrom(time.Now())
+}
+
+// InfDuration is the duration returned by Delay when a Reservation is not OK.
+const InfDuration = time.Duration(1<<63 - 1)
+
+// DelayFrom returns the duration for which the reservation holder must wait
+// before taking the reserved action. Zero duration means act immediately.
+// InfDuration means the limiter cannot grant the tokens requested in this
+// Reservation within the maximum wait time.
+func (r *Reservation) DelayFrom(now time.Time) time.Duration {
+ if !r.ok {
+ return InfDuration
+ }
+ delay := r.timeToAct.Sub(now)
+ if delay < 0 {
+ return 0
+ }
+ return delay
+}
+
+// Cancel is shorthand for CancelAt(time.Now()).
+func (r *Reservation) Cancel() {
+ r.CancelAt(time.Now())
+ return
+}
+
+// CancelAt indicates that the reservation holder will not perform the reserved action
+// and reverses the effects of this Reservation on the rate limit as much as possible,
+// considering that other reservations may have already been made.
+func (r *Reservation) CancelAt(now time.Time) {
+ if !r.ok {
+ return
+ }
+
+ r.lim.mu.Lock()
+ defer r.lim.mu.Unlock()
+
+ if r.lim.limit == Inf || r.tokens == 0 || r.timeToAct.Before(now) {
+ return
+ }
+
+ // calculate tokens to restore
+ // The duration between lim.lastEvent and r.timeToAct tells us how many tokens were reserved
+ // after r was obtained. These tokens should not be restored.
+ restoreTokens := float64(r.tokens) - r.limit.tokensFromDuration(r.lim.lastEvent.Sub(r.timeToAct))
+ if restoreTokens <= 0 {
+ return
+ }
+ // advance time to now
+ now, _, tokens := r.lim.advance(now)
+ // calculate new number of tokens
+ tokens += restoreTokens
+ if burst := float64(r.lim.burst); tokens > burst {
+ tokens = burst
+ }
+ // update state
+ r.lim.last = now
+ r.lim.tokens = tokens
+ if r.timeToAct == r.lim.lastEvent {
+ prevEvent := r.timeToAct.Add(r.limit.durationFromTokens(float64(-r.tokens)))
+ if !prevEvent.Before(now) {
+ r.lim.lastEvent = prevEvent
+ }
+ }
+
+ return
+}
+
+// Reserve is shorthand for ReserveN(time.Now(), 1).
+func (lim *Limiter) Reserve() *Reservation {
+ return lim.ReserveN(time.Now(), 1)
+}
+
+// ReserveN returns a Reservation that indicates how long the caller must wait before n events happen.
+// The Limiter takes this Reservation into account when allowing future events.
+// ReserveN returns false if n exceeds the Limiter's burst size.
+// Usage example:
+// r := lim.ReserveN(time.Now(), 1)
+// if !r.OK() {
+// // Not allowed to act! Did you remember to set lim.burst to be > 0 ?
+// return
+// }
+// time.Sleep(r.Delay())
+// Act()
+// Use this method if you wish to wait and slow down in accordance with the rate limit without dropping events.
+// If you need to respect a deadline or cancel the delay, use Wait instead.
+// To drop or skip events exceeding rate limit, use Allow instead.
+func (lim *Limiter) ReserveN(now time.Time, n int) *Reservation {
+ r := lim.reserveN(now, n, InfDuration)
+ return &r
+}
+
+// contextContext is a temporary(?) copy of the context.Context type
+// to support both Go 1.6 using golang.org/x/net/context and Go 1.7+
+// with the built-in context package. If people ever stop using Go 1.6
+// we can remove this.
+type contextContext interface {
+ Deadline() (deadline time.Time, ok bool)
+ Done() <-chan struct{}
+ Err() error
+ Value(key interface{}) interface{}
+}
+
+// Wait is shorthand for WaitN(ctx, 1).
+func (lim *Limiter) wait(ctx contextContext) (err error) {
+ return lim.WaitN(ctx, 1)
+}
+
+// WaitN blocks until lim permits n events to happen.
+// It returns an error if n exceeds the Limiter's burst size, the Context is
+// canceled, or the expected wait time exceeds the Context's Deadline.
+// The burst limit is ignored if the rate limit is Inf.
+func (lim *Limiter) waitN(ctx contextContext, n int) (err error) {
+ if n > lim.burst && lim.limit != Inf {
+ return fmt.Errorf("rate: Wait(n=%d) exceeds limiter's burst %d", n, lim.burst)
+ }
+ // Check if ctx is already cancelled
+ select {
+ case <-ctx.Done():
+ return ctx.Err()
+ default:
+ }
+ // Determine wait limit
+ now := time.Now()
+ waitLimit := InfDuration
+ if deadline, ok := ctx.Deadline(); ok {
+ waitLimit = deadline.Sub(now)
+ }
+ // Reserve
+ r := lim.reserveN(now, n, waitLimit)
+ if !r.ok {
+ return fmt.Errorf("rate: Wait(n=%d) would exceed context deadline", n)
+ }
+ // Wait if necessary
+ delay := r.DelayFrom(now)
+ if delay == 0 {
+ return nil
+ }
+ t := time.NewTimer(delay)
+ defer t.Stop()
+ select {
+ case <-t.C:
+ // We can proceed.
+ return nil
+ case <-ctx.Done():
+ // Context was canceled before we could proceed. Cancel the
+ // reservation, which may permit other events to proceed sooner.
+ r.Cancel()
+ return ctx.Err()
+ }
+}
+
+// SetLimit is shorthand for SetLimitAt(time.Now(), newLimit).
+func (lim *Limiter) SetLimit(newLimit Limit) {
+ lim.SetLimitAt(time.Now(), newLimit)
+}
+
+// SetLimitAt sets a new Limit for the limiter. The new Limit, and Burst, may be violated
+// or underutilized by those which reserved (using Reserve or Wait) but did not yet act
+// before SetLimitAt was called.
+func (lim *Limiter) SetLimitAt(now time.Time, newLimit Limit) {
+ lim.mu.Lock()
+ defer lim.mu.Unlock()
+
+ now, _, tokens := lim.advance(now)
+
+ lim.last = now
+ lim.tokens = tokens
+ lim.limit = newLimit
+}
+
+// reserveN is a helper method for AllowN, ReserveN, and WaitN.
+// maxFutureReserve specifies the maximum reservation wait duration allowed.
+// reserveN returns Reservation, not *Reservation, to avoid allocation in AllowN and WaitN.
+func (lim *Limiter) reserveN(now time.Time, n int, maxFutureReserve time.Duration) Reservation {
+ lim.mu.Lock()
+
+ if lim.limit == Inf {
+ lim.mu.Unlock()
+ return Reservation{
+ ok: true,
+ lim: lim,
+ tokens: n,
+ timeToAct: now,
+ }
+ }
+
+ now, last, tokens := lim.advance(now)
+
+ // 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 = now.Add(waitDuration)
+ }
+
+ // Update state
+ if ok {
+ lim.last = now
+ lim.tokens = tokens
+ lim.lastEvent = r.timeToAct
+ } else {
+ lim.last = last
+ }
+
+ lim.mu.Unlock()
+ return r
+}
+
+// advance calculates and returns an updated state for lim resulting from the passage of time.
+// lim is not changed.
+func (lim *Limiter) advance(now time.Time) (newNow time.Time, newLast time.Time, newTokens float64) {
+ last := lim.last
+ if now.Before(last) {
+ last = now
+ }
+
+ // Avoid making delta overflow below when last is very old.
+ maxElapsed := lim.limit.durationFromTokens(float64(lim.burst) - lim.tokens)
+ elapsed := now.Sub(last)
+ if elapsed > maxElapsed {
+ elapsed = maxElapsed
+ }
+
+ // Calculate the new number of tokens, due to time that passed.
+ delta := lim.limit.tokensFromDuration(elapsed)
+ tokens := lim.tokens + delta
+ if burst := float64(lim.burst); tokens > burst {
+ tokens = burst
+ }
+
+ return now, last, tokens
+}
+
+// durationFromTokens is a unit conversion function from the number of tokens to the duration
+// of time it takes to accumulate them at a rate of limit tokens per second.
+func (limit Limit) durationFromTokens(tokens float64) time.Duration {
+ seconds := tokens / float64(limit)
+ return time.Nanosecond * time.Duration(1e9*seconds)
+}
+
+// tokensFromDuration is a unit conversion function from a time duration to the number of tokens
+// which could be accumulated during that duration at a rate of limit tokens per second.
+func (limit Limit) tokensFromDuration(d time.Duration) float64 {
+ return d.Seconds() * float64(limit)
+}
--- /dev/null
+// Copyright 2017 The Go Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+// +build !go1.7
+
+package rate
+
+import "golang.org/x/net/context"
+
+// Wait is shorthand for WaitN(ctx, 1).
+func (lim *Limiter) Wait(ctx context.Context) (err error) {
+ return lim.waitN(ctx, 1)
+}
+
+// WaitN blocks until lim permits n events to happen.
+// It returns an error if n exceeds the Limiter's burst size, the Context is
+// canceled, or the expected wait time exceeds the Context's Deadline.
+func (lim *Limiter) WaitN(ctx context.Context, n int) (err error) {
+ return lim.waitN(ctx, n)
+}
--- /dev/null
+// Copyright 2017 The Go Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+// +build go1.7
+
+package rate
+
+import "context"
+
+// Wait is shorthand for WaitN(ctx, 1).
+func (lim *Limiter) Wait(ctx context.Context) (err error) {
+ return lim.waitN(ctx, 1)
+}
+
+// WaitN blocks until lim permits n events to happen.
+// It returns an error if n exceeds the Limiter's burst size, the Context is
+// canceled, or the expected wait time exceeds the Context's Deadline.
+func (lim *Limiter) WaitN(ctx context.Context, n int) (err error) {
+ return lim.waitN(ctx, n)
+}
--- /dev/null
+// Copyright 2015 The Go Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+// +build go1.7
+
+package rate
+
+import (
+ "context"
+ "math"
+ "runtime"
+ "sync"
+ "sync/atomic"
+ "testing"
+ "time"
+)
+
+func TestLimit(t *testing.T) {
+ if Limit(10) == Inf {
+ t.Errorf("Limit(10) == Inf should be false")
+ }
+}
+
+func closeEnough(a, b Limit) bool {
+ return (math.Abs(float64(a)/float64(b)) - 1.0) < 1e-9
+}
+
+func TestEvery(t *testing.T) {
+ cases := []struct {
+ interval time.Duration
+ lim Limit
+ }{
+ {0, Inf},
+ {-1, Inf},
+ {1 * time.Nanosecond, Limit(1e9)},
+ {1 * time.Microsecond, Limit(1e6)},
+ {1 * time.Millisecond, Limit(1e3)},
+ {10 * time.Millisecond, Limit(100)},
+ {100 * time.Millisecond, Limit(10)},
+ {1 * time.Second, Limit(1)},
+ {2 * time.Second, Limit(0.5)},
+ {time.Duration(2.5 * float64(time.Second)), Limit(0.4)},
+ {4 * time.Second, Limit(0.25)},
+ {10 * time.Second, Limit(0.1)},
+ {time.Duration(math.MaxInt64), Limit(1e9 / float64(math.MaxInt64))},
+ }
+ for _, tc := range cases {
+ lim := Every(tc.interval)
+ if !closeEnough(lim, tc.lim) {
+ t.Errorf("Every(%v) = %v want %v", tc.interval, lim, tc.lim)
+ }
+ }
+}
+
+const (
+ d = 100 * time.Millisecond
+)
+
+var (
+ t0 = time.Now()
+ t1 = t0.Add(time.Duration(1) * d)
+ t2 = t0.Add(time.Duration(2) * d)
+ t3 = t0.Add(time.Duration(3) * d)
+ t4 = t0.Add(time.Duration(4) * d)
+ t5 = t0.Add(time.Duration(5) * d)
+ t9 = t0.Add(time.Duration(9) * d)
+)
+
+type allow struct {
+ t time.Time
+ n int
+ ok bool
+}
+
+func run(t *testing.T, lim *Limiter, allows []allow) {
+ for i, allow := range allows {
+ ok := lim.AllowN(allow.t, allow.n)
+ if ok != allow.ok {
+ t.Errorf("step %d: lim.AllowN(%v, %v) = %v want %v",
+ i, allow.t, allow.n, ok, allow.ok)
+ }
+ }
+}
+
+func TestLimiterBurst1(t *testing.T) {
+ run(t, NewLimiter(10, 1), []allow{
+ {t0, 1, true},
+ {t0, 1, false},
+ {t0, 1, false},
+ {t1, 1, true},
+ {t1, 1, false},
+ {t1, 1, false},
+ {t2, 2, false}, // burst size is 1, so n=2 always fails
+ {t2, 1, true},
+ {t2, 1, false},
+ })
+}
+
+func TestLimiterBurst3(t *testing.T) {
+ run(t, NewLimiter(10, 3), []allow{
+ {t0, 2, true},
+ {t0, 2, false},
+ {t0, 1, true},
+ {t0, 1, false},
+ {t1, 4, false},
+ {t2, 1, true},
+ {t3, 1, true},
+ {t4, 1, true},
+ {t4, 1, true},
+ {t4, 1, false},
+ {t4, 1, false},
+ {t9, 3, true},
+ {t9, 0, true},
+ })
+}
+
+func TestLimiterJumpBackwards(t *testing.T) {
+ run(t, NewLimiter(10, 3), []allow{
+ {t1, 1, true}, // start at t1
+ {t0, 1, true}, // jump back to t0, two tokens remain
+ {t0, 1, true},
+ {t0, 1, false},
+ {t0, 1, false},
+ {t1, 1, true}, // got a token
+ {t1, 1, false},
+ {t1, 1, false},
+ {t2, 1, true}, // got another token
+ {t2, 1, false},
+ {t2, 1, false},
+ })
+}
+
+func TestSimultaneousRequests(t *testing.T) {
+ const (
+ limit = 1
+ burst = 5
+ numRequests = 15
+ )
+ var (
+ wg sync.WaitGroup
+ numOK = uint32(0)
+ )
+
+ // Very slow replenishing bucket.
+ lim := NewLimiter(limit, burst)
+
+ // Tries to take a token, atomically updates the counter and decreases the wait
+ // group counter.
+ f := func() {
+ defer wg.Done()
+ if ok := lim.Allow(); ok {
+ atomic.AddUint32(&numOK, 1)
+ }
+ }
+
+ wg.Add(numRequests)
+ for i := 0; i < numRequests; i++ {
+ go f()
+ }
+ wg.Wait()
+ if numOK != burst {
+ t.Errorf("numOK = %d, want %d", numOK, burst)
+ }
+}
+
+func TestLongRunningQPS(t *testing.T) {
+ if testing.Short() {
+ t.Skip("skipping in short mode")
+ }
+ if runtime.GOOS == "openbsd" {
+ t.Skip("low resolution time.Sleep invalidates test (golang.org/issue/14183)")
+ return
+ }
+
+ // The test runs for a few seconds executing many requests and then checks
+ // that overall number of requests is reasonable.
+ const (
+ limit = 100
+ burst = 100
+ )
+ var numOK = int32(0)
+
+ lim := NewLimiter(limit, burst)
+
+ var wg sync.WaitGroup
+ f := func() {
+ if ok := lim.Allow(); ok {
+ atomic.AddInt32(&numOK, 1)
+ }
+ wg.Done()
+ }
+
+ start := time.Now()
+ end := start.Add(5 * time.Second)
+ for time.Now().Before(end) {
+ wg.Add(1)
+ go f()
+
+ // This will still offer ~500 requests per second, but won't consume
+ // outrageous amount of CPU.
+ time.Sleep(2 * time.Millisecond)
+ }
+ wg.Wait()
+ elapsed := time.Since(start)
+ ideal := burst + (limit * float64(elapsed) / float64(time.Second))
+
+ // We should never get more requests than allowed.
+ if want := int32(ideal + 1); numOK > want {
+ t.Errorf("numOK = %d, want %d (ideal %f)", numOK, want, ideal)
+ }
+ // We should get very close to the number of requests allowed.
+ if want := int32(0.999 * ideal); numOK < want {
+ t.Errorf("numOK = %d, want %d (ideal %f)", numOK, want, ideal)
+ }
+}
+
+type request struct {
+ t time.Time
+ n int
+ act time.Time
+ ok bool
+}
+
+// dFromDuration converts a duration to a multiple of the global constant d
+func dFromDuration(dur time.Duration) int {
+ // Adding a millisecond to be swallowed by the integer division
+ // because we don't care about small inaccuracies
+ return int((dur + time.Millisecond) / d)
+}
+
+// dSince returns multiples of d since t0
+func dSince(t time.Time) int {
+ return dFromDuration(t.Sub(t0))
+}
+
+func runReserve(t *testing.T, lim *Limiter, req request) *Reservation {
+ return runReserveMax(t, lim, req, InfDuration)
+}
+
+func runReserveMax(t *testing.T, lim *Limiter, req request, maxReserve time.Duration) *Reservation {
+ r := lim.reserveN(req.t, req.n, maxReserve)
+ if r.ok && (dSince(r.timeToAct) != dSince(req.act)) || r.ok != req.ok {
+ t.Errorf("lim.reserveN(t%d, %v, %v) = (t%d, %v) want (t%d, %v)",
+ dSince(req.t), req.n, maxReserve, dSince(r.timeToAct), r.ok, dSince(req.act), req.ok)
+ }
+ return &r
+}
+
+func TestSimpleReserve(t *testing.T) {
+ lim := NewLimiter(10, 2)
+
+ runReserve(t, lim, request{t0, 2, t0, true})
+ runReserve(t, lim, request{t0, 2, t2, true})
+ runReserve(t, lim, request{t3, 2, t4, true})
+}
+
+func TestMix(t *testing.T) {
+ lim := NewLimiter(10, 2)
+
+ runReserve(t, lim, request{t0, 3, t1, false}) // should return false because n > Burst
+ runReserve(t, lim, request{t0, 2, t0, true})
+ run(t, lim, []allow{{t1, 2, false}}) // not enought tokens - don't allow
+ runReserve(t, lim, request{t1, 2, t2, true})
+ run(t, lim, []allow{{t1, 1, false}}) // negative tokens - don't allow
+ run(t, lim, []allow{{t3, 1, true}})
+}
+
+func TestCancelInvalid(t *testing.T) {
+ lim := NewLimiter(10, 2)
+
+ runReserve(t, lim, request{t0, 2, t0, true})
+ r := runReserve(t, lim, request{t0, 3, t3, false})
+ r.CancelAt(t0) // should have no effect
+ runReserve(t, lim, request{t0, 2, t2, true}) // did not get extra tokens
+}
+
+func TestCancelLast(t *testing.T) {
+ lim := NewLimiter(10, 2)
+
+ runReserve(t, lim, request{t0, 2, t0, true})
+ r := runReserve(t, lim, request{t0, 2, t2, true})
+ r.CancelAt(t1) // got 2 tokens back
+ runReserve(t, lim, request{t1, 2, t2, true})
+}
+
+func TestCancelTooLate(t *testing.T) {
+ lim := NewLimiter(10, 2)
+
+ runReserve(t, lim, request{t0, 2, t0, true})
+ r := runReserve(t, lim, request{t0, 2, t2, true})
+ r.CancelAt(t3) // too late to cancel - should have no effect
+ runReserve(t, lim, request{t3, 2, t4, true})
+}
+
+func TestCancel0Tokens(t *testing.T) {
+ lim := NewLimiter(10, 2)
+
+ runReserve(t, lim, request{t0, 2, t0, true})
+ r := runReserve(t, lim, request{t0, 1, t1, true})
+ runReserve(t, lim, request{t0, 1, t2, true})
+ r.CancelAt(t0) // got 0 tokens back
+ runReserve(t, lim, request{t0, 1, t3, true})
+}
+
+func TestCancel1Token(t *testing.T) {
+ lim := NewLimiter(10, 2)
+
+ runReserve(t, lim, request{t0, 2, t0, true})
+ r := runReserve(t, lim, request{t0, 2, t2, true})
+ runReserve(t, lim, request{t0, 1, t3, true})
+ r.CancelAt(t2) // got 1 token back
+ runReserve(t, lim, request{t2, 2, t4, true})
+}
+
+func TestCancelMulti(t *testing.T) {
+ lim := NewLimiter(10, 4)
+
+ runReserve(t, lim, request{t0, 4, t0, true})
+ rA := runReserve(t, lim, request{t0, 3, t3, true})
+ runReserve(t, lim, request{t0, 1, t4, true})
+ rC := runReserve(t, lim, request{t0, 1, t5, true})
+ rC.CancelAt(t1) // get 1 token back
+ rA.CancelAt(t1) // get 2 tokens back, as if C was never reserved
+ runReserve(t, lim, request{t1, 3, t5, true})
+}
+
+func TestReserveJumpBack(t *testing.T) {
+ lim := NewLimiter(10, 2)
+
+ runReserve(t, lim, request{t1, 2, t1, true}) // start at t1
+ runReserve(t, lim, request{t0, 1, t1, true}) // should violate Limit,Burst
+ runReserve(t, lim, request{t2, 2, t3, true})
+}
+
+func TestReserveJumpBackCancel(t *testing.T) {
+ lim := NewLimiter(10, 2)
+
+ runReserve(t, lim, request{t1, 2, t1, true}) // start at t1
+ r := runReserve(t, lim, request{t1, 2, t3, true})
+ runReserve(t, lim, request{t1, 1, t4, true})
+ r.CancelAt(t0) // cancel at t0, get 1 token back
+ runReserve(t, lim, request{t1, 2, t4, true}) // should violate Limit,Burst
+}
+
+func TestReserveSetLimit(t *testing.T) {
+ lim := NewLimiter(5, 2)
+
+ runReserve(t, lim, request{t0, 2, t0, true})
+ runReserve(t, lim, request{t0, 2, t4, true})
+ lim.SetLimitAt(t2, 10)
+ runReserve(t, lim, request{t2, 1, t4, true}) // violates Limit and Burst
+}
+
+func TestReserveSetLimitCancel(t *testing.T) {
+ lim := NewLimiter(5, 2)
+
+ runReserve(t, lim, request{t0, 2, t0, true})
+ r := runReserve(t, lim, request{t0, 2, t4, true})
+ lim.SetLimitAt(t2, 10)
+ r.CancelAt(t2) // 2 tokens back
+ runReserve(t, lim, request{t2, 2, t3, true})
+}
+
+func TestReserveMax(t *testing.T) {
+ lim := NewLimiter(10, 2)
+ maxT := d
+
+ runReserveMax(t, lim, request{t0, 2, t0, true}, maxT)
+ runReserveMax(t, lim, request{t0, 1, t1, true}, maxT) // reserve for close future
+ runReserveMax(t, lim, request{t0, 1, t2, false}, maxT) // time to act too far in the future
+}
+
+type wait struct {
+ name string
+ ctx context.Context
+ n int
+ delay int // in multiples of d
+ nilErr bool
+}
+
+func runWait(t *testing.T, lim *Limiter, w wait) {
+ start := time.Now()
+ err := lim.WaitN(w.ctx, w.n)
+ delay := time.Now().Sub(start)
+ if (w.nilErr && err != nil) || (!w.nilErr && err == nil) || w.delay != dFromDuration(delay) {
+ errString := "<nil>"
+ if !w.nilErr {
+ errString = "<non-nil error>"
+ }
+ t.Errorf("lim.WaitN(%v, lim, %v) = %v with delay %v ; want %v with delay %v",
+ w.name, w.n, err, delay, errString, d*time.Duration(w.delay))
+ }
+}
+
+func TestWaitSimple(t *testing.T) {
+ lim := NewLimiter(10, 3)
+
+ ctx, cancel := context.WithCancel(context.Background())
+ cancel()
+ runWait(t, lim, wait{"already-cancelled", ctx, 1, 0, false})
+
+ runWait(t, lim, wait{"exceed-burst-error", context.Background(), 4, 0, false})
+
+ runWait(t, lim, wait{"act-now", context.Background(), 2, 0, true})
+ runWait(t, lim, wait{"act-later", context.Background(), 3, 2, true})
+}
+
+func TestWaitCancel(t *testing.T) {
+ lim := NewLimiter(10, 3)
+
+ ctx, cancel := context.WithCancel(context.Background())
+ runWait(t, lim, wait{"act-now", ctx, 2, 0, true}) // after this lim.tokens = 1
+ go func() {
+ time.Sleep(d)
+ cancel()
+ }()
+ runWait(t, lim, wait{"will-cancel", ctx, 3, 1, false})
+ // should get 3 tokens back, and have lim.tokens = 2
+ t.Logf("tokens:%v last:%v lastEvent:%v", lim.tokens, lim.last, lim.lastEvent)
+ runWait(t, lim, wait{"act-now-after-cancel", context.Background(), 2, 0, true})
+}
+
+func TestWaitTimeout(t *testing.T) {
+ lim := NewLimiter(10, 3)
+
+ ctx, cancel := context.WithTimeout(context.Background(), d)
+ defer cancel()
+ runWait(t, lim, wait{"act-now", ctx, 2, 0, true})
+ runWait(t, lim, wait{"w-timeout-err", ctx, 3, 0, false})
+}
+
+func TestWaitInf(t *testing.T) {
+ lim := NewLimiter(Inf, 0)
+
+ runWait(t, lim, wait{"exceed-burst-no-error", context.Background(), 3, 0, true})
+}
+
+func BenchmarkAllowN(b *testing.B) {
+ lim := NewLimiter(Every(1*time.Second), 1)
+ now := time.Now()
+ b.ReportAllocs()
+ b.ResetTimer()
+ b.RunParallel(func(pb *testing.PB) {
+ for pb.Next() {
+ lim.AllowN(now, 1)
+ }
+ })
+}
+
+func BenchmarkWaitNNoDelay(b *testing.B) {
+ lim := NewLimiter(Limit(b.N), b.N)
+ ctx := context.Background()
+ b.ReportAllocs()
+ b.ResetTimer()
+ for i := 0; i < b.N; i++ {
+ lim.WaitN(ctx, 1)
+ }
+}
--- /dev/null
+Subproject commit ced884f4700c5c9657569036e287a0653481574b
import (
"sync"
+ "time"
- "github.com/juju/ratelimit"
+ "golang.org/x/time/rate"
)
type RateLimiter interface {
Accept()
// Stop stops the rate limiter, subsequent calls to CanAccept will return false
Stop()
- // Saturation returns a percentage number which describes how saturated
- // this rate limiter is.
- // Usually we use token bucket rate limiter. In that case,
- // 1.0 means no tokens are available; 0.0 means we have a full bucket of tokens to use.
- Saturation() float64
+ // QPS returns QPS of this rate limiter
+ QPS() float32
}
type tokenBucketRateLimiter struct {
- limiter *ratelimit.Bucket
+ limiter *rate.Limiter
+ clock Clock
+ qps float32
}
// NewTokenBucketRateLimiter creates a rate limiter which implements a token bucket approach.
// The bucket is initially filled with 'burst' tokens, and refills at a rate of 'qps'.
// The maximum number of tokens in the bucket is capped at 'burst'.
func NewTokenBucketRateLimiter(qps float32, burst int) RateLimiter {
- limiter := ratelimit.NewBucketWithRate(float64(qps), int64(burst))
- return &tokenBucketRateLimiter{limiter}
+ limiter := rate.NewLimiter(rate.Limit(qps), burst)
+ return newTokenBucketRateLimiter(limiter, realClock{}, qps)
}
-func (t *tokenBucketRateLimiter) TryAccept() bool {
- return t.limiter.TakeAvailable(1) == 1
+// An injectable, mockable clock interface.
+type Clock interface {
+ Now() time.Time
+ Sleep(time.Duration)
}
-func (t *tokenBucketRateLimiter) Saturation() float64 {
- capacity := t.limiter.Capacity()
- avail := t.limiter.Available()
- return float64(capacity-avail) / float64(capacity)
+type realClock struct{}
+
+func (realClock) Now() time.Time {
+ return time.Now()
+}
+func (realClock) Sleep(d time.Duration) {
+ time.Sleep(d)
+}
+
+// NewTokenBucketRateLimiterWithClock is identical to NewTokenBucketRateLimiter
+// but allows an injectable clock, for testing.
+func NewTokenBucketRateLimiterWithClock(qps float32, burst int, c Clock) RateLimiter {
+ limiter := rate.NewLimiter(rate.Limit(qps), burst)
+ return newTokenBucketRateLimiter(limiter, c, qps)
+}
+
+func newTokenBucketRateLimiter(limiter *rate.Limiter, c Clock, qps float32) RateLimiter {
+ return &tokenBucketRateLimiter{
+ limiter: limiter,
+ clock: c,
+ qps: qps,
+ }
+}
+
+func (t *tokenBucketRateLimiter) TryAccept() bool {
+ return t.limiter.AllowN(t.clock.Now(), 1)
}
// Accept will block until a token becomes available
func (t *tokenBucketRateLimiter) Accept() {
- t.limiter.Wait(1)
+ now := t.clock.Now()
+ t.clock.Sleep(t.limiter.ReserveN(now, 1).DelayFrom(now))
}
func (t *tokenBucketRateLimiter) Stop() {
}
+func (t *tokenBucketRateLimiter) QPS() float32 {
+ return t.qps
+}
+
type fakeAlwaysRateLimiter struct{}
func NewFakeAlwaysRateLimiter() RateLimiter {
return true
}
-func (t *fakeAlwaysRateLimiter) Saturation() float64 {
- return 0
-}
-
func (t *fakeAlwaysRateLimiter) Stop() {}
func (t *fakeAlwaysRateLimiter) Accept() {}
+func (t *fakeAlwaysRateLimiter) QPS() float32 {
+ return 1
+}
+
type fakeNeverRateLimiter struct {
wg sync.WaitGroup
}
func NewFakeNeverRateLimiter() RateLimiter {
- wg := sync.WaitGroup{}
- wg.Add(1)
- return &fakeNeverRateLimiter{
- wg: wg,
- }
+ rl := fakeNeverRateLimiter{}
+ rl.wg.Add(1)
+ return &rl
}
func (t *fakeNeverRateLimiter) TryAccept() bool {
return false
}
-func (t *fakeNeverRateLimiter) Saturation() float64 {
- return 1
-}
-
func (t *fakeNeverRateLimiter) Stop() {
t.wg.Done()
}
func (t *fakeNeverRateLimiter) Accept() {
t.wg.Wait()
}
+
+func (t *fakeNeverRateLimiter) QPS() float32 {
+ return 1
+}
"revision": "baa956b6864800c46094b68723bbe316c65035be",
"revisionTime": "2016-07-12T10:00:20Z"
},
- {
- "checksumSHA1": "sKheT5xw89Tbu2Q071FQO27CVmE=",
- "origin": "k8s.io/kubernetes/vendor/github.com/juju/ratelimit",
- "path": "github.com/juju/ratelimit",
- "revision": "baa956b6864800c46094b68723bbe316c65035be",
- "revisionTime": "2016-07-12T10:00:20Z"
- },
{
"checksumSHA1": "Q2vw4HZBbnU8BLFt8VrzStwqSJg=",
"origin": "k8s.io/kubernetes/vendor/github.com/matttproud/golang_protobuf_extensions/pbutil",
Code Review / oom / registrator.git / commitdiff