cubefs

1
package datanode
2

3
import (
4
	"context"
5
	"fmt"
6

7
	"golang.org/x/time/rate"
8
)
9

10
var (
11
	deleteLimiteRater      = rate.NewLimiter(rate.Inf, defaultMarkDeleteLimitBurst)
12
	MaxExtentRepairLimit   = 20000
13
	MinExtentRepairLimit   = 5
14
	CurExtentRepairLimit   = MaxExtentRepairLimit
15
	extentRepairLimitRater chan struct{}
16
)
17

18
func initRepairLimit() {
19
	extentRepairLimitRater = make(chan struct{}, MaxExtentRepairLimit)
20
	for i := 0; i < MaxExtentRepairLimit; i++ {
21
		extentRepairLimitRater <- struct{}{}
22
	}
23
}
24

25
func requestDoExtentRepair() (err error) {
26
	err = fmt.Errorf("repair limit, cannot do extentRepair")
27

28
	select {
29
	case <-extentRepairLimitRater:
30
		return nil
31
	default:
32
		return
33
	}
34
}
35

36
func fininshDoExtentRepair() {
37
	select {
38
	case extentRepairLimitRater <- struct{}{}:
39
		return
40
	default:
41
		return
42
	}
43
}
44

45
func setDoExtentRepair(value int) {
46
	if value <= 0 {
47
		value = MaxExtentRepairLimit
48
	}
49

50
	if value > MaxExtentRepairLimit {
51
		value = MaxExtentRepairLimit
52
	}
53

54
	if value < MinExtentRepairLimit {
55
		value = MinExtentRepairLimit
56
	}
57

58
	if CurExtentRepairLimit != value {
59
		CurExtentRepairLimit = value
60
		close(extentRepairLimitRater)
61
		extentRepairLimitRater = make(chan struct{}, CurExtentRepairLimit)
62
		for i := 0; i < CurExtentRepairLimit; i++ {
63
			extentRepairLimitRater <- struct{}{}
64
		}
65
	}
66
}
67

68
func DeleteLimiterWait() {
69
	ctx := context.Background()
70
	deleteLimiteRater.Wait(ctx)
71
}
72

73
func setLimiter(limiter *rate.Limiter, limitValue uint64) {
74
	r := limitValue
75
	l := rate.Limit(r)
76
	if r == 0 {
77
		l = rate.Inf
78
	}
79
	limiter.SetLimit(l)
80
}
81