// SheddingHandler returns a middleware that does load shedding.
func SheddingHandler(shedder load.Shedder, metrics *stat.Metrics) func(http.Handler) http.Handler {
	if shedder == nil {
		return func(next http.Handler) http.Handler {
			return next
		}
	}
	ensureSheddingStat() // 負(fù)責(zé)每分鐘打印shedding相關(guān)的數(shù)據(jù)
	return func(next http.Handler) http.Handler {
		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
			sheddingStat.IncrementTotal()
			promise, err := shedder.Allow() // 判斷是否允許此請(qǐng)求進(jìn)入下一步
			if err != nil {
				metrics.AddDrop() // drop掉請(qǐng)求，在中間件層面就拒絕了請(qǐng)求
				sheddingStat.IncrementDrop()
				logx.Errorf("[http] dropped, %s - %s - %s",
					r.RequestURI, httpx.GetRemoteAddr(r), r.UserAgent())
				w.WriteHeader(http.StatusServiceUnavailable)// 返回503，提示服務(wù)不可用
				return
			}
			cw := response.NewWithCodeResponseWriter(w)
			defer func() {
				if cw.Code == http.StatusServiceUnavailable {
					promise.Fail() // 相關(guān)指標(biāo)記錄
				} else {
					sheddingStat.IncrementPass()
					promise.Pass() // 相關(guān)指標(biāo)記錄
				}
			}()
			next.ServeHTTP(cw, r)
		})
	}
}

// Allow implements Shedder.Allow.
func (as *adaptiveShedder) Allow() (Promise, error) {
	if as.shouldDrop() {// 判斷是否應(yīng)該丟棄
		as.droppedRecently.Set(true)
		return nil, ErrServiceOverloaded// 丟棄
	}
	as.addFlying(1) // 通過(guò)校驗(yàn)
	return &promise{
		start:   timex.Now(),
		shedder: as,
	}, nil
}

func (as *adaptiveShedder) shouldDrop() bool {
	if as.systemOverloaded() || as.stillHot() {// 如果任一滿足，這個(gè)請(qǐng)求都會(huì)被過(guò)載
		if as.highThru() {
			flying := atomic.LoadInt64(&as.flying)
			as.avgFlyingLock.Lock()
			avgFlying := as.avgFlying
			as.avgFlyingLock.Unlock()
			msg := fmt.Sprintf(
				"dropreq, cpu: %d, maxPass: %d, minRt: %.2f, hot: %t, flying: %d, avgFlying: %.2f",
				stat.CpuUsage(), as.maxPass(), as.minRt(), as.stillHot(), flying, avgFlying)
			logx.Error(msg)
			stat.Report(msg)
			return true
		}
	}
	return false
}
func (as *adaptiveShedder) systemOverloaded() bool {
	if !systemOverloadChecker(as.cpuThreshold) { // 校驗(yàn)CPU的負(fù)載是否超出設(shè)定值
		return false
	}
	as.overloadTime.Set(timex.Now())// 超出設(shè)定值，記錄當(dāng)前的時(shí)間（這主要是為了后續(xù)流量減小，系統(tǒng)的恢復(fù)用）
	return true
}
func (as *adaptiveShedder) stillHot() bool {
	if !as.droppedRecently.True() {// 如果這個(gè)請(qǐng)求之前有請(qǐng)求被drop這里值為true，反之為false
		return false// 之前的請(qǐng)求沒有被drop表示系統(tǒng)可能沒有遇到過(guò)載的問(wèn)題，返回false
	}
	overloadTime := as.overloadTime.Load()// 如果之前有請(qǐng)求被drop，表示存在過(guò)載
	if overloadTime == 0 {// 看看是否有記錄過(guò)載的時(shí)間
		return false
	}
	if timex.Since(overloadTime) < coolOffDuration {// 如果小于冷卻時(shí)間，表示系統(tǒng)依然是過(guò)載狀態(tài)
		return true
	}
	as.droppedRecently.Set(false)// 表示CPU過(guò)載，上一次過(guò)載過(guò)了冷卻器，這個(gè)請(qǐng)求可以繼續(xù)執(zhí)行，設(shè)置為false
	return false
}

const (
	// 250ms and 0.95 as beta will count the average cpu load for past 5 seconds
	cpuRefreshInterval = time.Millisecond * 250
	allRefreshInterval = time.Minute
	// moving average beta hyperparameter
	beta = 0.95
)
var cpuUsage int64
func init() {
	go func() {
		cpuTicker := time.NewTicker(cpuRefreshInterval)
		defer cpuTicker.Stop()
		allTicker := time.NewTicker(allRefreshInterval)
		defer allTicker.Stop()
		for {
			select {
			case <-cpuTicker.C:
				threading.RunSafe(func() {
					curUsage := internal.RefreshCpu() // 刷新CPU使用率數(shù)據(jù)
					prevUsage := atomic.LoadInt64(&cpuUsage)
					// cpu = cpu??1 * beta + cpu? * (1 - beta)
					usage := int64(float64(prevUsage)*beta + float64(curUsage)*(1-beta))
					atomic.StoreInt64(&cpuUsage, usage)
				})
			case <-allTicker.C:
				if logEnabled.True() {
					printUsage()
				}
			}
		}
	}()
}

func (as *adaptiveShedder) addFlying(delta int64) {
	flying := atomic.AddInt64(&as.flying, delta)// 請(qǐng)求通過(guò)檢驗(yàn)進(jìn)入后會(huì)加1，請(qǐng)求被服務(wù)處理完后會(huì)減1
	if delta < 0 {
		as.avgFlyingLock.Lock()
		// 平均請(qǐng)求數(shù)計(jì)算為當(dāng)前平均請(qǐng)求數(shù)*0.9 + 當(dāng)前運(yùn)行請(qǐng)求數(shù)*0.1
		as.avgFlying = as.avgFlying*flyingBeta + float64(flying)*(1-flyingBeta)
		as.avgFlyingLock.Unlock()
	}
}
func (as *adaptiveShedder) highThru() bool {
	as.avgFlyingLock.Lock()
	avgFlying := as.avgFlying // 運(yùn)行中的平均請(qǐng)求數(shù)
	as.avgFlyingLock.Unlock()
	maxFlight := as.maxFlight()// 運(yùn)行的最大的請(qǐng)求數(shù)
	// 如果運(yùn)行的平均請(qǐng)求數(shù)>最大的請(qǐng)求數(shù)且當(dāng)前運(yùn)行的請(qǐng)求數(shù)>最大的請(qǐng)求數(shù)，表示依舊高負(fù)載
	return int64(avgFlying) > maxFlight && atomic.LoadInt64(&as.flying) > maxFlight
}
func (as *adaptiveShedder) maxFlight() int64 {
	// windows = buckets per second
	// maxQPS = maxPASS * windows
	// minRT = min average response time in milliseconds
	// maxQPS * minRT / milliseconds_per_second
	// 最大的運(yùn)行數(shù)的計(jì)算為最大請(qǐng)求數(shù)*窗口的長(zhǎng)度*最小的處理時(shí)間
	return int64(math.Max(1, float64(as.maxPass()*as.windows)*(as.minRt()/1e3)))
}

func (as *adaptiveShedder) maxPass() int64 {
	var result float64 = 1
	as.passCounter.Reduce(func(b *collection.Bucket) {
		if b.Sum > result {
			result = b.Sum
		}
	})
	return int64(result)
}
func (as *adaptiveShedder) minRt() float64 {
	result := defaultMinRt
	as.rtCounter.Reduce(func(b *collection.Bucket) {
		if b.Count <= 0 {
			return
		}
		avg := math.Round(b.Sum / float64(b.Count))
		if avg < result {
			result = avg
		}
	})
	return result
}