scheduler

Documentation

Index

• type BasicTokenBucket
  • func NewBasicTokenBucket(now time.Time, fillRate float64, metrics *TokenBucketMetrics) *BasicTokenBucket
  • func (btb *BasicTokenBucket) GetFillRate() float64
  • func (btb *BasicTokenBucket) PreprocessRequest(now time.Time, rContext RequestContext) bool
  • func (btb *BasicTokenBucket) SetFillRate(now time.Time, fillRate float64)
  • func (btb *BasicTokenBucket) Take(now time.Time, timeout time.Duration, tokens float64) (time.Duration, bool)
  • func (btb *BasicTokenBucket) TakeIfAvailable(now time.Time, tokens float64) bool
• type RequestContext
• type Scheduler
  • func NewWFQScheduler(timeout time.Duration, tokenManger TokenManager, clk clockwork.Clock, ...) Scheduler
• type TokenBucketLoadShed
  • func NewTokenBucketLoadShed(now time.Time, metrics *TokenBucketLoadShedMetrics) *TokenBucketLoadShed
  • func (tbls *TokenBucketLoadShed) LoadShedFactor() float64
  • func (tbls *TokenBucketLoadShed) PreprocessRequest(now time.Time, rContext RequestContext) bool
  • func (tbls *TokenBucketLoadShed) SetLoadShedFactor(now time.Time, lsf float64)
  • func (tbls *TokenBucketLoadShed) Take(now time.Time, timeout time.Duration, tokens float64) (time.Duration, bool)
  • func (tbls *TokenBucketLoadShed) TakeIfAvailable(now time.Time, tokens float64) bool
• type TokenBucketLoadShedMetrics
• type TokenBucketMetrics
• type TokenManager
• type WFQMetrics
• type WFQScheduler
  • func (sched *WFQScheduler) GetPendingFlows() int
  • func (sched *WFQScheduler) GetPendingRequests() int
  • func (sched *WFQScheduler) Schedule(rContext RequestContext) bool
• type WindowedCounter
  • func NewWindowedCounter(now time.Time, totalSlots uint8, slotDuration time.Duration) *WindowedCounter
  • func (counter *WindowedCounter) AddTokens(now time.Time, tokens uint64) bool
  • func (counter *WindowedCounter) CalculateTokenRate() float64
  • func (counter *WindowedCounter) IsBootstrapping() bool

Types

type BasicTokenBucket

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

BasicTokenBucket is a basic token bucket implementation.

func NewBasicTokenBucket

func NewBasicTokenBucket(now time.Time, fillRate float64, metrics *TokenBucketMetrics) *BasicTokenBucket

NewBasicTokenBucket creates a new BasicTokenBucket with adjusted fill rate.

func (*BasicTokenBucket) GetFillRate

func (btb *BasicTokenBucket) GetFillRate() float64

GetFillRate returns the fill rate of the BasicTokenBucket.

func (*BasicTokenBucket) PreprocessRequest

func (btb *BasicTokenBucket) PreprocessRequest(now time.Time, rContext RequestContext) bool

PreprocessRequest is a no-op for BasicTokenBucket and by default, it rejects the request.

func (*BasicTokenBucket) SetFillRate

func (btb *BasicTokenBucket) SetFillRate(now time.Time, fillRate float64)

SetFillRate adjusts the fill rate of the BasicTokenBucket.

func (*BasicTokenBucket) Take

func (btb *BasicTokenBucket) Take(now time.Time, timeout time.Duration, tokens float64) (time.Duration, bool)

Take takes tokens from the basic token bucket even if available tokens are less than asked. If tokens are not available at the moment, it will return amount of wait time and checks whether the operation was successful or not.

func (*BasicTokenBucket) TakeIfAvailable

func (btb *BasicTokenBucket) TakeIfAvailable(now time.Time, tokens float64) bool

TakeIfAvailable takes tokens from the basic token bucket if available, otherwise return false.

type RequestContext

type RequestContext struct {
	FairnessLabel string // for enforcing fairness
	Tokens        uint64 // expected latency for this request
	Priority      uint8  // larger values represent higher priority
	Timeout       time.Duration
}

RequestContext is metadata for request in a flow that is to be allowed or dropped based on controlled delay and queue limits.

type Scheduler

type Scheduler interface {
	// Schedule sends RequestContext to the underlying scheduler and returns a boolean value,
	// where true means accept and false means reject.
	Schedule(rContext RequestContext) bool
}

Scheduler : Interface for schedulers.

func NewWFQScheduler

func NewWFQScheduler(timeout time.Duration, tokenManger TokenManager, clk clockwork.Clock, metrics *WFQMetrics) Scheduler

NewWFQScheduler creates a new weighted fair queue scheduler, timeout -- timeout for requests.

type TokenBucketLoadShed

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

TokenBucketLoadShed is a token bucket with load shedding.

func NewTokenBucketLoadShed

func NewTokenBucketLoadShed(now time.Time, metrics *TokenBucketLoadShedMetrics) *TokenBucketLoadShed

NewTokenBucketLoadShed creates a new TokenBucketLoadShed.

func (*TokenBucketLoadShed) LoadShedFactor

func (tbls *TokenBucketLoadShed) LoadShedFactor() float64

LoadShedFactor returns the current load shed factor.

func (*TokenBucketLoadShed) PreprocessRequest

func (tbls *TokenBucketLoadShed) PreprocessRequest(now time.Time, rContext RequestContext) bool

PreprocessRequest preprocesses a request and makes decision whether to accept or reject the request.

func (*TokenBucketLoadShed) SetLoadShedFactor

func (tbls *TokenBucketLoadShed) SetLoadShedFactor(now time.Time, lsf float64)

SetLoadShedFactor sets the load shed factor number between [0,1] --> 0 = no load shedding, 1 = load shed 100%.

func (*TokenBucketLoadShed) Take

func (tbls *TokenBucketLoadShed) Take(now time.Time, timeout time.Duration, tokens float64) (time.Duration, bool)

Take takes tokens from the token bucket even if available tokens are less than asked. If tokens are not available at the moment, it will return amount of wait time and checks whether the operation was successful or not.

func (*TokenBucketLoadShed) TakeIfAvailable

func (tbls *TokenBucketLoadShed) TakeIfAvailable(now time.Time, tokens float64) bool

TakeIfAvailable takes tokens from the token bucket if available, otherwise return false.

type TokenBucketLoadShedMetrics

type TokenBucketLoadShedMetrics struct {
	LSFGauge           prometheus.Gauge
	TokenBucketMetrics *TokenBucketMetrics
}

TokenBucketLoadShedMetrics holds metrics related to internals of TokenBucketLoadShed.

type TokenBucketMetrics

type TokenBucketMetrics struct {
	FillRateGauge        prometheus.Gauge
	BucketCapacityGauge  prometheus.Gauge
	AvailableTokensGauge prometheus.Gauge
}

TokenBucketMetrics holds metrics related to internals of TokenBucket.

type TokenManager

type TokenManager interface {
	// Take tokens if available, otherwise return false
	TakeIfAvailable(now time.Time, tokens float64) bool
	// Take tokens even if available tokens are less than asked - returns wait time if tokens are not available immediately. The other return value conveys whether the operation was successful or not.
	Take(now time.Time, timeout time.Duration, tokens float64) (time.Duration, bool)
	// Provides TokenManager the request that the scheduler processing -- some TokenManager implementations use this level of visibility for their algorithms. Return value decides whether the request has to be accepted right away in case TokenManger is not yet ready or configured to accept all traffic (short circuit).
	PreprocessRequest(now time.Time, rContext RequestContext) bool
}

TokenManager : Interface for token managers.

type WFQMetrics

type WFQMetrics struct {
	FlowsGauge        prometheus.Gauge
	HeapRequestsGauge prometheus.Gauge
}

WFQMetrics holds metrics related to internal workings of WFQScheduler.

type WFQScheduler

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

WFQScheduler : WFQ + CoDel.

func (*WFQScheduler) GetPendingFlows

func (sched *WFQScheduler) GetPendingFlows() int

GetPendingFlows returns the number of flows in the scheduler.

func (*WFQScheduler) GetPendingRequests

func (sched *WFQScheduler) GetPendingRequests() int

GetPendingRequests returns the number of requests in the scheduler.

func (*WFQScheduler) Schedule

func (sched *WFQScheduler) Schedule(rContext RequestContext) bool

Schedule blocks until the request is scheduled or until timeout (CoDel). Return value - true: Accept, false: Reject.

type WindowedCounter

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

WindowedCounter is a token bucket with a windowed counter.

func NewWindowedCounter

func NewWindowedCounter(now time.Time, totalSlots uint8, slotDuration time.Duration) *WindowedCounter

NewWindowedCounter creates a new WindowedCounter with extra slot for the current window.

func (*WindowedCounter) AddTokens

func (counter *WindowedCounter) AddTokens(now time.Time, tokens uint64) bool

AddTokens to the counter. Return value is true when counter shifted slots and the all the slots in the counter is valid.

func (*WindowedCounter) CalculateTokenRate

func (counter *WindowedCounter) CalculateTokenRate() float64

CalculateTokenRate returns the calculated token rate in the current window.

func (*WindowedCounter) IsBootstrapping

func (counter *WindowedCounter) IsBootstrapping() bool

IsBootstrapping checks whether the counter is in bootstrapping mode.