func (c *RaftCluster) DropCacheRegion(id uint64) {
	c.RLock()
	defer c.RUnlock()
	if region := c.GetRegion(id); region != nil {
		c.core.RemoveRegion(region)
	}
}
DropCacheRegion removes a region from the cache.
package cluster
import (
	"context"
	"fmt"
	"net/http"
	"strconv"
	"sync"
	"time"
	"github.com/coreos/go-semver/semver"
	"github.com/gogo/protobuf/proto"
	"github.com/pingcap/errors"
	"github.com/pingcap/failpoint"
	"github.com/pingcap/kvproto/pkg/metapb"
	"github.com/pingcap/kvproto/pkg/pdpb"
	"github.com/pingcap/log"
	"github.com/tikv/pd/pkg/cache"
	"github.com/tikv/pd/pkg/component"
	"github.com/tikv/pd/pkg/errs"
	"github.com/tikv/pd/pkg/etcdutil"
	"github.com/tikv/pd/pkg/keyutil"
	"github.com/tikv/pd/pkg/logutil"
	"github.com/tikv/pd/pkg/typeutil"
	"github.com/tikv/pd/server/config"
	"github.com/tikv/pd/server/core"
	"github.com/tikv/pd/server/core/storelimit"
	"github.com/tikv/pd/server/id"
	syncer "github.com/tikv/pd/server/region_syncer"
	"github.com/tikv/pd/server/replication"
	"github.com/tikv/pd/server/schedule"
	"github.com/tikv/pd/server/schedule/checker"
	"github.com/tikv/pd/server/schedule/hbstream"
	"github.com/tikv/pd/server/schedule/labeler"
	"github.com/tikv/pd/server/schedule/placement"
	"github.com/tikv/pd/server/statistics"
	"github.com/tikv/pd/server/versioninfo"
	"go.etcd.io/etcd/clientv3"
	"go.uber.org/zap"
)
var backgroundJobInterval = 10 * time.Second
const (
	clientTimeout              = 3 * time.Second
	defaultChangedRegionsLimit = 10000
	persistLimitRetryTimes = 5
	persistLimitWaitTime   = 100 * time.Millisecond
)
type Server interface {
	GetAllocator() id.Allocator
	GetConfig() *config.Config
	GetPersistOptions() *config.PersistOptions
	GetStorage() *core.Storage
	GetHBStreams() *hbstream.HeartbeatStreams
	GetRaftCluster() *RaftCluster
	GetBasicCluster() *core.BasicCluster
	ReplicateFileToAllMembers(ctx context.Context, name string, data []byte) error
}
type RaftCluster struct {
	sync.RWMutex
	ctx context.Context
	running bool
	clusterID   uint64
	clusterRoot string
	core    *core.BasicCluster
	meta    *metapb.Cluster
	opt     *config.PersistOptions
	storage *core.Storage
	id      id.Allocator
	limiter *StoreLimiter
	prepareChecker *prepareChecker
	changedRegions chan *core.RegionInfo
	labelLevelStats *statistics.LabelStatistics
	regionStats     *statistics.RegionStatistics
	hotStat         *statistics.HotStat
	coordinator      *coordinator
	suspectRegions   *cache.TTLUint64  
	suspectKeyRanges *cache.TTLString  
	wg           sync.WaitGroup
	quit         chan struct{}
	regionSyncer *syncer.RegionSyncer
	ruleManager   *placement.RuleManager
	regionLabeler *labeler.RegionLabeler
	etcdClient    *clientv3.Client
	httpClient    *http.Client
	replicationMode *replication.ModeManager
	traceRegionFlow bool
	componentManager *component.Manager
}
type Status struct {
	RaftBootstrapTime time.Time `json:"raft_bootstrap_time,omitempty"`
	IsInitialized     bool      `json:"is_initialized"`
	ReplicationStatus string    `json:"replication_status"`
}
func NewRaftCluster(ctx context.Context, root string, clusterID uint64, regionSyncer *syncer.RegionSyncer, etcdClient *clientv3.Client, httpClient *http.Client) *RaftCluster {
	return &RaftCluster{
		ctx:          ctx,
		running:      false,
		clusterID:    clusterID,
		clusterRoot:  root,
		regionSyncer: regionSyncer,
		httpClient:   httpClient,
		etcdClient:   etcdClient,
	}
}
func (c *RaftCluster) LoadClusterStatus() (*Status, error) {
	bootstrapTime, err := c.loadBootstrapTime()
	if err != nil {
		return nil, err
	}
	var isInitialized bool
	if bootstrapTime != typeutil.ZeroTime {
		isInitialized = c.isInitialized()
	}
	var replicationStatus string
	if c.replicationMode != nil {
		replicationStatus = c.replicationMode.GetReplicationStatus().String()
	}
	return &Status{
		RaftBootstrapTime: bootstrapTime,
		IsInitialized:     isInitialized,
		ReplicationStatus: replicationStatus,
	}, nil
}
func (c *RaftCluster) isInitialized() bool {
	if c.core.GetRegionCount() > 1 {
		return true
	}
	region := c.core.SearchRegion(nil)
	return region != nil &&
		len(region.GetVoters()) >= int(c.GetReplicationConfig().MaxReplicas) &&
		len(region.GetPendingPeers()) == 0
}
func (c *RaftCluster) GetReplicationConfig() *config.ReplicationConfig {
	cfg := &config.ReplicationConfig{}
	*cfg = *c.opt.GetReplicationConfig()
	return cfg
}
func (c *RaftCluster) loadBootstrapTime() (time.Time, error) {
	var t time.Time
	data, err := c.storage.Load(c.storage.ClusterStatePath("raft_bootstrap_time"))
	if err != nil {
		return t, err
	}
	if data == "" {
		return t, nil
	}
	return typeutil.ParseTimestamp([]byte(data))
}
func (c *RaftCluster) InitCluster(id id.Allocator, opt *config.PersistOptions, storage *core.Storage, basicCluster *core.BasicCluster) {
	c.core = basicCluster
	c.opt = opt
	c.storage = storage
	c.id = id
	c.labelLevelStats = statistics.NewLabelStatistics()
	c.hotStat = statistics.NewHotStat(c.ctx, c.quit)
	c.prepareChecker = newPrepareChecker()
	c.changedRegions = make(chan *core.RegionInfo, defaultChangedRegionsLimit)
	c.suspectRegions = cache.NewIDTTL(c.ctx, time.Minute, 3*time.Minute)
	c.suspectKeyRanges = cache.NewStringTTL(c.ctx, time.Minute, 3*time.Minute)
	c.traceRegionFlow = opt.GetPDServerConfig().TraceRegionFlow
}
func (c *RaftCluster) Start(s Server) error {
	c.Lock()
	defer c.Unlock()
	if c.running {
		log.Warn("raft cluster has already been started")
		return nil
	}
	c.quit = make(chan struct{})
	c.InitCluster(s.GetAllocator(), s.GetPersistOptions(), s.GetStorage(), s.GetBasicCluster())
	cluster, err := c.LoadClusterInfo()
	if err != nil {
		return err
	}
	if cluster == nil {
		return nil
	}
	c.ruleManager = placement.NewRuleManager(c.storage, c, c.GetOpts())
	if c.opt.IsPlacementRulesEnabled() {
		err = c.ruleManager.Initialize(c.opt.GetMaxReplicas(), c.opt.GetLocationLabels())
		if err != nil {
			return err
		}
	}
	c.regionLabeler, err = labeler.NewRegionLabeler(c.storage)
	if err != nil {
		return err
	}
	c.componentManager = component.NewManager(c.storage)
	_, err = c.storage.LoadComponent(&c.componentManager)
	if err != nil {
		return err
	}
	c.replicationMode, err = replication.NewReplicationModeManager(s.GetConfig().ReplicationMode, s.GetStorage(), cluster, s)
	if err != nil {
		return err
	}
	c.coordinator = newCoordinator(c.ctx, cluster, s.GetHBStreams())
	c.regionStats = statistics.NewRegionStatistics(c.opt, c.ruleManager)
	c.limiter = NewStoreLimiter(s.GetPersistOptions())
	c.wg.Add(5)
	go c.runCoordinator()
	failpoint.Inject("highFrequencyClusterJobs", func() {
		backgroundJobInterval = 100 * time.Microsecond
	})
	go c.runBackgroundJobs(backgroundJobInterval)
	go c.runStatsBackgroundJobs()
	go c.syncRegions()
	go c.runReplicationMode()
	c.running = true
	return nil
}
func (c *RaftCluster) LoadClusterInfo() (*RaftCluster, error) {
	c.meta = &metapb.Cluster{}
	ok, err := c.storage.LoadMeta(c.meta)
	if err != nil {
		return nil, err
	}
	if !ok {
		return nil, nil
	}
	start := time.Now()
	if err := c.storage.LoadStores(c.core.PutStore); err != nil {
		return nil, err
	}
	log.Info("load stores",
		zap.Int("count", c.GetStoreCount()),
		zap.Duration("cost", time.Since(start)),
	)
	start = time.Now()
	if err := c.storage.LoadRegionsOnce(c.ctx, c.core.CheckAndPutRegion); err != nil {
		return nil, err
	}
	log.Info("load regions",
		zap.Int("count", c.core.GetRegionCount()),
		zap.Duration("cost", time.Since(start)),
	)
	for _, store := range c.GetStores() {
		c.hotStat.GetOrCreateRollingStoreStats(store.GetID())
	}
	return c, nil
}
func (c *RaftCluster) runBackgroundJobs(interval time.Duration) {
	defer logutil.LogPanic()
	defer c.wg.Done()
	ticker := time.NewTicker(interval)
	defer ticker.Stop()
	for {
		select {
		case <-c.quit:
			log.Info("metrics are reset")
			c.resetMetrics()
			log.Info("background jobs has been stopped")
			return
		case <-ticker.C:
			c.checkStores()
			c.collectMetrics()
			c.coordinator.opController.PruneHistory()
		}
	}
}
func (c *RaftCluster) runStatsBackgroundJobs() {
	defer logutil.LogPanic()
	defer c.wg.Done()
	ticker := time.NewTicker(statistics.RegionsStatsObserveInterval)
	defer ticker.Stop()
	for {
		select {
		case <-c.quit:
			log.Info("statistics background jobs has been stopped")
			return
		case <-ticker.C:
			c.RLock()
			storeIDs, writeBytesRates, writeKeysRates := c.core.GetStoresWriteRate()
			c.RUnlock()
			c.Lock()
			c.hotStat.ObserveRegionsStats(storeIDs, writeBytesRates, writeKeysRates)
			c.Unlock()
		}
	}
}
func (c *RaftCluster) runCoordinator() {
	defer logutil.LogPanic()
	defer c.wg.Done()
	defer func() {
		c.coordinator.wg.Wait()
		log.Info("coordinator has been stopped")
	}()
	c.coordinator.run()
	<-c.coordinator.ctx.Done()
	log.Info("coordinator is stopping")
}
func (c *RaftCluster) syncRegions() {
	defer logutil.LogPanic()
	defer c.wg.Done()
	c.regionSyncer.RunServer(c.changedRegionNotifier(), c.quit)
}
func (c *RaftCluster) runReplicationMode() {
	defer logutil.LogPanic()
	defer c.wg.Done()
	c.replicationMode.Run(c.quit)
}
func (c *RaftCluster) Stop() {
	c.Lock()
	if !c.running {
		c.Unlock()
		return
	}
	c.running = false
	c.coordinator.stop()
	close(c.quit)
	c.Unlock()
	c.wg.Wait()
	log.Info("raftcluster is stopped")
}
func (c *RaftCluster) IsRunning() bool {
	c.RLock()
	defer c.RUnlock()
	return c.running
}
func (c *RaftCluster) GetOperatorController() *schedule.OperatorController {
	c.RLock()
	defer c.RUnlock()
	return c.coordinator.opController
}
func (c *RaftCluster) GetBasicCluster() *core.BasicCluster {
	c.RLock()
	defer c.RUnlock()
	return c.core
}
func (c *RaftCluster) GetRegionScatter() *schedule.RegionScatterer {
	c.RLock()
	defer c.RUnlock()
	return c.coordinator.regionScatterer
}
func (c *RaftCluster) GetRegionSplitter() *schedule.RegionSplitter {
	c.RLock()
	defer c.RUnlock()
	return c.coordinator.regionSplitter
}
func (c *RaftCluster) GetHeartbeatStreams() *hbstream.HeartbeatStreams {
	c.RLock()
	defer c.RUnlock()
	return c.coordinator.hbStreams
}
func (c *RaftCluster) GetCoordinator() *coordinator {
	c.RLock()
	defer c.RUnlock()
	return c.coordinator
}
func (c *RaftCluster) GetRegionSyncer() *syncer.RegionSyncer {
	c.RLock()
	defer c.RUnlock()
	return c.regionSyncer
}
func (c *RaftCluster) GetReplicationMode() *replication.ModeManager {
	c.RLock()
	defer c.RUnlock()
	return c.replicationMode
}
func (c *RaftCluster) GetStorage() *core.Storage {
	c.RLock()
	defer c.RUnlock()
	return c.storage
}
func (c *RaftCluster) SetStorage(s *core.Storage) {
	c.Lock()
	defer c.Unlock()
	c.storage = s
}
func (c *RaftCluster) GetOpts() *config.PersistOptions {
	return c.opt
}
func (c *RaftCluster) AddSuspectRegions(regionIDs ...uint64) {
	c.Lock()
	defer c.Unlock()
	for _, regionID := range regionIDs {
		c.suspectRegions.Put(regionID, nil)
	}
}
func (c *RaftCluster) GetSuspectRegions() []uint64 {
	c.RLock()
	defer c.RUnlock()
	return c.suspectRegions.GetAllID()
}
func (c *RaftCluster) RemoveSuspectRegion(id uint64) {
	c.Lock()
	defer c.Unlock()
	c.suspectRegions.Remove(id)
}
func (c *RaftCluster) AddSuspectKeyRange(start, end []byte) {
	c.Lock()
	defer c.Unlock()
	c.suspectKeyRanges.Put(keyutil.BuildKeyRangeKey(start, end), [2][]byte{start, end})
}
func (c *RaftCluster) PopOneSuspectKeyRange() ([2][]byte, bool) {
	c.Lock()
	defer c.Unlock()
	_, value, success := c.suspectKeyRanges.Pop()
	if !success {
		return [2][]byte{}, false
	}
	v, ok := value.([2][]byte)
	if !ok {
		return [2][]byte{}, false
	}
	return v, true
}
func (c *RaftCluster) ClearSuspectKeyRanges() {
	c.Lock()
	defer c.Unlock()
	c.suspectKeyRanges.Clear()
}
func (c *RaftCluster) HandleStoreHeartbeat(stats *pdpb.StoreStats) error {
	c.Lock()
	defer c.Unlock()
	storeID := stats.GetStoreId()
	store := c.GetStore(storeID)
	if store == nil {
		return errors.Errorf("store %v not found", storeID)
	}
	newStore := store.Clone(core.SetStoreStats(stats), core.SetLastHeartbeatTS(time.Now()))
	if newStore.IsLowSpace(c.opt.GetLowSpaceRatio()) {
		log.Warn("store does not have enough disk space",
			zap.Uint64("store-id", newStore.GetID()),
			zap.Uint64("capacity", newStore.GetCapacity()),
			zap.Uint64("available", newStore.GetAvailable()))
	}
	if newStore.NeedPersist() && c.storage != nil {
		if err := c.storage.SaveStore(newStore.GetMeta()); err != nil {
			log.Error("failed to persist store", zap.Uint64("store-id", newStore.GetID()), errs.ZapError(err))
		} else {
			newStore = newStore.Clone(core.SetLastPersistTime(time.Now()))
		}
	}
	if store := c.core.GetStore(newStore.GetID()); store != nil {
		statistics.UpdateStoreHeartbeatMetrics(store)
	}
	c.core.PutStore(newStore)
	c.hotStat.Observe(newStore.GetID(), newStore.GetStoreStats())
	c.hotStat.FilterUnhealthyStore(c)
	reportInterval := stats.GetInterval()
	interval := reportInterval.GetEndTimestamp() - reportInterval.GetStartTimestamp()
	if c.limiter != nil && c.opt.GetStoreLimitMode() == "auto" {
		c.limiter.Collect(newStore.GetStoreStats())
	}
	regionIDs := make(map[uint64]struct{}, len(stats.GetPeerStats()))
	for _, peerStat := range stats.GetPeerStats() {
		regionID := peerStat.GetRegionId()
		regionIDs[regionID] = struct{}{}
		region := c.GetRegion(regionID)
		if region == nil {
			log.Warn("discard hot peer stat for unknown region",
				zap.Uint64("region-id", regionID),
				zap.Uint64("store-id", storeID))
			continue
		}
		peer := region.GetStorePeer(storeID)
		if peer == nil {
			log.Warn("discard hot peer stat for unknown region peer",
				zap.Uint64("region-id", regionID),
				zap.Uint64("store-id", storeID))
			continue
		}
		readQueryNum := core.GetReadQueryNum(peerStat.GetQueryStats())
		loads := []float64{
			statistics.RegionReadBytes:  float64(peerStat.GetReadBytes()),
			statistics.RegionReadKeys:   float64(peerStat.GetReadKeys()),
			statistics.RegionReadQuery:  float64(readQueryNum),
			statistics.RegionWriteBytes: 0,
			statistics.RegionWriteKeys:  0,
			statistics.RegionWriteQuery: 0,
		}
		peerInfo := core.NewPeerInfo(peer, loads, interval)
		c.hotStat.CheckReadAsync(statistics.NewCheckPeerTask(peerInfo, region))
	}
	c.hotStat.CheckReadAsync(statistics.NewCollectUnReportedPeerTask(storeID, regionIDs, interval))
	return nil
}
var regionGuide = core.GenerateRegionGuideFunc(true)
func (c *RaftCluster) processRegionHeartbeat(region *core.RegionInfo) error {
	c.RLock()
	storage := c.storage
	coreCluster := c.core
	hotStat := c.hotStat
	c.RUnlock()
	origin, err := coreCluster.PreCheckPutRegion(region)
	if err != nil {
		return err
	}
	hotStat.CheckWriteAsync(statistics.NewCheckExpiredItemTask(region))
	hotStat.CheckReadAsync(statistics.NewCheckExpiredItemTask(region))
	reportInterval := region.GetInterval()
	interval := reportInterval.GetEndTimestamp() - reportInterval.GetStartTimestamp()
	for _, peer := range region.GetPeers() {
		peerInfo := core.NewPeerInfo(peer, region.GetWriteLoads(), interval)
		hotStat.CheckWriteAsync(statistics.NewCheckPeerTask(peerInfo, region))
	}
	isNew, saveKV, saveCache, needSync := regionGuide(region, origin)
	if !saveKV && !saveCache && !isNew {
		return nil
	}
	failpoint.Inject("concurrentRegionHeartbeat", func() {
		time.Sleep(500 * time.Millisecond)
	})
	var overlaps []*core.RegionInfo
	c.Lock()
	if saveCache {
		if _, err := c.core.PreCheckPutRegion(region); err != nil {
			c.Unlock()
			return err
		}
		overlaps = c.core.PutRegion(region)
		for _, item := range overlaps {
			if c.regionStats != nil {
				c.regionStats.ClearDefunctRegion(item.GetID())
			}
			c.labelLevelStats.ClearDefunctRegion(item.GetID())
		}
		storeMap := make(map[uint64]struct{})
		for _, p := range region.GetPeers() {
			storeMap[p.GetStoreId()] = struct{}{}
		}
		if origin != nil {
			for _, p := range origin.GetPeers() {
				storeMap[p.GetStoreId()] = struct{}{}
			}
		}
		for key := range storeMap {
			c.updateStoreStatusLocked(key)
		}
		regionEventCounter.WithLabelValues("update_cache").Inc()
	}
	if isNew {
		c.prepareChecker.collect(region)
	}
	if c.regionStats != nil {
		c.regionStats.Observe(region, c.getRegionStoresLocked(region))
	}
	changedRegions := c.changedRegions
	c.Unlock()
	if storage != nil {
		for _, item := range overlaps {
			if err := storage.DeleteRegion(item.GetMeta()); err != nil {
				log.Error("failed to delete region from storage",
					zap.Uint64("region-id", item.GetID()),
					logutil.ZapRedactStringer("region-meta", core.RegionToHexMeta(item.GetMeta())),
					errs.ZapError(err))
			}
		}
		if saveKV {
			if err := storage.SaveRegion(region.GetMeta()); err != nil {
				log.Error("failed to save region to storage",
					zap.Uint64("region-id", region.GetID()),
					logutil.ZapRedactStringer("region-meta", core.RegionToHexMeta(region.GetMeta())),
					errs.ZapError(err))
			}
			regionEventCounter.WithLabelValues("update_kv").Inc()
		}
	}
	if saveKV || needSync {
		select {
		case changedRegions <- region:
		default:
		}
	}
	return nil
}
func (c *RaftCluster) updateStoreStatusLocked(id uint64) {
	leaderCount := c.core.GetStoreLeaderCount(id)
	regionCount := c.core.GetStoreRegionCount(id)
	pendingPeerCount := c.core.GetStorePendingPeerCount(id)
	leaderRegionSize := c.core.GetStoreLeaderRegionSize(id)
	regionSize := c.core.GetStoreRegionSize(id)
	c.core.UpdateStoreStatus(id, leaderCount, regionCount, pendingPeerCount, leaderRegionSize, regionSize)
}
func (c *RaftCluster) getClusterID() uint64 {
	c.RLock()
	defer c.RUnlock()
	return c.meta.GetId()
}
func (c *RaftCluster) putMetaLocked(meta *metapb.Cluster) error {
	if c.storage != nil {
		if err := c.storage.SaveMeta(meta); err != nil {
			return err
		}
	}
	c.meta = meta
	return nil
}
func (c *RaftCluster) GetRegionByKey(regionKey []byte) *core.RegionInfo {
	return c.core.SearchRegion(regionKey)
}
func (c *RaftCluster) GetPrevRegionByKey(regionKey []byte) *core.RegionInfo {
	return c.core.SearchPrevRegion(regionKey)
}
func (c *RaftCluster) ScanRegions(startKey, endKey []byte, limit int) []*core.RegionInfo {
	return c.core.ScanRange(startKey, endKey, limit)
}
func (c *RaftCluster) GetRegion(regionID uint64) *core.RegionInfo {
	return c.core.GetRegion(regionID)
}
func (c *RaftCluster) GetMetaRegions() []*metapb.Region {
	return c.core.GetMetaRegions()
}
func (c *RaftCluster) GetRegions() []*core.RegionInfo {
	return c.core.GetRegions()
}
func (c *RaftCluster) GetRegionCount() int {
	return c.core.GetRegionCount()
}
func (c *RaftCluster) GetStoreRegions(storeID uint64) []*core.RegionInfo {
	return c.core.GetStoreRegions(storeID)
}
func (c *RaftCluster) RandLeaderRegion(storeID uint64, ranges []core.KeyRange, opts ...core.RegionOption) *core.RegionInfo {
	return c.core.RandLeaderRegion(storeID, ranges, opts...)
}
func (c *RaftCluster) RandFollowerRegion(storeID uint64, ranges []core.KeyRange, opts ...core.RegionOption) *core.RegionInfo {
	return c.core.RandFollowerRegion(storeID, ranges, opts...)
}
func (c *RaftCluster) RandPendingRegion(storeID uint64, ranges []core.KeyRange, opts ...core.RegionOption) *core.RegionInfo {
	return c.core.RandPendingRegion(storeID, ranges, opts...)
}
func (c *RaftCluster) RandLearnerRegion(storeID uint64, ranges []core.KeyRange, opts ...core.RegionOption) *core.RegionInfo {
	return c.core.RandLearnerRegion(storeID, ranges, opts...)
}
func (c *RaftCluster) GetLeaderStore(region *core.RegionInfo) *core.StoreInfo {
	return c.core.GetLeaderStore(region)
}
func (c *RaftCluster) GetFollowerStores(region *core.RegionInfo) []*core.StoreInfo {
	return c.core.GetFollowerStores(region)
}
func (c *RaftCluster) GetRegionStores(region *core.RegionInfo) []*core.StoreInfo {
	return c.core.GetRegionStores(region)
}
func (c *RaftCluster) GetStoreCount() int {
	return c.core.GetStoreCount()
}
func (c *RaftCluster) GetStoreRegionCount(storeID uint64) int {
	return c.core.GetStoreRegionCount(storeID)
}
func (c *RaftCluster) GetAverageRegionSize() int64 {
	return c.core.GetAverageRegionSize()
}
func (c *RaftCluster) GetRegionStats(startKey, endKey []byte) *statistics.RegionStats {
	c.RLock()
	defer c.RUnlock()
	return statistics.GetRegionStats(c.core.ScanRange(startKey, endKey, -1))
}
func (c *RaftCluster) GetStoresStats() *statistics.StoresStats {
	c.RLock()
	defer c.RUnlock()
	return c.hotStat.StoresStats
}