storelimit

Documentation

Index

• Constants
• Variables
• type SlidingWindows
  • func NewSlidingWindows() *SlidingWindows
  • func (s *SlidingWindows) Ack(token int64, typ Type)
  • func (s *SlidingWindows) Available(_ int64, typ Type, level constant.PriorityLevel) bool
  • func (s *SlidingWindows) Feedback(e float64)
  • func (s *SlidingWindows) GetCap() int64
  • func (s *SlidingWindows) GetUsed() []int64
  • func (*SlidingWindows) Reset(_ float64, _ Type)
  • func (s *SlidingWindows) Take(token int64, typ Type, level constant.PriorityLevel) bool
  • func (*SlidingWindows) Version() string
• type StoreLimit
  • func NewStoreRateLimit(ratePerSec float64) StoreLimit
• type StoreRateLimit
  • func (*StoreRateLimit) Ack(_ int64, _ Type)
  • func (l *StoreRateLimit) Available(cost int64, typ Type, _ constant.PriorityLevel) bool
  • func (*StoreRateLimit) Feedback(_ float64)
  • func (l *StoreRateLimit) Rate(typ Type) float64
  • func (l *StoreRateLimit) Reset(rate float64, typ Type)
  • func (l *StoreRateLimit) Take(cost int64, typ Type, _ constant.PriorityLevel) bool
  • func (*StoreRateLimit) Version() string
• type Type
  • func (t Type) String() string

Constants

const (
	// VersionV1 represents the rate limit version of the store limit
	VersionV1 = "v1"
	// VersionV2 represents the sliding window version of the store limit
	VersionV2 = "v2"
)

const (
	// SmallRegionThreshold is used to represent a region which can be regarded as a small region once the size is small than it.
	SmallRegionThreshold int64 = 20
	// Unlimited is used to control the store limit. Here uses a big enough number to represent unlimited.
	Unlimited = float64(100000000)
)

Variables

var RegionInfluence = []int64{
	AddPeer:      influence,
	RemovePeer:   influence,
	SendSnapshot: influence,
}

RegionInfluence represents the influence of an operator step, which is used by store limit.

var SmallRegionInfluence = []int64{
	AddPeer:    smallInfluence,
	RemovePeer: smallInfluence,
}

SmallRegionInfluence represents the influence of an operator step when the region size is smaller than smallRegionThreshold, which is used by store limit.

var TypeNameValue = map[string]Type{
	"add-peer":      AddPeer,
	"remove-peer":   RemovePeer,
	"send-snapshot": SendSnapshot,
}

TypeNameValue indicates the name of store limit type and the enum value

Types

type SlidingWindows

type SlidingWindows struct {
	// contains filtered or unexported fields
}

SlidingWindows is a multi sliding windows

func NewSlidingWindows

func NewSlidingWindows() *SlidingWindows

NewSlidingWindows is the construct of SlidingWindows.

func (*SlidingWindows) Ack

func (s *SlidingWindows) Ack(token int64, typ Type)

Ack indicates that some executing operator has been finished. The order of refilling windows is from high to low. It will refill the highest window first.

func (*SlidingWindows) Available

func (s *SlidingWindows) Available(_ int64, typ Type, level constant.PriorityLevel) bool

Available returns whether the token can be taken. The order of checking windows is from low to high. It checks the given window finally if the lower window has no free size.

func (*SlidingWindows) Feedback

func (s *SlidingWindows) Feedback(e float64)

Feedback is used to update the capacity of the sliding windows.

func (*SlidingWindows) GetCap

func (s *SlidingWindows) GetCap() int64

GetCap returns the capacity of the sliding windows.

func (*SlidingWindows) GetUsed

func (s *SlidingWindows) GetUsed() []int64

GetUsed returns the used size in the sliding windows.

func (*SlidingWindows) Reset

func (*SlidingWindows) Reset(_ float64, _ Type)

Reset does nothing because the capacity depends on the feedback.

func (*SlidingWindows) Take

func (s *SlidingWindows) Take(token int64, typ Type, level constant.PriorityLevel) bool

Take tries to take the token. It will consume the given window finally if the lower window has no free size.

func (*SlidingWindows) Version

func (*SlidingWindows) Version() string

Version returns v2

type StoreLimit

type StoreLimit interface {
	// Available returns true if the store can accept the operator
	Available(cost int64, typ Type, level constant.PriorityLevel) bool
	// Take takes the cost of the operator, it returns false if the store can't accept any operators.
	Take(count int64, typ Type, level constant.PriorityLevel) bool
	// Reset resets the store limit
	Reset(rate float64, typ Type)
	// Feedback update limit capacity by auto-tuning.
	Feedback(e float64)
	// Ack put back the cost into the limit for the next waiting operator after the operator is finished.
	// only snapshot type can use this method.
	Ack(cost int64, typ Type)
	// Version returns the version of the store limit
	Version() string
}

StoreLimit is an interface to control the operator rate of store TODO: add a method to control the rate of store the normal control flow is: 1. check the store limit with Available in checker or scheduler. 2. check the store limit with Available in operator controller again. the different between 1 and 2 is that 1 maybe not use the operator level. 3. take the cost of operator with Take in operator controller. 4. ack will put back the cost into the limit for the next waiting operator after the operator is finished. the cost is the operator influence, so the influence should be same in the life of the operator.

func NewStoreRateLimit

func NewStoreRateLimit(ratePerSec float64) StoreLimit

NewStoreRateLimit creates a StoreRateLimit.

type StoreRateLimit

type StoreRateLimit struct {
	// contains filtered or unexported fields
}

StoreRateLimit is a rate limiter for store.

func (*StoreRateLimit) Ack

func (*StoreRateLimit) Ack(_ int64, _ Type)

Ack does nothing.

func (*StoreRateLimit) Available

func (l *StoreRateLimit) Available(cost int64, typ Type, _ constant.PriorityLevel) bool

Available returns the number of available tokens. notice that the priority level is not used.

func (*StoreRateLimit) Feedback

func (*StoreRateLimit) Feedback(_ float64)

Feedback does nothing.

func (*StoreRateLimit) Rate

func (l *StoreRateLimit) Rate(typ Type) float64

Rate returns the capacity of the store limit.

func (*StoreRateLimit) Reset

func (l *StoreRateLimit) Reset(rate float64, typ Type)

Reset resets the rate limit.

func (*StoreRateLimit) Take

func (l *StoreRateLimit) Take(cost int64, typ Type, _ constant.PriorityLevel) bool

Take takes count tokens from the bucket without blocking. notice that the priority level is not used.

func (*StoreRateLimit) Version

func (*StoreRateLimit) Version() string

Version returns v1

type Type

type Type int

Type indicates the type of store limit

const (
	// AddPeer indicates the type of store limit that limits the adding peer rate
	AddPeer Type = iota
	// RemovePeer indicates the type of store limit that limits the removing peer rate
	RemovePeer
	// SendSnapshot indicates the type of sending snapshot.
	SendSnapshot
)

func (Type) String

func (t Type) String() string

String returns the representation of the Type