func (bs *balanceSolver) filterHotPeers() []*statistics.HotPeerStat {
	ret := bs.cur.srcDetail.HotPeers
	maxPeerNum := bs.sche.conf.GetMaxPeerNumber()
	appendItem := func(items []*statistics.HotPeerStat, item *statistics.HotPeerStat) []*statistics.HotPeerStat {
		minHotDegree := bs.cluster.GetOpts().GetHotRegionCacheHitsThreshold()
		if _, ok := bs.sche.regionPendings[item.ID()]; !ok && !item.IsNeedCoolDownTransferLeader(minHotDegree) {
			items = append(items, item)
		}
		return items
	}
	if len(ret) <= maxPeerNum {
		nret := make([]*statistics.HotPeerStat, 0, len(ret))
		for _, peer := range ret {
			nret = appendItem(nret, peer)
		}
		return nret
	}
	union := bs.sortHotPeers(ret, maxPeerNum)
	ret = make([]*statistics.HotPeerStat, 0, len(union))
	for peer := range union {
		ret = appendItem(ret, peer)
	}
	return ret
}
filterHotPeers filtered hot peers from statistics.HotPeerStat and deleted the peer if its region is in pending status.
The returned hotPeer count in controlled by `max-peer-number`.
package schedulers
import (
	"fmt"
	"math"
	"math/rand"
	"net/http"
	"sort"
	"strconv"
	"sync"
	"time"
	"github.com/pingcap/kvproto/pkg/metapb"
	"github.com/pingcap/log"
	"github.com/tikv/pd/pkg/errs"
	"github.com/tikv/pd/pkg/slice"
	"github.com/tikv/pd/server/core"
	"github.com/tikv/pd/server/schedule"
	"github.com/tikv/pd/server/schedule/filter"
	"github.com/tikv/pd/server/schedule/operator"
	"github.com/tikv/pd/server/schedule/opt"
	"github.com/tikv/pd/server/statistics"
	"go.uber.org/zap"
)
func init() {
	schedule.RegisterSliceDecoderBuilder(HotRegionType, func(args []string) schedule.ConfigDecoder {
		return func(v interface{}) error {
			return nil
		}
	})
	schedule.RegisterScheduler(HotRegionType, func(opController *schedule.OperatorController, storage *core.Storage, decoder schedule.ConfigDecoder) (schedule.Scheduler, error) {
		conf := initHotRegionScheduleConfig()
		var data map[string]interface{}
		if err := decoder(&data); err != nil {
			return nil, err
		}
		if len(data) != 0 {
			conf.apply(compatibleConfig)
			if err := decoder(conf); err != nil {
				return nil, err
			}
		}
		conf.storage = storage
		return newHotScheduler(opController, conf), nil
	})
}
const (
	HotRegionName = "balance-hot-region-scheduler"
	HotRegionType = "hot-region"
	HotReadRegionType = "hot-read-region"
	HotWriteRegionType = "hot-write-region"
	minHotScheduleInterval = time.Second
	maxHotScheduleInterval = 20 * time.Second
)
var (
	schedulePeerPr = 0.66
	pendingAmpFactor = 2.0
)
type hotScheduler struct {
	name string
	*BaseScheduler
	sync.RWMutex
	types []rwType
	r     *rand.Rand
	regionPendings map[uint64]*pendingInfluence
	stInfos map[uint64]*storeSummaryInfo
	stLoadInfos [resourceTypeLen]map[uint64]*storeLoadDetail
	conf *hotRegionSchedulerConfig
}
func newHotScheduler(opController *schedule.OperatorController, conf *hotRegionSchedulerConfig) *hotScheduler {
	base := NewBaseScheduler(opController)
	ret := &hotScheduler{
		name:           HotRegionName,
		BaseScheduler:  base,
		types:          []rwType{write, read},
		r:              rand.New(rand.NewSource(time.Now().UnixNano())),
		regionPendings: make(map[uint64]*pendingInfluence),
		conf:           conf,
	}
	for ty := resourceType(0); ty < resourceTypeLen; ty++ {
		ret.stLoadInfos[ty] = map[uint64]*storeLoadDetail{}
	}
	return ret
}
func (h *hotScheduler) GetName() string {
	return h.name
}
func (h *hotScheduler) GetType() string {
	return HotRegionType
}
func (h *hotScheduler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	h.conf.ServeHTTP(w, r)
}
func (h *hotScheduler) GetMinInterval() time.Duration {
	return minHotScheduleInterval
}
func (h *hotScheduler) GetNextInterval(interval time.Duration) time.Duration {
	return intervalGrow(h.GetMinInterval(), maxHotScheduleInterval, exponentialGrowth)
}
func (h *hotScheduler) IsScheduleAllowed(cluster opt.Cluster) bool {
	allowed := h.OpController.OperatorCount(operator.OpHotRegion) < cluster.GetOpts().GetHotRegionScheduleLimit()
	if !allowed {
		operator.OperatorLimitCounter.WithLabelValues(h.GetType(), operator.OpHotRegion.String()).Inc()
	}
	return allowed
}
func (h *hotScheduler) Schedule(cluster opt.Cluster) []*operator.Operator {
	schedulerCounter.WithLabelValues(h.GetName(), "schedule").Inc()
	return h.dispatch(h.types[h.r.Int()%len(h.types)], cluster)
}
func (h *hotScheduler) dispatch(typ rwType, cluster opt.Cluster) []*operator.Operator {
	h.Lock()
	defer h.Unlock()
	h.prepareForBalance(typ, cluster)
	switch typ {
	case read:
		return h.balanceHotReadRegions(cluster)
	case write:
		return h.balanceHotWriteRegions(cluster)
	}
	return nil
}
func (h *hotScheduler) prepareForBalance(typ rwType, cluster opt.Cluster) {
	h.stInfos = summaryStoreInfos(cluster)
	h.summaryPendingInfluence()
	storesLoads := cluster.GetStoresLoads()
	isTraceRegionFlow := cluster.GetOpts().IsTraceRegionFlow()
	switch typ {
	case read:
		regionRead := cluster.RegionReadStats()
		h.stLoadInfos[readLeader] = summaryStoresLoad(
			h.stInfos,
			storesLoads,
			regionRead,
			isTraceRegionFlow,
			read, core.LeaderKind)
		h.stLoadInfos[readPeer] = summaryStoresLoad(
			h.stInfos,
			storesLoads,
			regionRead,
			isTraceRegionFlow,
			read, core.RegionKind)
	case write:
		regionWrite := cluster.RegionWriteStats()
		h.stLoadInfos[writeLeader] = summaryStoresLoad(
			h.stInfos,
			storesLoads,
			regionWrite,
			isTraceRegionFlow,
			write, core.LeaderKind)
		h.stLoadInfos[writePeer] = summaryStoresLoad(
			h.stInfos,
			storesLoads,
			regionWrite,
			isTraceRegionFlow,
			write, core.RegionKind)
	}
}
func (h *hotScheduler) summaryPendingInfluence() {
	for id, p := range h.regionPendings {
		from := h.stInfos[p.from]
		to := h.stInfos[p.to]
		maxZombieDur := p.maxZombieDuration
		weight, needGC := h.calcPendingInfluence(p.op, maxZombieDur)
		if needGC {
			delete(h.regionPendings, id)
			schedulerStatus.WithLabelValues(h.GetName(), "pending_op_infos").Dec()
			log.Debug("gc pending influence in hot region scheduler",
				zap.Uint64("region-id", id),
				zap.Time("create", p.op.GetCreateTime()),
				zap.Time("now", time.Now()),
				zap.Duration("zombie", maxZombieDur))
			continue
		}
		if from != nil && weight > 0 {
			from.addInfluence(&p.origin, -weight)
		}
		if to != nil && weight > 0 {
			to.addInfluence(&p.origin, weight)
		}
	}
}
func (h *hotScheduler) tryAddPendingInfluence(op *operator.Operator, srcStore, dstStore uint64, infl Influence, maxZombieDur time.Duration) bool {
	regionID := op.RegionID()
	_, ok := h.regionPendings[regionID]
	if ok {
		schedulerStatus.WithLabelValues(h.GetName(), "pending_op_fails").Inc()
		return false
	}
	influence := newPendingInfluence(op, srcStore, dstStore, infl, maxZombieDur)
	h.regionPendings[regionID] = influence
	schedulerStatus.WithLabelValues(h.GetName(), "pending_op_infos").Inc()
	return true
}
func (h *hotScheduler) balanceHotReadRegions(cluster opt.Cluster) []*operator.Operator {
	leaderSolver := newBalanceSolver(h, cluster, read, transferLeader)
	leaderOps := leaderSolver.solve()
	peerSolver := newBalanceSolver(h, cluster, read, movePeer)
	peerOps := peerSolver.solve()
	if len(leaderOps) == 0 && len(peerOps) == 0 {
		schedulerCounter.WithLabelValues(h.GetName(), "skip").Inc()
		return nil
	}
	if len(leaderOps) == 0 {
		if peerSolver.tryAddPendingInfluence() {
			return peerOps
		}
		schedulerCounter.WithLabelValues(h.GetName(), "skip").Inc()
		return nil
	}
	if len(peerOps) == 0 {
		if leaderSolver.tryAddPendingInfluence() {
			return leaderOps
		}
		schedulerCounter.WithLabelValues(h.GetName(), "skip").Inc()
		return nil
	}
	leaderSolver.cur = leaderSolver.best
	if leaderSolver.betterThan(peerSolver.best) {
		if leaderSolver.tryAddPendingInfluence() {
			return leaderOps
		}
		if peerSolver.tryAddPendingInfluence() {
			return peerOps
		}
	} else {
		if peerSolver.tryAddPendingInfluence() {
			return peerOps
		}
		if leaderSolver.tryAddPendingInfluence() {
			return leaderOps
		}
	}
	schedulerCounter.WithLabelValues(h.GetName(), "skip").Inc()
	return nil
}
func (h *hotScheduler) balanceHotWriteRegions(cluster opt.Cluster) []*operator.Operator {
	s := h.r.Intn(100)
	switch {
	case s < int(schedulePeerPr*100):
		peerSolver := newBalanceSolver(h, cluster, write, movePeer)
		ops := peerSolver.solve()
		if len(ops) > 0 && peerSolver.tryAddPendingInfluence() {
			return ops
		}
	default:
	}
	leaderSolver := newBalanceSolver(h, cluster, write, transferLeader)
	ops := leaderSolver.solve()
	if len(ops) > 0 && leaderSolver.tryAddPendingInfluence() {
		return ops
	}
	schedulerCounter.WithLabelValues(h.GetName(), "skip").Inc()
	return nil
}
type balanceSolver struct {
	sche         *hotScheduler
	cluster      opt.Cluster
	stLoadDetail map[uint64]*storeLoadDetail
	rwTy         rwType
	opTy         opType
	cur  *solution
	best *solution
	ops  []*operator.Operator
	infl Influence
	maxSrc   *storeLoad
	minDst   *storeLoad
	rankStep *storeLoad
	firstPriority  int
	secondPriority int
	firstPriorityIsBetter  bool
	secondPriorityIsBetter bool
}
type solution struct {
	srcDetail   *storeLoadDetail
	srcPeerStat *statistics.HotPeerStat
	region      *core.RegionInfo
	dstDetail   *storeLoadDetail
	progressiveRank int64
}
func (bs *balanceSolver) init() {
	switch toResourceType(bs.rwTy, bs.opTy) {
	case writePeer:
		bs.stLoadDetail = bs.sche.stLoadInfos[writePeer]
	case writeLeader:
		bs.stLoadDetail = bs.sche.stLoadInfos[writeLeader]
	case readLeader:
		bs.stLoadDetail = bs.sche.stLoadInfos[readLeader]
	case readPeer:
		bs.stLoadDetail = bs.sche.stLoadInfos[readPeer]
	}
	bs.maxSrc = &storeLoad{Loads: make([]float64, statistics.DimLen)}
	bs.minDst = &storeLoad{
		Loads: make([]float64, statistics.DimLen),
		Count: math.MaxFloat64,
	}
	for i := range bs.minDst.Loads {
		bs.minDst.Loads[i] = math.MaxFloat64
	}
	maxCur := &storeLoad{Loads: make([]float64, statistics.DimLen)}
	for _, detail := range bs.stLoadDetail {
		bs.maxSrc = maxLoad(bs.maxSrc, detail.LoadPred.min())
		bs.minDst = minLoad(bs.minDst, detail.LoadPred.max())
		maxCur = maxLoad(maxCur, &detail.LoadPred.Current)
	}
	rankStepRatios := []float64{
		statistics.ByteDim:  bs.sche.conf.GetByteRankStepRatio(),
		statistics.KeyDim:   bs.sche.conf.GetKeyRankStepRatio(),
		statistics.QueryDim: bs.sche.conf.GetQueryRateRankStepRatio()}
	stepLoads := make([]float64, statistics.DimLen)
	for i := range stepLoads {
		stepLoads[i] = maxCur.Loads[i] * rankStepRatios[i]
	}
	bs.rankStep = &storeLoad{
		Loads: stepLoads,
		Count: maxCur.Count * bs.sche.conf.GetCountRankStepRatio(),
	}
	bs.firstPriority, bs.secondPriority = prioritiesToDim(bs.getPriorities())
}
func (bs *balanceSolver) isSelectedDim(dim int) bool {
	return dim == bs.firstPriority || dim == bs.secondPriority
}
func (bs *balanceSolver) getPriorities() []string {
	querySupport := bs.sche.conf.checkQuerySupport(bs.cluster)
	switch bs.rwTy {
	case read:
		return adjustConfig(querySupport, bs.sche.conf.GetReadPriorities(), getReadPriorities)
	case write:
		switch bs.opTy {
		case transferLeader:
			return adjustConfig(querySupport, bs.sche.conf.GetWriteLeaderPriorities(), getWriteLeaderPriorities)
		case movePeer:
			return adjustConfig(querySupport, bs.sche.conf.GetWritePeerPriorities(), getWritePeerPriorities)
		}
	}
	log.Error("illegal type or illegal operator while getting the priority", zap.String("type", bs.rwTy.String()), zap.String("operator", bs.opTy.String()))
	return []string{}
}
func newBalanceSolver(sche *hotScheduler, cluster opt.Cluster, rwTy rwType, opTy opType) *balanceSolver {
	solver := &balanceSolver{
		sche:    sche,
		cluster: cluster,
		rwTy:    rwTy,
		opTy:    opTy,
	}
	solver.init()
	return solver
}
func (bs *balanceSolver) isValid() bool {
	if bs.cluster == nil || bs.sche == nil || bs.stLoadDetail == nil {
		return false
	}
	switch bs.rwTy {
	case write, read:
	default:
		return false
	}
	switch bs.opTy {
	case movePeer, transferLeader:
	default:
		return false
	}
	return true
}
func (bs *balanceSolver) solve() []*operator.Operator {
	if !bs.isValid() {
		return nil
	}
	bs.cur = &solution{}
	for _, srcDetail := range bs.filterSrcStores() {
		bs.cur.srcDetail = srcDetail
		for _, srcPeerStat := range bs.filterHotPeers() {
			bs.cur.srcPeerStat = srcPeerStat
			bs.cur.region = bs.getRegion()
			if bs.cur.region == nil {
				continue
			}
			for _, dstDetail := range bs.filterDstStores() {
				bs.cur.dstDetail = dstDetail
				bs.calcProgressiveRank()
				if bs.cur.progressiveRank < 0 && bs.betterThan(bs.best) {
					if newOp, newInfl := bs.buildOperator(); newOp != nil {
						bs.ops = []*operator.Operator{newOp}
						bs.infl = *newInfl
						clone := *bs.cur
						bs.best = &clone
					}
				}
			}
		}
	}
	return bs.ops
}
func (bs *balanceSolver) tryAddPendingInfluence() bool {
	if bs.best == nil || len(bs.ops) == 0 {
		return false
	}
	if bs.best.srcDetail.Info.IsTiFlash != bs.best.dstDetail.Info.IsTiFlash {
		schedulerCounter.WithLabelValues(bs.sche.GetName(), "not-same-engine").Inc()
		return false
	}
	var maxZombieDur time.Duration
	switch {
	case bs.isForWriteLeader():
		maxZombieDur = bs.sche.conf.GetRegionsStatZombieDuration()
	case bs.isForWritePeer():
		if bs.best.srcDetail.Info.IsTiFlash {
			maxZombieDur = bs.sche.conf.GetRegionsStatZombieDuration()
		} else {
			maxZombieDur = bs.sche.conf.GetStoreStatZombieDuration()
		}
	default:
		maxZombieDur = bs.sche.conf.GetStoreStatZombieDuration()
	}
	return bs.sche.tryAddPendingInfluence(bs.ops[0], bs.best.srcDetail.getID(), bs.best.dstDetail.getID(), bs.infl, maxZombieDur)
}
func (bs *balanceSolver) isForWriteLeader() bool {
	return bs.rwTy == write && bs.opTy == transferLeader
}
func (bs *balanceSolver) isForWritePeer() bool {
	return bs.rwTy == write && bs.opTy == movePeer
}
func (bs *balanceSolver) filterSrcStores() map[uint64]*storeLoadDetail {
	ret := make(map[uint64]*storeLoadDetail)
	confSrcToleranceRatio := bs.sche.conf.GetSrcToleranceRatio()
	confEnableForTiFlash := bs.sche.conf.GetEnableForTiFlash()
	for id, detail := range bs.stLoadDetail {
		srcToleranceRatio := confSrcToleranceRatio
		if detail.Info.IsTiFlash {
			if !confEnableForTiFlash {
				continue
			}
			if bs.rwTy != write || bs.opTy != movePeer {
				continue
			}
			srcToleranceRatio += tiflashToleranceRatioCorrection
		}
		if len(detail.HotPeers) == 0 {
			continue
		}
		if bs.checkSrcByDimPriorityAndTolerance(detail.LoadPred.min(), &detail.LoadPred.Expect, srcToleranceRatio) {
			ret[id] = detail
			hotSchedulerResultCounter.WithLabelValues("src-store-succ", strconv.FormatUint(id, 10)).Inc()
		} else {
			hotSchedulerResultCounter.WithLabelValues("src-store-failed", strconv.FormatUint(id, 10)).Inc()
		}
	}
	return ret
}
func (bs *balanceSolver) checkSrcByDimPriorityAndTolerance(minLoad, expectLoad *storeLoad, toleranceRatio float64) bool {
	if bs.sche.conf.IsStrictPickingStoreEnabled() {
		return slice.AllOf(minLoad.Loads, func(i int) bool {
			if bs.isSelectedDim(i) {
				return minLoad.Loads[i] > toleranceRatio*expectLoad.Loads[i]
			}
			return true
		})
	}
	return minLoad.Loads[bs.firstPriority] > toleranceRatio*expectLoad.Loads[bs.firstPriority]
}