Discussion on flow restriction

I/O Intensive and CPU intensive

about IO/Network Bottleneck class program , Its performance is network card / disk IO Will precede CPU Full , In this case, even if the optimization CPU The use of can not improve the throughput of the whole system , It can only improve the reading and writing speed of the disk , Increase memory size , Increase the bandwidth of the network card to improve the overall performance . and CPU Bottleneck class program , Before the storage and network card are full CPU Take the lead in arriving at 100%,CPU Busy with various computing tasks ,IO The equipment is relatively idle .

No matter what type of service , When the resource is used to its limit, it will lead to request accumulation , Overtime , System hang die , Eventually hurt the end user . For distributed Web For service , The bottleneck is not always inside the system , It may also be outside . Non computationally intensive systems often fall behind in relational databases , And then Web The module itself is far from reaching the bottleneck .

Common means of flow restriction

Leaky bucket

A leaky bucket means that we have a bucket that is always full of water , Every fixed period of time, a drop of water leaks out . If you get this drop of water , Then you can continue the service request , If not , Then you need to wait for the next drop of water

Token bucket

Token bucket refers to adding tokens to the bucket at a constant speed , When requesting a service, you need to get a token from the bucket , The number of tokens can be adjusted according to the resources consumed . If there is no token , You can choose to wait , Or give up

These two methods , The outflow rate of the leaky bucket is fixed , The token bucket can be accessed as long as there is a token in the bucket , Even if the bucket allows a certain degree of concurrency . When there is no token in the bucket, the token bucket will degenerate into a leaky bucket model .

Open source github.com/juju/ratelimit Several different token bucket filling methods are provided

Default

func NewBucket(fillInterval time.Duration, capacity int64) *Bucket

• fillInterval It refers to how long it takes to put a token into the bucket
• capacity Is the capacity of the barrel , The part exceeding the capacity of the barrel will be directly discarded
• The bucket is initially full

Batch release tokens

func NewBucketWithRate(rate float64, capacity int64) *Bucket

Every time this interface puts a token into the bucket , Yes quantum A token , Not a token . According to the proportion provided , Number of tokens filled per second . for example capacity yes 100, and rate yes 0.1, It will fill every second 100*0.1=10 A token

Get token in bucket

func (tb *Bucket) Take(count int64) time.Duration {
    }
func (tb *Bucket) TakeAvailable(count int64) int64 {
    }
func (tb *Bucket) TakeMaxDuration(count int64, maxWait time.Duration) (time.Duration, bool) {
    }
func (tb *Bucket) Wait(count int64) {
    }
func (tb *Bucket) WaitMaxDuration(count int64, maxWait time.Duration) bool {
    }

principle

functionally , Token bucket model is the addition and subtraction of global count , Using counting requires us to add a read-write lock . stay Go In language , Can pass buffer channel To complete the operation of adding and subtracting tokens ：

var tokenBucket = make(chan struct{
    }, capacity)

Every once in a while tokenBucket Add token, If bucket Is already full , Then just give up . Here is a production consumer model

package main

import (
    "fmt"
    "math/rand"
    "time"
)

var tokenBucket = make(chan struct{
    }, kCapcity)

const (
    kFillInterval = 1 * time.Millisecond
    kCapcity      = 100
)

func main() {
    

    fillToken := func() {
    
        ticker := time.NewTicker(kFillInterval)
        for {
    
            select {
    
            case <-ticker.C:
                select {
    
                case tokenBucket <- struct{
    }{
    }:
                default:
                }
                fmt.Println("current token cnt:", len(tokenBucket), time.Now())
            }
        }

    }

    getRandom := func() int {
    
        seed := rand.Intn(10)
        return seed
    }

    getToken := func() {
    
        ticker := time.NewTicker(15 * time.Millisecond)
        for {
    
            select {
    
            case <-ticker.C:
                num := getRandom()

                for i := 0; i < num; i++ {
    
                    <-tokenBucket
                }
            }
        }

    }

    go fillToken()
    go getToken()

    time.Sleep(time.Hour)
}

Lazy evaluation

Let's think about , The token bucket puts tokens into the bucket at regular intervals , If we write down the last time The time to put the token is t1, And the number of tokens at that time k1, The time interval for putting tokens is ti, Every time you put it in the token bucket discharge x A token , The token bucket capacity is cap. Now if someone calls getToken Come and get it n Order card , We remember this moment as t2. stay t2 moment , How many tokens should there theoretically be in the token bucket ？ The pseudocode is as follows ：

cur = k1 + ((t2 - t1)/ti) * x
cur = cur > cap ? cap : cur

We use the time difference between two time points , Combined with other parameters , In theory, you can... Before taking the token I know how many tokens are in the bucket . That's laborious, like the front of this section channel Infill token Gymnastics do , It's not necessary in theory . Just every time Take When , Then check the token in the token bucket token Numerical progression Line simple calculation , You can get the correct number of tokens . After getting the correct number of tokens , Then carry out the actual Take Just operate , This Take Operation only needs Simply subtract the number of tokens , Remember to lock to ensure concurrency security . github.com/juju/ratelimit That's what this library does .