BBR自适应限流算法实现

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

// BBR implements bbr-like limiter.
// It is inspired by sentinel.
// https://github.com/alibaba//wiki/%E7%B3%BB%E7%BB%9F%E8%87%AA%E9%80%82%E5%BA%94%E9%99%90%E6%B5%81
type BBR struct {
	cpu             cpuGetter
	// passStat：请求处理成功的量（滑动窗口计数器）
	passStat        metric.RollingCounter
	// rtStat：请求成功的响应耗时（滑动窗口计数器）
	rtStat          metric.RollingCounter
	inFlight        int64
	winBucketPerSec int64
	bucketDuration  time.Duration
	winSize         int
	conf            *Config
	// 对应于公式的 prevDrop
	prevDrop        atomic.Value
	maxPASSCache    atomic.Value
	minRtCache      atomic.Value
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20

const (
	// Success opertion type: success
	Success Op = iota
	// Ignore opertion type: ignore
	Ignore
	// Drop opertion type: drop
	Drop
)

// DoneInfo done info.
type DoneInfo struct {
	Err error
	Op  Op
}

// Limiter limit interface.
type Limiter interface {
	//func(info DoneInfo) 调用参数为 Doneinfo，返回值无
	Allow(ctx context.Context, opts ...AllowOption) (func(info DoneInfo), error)
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

var (
	cpu         int64
	decay       = 0.95
	initTime    = time.Now()
	defaultConf = &Config{
		Window:       time.Second * 10,
		WinBucket:    100,
		CPUThreshold: 800,
	}
)

type cpuGetter func() int64

1
2
3
4
5

// Group represents a class of BBRLimiter and forms a namespace in which
// units of BBRLimiter.
type Group struct {
	group *group.Group
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

// NewGroup new a limiter group container, if conf nil use default conf.
func NewGroup(conf *Config)*Group {
	if conf == nil {
		// 使用默认 group 配置
		conf = defaultConf
	}
	// 创建 group
	group := group.NewGroup(func() interface{} {
		return newLimiter(conf)
	})
	return &Group{
		group: group,
	}
}

1
2
3
4
5

// Get get a limiter by a specified key, if limiter not exists then make a new one.
func (g *Group) Get(key string) limit.Limiter {
	limiter := g.group.Get(key)
	return limiter.(limit.Limiter)
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

func newLimiter(conf *Config) limit.Limiter {
	if conf == nil {
		conf = defaultConf
	}
	size := conf.WinBucket
	bucketDuration := conf.Window / time.Duration(conf.WinBucket)

	//
	passStat := metric.NewRollingCounter(metric.RollingCounterOpts{Size: size, BucketDuration: bucketDuration})
	rtStat := metric.NewRollingCounter(metric.RollingCounterOpts{Size: size, BucketDuration: bucketDuration})
	// 定义了 cpu 为 func，直接返回全局变量 cpu 的值（atomic 方式）
	cpu := func() int64 {
		return atomic.LoadInt64(&cpu)
	}
	limiter := &BBR{
		cpu:             cpu,
		conf:            conf,
		passStat:        passStat,
		rtStat:          rtStat,
		winBucketPerSec: int64(time.Second) / (int64(conf.Window) / int64(conf.WinBucket)),
	}
	return limiter
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

func init() {
	go cpuproc()
}

// cpu = cpuᵗ⁻¹ * decay + cpuᵗ * (1 - decay)
func cpuproc() {
	ticker := time.NewTicker(time.Millisecond * 250)
	defer func() {
		ticker.Stop()
		if err := recover(); err != nil {
			log.Error("rate.limit.cpuproc() err(%+v)", err)
			// 在 recover 中重启 CPU 监控
			go cpuproc()
		}
	}()

	// EMA algorithm: https://blog.csdn.net/m0_38106113/article/details/81542863
	for range ticker.C {
		stat := &cpustat.Stat{}
		cpustat.ReadStat(stat)
		// 获取前一个周期的 cpu 数据
		prevCpu := atomic.LoadInt64(&cpu)
		//decay = 0.95
		curCpu := int64(float64(prevCpu)*decay + float64(stat.Usage)*(1.0-decay))
		// 保存新的校准后的 CPU 数据
		atomic.StoreInt64(&cpu, curCpu)
	}
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126

package cpu

import (
	"bufio"
	"fmt"
	"io"
	"os"
	"path"
	"strconv"
	"strings"
)

const cgroupRootDir = "/sys/fs/cgroup"

// cgroup Linux cgroup
type cgroup struct {
	cgroupSet map[string]string
}

// CPUCFSQuotaUs cpu.cfs_quota_us
func (c *cgroup) CPUCFSQuotaUs() (int64, error) {
	data, err := readFile(path.Join(c.cgroupSet["cpu"], "cpu.cfs_quota_us"))
	if err != nil {
		return 0, err
	}
	return strconv.ParseInt(data, 10, 64)
}

// CPUCFSPeriodUs cpu.cfs_period_us
func (c *cgroup) CPUCFSPeriodUs() (uint64, error) {
	data, err := readFile(path.Join(c.cgroupSet["cpu"], "cpu.cfs_period_us"))
	if err != nil {
		return 0, err
	}
	return parseUint(data)
}

// CPUAcctUsage cpuacct.usage
func (c *cgroup) CPUAcctUsage() (uint64, error) {
	data, err := readFile(path.Join(c.cgroupSet["cpuacct"], "cpuacct.usage"))
	if err != nil {
		return 0, err
	}
	return parseUint(data)
}

// CPUAcctUsagePerCPU cpuacct.usage_percpu
func (c *cgroup) CPUAcctUsagePerCPU() ([]uint64, error) {
	data, err := readFile(path.Join(c.cgroupSet["cpuacct"], "cpuacct.usage_percpu"))
	if err != nil {
		return nil, err
	}
	var usage []uint64
	for _, v := range strings.Fields(string(data)) {
		var u uint64
		if u, err = parseUint(v); err != nil {
			return nil, err
		}
		// fix possible_cpu:https://www.ibm.com/support/knowledgecenter/en/linuxonibm/com.ibm.linux.z.lgdd/lgdd_r_posscpusparm.html
		if u != 0 {
			usage = append(usage, u)
		}
	}
	return usage, nil
}

// CPUSetCPUs cpuset.cpus
func (c *cgroup) CPUSetCPUs() ([]uint64, error) {
	data, err := readFile(path.Join(c.cgroupSet["cpuset"], "cpuset.cpus"))
	if err != nil {
		return nil, err
	}
	cpus, err := ParseUintList(data)
	if err != nil {
		return nil, err
	}
	var sets []uint64
	for k := range cpus {
		sets = append(sets, uint64(k))
	}
	return sets, nil
}

// CurrentcGroup get current process cgroup
func currentcGroup() (*cgroup, error) {
	pid := os.Getpid()
	cgroupFile := fmt.Sprintf("/proc/%d/cgroup", pid)
	cgroupSet := make(map[string]string)
	fp, err := os.Open(cgroupFile)
	if err != nil {
		return nil, err
	}
	defer fp.Close()
	buf := bufio.NewReader(fp)
	for {
		line, err := buf.ReadString('\n')
		if err != nil {
			if err == io.EOF {
				break
			}
			return nil, err
		}
		col := strings.Split(strings.TrimSpace(line), ":")
		if len(col) != 3 {
			return nil, fmt.Errorf("invalid cgroup format %s", line)
		}
		dir := col[2]
		// When dir is not equal to /, it must be in docker
		if dir != "/" {
			cgroupSet[col[1]] = path.Join(cgroupRootDir, col[1])
			if strings.Contains(col[1], ",") {
				for _, k := range strings.Split(col[1], ",") {
					cgroupSet[k] = path.Join(cgroupRootDir, k)
				}
			}
		} else {
			cgroupSet[col[1]] = path.Join(cgroupRootDir, col[1], col[2])
			if strings.Contains(col[1], ",") {
				for _, k := range strings.Split(col[1], ",") {
					cgroupSet[k] = path.Join(cgroupRootDir, k, col[2])
				}
			}
		}
	}
	return &cgroup{cgroupSet: cgroupSet}, nil
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

// Limit is a server interceptor that detects and rejects overloaded traffic.
func (b *RateLimiter) Limit() grpc.UnaryServerInterceptor {
return func(ctx context.Context, req interface{}, args*grpc.UnaryServerInfo, handler grpc.UnaryHandler) (resp interface{}, err error) {
		uri := args.FullMethod
		limiter := b.group.Get(uri)
		done, err := limiter.Allow(ctx)
		if err != nil {
			_metricServerBBR.Inc(uri)
			return
		}
		defer func() {
			done(limit.DoneInfo{Op: limit.Success})
			b.printStats(uri, limiter)
		}()
		resp, err = handler(ctx, req)
		return
	}
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

func(do limit.DoneInfo) {
	// 获取执行过程的耗时
	rt := int64((time.Since(initTime) - stime) / time.Millisecond)
	// 向 RollingCounter 中添加 rt（这是把 RollingCounter 当做累加器使用）
	l.rtStat.Add(rt)
	atomic.AddInt64(&l.inFlight, -1)
	switch do.Op {
	case limit.Success:
		l.passStat.Add(1)
		return
	default:
		return
	}
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

// Allow checks all inbound traffic.
// Once overload is detected, it raises ecode.LimitExceed error.
func (l *BBR) Allow(ctx context.Context, opts ...limit.AllowOption) (func(info limit.DoneInfo), error) {
	allowOpts := limit.DefaultAllowOpts()
	for _, opt := range opts {
		opt.Apply(&allowOpts)
	}
	// shouldDrop 判断是否需要限流，如果true表示限流 之后重点讲
	if l.shouldDrop() {
		return nil, ecode.LimitExceed
	}
	atomic.AddInt64(&l.inFlight, 1) // 之前说的，正在处理数+1
	stime := time.Since(initTime) // 现在时间减去程序初始化时间 表示程序开始执行时刻
	return func(do limit.DoneInfo) { // allow返回函数 在中间件（拦截器）中handle执行完成后调用
		rt := int64((time.Since(initTime) - stime) / time.Millisecond) // 执行完handle的时间减去stime 表示 程序执行的总时间 单位ms
		l.rtStat.Add(rt)	// 把处理时间放进采样数据window
		atomic.AddInt64(&l.inFlight, -1) // 正在处理数-1 便是处理完成
		switch do.Op {
		case limit.Success:
			l.passStat.Add(1)	// 成功了，把通过数的采样数据window加1
			return
		default:
			return
		}
	}, nil
}
// Stats contains the metrics’s snapshot of bbr. type Stat struct { Cpu int64 InFlight int64 MaxInFlight int64 MinRt int64 MaxPass int64 }

// Config contains configs of bbr limiter. type Config struct { Enabled bool Window time.Duration WinBucket int Rule string Debug bool CPUThreshold int64 }

1
2
3
4
5

type Bucket struct {
	Points []float64 // 单个节点中的统计数据（数组）
	Count  int64     // 总数
	next   *Bucket   // 链表实现
}

1
2
3
4
5

// Append appends the given value to the bucket.
func (b *Bucket) Append(val float64) {
	b.Points = append(b.Points, val)
	b.Count++
}

1
2
3
4
5
6

// Add adds the given value to the point.
// 给 []float64 指定位置累加值
func (b*Bucket) Add(offset int, val float64) {
	b.Points[offset] += val
	b.Count++
}

1
2
3
4
5

// Next returns the next bucket.
// 遍历链表用（返回当前 node 的下一个 node）
func (b *Bucket) Next()*Bucket {
	return b.next
}

1
2
3
4

type Window struct {
	window []Bucket // 滑动窗口实现
	size   int
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

// NewWindow creates a new Window based on WindowOpts.
// 初始化滑动窗口
func NewWindow(opts WindowOpts) *Window {
	buckets := make([]Bucket, opts.Size)
	for offset := range buckets {
		// 初始化每个 bucket
		buckets[offset] = Bucket{Points: make([]float64, 0)}
		nextOffset := offset + 1
		if nextOffset == opts.Size {
			// 构建一个 queue（环）
			nextOffset = 0
		}

		// 初始化为一个链表（首尾相接）
		buckets[offset].next = &buckets[nextOffset]
	}
	return &Window{window: buckets, size: opts.Size}
}

1
2
3
4
5

// Append appends the given value to the bucket where index equals the given offset.
func (w *Window) Append(offset int, val float64) {
	// 调用了 Bucket 的 Append 方法
	w.window[offset].Append(val)
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

// Add adds the given value to the latest point within bucket where index equals the given offset.
func (w*Window) Add(offset int, val float64) {
	if w.window[offset].Count == 0 {
		// 如果 bucket 是空的（没有统计值），直接 Append
		w.window[offset].Append(val)
		return
	}
	// bucket 非空，在 Point[] 的 0 号位置累加值
	w.window[offset].Add(0, val)
}

1
2
3
4
5
6
7
8

// 需要提供传入的 offset 和 count
func (w *Window) Iterator(offset int, count int) Iterator {
	// 构建 window 的 Iterator（迭代器），方便统计和遍历
	return Iterator{
		count: count,
		cur:   &w.window[offset],
	}
}

1
2
3
4
5
6

// Iterator iterates the buckets within the window.
type Iterator struct {
	count         int // 遍历完成的条件（i.count != i.iteratedCount）
	iteratedCount int
	cur           *Bucket // 当前迭代器的位置
}

1
2
3
4
5

// 这里很重要，Iter 的迭代规则是，移动 count 的次数 == 当前移动次数
// Next returns true util all of the buckets has been iterated.
func (i *Iterator) Next() bool {
	return i.count != i.iteratedCount
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

// Bucket gets current bucket.
func (i *Iterator) Bucket() Bucket {
	if !(i.Next()) {
		panic(fmt.Errorf("stat/metric: iteration out of range iteratedCount: %d count: %d", i.iteratedCount, i.count))
	}
	bucket :=*i.cur
	i.iteratedCount++	// 累加计数器
	i.cur = i.cur.Next()
	return bucket
}

1
2
3
4
5
6
7
8
9

type RollingPolicy struct {
	mu     sync.RWMutex		//（子协程）并发的 add 操作必须加锁！
	size   int			// 滑动窗口的 size
	window *Window			// 滑动窗口
	offset int

	bucketDuration time.Duration		// 一个桶代表多长时间
	lastAppendTime time.Time		// 滑动窗口的 START 位置
}

1
2
3
4
5
6
7
8

func (r *RollingPolicy) timespan() int {
	v := int(time.Since(r.lastAppendTime) / r.bucketDuration)
	if v > -1 { // maybe time backwards
		return v
	}
	// 时间调整了
	return r.size
}

1
2
3
4
5
6
7
8
9

// Append appends the given points to the window.
func (r *RollingPolicy) Append(val float64) {
	r.add(r.window.Append, val)
}

// Add adds the given value to the latest point within bucket.
func (r *RollingPolicy) Add(val float64) {
	r.add(r.window.Add, val)
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

func (r *RollingPolicy) add(f func(offset int, val float64), val float64) {
	r.mu.Lock()
	// 计算时间跨度（跨过了几个 bucket）
	timespan := r.timespan()
	if timespan > 0 {
		// 当 timespan>0 时，表示有跨度
		// 更新当前 append 时间
		r.lastAppendTime = r.lastAppendTime.Add(time.Duration(timespan* int(r.bucketDuration)))
		offset := r.offset
		// reset the expired buckets
		s := offset + 1 //s 指向下一个位置
		if timespan > r.size {
			// 如果跨度超过了 window 的大小，timespan 最大为 window 的 size
			timespan = r.size
		}
		e, e1 := s+timespan, 0 // e: reset offset must start from offset+1
		if e > r.size {
			e1 = e - r.size
			e = r.size
		}
		for i := s; i < e; i++ {
			// 清理 offset1---> s+timespan 的之间的 bucket
			r.window.ResetBucket(i)
			offset = i
		}
		for i := 0; i < e1; i++ {
			// 如果超过一个跨度，那么说明时间跨度在两个区间上
			r.window.ResetBucket(i)
			offset = i
		}
		r.offset = offset
	}
	// 添加到 offset 位置
	//（当 timespan==0 时，说明，当前时间未出现 span，直接操作 r.offset 位置即可）
	f(r.offset, val)
	r.mu.Unlock()
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

// Reduce applies the reduction function to all buckets within the window.
func (r *RollingPolicy) Reduce(f func(Iterator) float64) (val float64) {
	r.mu.RLock()
	timespan := r.timespan()
	if count := r.size - timespan; count > 0 {
		offset := r.offset + timespan + 1
		if offset >= r.size {
			offset = offset - r.size
		}
		// 计算得到遍历的开始位置 offset 和遍历长度 count
		//f 的参数，就是 Iterator 的结果
		val = f(r.window.Iterator(offset, count))
	}
	r.mu.RUnlock()
	return val
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

type rollingCounter struct {
	policy *RollingPolicy
}


type RollingCounter interface {
	Metric			// with prometheus
	Aggregation
	Timespan() int
	// Reduce applies the reduction function to all buckets within the window.
	Reduce(func(Iterator) float64) float64
}

1
2
3
4
5
6
7

func (r *rollingCounter) Add(val int64) {
    // 计数器不能为负数
	if val < 0 {
		panic(fmt.Errorf("stat/metric: cannot decrease in value. val: %d", val))
	}
	r.policy.Add(float64(val))
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

type rollingGauge struct {
	policy *RollingPolicy
}

// RollingGauge represents a ring window based on time duration.
// e.g. [[1, 2], [1, 2, 3], [1,2, 3, 4]]
type RollingGauge interface {
	Metric
	Aggregation
	// Reduce applies the reduction function to all buckets within the window.
	Reduce(func(Iterator) float64) float64
}

1
2
3
4
5
6
7

func (r *rollingGauge) Add(val int64) {
	r.policy.Append(float64(val))
}
// Append appends the given points to the window.
func (r *RollingPolicy) Append(val float64) {
	r.add(r.window.Append, val)
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

// 单个采样窗口的最大的请求数, 默认的采样窗口是10s, 采样bucket数量100
func (l *BBR) maxPASS() int64 {
	rawMaxPass := atomic.LoadInt64(&l.rawMaxPASS)
	if rawMaxPass > 0 && l.passStat.Timespan() < 1 {
		// 当前的时间跨度未超过一个单位
		return rawMaxPass
	}

	// 通过 Reduce 方法来获取最大的 Pass 值
	// 遍历100个采样bucket, 找到采样bucket中最大的请求数
	rawMaxPass = int64(l.passStat.Reduce(func(iterator metric.Iterator) float64 {
		var result = 1.0
		for i := 1; iterator.Next() && i < l.conf.WinBucket; i++ {
			bucket := iterator.Bucket()
			count := 0.0
			for _, p := range bucket.Points {
				// 叠加 bucket.Points，注意 Points 的类型是 Points []float64
				count += p
			}
			result = math.Max(result, count)
		}
		return result
	}))
	if rawMaxPass == 0 {
		rawMaxPass = 1
	}
	// 存储在 rawMaxPASS 中并返回
	atomic.StoreInt64(&l.rawMaxPASS, rawMaxPass)
	return rawMaxPass
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

// 单个采样窗口中最小的响应时间
func (l *BBR) minRT() int64 {
	rawMinRT := atomic.LoadInt64(&l.rawMinRt)
	if rawMinRT > 0 && l.rtStat.Timespan() < 1 {
		return rawMinRT
	}
	// 遍历100个采样bucket, 找到采样bucket中最小的响应时间
	rawMinRT = int64(math.Ceil(l.rtStat.Reduce(func(iterator metric.Iterator) float64 {
		var result = math.MaxFloat64
		for i := 1; iterator.Next() && i < l.conf.WinBucket; i++ {
			bucket := iterator.Bucket()
			if len(bucket.Points) == 0 {
				continue
			}
			total := 0.0
			for _, p := range bucket.Points {
				total += p
			}
			avg := total / float64(bucket.Count)
			result = math.Min(result, avg)
		}
		return result
	})))
	if rawMinRT <= 0 {
		rawMinRT = 1
	}
	atomic.StoreInt64(&l.rawMinRt, rawMinRT)
	return rawMinRT
}

1

cpu > 800 AND (Now - PrevDrop) < 1s AND maxFlight < InFlight

1
2
3

func (l *BBR) maxFlight() int64 {
	return int64(math.Floor(float64(l.maxPASS()*l.minRT()*l.winBucketPerSec)/1000.0 + 0.5))
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

func (l *BBR) shouldDrop() bool {
	// 判断目前cpu的使用率是否达到设置的CPU的限制, 默认值800
	if l.cpu() < l.conf.CPUThreshold {
		// 如果上一次舍弃请求的时间是0, 那么说明没有限流的需求, 直接返回
		prevDrop, _ := l.prevDrop.Load().(time.Duration)
		if prevDrop == 0 {
			return false
		}
		// 如果上一次请求的时间与当前的请求时间小于1s, 那么说明有限流的需求
		if time.Since(initTime)-prevDrop <= time.Second {
			if atomic.LoadInt32(&l.prevDropHit) == 0 {
				atomic.StoreInt32(&l.prevDropHit, 1)
			}
			// 增加正在处理的请求的数量
			inFlight := atomic.LoadInt64(&l.inFlight)
			// 判断正在处理的请求数是否达到系统的最大的请求数量
			return inFlight > 1 && inFlight > l.maxFlight()
		}
		// 清空当前的prevDrop
		l.prevDrop.Store(time.Duration(0))
		return false
	}
	// 增加正在处理的请求的数量
	inFlight := atomic.LoadInt64(&l.inFlight)
	// 判断正在处理的请求数是否达到系统的最大的请求数量
	drop := inFlight > 1 && inFlight > l.maxFlight()
	if drop {
		prevDrop, _ := l.prevDrop.Load().(time.Duration)
		// 如果判断达到了最大请求数量, 并且当前有限流需求
		if prevDrop != 0 {
			return drop
		}
		l.prevDrop.Store(time.Since(initTime))
	}
	return drop
}