package aqm import ( "context" "math" "sync" "time" "github.com/go-kratos/kratos/pkg/ecode" ) // Config codel config. type Config struct { Target int64 // target queue delay (default 20 ms). Internal int64 // sliding minimum time window width (default 500 ms) MaxOutstanding int64 //max num of concurrent acquires } // Stat is the Statistics of codel. type Stat struct { Dropping bool FaTime int64 DropNext int64 Packets int } type packet struct { ch chan bool ts int64 } var defaultConf = &Config{ Target: 50, Internal: 500, MaxOutstanding: 40, } // Queue queue is CoDel req buffer queue. type Queue struct { pool sync.Pool packets chan packet mux sync.RWMutex conf *Config count int64 dropping bool // Equal to 1 if in drop state faTime int64 // Time when we'll declare we're above target (0 if below) dropNext int64 // Packets dropped since going into drop state outstanding int64 } // Default new a default codel queue. func Default() *Queue { return New(defaultConf) } // New new codel queue. func New(conf *Config) *Queue { if conf == nil { conf = defaultConf } q := &Queue{ packets: make(chan packet, 2048), conf: conf, } q.pool.New = func() interface{} { return make(chan bool) } return q } // Reload set queue config. func (q *Queue) Reload(c *Config) { if c == nil || c.Internal <= 0 || c.Target <= 0 || c.MaxOutstanding <= 0 { return } // TODO codel queue size q.mux.Lock() q.conf = c q.mux.Unlock() } // Stat return the statistics of codel func (q *Queue) Stat() Stat { q.mux.Lock() defer q.mux.Unlock() return Stat{ Dropping: q.dropping, FaTime: q.faTime, DropNext: q.dropNext, Packets: len(q.packets), } } // Push req into CoDel request buffer queue. // if return error is nil,the caller must call q.Done() after finish request handling func (q *Queue) Push(ctx context.Context) (err error) { q.mux.Lock() if q.outstanding < q.conf.MaxOutstanding && len(q.packets) == 0 { q.outstanding ++ q.mux.Unlock() return } q.mux.Unlock() r := packet{ ch: q.pool.Get().(chan bool), ts: time.Now().UnixNano() / int64(time.Millisecond), } select { case q.packets <- r: default: err = ecode.LimitExceed q.pool.Put(r.ch) } if err == nil { select { case drop := <-r.ch: if drop { err = ecode.LimitExceed } q.pool.Put(r.ch) case <-ctx.Done(): err = ecode.Deadline } } return } // Pop req from CoDel request buffer queue. func (q *Queue) Pop() { q.mux.Lock() q.outstanding -- if q.outstanding < 0 { q.outstanding = 0 q.mux.Unlock() return } defer q.mux.Unlock() for { select { case p := <-q.packets: drop := q.judge(p) select { case p.ch <- drop: if !drop { return } default: q.pool.Put(p.ch) } default: return } } } func (q *Queue) controlLaw(now int64) int64 { q.dropNext = now + int64(float64(q.conf.Internal)/math.Sqrt(float64(q.count))) return q.dropNext } // judge decide if the packet should drop or not. func (q *Queue) judge(p packet) (drop bool) { now := time.Now().UnixNano() / int64(time.Millisecond) sojurn := now - p.ts if sojurn < q.conf.Target { q.faTime = 0 } else if q.faTime == 0 { q.faTime = now + q.conf.Internal } else if now >= q.faTime { drop = true } if q.dropping { if !drop { // sojourn time below target - leave dropping state q.dropping = false } else if now > q.dropNext { q.count++ q.dropNext = q.controlLaw(q.dropNext) return } } else if drop && (now-q.dropNext < q.conf.Internal || now-q.faTime >= q.conf.Internal) { q.dropping = true // If we're in a drop cycle, the drop rate that controlled the queue // on the last cycle is a good starting point to control it now. if now-q.dropNext < q.conf.Internal { if q.count > 2 { q.count = q.count - 2 } else { q.count = 1 } } else { q.count = 1 } q.dropNext = q.controlLaw(now) return } q.outstanding ++ drop = false return }