func (c *Cluster) GiveUpLeader(regionID uint64) {
	c.ChangeLeader(regionID, 0)
}
GiveUpLeader sets the Region's leader to 0. The Region will have no leader
before calling ChangeLeader().
package mocktikv
import (
	"bytes"
	"context"
	"math"
	"sort"
	"sync"
	"time"
	"github.com/golang/protobuf/proto"
	"github.com/pingcap/kvproto/pkg/kvrpcpb"
	"github.com/pingcap/kvproto/pkg/metapb"
	pd "github.com/tikv/pd/client"
)
type Cluster struct {
	sync.RWMutex
	id      uint64
	stores  map[uint64]*Store
	regions map[uint64]*Region
	mvccStore MVCCStore
	delayEvents map[delayKey]time.Duration
	delayMu     sync.Mutex
}
type delayKey struct {
	startTS  uint64
	regionID uint64
}
func NewCluster(mvccStore MVCCStore) *Cluster {
	return &Cluster{
		stores:      make(map[uint64]*Store),
		regions:     make(map[uint64]*Region),
		delayEvents: make(map[delayKey]time.Duration),
		mvccStore:   mvccStore,
	}
}
func (c *Cluster) AllocID() uint64 {
	c.Lock()
	defer c.Unlock()
	return c.allocID()
}
func (c *Cluster) AllocIDs(n int) []uint64 {
	c.Lock()
	defer c.Unlock()
	var ids []uint64
	for len(ids) < n {
		ids = append(ids, c.allocID())
	}
	return ids
}
func (c *Cluster) allocID() uint64 {
	c.id++
	return c.id
}
func (c *Cluster) GetAllRegions() []*Region {
	regions := make([]*Region, 0, len(c.regions))
	for _, region := range c.regions {
		regions = append(regions, region)
	}
	return regions
}
func (c *Cluster) GetStore(storeID uint64) *metapb.Store {
	c.RLock()
	defer c.RUnlock()
	if store := c.stores[storeID]; store != nil {
		return proto.Clone(store.meta).(*metapb.Store)
	}
	return nil
}
func (c *Cluster) GetAllStores() []*metapb.Store {
	c.RLock()
	defer c.RUnlock()
	stores := make([]*metapb.Store, 0, len(c.stores))
	for _, store := range c.stores {
		stores = append(stores, proto.Clone(store.meta).(*metapb.Store))
	}
	return stores
}
func (c *Cluster) StopStore(storeID uint64) {
	c.Lock()
	defer c.Unlock()
	if store := c.stores[storeID]; store != nil {
		store.meta.State = metapb.StoreState_Offline
	}
}
func (c *Cluster) StartStore(storeID uint64) {
	c.Lock()
	defer c.Unlock()
	if store := c.stores[storeID]; store != nil {
		store.meta.State = metapb.StoreState_Up
	}
}
func (c *Cluster) CancelStore(storeID uint64) {
	c.Lock()
	defer c.Unlock()
	if store := c.stores[storeID]; store != nil {
		store.cancel = true
	}
}
func (c *Cluster) UnCancelStore(storeID uint64) {
	c.Lock()
	defer c.Unlock()
	if store := c.stores[storeID]; store != nil {
		store.cancel = false
	}
}
func (c *Cluster) GetStoreByAddr(addr string) *metapb.Store {
	c.RLock()
	defer c.RUnlock()
	for _, s := range c.stores {
		if s.meta.GetAddress() == addr {
			return proto.Clone(s.meta).(*metapb.Store)
		}
	}
	return nil
}
func (c *Cluster) GetAndCheckStoreByAddr(addr string) (ss []*metapb.Store, err error) {
	c.RLock()
	defer c.RUnlock()
	for _, s := range c.stores {
		if s.cancel {
			err = context.Canceled
			return
		}
		if s.meta.GetAddress() == addr {
			ss = append(ss, proto.Clone(s.meta).(*metapb.Store))
		}
	}
	return
}
func (c *Cluster) AddStore(storeID uint64, addr string, labels ...*metapb.StoreLabel) {
	c.Lock()
	defer c.Unlock()
	c.stores[storeID] = newStore(storeID, addr, labels...)
}
func (c *Cluster) RemoveStore(storeID uint64) {
	c.Lock()
	defer c.Unlock()
	delete(c.stores, storeID)
}
func (c *Cluster) MarkTombstone(storeID uint64) {
	c.Lock()
	defer c.Unlock()
	nm := *c.stores[storeID].meta
	nm.State = metapb.StoreState_Tombstone
	c.stores[storeID].meta = &nm
}
func (c *Cluster) UpdateStoreAddr(storeID uint64, addr string, labels ...*metapb.StoreLabel) {
	c.Lock()
	defer c.Unlock()
	c.stores[storeID] = newStore(storeID, addr, labels...)
}
func (c *Cluster) GetRegion(regionID uint64) (*metapb.Region, uint64) {
	c.RLock()
	defer c.RUnlock()
	r := c.regions[regionID]
	if r == nil {
		return nil, 0
	}
	return proto.Clone(r.Meta).(*metapb.Region), r.leader
}
func (c *Cluster) GetRegionByKey(key []byte) (*metapb.Region, *metapb.Peer) {
	c.RLock()
	defer c.RUnlock()
	return c.getRegionByKeyNoLock(key)
}
func (c *Cluster) getRegionByKeyNoLock(key []byte) (*metapb.Region, *metapb.Peer) {
	for _, r := range c.regions {
		if regionContains(r.Meta.StartKey, r.Meta.EndKey, key) {
			return proto.Clone(r.Meta).(*metapb.Region), proto.Clone(r.leaderPeer()).(*metapb.Peer)
		}
	}
	return nil, nil
}
func (c *Cluster) GetPrevRegionByKey(key []byte) (*metapb.Region, *metapb.Peer) {
	c.RLock()
	defer c.RUnlock()
	currentRegion, _ := c.getRegionByKeyNoLock(key)
	if len(currentRegion.StartKey) == 0 {
		return nil, nil
	}
	for _, r := range c.regions {
		if bytes.Equal(r.Meta.EndKey, currentRegion.StartKey) {
			return proto.Clone(r.Meta).(*metapb.Region), proto.Clone(r.leaderPeer()).(*metapb.Peer)
		}
	}
	return nil, nil
}
func (c *Cluster) GetRegionByID(regionID uint64) (*metapb.Region, *metapb.Peer) {
	c.RLock()
	defer c.RUnlock()
	for _, r := range c.regions {
		if r.Meta.GetId() == regionID {
			return proto.Clone(r.Meta).(*metapb.Region), proto.Clone(r.leaderPeer()).(*metapb.Peer)
		}
	}
	return nil, nil
}
func (c *Cluster) ScanRegions(startKey, endKey []byte, limit int) []*pd.Region {
	c.RLock()
	defer c.RUnlock()
	regions := make([]*Region, 0, len(c.regions))
	for _, region := range c.regions {
		regions = append(regions, region)
	}
	sort.Slice(regions, func(i, j int) bool {
		return bytes.Compare(regions[i].Meta.GetStartKey(), regions[j].Meta.GetStartKey()) < 0
	})
	startPos := sort.Search(len(regions), func(i int) bool {
		if len(regions[i].Meta.GetEndKey()) == 0 {
			return true
		}
		return bytes.Compare(regions[i].Meta.GetEndKey(), startKey) > 0
	})
	regions = regions[startPos:]
	if len(endKey) > 0 {
		endPos := sort.Search(len(regions), func(i int) bool {
			return bytes.Compare(regions[i].Meta.GetStartKey(), endKey) >= 0
		})
		if endPos > 0 {
			regions = regions[:endPos]
		}
	}
	if limit > 0 && len(regions) > limit {
		regions = regions[:limit]
	}
	result := make([]*pd.Region, 0, len(regions))
	for _, region := range regions {
		leader := region.leaderPeer()
		if leader == nil {
			leader = &metapb.Peer{}
		} else {
			leader = proto.Clone(leader).(*metapb.Peer)
		}
		r := &pd.Region{
			Meta:   proto.Clone(region.Meta).(*metapb.Region),
			Leader: leader,
		}
		result = append(result, r)
	}
	return result
}
func (c *Cluster) Bootstrap(regionID uint64, storeIDs, peerIDs []uint64, leaderPeerID uint64) {
	c.Lock()
	defer c.Unlock()
	if len(storeIDs) != len(peerIDs) {
		panic("len(storeIDs) != len(peerIDs)")
	}
	c.regions[regionID] = newRegion(regionID, storeIDs, peerIDs, leaderPeerID)
}
func (c *Cluster) AddPeer(regionID, storeID, peerID uint64) {
	c.Lock()
	defer c.Unlock()
	c.regions[regionID].addPeer(peerID, storeID)
}
func (c *Cluster) RemovePeer(regionID, storeID uint64) {
	c.Lock()
	defer c.Unlock()
	c.regions[regionID].removePeer(storeID)
}
func (c *Cluster) ChangeLeader(regionID, leaderPeerID uint64) {
	c.Lock()
	defer c.Unlock()
	c.regions[regionID].changeLeader(leaderPeerID)
}