<commit_before>// Copyright 2018 PingCAP, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// See the License for the specific language governing permissions and
// limitations under the License.

package schedule

import (
	"math/rand"
	"time"

	"github.com/pingcap/pd/server/cache"
	"github.com/pingcap/pd/server/core"
)

// FlowKind is a identify Flow types.
type FlowKind uint32

// Flags for flow.
const (
	WriteFlow FlowKind = iota
	ReadFlow
)

// HotSpotCache is a cache hold hot regions.
type HotSpotCache struct {
	writeFlow cache.Cache
	readFlow  cache.Cache
}

func newHotSpotCache() *HotSpotCache {
	return &HotSpotCache{
		writeFlow: cache.NewCache(statCacheMaxLen, cache.TwoQueueCache),
		readFlow:  cache.NewCache(statCacheMaxLen, cache.TwoQueueCache),
	}
}

// CheckWrite checks the write status, returns whether need update statistics and item.
func (w *HotSpotCache) CheckWrite(region *core.RegionInfo, stores *core.StoresInfo) (bool, *core.RegionStat) {
	var (
		WrittenBytesPerSec uint64
		value              *core.RegionStat
	)
	v, isExist := w.writeFlow.Peek(region.GetId())
	if isExist && !Simulating {
		value = v.(*core.RegionStat)
		interval := time.Since(value.LastUpdateTime).Seconds()
		if interval < minHotRegionReportInterval {
			return false, nil
		}
		WrittenBytesPerSec = uint64(float64(region.WrittenBytes) / interval)
	} else {
		WrittenBytesPerSec = uint64(float64(region.WrittenBytes) / float64(RegionHeartBeatReportInterval))
	}

	hotRegionThreshold := calculateWriteHotThreshold(stores)
	return w.isNeedUpdateStatCache(region, WrittenBytesPerSec, hotRegionThreshold, value, WriteFlow)
}

// CheckRead checks the read status, returns whether need update statistics and item.
func (w *HotSpotCache) CheckRead(region *core.RegionInfo, stores *core.StoresInfo) (bool, *core.RegionStat) {
	var (
		ReadBytesPerSec uint64
		value           *core.RegionStat
	)
	v, isExist := w.readFlow.Peek(region.GetId())
	if isExist && !Simulating {
		value = v.(*core.RegionStat)
		interval := time.Since(value.LastUpdateTime).Seconds()
		if interval < minHotRegionReportInterval {
			return false, nil
		}
		ReadBytesPerSec = uint64(float64(region.ReadBytes) / interval)
	} else {
		ReadBytesPerSec = uint64(float64(region.ReadBytes) / float64(RegionHeartBeatReportInterval))
	}
	hotRegionThreshold := calculateReadHotThreshold(stores)
	return w.isNeedUpdateStatCache(region, ReadBytesPerSec, hotRegionThreshold, value, ReadFlow)
}

func (w *HotSpotCache) incMetrics(name string, kind FlowKind) {
	switch kind {
	case WriteFlow:
		hotCacheStatusGauge.WithLabelValues(name, "write").Inc()
	case ReadFlow:
		hotCacheStatusGauge.WithLabelValues(name, "read").Inc()
	}
}

func calculateWriteHotThreshold(stores *core.StoresInfo) uint64 {
	// hotRegionThreshold is use to pick hot region
	// suppose the number of the hot Regions is statCacheMaxLen
	// and we use total written Bytes past storeHeartBeatReportInterval seconds to divide the number of hot Regions
	// divide 2 because the store reports data about two times than the region record write to rocksdb
	divisor := float64(statCacheMaxLen) * 2
	hotRegionThreshold := uint64(float64(stores.TotalBytesWriteRate()) / divisor)

	if hotRegionThreshold < hotWriteRegionMinFlowRate {
		hotRegionThreshold = hotWriteRegionMinFlowRate
	}
	return hotRegionThreshold<commit_msg>fix hot write and read (#1150)<commit_after>// Copyright 2018 PingCAP, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// See the License for the specific language governing permissions and
// limitations under the License.

package schedule

import (
	"math/rand"
	"time"

	"github.com/pingcap/pd/server/cache"
	"github.com/pingcap/pd/server/core"
)

// FlowKind is a identify Flow types.
type FlowKind uint32

// Flags for flow.
const (
	WriteFlow FlowKind = iota
	ReadFlow
)

// HotSpotCache is a cache hold hot regions.
type HotSpotCache struct {
	writeFlow cache.Cache
	readFlow  cache.Cache
}

func newHotSpotCache() *HotSpotCache {
	return &HotSpotCache{
		writeFlow: cache.NewCache(statCacheMaxLen, cache.TwoQueueCache),
		readFlow:  cache.NewCache(statCacheMaxLen, cache.TwoQueueCache),
	}
}

// CheckWrite checks the write status, returns whether need update statistics and item.
func (w *HotSpotCache) CheckWrite(region *core.RegionInfo, stores *core.StoresInfo) (bool, *core.RegionStat) {
	var (
		WrittenBytesPerSec uint64
		value              *core.RegionStat
	)

	WrittenBytesPerSec = uint64(float64(region.WrittenBytes) / float64(RegionHeartBeatReportInterval))

	v, isExist := w.writeFlow.Peek(region.GetId())
	if isExist {
		value = v.(*core.RegionStat)
		if !Simulating {
			interval := time.Since(value.LastUpdateTime).Seconds()
			if interval < minHotRegionReportInterval {
				return false, nil
			}
			WrittenBytesPerSec = uint64(float64(region.WrittenBytes) / interval)
		}
	}

	hotRegionThreshold := calculateWriteHotThreshold(stores)
	return w.isNeedUpdateStatCache(region, WrittenBytesPerSec, hotRegionThreshold, value, WriteFlow)
}

// CheckRead checks the read status, returns whether need update statistics and item.
func (w *HotSpotCache) CheckRead(region *core.RegionInfo, stores *core.StoresInfo) (bool, *core.RegionStat) {
	var (
		ReadBytesPerSec uint64
		value           *core.RegionStat
	)

	ReadBytesPerSec = uint64(float64(region.ReadBytes) / float64(RegionHeartBeatReportInterval))

	v, isExist := w.readFlow.Peek(region.GetId())
	if isExist {
		value = v.(*core.RegionStat)
		if !Simulating {
			interval := time.Since(value.LastUpdateTime).Seconds()
			if interval < minHotRegionReportInterval {
				return false, nil
			}
			ReadBytesPerSec = uint64(float64(region.ReadBytes) / interval)
		}
	}

	hotRegionThreshold := calculateReadHotThreshold(stores)
	return w.isNeedUpdateStatCache(region, ReadBytesPerSec, hotRegionThreshold, value, ReadFlow)
}

func (w *HotSpotCache) incMetrics(name string, kind FlowKind) {
	switch kind {
	case WriteFlow:
		hotCacheStatusGauge.WithLabelValues(name, "write").Inc()
	case ReadFlow:
		hotCacheStatusGauge.WithLabelValues(name, "read").Inc()
	}
}

func calculateWriteHotThreshold(stores *core.StoresInfo) uint64 {
	// hotRegionThreshold is use to pick hot region
	// suppose the number of the hot Regions is statCacheMaxLen
	// and we use total written Bytes past storeHeartBeatReportInterval seconds to divide the number of hot Regions
	// divide 2 because the store reports data about two times than the region record write to rocksdb
	divisor := float64(statCacheMaxLen) * 2
	hotRegionThreshold := uint64(float64(stores.TotalBytesWriteRate()) / divisor)

	if hotRegionThreshold < hotWriteRegionMinFlowRate {
		hotRegionThreshold = hotWriteRegionMinFlowRate
	}
	return hotRegionThreshold
{'fraction_non_alphanumeric': 0.09417344173441734, 'fraction_numerical': 0.012466124661246613, 'mean_word_length': 4.612927756653993, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7530340', 'n_tokens_mistral': 2393, 'n_tokens_neox': 2213, 'n_words': 640}