raftstore

package
v0.0.0-...-b180c9c Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 10, 2020 License: Apache-2.0 Imports: 52 Imported by: 0

Documentation ¶

Index ¶

Constants ¶

View Source 
const (
	DefaultApplyWBSize = 4 * 1024

	WriteTypeFlagPut      = 'P'
	WriteTypeFlagDelete   = 'D'
	WriteTypeFlagLock     = 'L'
	WriteTypeFlagRollback = 'R'
)
View Source 
const (
	InitEpochVer     uint64 = 1
	InitEpochConfVer uint64 = 1
	KvTS             uint64 = 1
	RaftTS           uint64 = 0
)
View Source 
const (
	KB          uint64 = 1024
	MB          uint64 = 1024 * 1024
	SplitSizeMb uint64 = 96
)
View Source 
const (
	LocalPrefix byte = 0x01

	// We save two types region data in DB, for raft and other meta data.
	// When the store starts, we should iterate all region meta data to
	// construct peer, no need to travel large raft data, so we separate them
	// with different prefixes.
	RegionRaftPrefix byte = 0x02
	RegionMetaPrefix byte = 0x03
	RegionRaftLogLen      = 19 // REGION_RAFT_PREFIX_KEY + region_id + suffix + index

	// Following are the suffix after the local prefix.
	// For region id
	RaftLogSuffix           byte = 0x01
	RaftStateSuffix         byte = 0x02
	ApplyStateSuffix        byte = 0x03
	SnapshotRaftStateSuffix byte = 0x04

	// For region meta
	RegionStateSuffix byte = 0x01
)
View Source 
const (
	MaxCheckClusterBootstrappedRetryCount = 60
	CheckClusterBootstrapRetrySeconds     = 3
)
View Source 
const (
	// When we create a region peer, we should initialize its log term/index > 0,
	// so that we can force the follower peer to sync the snapshot first.
	RaftInitLogTerm  = 5
	RaftInitLogIndex = 5

	MaxSnapRetryCnt = 5

	MaxCacheCapacity = 1024 - 1
)
View Source 
const (
	NSEC_PER_MSEC uint64 = 1000000
	SEC_SHIFT     uint64 = 10
	MSEC_MASK     uint64 = (1 << SEC_SHIFT) - 1
)
View Source 
const InvalidID uint64 = 0
View Source 
const LockstoreFileName = "lockstore.dump"
View Source 
const MaxDeleteBatchSize int = 32 * 1024

In our tests, we found that if the batch size is too large, running deleteAllInRange will reduce OLTP QPS by 30% ~ 60%. We found that 32K is a proper choice.

View Source 
const RaftInvalidIndex uint64 = 0

Variables ¶

View Source 
var (
	MinKey = []byte{0}

	// Data key has two prefix, meta 'm' and table 't',
	// extra keys has prefix 'm' + 1 = 'n',
	// extra table keys has prefix 't' + 1 = 'u', end key would be 'v'.
	MinDataKey = []byte{'m'}
	MaxDataKey = []byte{'v'}

	RegionMetaMinKey = []byte{LocalPrefix, RegionMetaPrefix}
	RegionMetaMaxKey = []byte{LocalPrefix, RegionMetaPrefix + 1}
)

Functions ¶

func ApplyStateKey ¶ 

func ApplyStateKey(regionID uint64) []byte

func BindRespError ¶ 

func BindRespError(resp *raft_cmdpb.RaftCmdResponse, err error)

func BindRespTerm ¶ 

func BindRespTerm(resp *raft_cmdpb.RaftCmdResponse, term uint64)

func BootstrapStore ¶ 

func BootstrapStore(engines *Engines, clussterID, storeID uint64) error

func CheckKeyInRegion ¶ 

func CheckKeyInRegion(key []byte, region *metapb.Region) error

/ Check if key in region range [`start_key`, `end_key`).

func CheckKeyInRegionExclusive ¶ 

func CheckKeyInRegionExclusive(key []byte, region *metapb.Region) error

/ Check if key in region range (`start_key`, `end_key`).

func CheckKeyInRegionInclusive ¶ 

func CheckKeyInRegionInclusive(key []byte, region *metapb.Region) error

/ Check if key in region range [`start_key`, `end_key`].

func CheckRegionEpoch ¶ 

func CheckRegionEpoch(req *raft_cmdpb.RaftCmdRequest, region *metapb.Region, includeRegion bool) error

func ClearMeta ¶ 

func ClearMeta(engines *Engines, kvWB, raftWB *WriteBatch, regionID uint64, lastIndex uint64) error

func ClearPrepareBootstrap ¶ 

func ClearPrepareBootstrap(engines *Engines, regionID uint64) error

func ClearPrepareBootstrapState ¶ 

func ClearPrepareBootstrapState(engines *Engines) error

func CloneMsg ¶ 

func CloneMsg(origin, cloned proto.Message) error

func CompactRaftLog ¶ 

func CompactRaftLog(tag string, state *applyState, compactIndex, compactTerm uint64) error

CompactRaftLog discards all log entries prior to compact_index. We must guarantee that the compact_index is not greater than applied index.

func CreateRaftLogCompactionFilter ¶ 

func CreateRaftLogCompactionFilter(targetLevel int, startKey, endKey []byte) badger.CompactionFilter

func ErrResp ¶ 

func ErrResp(err error) *raft_cmdpb.RaftCmdResponse

func ErrRespRegionNotFound ¶ 

func ErrRespRegionNotFound(regionID uint64) *raft_cmdpb.RaftCmdResponse

func ErrRespStaleCommand ¶ 

func ErrRespStaleCommand(term uint64) *raft_cmdpb.RaftCmdResponse

func ErrRespWithTerm ¶ 

func ErrRespWithTerm(err error, term uint64) *raft_cmdpb.RaftCmdResponse

func GetChangePeerCmd ¶ 

func GetChangePeerCmd(msg *raft_cmdpb.RaftCmdRequest) *raft_cmdpb.ChangePeerRequest

func IsEpochStale ¶ 

func IsEpochStale(epoch *metapb.RegionEpoch, checkEpoch *metapb.RegionEpoch) bool

/ check whether epoch is staler than check_epoch.

func IsUrgentRequest ¶ 

func IsUrgentRequest(rlog raftlog.RaftLog) bool

/ We enable follower lazy commit to get a better performance. / But it may not be appropriate for some requests. This function / checks whether the request should be committed on all followers / as soon as possible.

func NewCustomWriteBatch ¶ 

func NewCustomWriteBatch(startTS, commitTS uint64, ctx *kvrpcpb.Context) mvcc.WriteBatch

func NewDBWriter ¶ 

func NewDBWriter(conf *config.Config, router *RaftstoreRouter) mvcc.DBWriter

func NewTestRaftWriter ¶ 

func NewTestRaftWriter(dbBundle *mvcc.DBBundle, engine *Engines) mvcc.DBWriter

func NotifyReqRegionRemoved ¶ 

func NotifyReqRegionRemoved(regionId uint64, cb *Callback)

func NotifyStaleReq ¶ 

func NotifyStaleReq(term uint64, cb *Callback)

func PeerEqual ¶ 

func PeerEqual(l, r *metapb.Peer) bool

func PrepareBootstrap ¶ 

func PrepareBootstrap(engins *Engines, storeID, regionID, peerID uint64) (*metapb.Region, error)

func Quorum ¶ 

func Quorum(total int) int

func RaftLogIndex ¶ 

func RaftLogIndex(key []byte) (uint64, error)

/ RaftLogIndex gets the log index from raft log key generated by `raft_log_key`.

func RaftLogKey ¶ 

func RaftLogKey(regionID, index uint64) []byte

func RaftStateKey ¶ 

func RaftStateKey(regionID uint64) []byte

func RaftstoreErrToPbError ¶ 

func RaftstoreErrToPbError(e error) *errorpb.Error

func RawEndKey ¶ 

func RawEndKey(region *metapb.Region) []byte

Get the `end_key` of current region in encoded form.

func RawStartKey ¶ 

func RawStartKey(region *metapb.Region) []byte

Get the `start_key` of current region in encoded form.

func RegionEqual ¶ 

func RegionEqual(l, r *metapb.Region) bool

func RegionMetaPrefixKey ¶ 

func RegionMetaPrefixKey(regionID uint64) []byte

func RegionRaftPrefixKey ¶ 

func RegionRaftPrefixKey(regionID uint64) []byte

func RegionStateKey ¶ 

func RegionStateKey(regionID uint64) []byte

func RestoreLockStore ¶ 

func RestoreLockStore(offset uint64, bundle *mvcc.DBBundle, raftDB *badger.DB) error

func SnapshotRaftStateKey ¶ 

func SnapshotRaftStateKey(regionID uint64) []byte

func TimeToU64 ¶ 

func TimeToU64(t time.Time) uint64

func U64ToTime ¶ 

func U64ToTime(u uint64) time.Time

func WritePeerState ¶ 

func WritePeerState(kvWB *WriteBatch, region *metapb.Region, state rspb.PeerState, mergeState *rspb.MergeState)

Types ¶

type ApplyOptions ¶ 

type ApplyOptions struct {
	DBBundle *mvcc.DBBundle
	Region   *metapb.Region
	Abort    *uint32
	Builder  *sstable.Builder
	WB       *WriteBatch
}

type ApplyResult ¶ 

type ApplyResult struct {
	HasPut      bool
	RegionState *rspb.RegionLocalState
}

type ApplySnapResult ¶ 

type ApplySnapResult struct {
	// PrevRegion is the region before snapshot applied
	PrevRegion *metapb.Region
	Region     *metapb.Region
}

type CFFile ¶ 

type CFFile struct {
	CF          CFName
	Path        string
	TmpPath     string
	ClonePath   string
	SstWriter   *rocksdb.SstFileWriter
	File        *os.File
	KVCount     int
	Size        uint64
	WrittenSize uint64
	Checksum    uint32
	WriteDigest hash.Hash32
}

type CFName ¶ 

type CFName = string
const (
	CFDefault CFName = "default"
	CFLock    CFName = "lock"
	CFWrite   CFName = "write"
	CFRaft    CFName = "raft"
)

type CacheQueryStats ¶ 

type CacheQueryStats struct {
	// contains filtered or unexported fields
}

type Callback ¶ 

type Callback struct {
	// contains filtered or unexported fields
}

func NewCallback ¶ 

func NewCallback() *Callback

func (*Callback) Done ¶ 

func (cb *Callback) Done(resp *raft_cmdpb.RaftCmdResponse)

type Config ¶ 

type Config struct {
	// true for high reliability, prevent data loss when power failure.
	SyncLog bool
	// minimizes disruption when a partitioned node rejoins the cluster by using a two phase election.
	Prevote    bool
	RaftdbPath string

	SnapPath string

	// store capacity. 0 means no limit.
	Capacity uint64

	// raft_base_tick_interval is a base tick interval (ms).
	RaftBaseTickInterval        time.Duration
	RaftHeartbeatTicks          int
	RaftElectionTimeoutTicks    int
	RaftMinElectionTimeoutTicks int
	RaftMaxElectionTimeoutTicks int
	RaftMaxSizePerMsg           uint64
	RaftMaxInflightMsgs         int

	// When the entry exceed the max size, reject to propose it.
	RaftEntryMaxSize uint64

	// Interval to gc unnecessary raft log (ms).
	RaftLogGCTickInterval time.Duration
	// A threshold to gc stale raft log, must >= 1.
	RaftLogGcThreshold uint64
	// When entry count exceed this value, gc will be forced trigger.
	RaftLogGcCountLimit uint64
	// When the approximate size of raft log entries exceed this value,
	// gc will be forced trigger.
	RaftLogGcSizeLimit uint64
	// When a peer is not responding for this time, leader will not keep entry cache for it.
	RaftEntryCacheLifeTime time.Duration
	// When a peer is newly added, reject transferring leader to the peer for a while.
	RaftRejectTransferLeaderDuration time.Duration

	// Interval (ms) to check region whether need to be split or not.
	SplitRegionCheckTickInterval time.Duration
	/// When size change of region exceed the diff since last check, it
	/// will be checked again whether it should be split.
	RegionSplitCheckDiff uint64
	/// Interval (ms) to check whether start compaction for a region.
	RegionCompactCheckInterval time.Duration
	// delay time before deleting a stale peer
	CleanStalePeerDelay time.Duration
	/// Number of regions for each time checking.
	RegionCompactCheckStep uint64
	/// Minimum number of tombstones to trigger manual compaction.
	RegionCompactMinTombstones uint64
	/// Minimum percentage of tombstones to trigger manual compaction.
	/// Should between 1 and 100.
	RegionCompactTombstonesPencent uint64
	PdHeartbeatTickInterval        time.Duration
	PdStoreHeartbeatTickInterval   time.Duration
	SnapMgrGcTickInterval          time.Duration
	SnapGcTimeout                  time.Duration

	NotifyCapacity  uint64
	MessagesPerTick uint64

	/// When a peer is not active for max_peer_down_duration,
	/// the peer is considered to be down and is reported to PD.
	MaxPeerDownDuration time.Duration

	/// If the leader of a peer is missing for longer than max_leader_missing_duration,
	/// the peer would ask pd to confirm whether it is valid in any region.
	/// If the peer is stale and is not valid in any region, it will destroy itself.
	MaxLeaderMissingDuration time.Duration
	/// Similar to the max_leader_missing_duration, instead it will log warnings and
	/// try to alert monitoring systems, if there is any.
	AbnormalLeaderMissingDuration time.Duration
	PeerStaleStateCheckInterval   time.Duration

	LeaderTransferMaxLogLag uint64

	SnapApplyBatchSize uint64

	// Interval (ms) to check region whether the data is consistent.
	ConsistencyCheckInterval time.Duration

	ReportRegionFlowInterval time.Duration

	// The lease provided by a successfully proposed and applied entry.
	RaftStoreMaxLeaderLease time.Duration

	// Right region derive origin region id when split.
	RightDeriveWhenSplit bool

	AllowRemoveLeader bool

	/// Max log gap allowed to propose merge.
	MergeMaxLogGap uint64

	/// Interval to re-propose merge.
	MergeCheckTickInterval time.Duration

	UseDeleteRange bool

	ApplyMaxBatchSize uint64
	ApplyPoolSize     uint64

	StoreMaxBatchSize uint64

	ConcurrentSendSnapLimit uint64
	ConcurrentRecvSnapLimit uint64

	GrpcInitialWindowSize uint64
	GrpcKeepAliveTime     time.Duration
	GrpcKeepAliveTimeout  time.Duration
	GrpcRaftConnNum       uint64

	Addr          string
	AdvertiseAddr string
	Labels        []StoreLabel

	SplitCheck *splitCheckConfig
}

func NewDefaultConfig ¶ 

func NewDefaultConfig() *Config

func (*Config) Validate ¶ 

func (c *Config) Validate() error

type ConsistencyState ¶ 

type ConsistencyState struct {
	LastCheckTime time.Time
	// (computed_result_or_to_be_verified, index, hash)
	Index uint64
	Hash  []byte
}

/ `ConsistencyState` is used for consistency check.

type DestroyPeerJob ¶ 

type DestroyPeerJob struct {
	Initialized bool
	AsyncRemove bool
	RegionId    uint64
	Peer        *metapb.Peer
}

type Engines ¶ 

type Engines struct {
	// contains filtered or unexported fields
}

func NewEngines ¶ 

func NewEngines(kvEngine *mvcc.DBBundle, raftEngine *badger.DB, kvPath, raftPath string) *Engines

func (*Engines) SyncKVWAL ¶ 

func (en *Engines) SyncKVWAL() error

func (*Engines) SyncRaftWAL ¶ 

func (en *Engines) SyncRaftWAL() error

func (*Engines) WriteKV ¶ 

func (en *Engines) WriteKV(wb *WriteBatch) error

func (*Engines) WriteRaft ¶ 

func (en *Engines) WriteRaft(wb *WriteBatch) error

type EntryCache ¶ 

type EntryCache struct {
	// contains filtered or unexported fields
}

type ErrEpochNotMatch ¶ 

type ErrEpochNotMatch struct {
	Message string
	Regions []*metapb.Region
}

func (*ErrEpochNotMatch) Error ¶ 

func (e *ErrEpochNotMatch) Error() string

type ErrKeyNotInRegion ¶ 

type ErrKeyNotInRegion struct {
	Key    []byte
	Region *metapb.Region
}

func (*ErrKeyNotInRegion) Error ¶ 

func (e *ErrKeyNotInRegion) Error() string

type ErrNotLeader ¶ 

type ErrNotLeader struct {
	RegionId uint64
	Leader   *metapb.Peer
}

func (*ErrNotLeader) Error ¶ 

func (e *ErrNotLeader) Error() string

type ErrRaftEntryTooLarge ¶ 

type ErrRaftEntryTooLarge struct {
	RegionId  uint64
	EntrySize uint64
}

func (*ErrRaftEntryTooLarge) Error ¶ 

func (e *ErrRaftEntryTooLarge) Error() string

type ErrRegionNotFound ¶ 

type ErrRegionNotFound struct {
	RegionId uint64
}

func (*ErrRegionNotFound) Error ¶ 

func (e *ErrRegionNotFound) Error() string

type ErrServerIsBusy ¶ 

type ErrServerIsBusy struct {
	Reason    string
	BackoffMs uint64
}

func (*ErrServerIsBusy) Error ¶ 

func (e *ErrServerIsBusy) Error() string

type ErrStaleCommand ¶ 

type ErrStaleCommand struct{}

func (*ErrStaleCommand) Error ¶ 

func (e *ErrStaleCommand) Error() string

type ErrStoreNotMatch ¶ 

type ErrStoreNotMatch struct {
	RequestStoreId uint64
	ActualStoreId  uint64
}

func (*ErrStoreNotMatch) Error ¶ 

func (e *ErrStoreNotMatch) Error() string

type GenSnapTask ¶ 

type GenSnapTask struct {
	// contains filtered or unexported fields
}

type GlobalContext ¶ 

type GlobalContext struct {
	// contains filtered or unexported fields
}

type IOLimiter ¶ 

type IOLimiter = rate.Limiter

func NewIOLimiter ¶ 

func NewIOLimiter(rateLimit int) *IOLimiter

func NewInfLimiter ¶ 

func NewInfLimiter() *IOLimiter

type InvokeContext ¶ 

type InvokeContext struct {
	RegionID   uint64
	RaftState  raftState
	ApplyState applyState

	SnapRegion *metapb.Region
	// contains filtered or unexported fields
}

func NewInvokeContext ¶ 

func NewInvokeContext(store *PeerStorage) *InvokeContext

type JobStatus ¶ 

type JobStatus = uint32
const (
	JobStatus_Pending JobStatus = 0 + iota
	JobStatus_Running
	JobStatus_Cancelling
	JobStatus_Cancelled
	JobStatus_Finished
	JobStatus_Failed
)

type LeaderChecker ¶ 

type LeaderChecker interface {
	IsLeader(ctx *kvrpcpb.Context, router *RaftstoreRouter) *errorpb.Error
}

type Lease ¶ 

type Lease struct {
	// contains filtered or unexported fields
}

/ Lease records an expired time, for examining the current moment is in lease or not. / It's dedicated to the Raft leader lease mechanism, contains either state of / 1. Suspect Timestamp / A suspicious leader lease timestamp, which marks the leader may still hold or lose / its lease until the clock time goes over this timestamp. / 2. Valid Timestamp / A valid leader lease timestamp, which marks the leader holds the lease for now. / The lease is valid until the clock time goes over this timestamp. / / ```text / Time / |----------------------------------> / ^ ^ / Now Suspect TS / State: | Suspect | Suspect / / |----------------------------------> / ^ ^ / Now Valid TS / State: | Valid | Expired / ``` / / Note: / - Valid timestamp would increase when raft log entries are applied in current term. / - Suspect timestamp would be set after the message `MsgTimeoutNow` is sent by current peer. / The message `MsgTimeoutNow` starts a leader transfer procedure. During this procedure, / current peer as an old leader may still hold its lease or lose it. / It's possible there is a new leader elected and current peer as an old leader / doesn't step down due to network partition from the new leader. In that case, / current peer lose its leader lease. / Within this suspect leader lease expire time, read requests could not be performed / locally. / - The valid leader lease should be `lease = max_lease - (commit_ts - send_ts)` / And the expired timestamp for that leader lease is `commit_ts + lease`, / which is `send_ts + max_lease` in short.

func NewLease ¶ 

func NewLease(maxLease time.Duration) *Lease

func (*Lease) Expire ¶ 

func (l *Lease) Expire()

func (*Lease) ExpireRemoteLease ¶ 

func (l *Lease) ExpireRemoteLease()

func (*Lease) Inspect ¶ 

func (l *Lease) Inspect(ts *time.Time) LeaseState

/ Inspect the lease state for the ts or now.

func (*Lease) MaybeNewRemoteLease ¶ 

func (l *Lease) MaybeNewRemoteLease(term uint64) *RemoteLease

/ Return a new `RemoteLease` if there is none.

func (*Lease) Renew ¶ 

func (l *Lease) Renew(sendTs time.Time)

/ Renew the lease to the bound.

func (*Lease) Suspect ¶ 

func (l *Lease) Suspect(sendTs time.Time)

/ Suspect the lease to the bound.

type LeaseState ¶ 

type LeaseState int
const (
	/// The lease is suspicious, may be invalid.
	LeaseState_Suspect LeaseState = 1 + iota
	/// The lease is valid.
	LeaseState_Valid
	/// The lease is expired.
	LeaseState_Expired
)

type LimitWriter ¶ 

type LimitWriter struct {
	// contains filtered or unexported fields
}

func (*LimitWriter) Write ¶ 

func (lw *LimitWriter) Write(b []byte) (int, error)

type MetaFile ¶ 

type MetaFile struct {
	Meta *rspb.SnapshotMeta
	Path string
	File *os.File

	// for writing snapshot
	TmpPath string
}

type Msg ¶ 

type Msg struct {
	Type     MsgType
	RegionID uint64
	Data     interface{}
}

func NewMsg ¶ 

func NewMsg(tp MsgType, data interface{}) Msg

func NewPeerMsg ¶ 

func NewPeerMsg(tp MsgType, regionID uint64, data interface{}) Msg

type MsgComputeHashResult ¶ 

type MsgComputeHashResult struct {
	Index uint64
	Hash  []byte
}

type MsgGCSnap ¶ 

type MsgGCSnap struct {
	Snaps []SnapKeyWithSending
}

type MsgHalfSplitRegion ¶ 

type MsgHalfSplitRegion struct {
	RegionEpoch *metapb.RegionEpoch
}

type MsgMergeResult ¶ 

type MsgMergeResult struct {
	TargetPeer *metapb.Peer
	Stale      bool
}

type MsgRaftCmd ¶ 

type MsgRaftCmd struct {
	SendTime time.Time
	Request  raftlog.RaftLog
	Callback *Callback
}

type MsgSignificant ¶ 

type MsgSignificant struct {
	Type           MsgSignificantType
	ToPeerID       uint64
	SnapshotStatus raft.SnapshotStatus
}

type MsgSignificantType ¶ 

type MsgSignificantType int
const (
	MsgSignificantTypeStatus      MsgSignificantType = 1
	MsgSignificantTypeUnreachable MsgSignificantType = 2
)

type MsgSplitRegion ¶ 

type MsgSplitRegion struct {
	RegionEpoch *metapb.RegionEpoch
	// It's an encoded key.
	// TODO: support meta key.
	SplitKeys [][]byte
	Callback  *Callback
}

type MsgStoreClearRegionSizeInRange ¶ 

type MsgStoreClearRegionSizeInRange struct {
	StartKey []byte
	EndKey   []byte
}

type MsgType ¶ 

type MsgType int64
const (
	MsgTypeNull                   MsgType = 0
	MsgTypeRaftMessage            MsgType = 1
	MsgTypeRaftCmd                MsgType = 2
	MsgTypeSplitRegion            MsgType = 3
	MsgTypeComputeResult          MsgType = 4
	MsgTypeRegionApproximateSize  MsgType = 5
	MsgTypeRegionApproximateKeys  MsgType = 6
	MsgTypeCompactionDeclineBytes MsgType = 7
	MsgTypeHalfSplitRegion        MsgType = 8
	MsgTypeMergeResult            MsgType = 9
	MsgTypeGcSnap                 MsgType = 10
	MsgTypeClearRegionSize        MsgType = 11
	MsgTypeTick                   MsgType = 12
	MsgTypeSignificantMsg         MsgType = 13
	MsgTypeStart                  MsgType = 14
	MsgTypeApplyRes               MsgType = 15
	MsgTypeNoop                   MsgType = 16

	MsgTypeStoreRaftMessage   MsgType = 101
	MsgTypeStoreSnapshotStats MsgType = 102
	// Clear region size and keys for all regions in the range, so we can force them to re-calculate
	// their size later.
	MsgTypeStoreClearRegionSizeInRange MsgType = 104
	MsgTypeStoreCompactedEvent         MsgType = 105
	MsgTypeStoreTick                   MsgType = 106
	MsgTypeStoreStart                  MsgType = 107

	MsgTypeFsmNormal  MsgType = 201
	MsgTypeFsmControl MsgType = 202

	MsgTypeApply             MsgType = 301
	MsgTypeApplyRegistration MsgType = 302
	MsgTypeApplyProposal     MsgType = 303
	MsgTypeApplyCatchUpLogs  MsgType = 304
	MsgTypeApplyLogsUpToDate MsgType = 305
	MsgTypeApplyDestroy      MsgType = 306
	MsgTypeApplySnapshot     MsgType = 307
)

type Node ¶ 

type Node struct {
	// contains filtered or unexported fields
}

func NewNode ¶ 

func NewNode(system *raftBatchSystem, store *metapb.Store, cfg *Config, pdClient pd.Client, observer PeerEventObserver) *Node

func (*Node) BootstrapCluster ¶ 

func (n *Node) BootstrapCluster(ctx context.Context, engines *Engines, firstRegion *metapb.Region) (newCluster bool, err error)

func (*Node) Start ¶ 

func (n *Node) Start(ctx context.Context, engines *Engines, trans Transport, snapMgr *SnapManager, pdWorker *worker, router *router) error

type Peer ¶ 

type Peer struct {
	Meta *metapb.Peer

	RaftGroup *raft.RawNode

	// Record the last instant of each peer's heartbeat response.
	PeerHeartbeats map[uint64]time.Time

	/// Record the instants of peers being added into the configuration.
	/// Remove them after they are not pending any more.
	PeersStartPendingTime map[uint64]time.Time
	RecentAddedPeer       *RecentAddedPeer

	/// an inaccurate difference in region size since last reset.
	SizeDiffHint uint64

	/// approximate size of the region.
	ApproximateSize *uint64
	/// approximate keys of the region.
	ApproximateKeys         *uint64
	CompactionDeclinedBytes uint64

	ConsistencyState *ConsistencyState

	Tag string

	// Index of last scheduled committed raft log.
	LastApplyingIdx  uint64
	LastCompactedIdx uint64

	// Approximate size of logs that is applied but not compacted yet.
	RaftLogSizeHint uint64

	PendingRemove bool

	PendingMergeState *rspb.MergeState

	PendingMergeApplyResult *WaitApplyResultState
	PeerStat                PeerStat
	// contains filtered or unexported fields
}

func NewPeer ¶ 

func NewPeer(storeId uint64, cfg *Config, engines *Engines, region *metapb.Region, regionSched chan<- task,
	peer *metapb.Peer) (*Peer, error)

func (*Peer) Activate ¶ 

func (p *Peer) Activate(applyMsgs *applyMsgs)

/ Register self to applyMsgs so that the peer is then usable. / Also trigger `RegionChangeEvent::Create` here.

func (*Peer) AnyNewPeerCatchUp ¶ 

func (p *Peer) AnyNewPeerCatchUp(peerId uint64) bool

/ Returns `true` if any new peer catches up with the leader in replicating logs. / And updates `PeersStartPendingTime` if needed.

func (*Peer) ApplyReads ¶ 

func (p *Peer) ApplyReads(kv *mvcc.DBBundle, ready *raft.Ready)

func (*Peer) CheckPeers ¶ 

func (p *Peer) CheckPeers()

/ Checks and updates `peer_heartbeats` for the peer.

func (*Peer) CheckStaleState ¶ 

func (p *Peer) CheckStaleState(cfg *Config) StaleState

func (*Peer) CollectDownPeers ¶ 

func (p *Peer) CollectDownPeers(maxDuration time.Duration) []*pdpb.PeerStats

/ Collects all down peers.

func (*Peer) CollectPendingPeers ¶ 

func (p *Peer) CollectPendingPeers() []*metapb.Peer

/ Collects all pending peers and update `peers_start_pending_time`.

func (*Peer) Destroy ¶ 

func (p *Peer) Destroy(engine *Engines, keepData bool) error

/ Does the real destroy task which includes: / 1. Set the region to tombstone; / 2. Clear data; / 3. Notify all pending requests.

func (*Peer) GetMinProgress ¶ 

func (p *Peer) GetMinProgress() uint64

func (*Peer) GetRaftStatus ¶ 

func (p *Peer) GetRaftStatus() *raft.Status

func (*Peer) GetRole ¶ 

func (p *Peer) GetRole() raft.StateType

func (*Peer) HandleRaftReadyAppend ¶ 

func (p *Peer) HandleRaftReadyAppend(trans Transport, applyMsgs *applyMsgs, kvWB, raftWB *WriteBatch, observer PeerEventObserver) *ReadyICPair

func (*Peer) HandleRaftReadyApply ¶ 

func (p *Peer) HandleRaftReadyApply(kv *mvcc.DBBundle, applyMsgs *applyMsgs, ready *raft.Ready)

func (*Peer) HasPendingSnapshot ¶ 

func (p *Peer) HasPendingSnapshot() bool

/ Returns `true` if the raft group has replicated a snapshot but not committed it yet.

func (*Peer) HeartbeatPd ¶ 

func (p *Peer) HeartbeatPd(pdScheduler chan<- task)

func (*Peer) IsApplyingSnapshot ¶ 

func (p *Peer) IsApplyingSnapshot() bool

func (*Peer) IsLeader ¶ 

func (p *Peer) IsLeader() bool

func (*Peer) LeaderId ¶ 

func (p *Peer) LeaderId() uint64

func (*Peer) MaybeCampaign ¶ 

func (p *Peer) MaybeCampaign(parentIsLeader bool) bool

func (*Peer) MaybeDestroy ¶ 

func (p *Peer) MaybeDestroy() *DestroyPeerJob

/ Tries to destroy itself. Returns a job (if needed) to do more cleaning tasks.

func (*Peer) MaybeRenewLeaderLease ¶ 

func (p *Peer) MaybeRenewLeaderLease(ts time.Time)

Try to renew leader lease.

func (*Peer) OnRoleChanged ¶ 

func (p *Peer) OnRoleChanged(observer PeerEventObserver, ready *raft.Ready)

func (*Peer) PeerId ¶ 

func (p *Peer) PeerId() uint64

func (*Peer) PostApply ¶ 

func (p *Peer) PostApply(kv *mvcc.DBBundle, applyState applyState, appliedIndexTerm uint64, merged bool, applyMetrics applyMetrics) bool

func (*Peer) PostPropose ¶ 

func (p *Peer) PostPropose(meta *ProposalMeta, isConfChange bool, cb *Callback)

func (*Peer) PostRaftReadyPersistent ¶ 

func (p *Peer) PostRaftReadyPersistent(trans Transport, applyMsgs *applyMsgs, ready *raft.Ready, invokeCtx *InvokeContext) *ApplySnapResult

func (*Peer) PostSplit ¶ 

func (p *Peer) PostSplit()

func (*Peer) PrePropose ¶ 

func (p *Peer) PrePropose(cfg *Config, rlog raftlog.RaftLog) (*ProposalContext, error)

func (*Peer) Propose ¶ 

func (p *Peer) Propose(kv *mvcc.DBBundle, cfg *Config, cb *Callback, rlog raftlog.RaftLog, errResp *raft_cmdpb.RaftCmdResponse) bool

Propose a request.

Return true means the request has been proposed successfully.

func (*Peer) ProposeConfChange ¶ 

func (p *Peer) ProposeConfChange(cfg *Config, req *raft_cmdpb.RaftCmdRequest) (uint64, error)

Fails in such cases: 1. A pending conf change has not been applied yet; 2. Removing the leader is not allowed in the configuration; 3. The conf change makes the raft group not healthy; 4. The conf change is dropped by raft group internally.

func (*Peer) ProposeNormal ¶ 

func (p *Peer) ProposeNormal(cfg *Config, rlog raftlog.RaftLog) (uint64, error)

func (*Peer) ProposeTransferLeader ¶ 

func (p *Peer) ProposeTransferLeader(cfg *Config, req *raft_cmdpb.RaftCmdRequest, cb *Callback) bool

Return true if the transfer leader request is accepted.

func (*Peer) ReadyToHandlePendingSnap ¶ 

func (p *Peer) ReadyToHandlePendingSnap() bool

func (*Peer) Region ¶ 

func (p *Peer) Region() *metapb.Region

func (*Peer) Send ¶ 

func (p *Peer) Send(trans Transport, msgs []eraftpb.Message) error

func (*Peer) SetRegion ¶ 

func (p *Peer) SetRegion(region *metapb.Region)

/ Set the region of a peer. / / This will update the region of the peer, caller must ensure the region / has been preserved in a durable device.

func (*Peer) Step ¶ 

func (p *Peer) Step(m *eraftpb.Message) error

/ Steps the raft message.

func (*Peer) Stop ¶ 

func (p *Peer) Stop()

func (*Peer) Store ¶ 

func (p *Peer) Store() *PeerStorage

func (*Peer) TakeApplyProposals ¶ 

func (p *Peer) TakeApplyProposals() *regionProposal

func (*Peer) Term ¶ 

func (p *Peer) Term() uint64

type PeerEventContext ¶ 

type PeerEventContext struct {
	LeaderChecker LeaderChecker
	RegionId      uint64
}

type PeerEventObserver ¶ 

type PeerEventObserver interface {
	// OnPeerCreate will be invoked when there is a new peer created.
	OnPeerCreate(ctx *PeerEventContext, region *metapb.Region)
	// OnPeerApplySnap will be invoked when there is a replicate peer's snapshot applied.
	OnPeerApplySnap(ctx *PeerEventContext, region *metapb.Region)
	// OnPeerDestroy will be invoked when a peer is destroyed.
	OnPeerDestroy(ctx *PeerEventContext)
	// OnSplitRegion will be invoked when region split into new regions with corresponding peers.
	OnSplitRegion(derived *metapb.Region, regions []*metapb.Region, peers []*PeerEventContext)
	// OnRegionConfChange will be invoked after conf change updated region's epoch.
	OnRegionConfChange(ctx *PeerEventContext, epoch *metapb.RegionEpoch)
	// OnRoleChange will be invoked after peer state has changed
	OnRoleChange(regionId uint64, newState raft.StateType)
}

type PeerStat ¶ 

type PeerStat struct {
	WrittenBytes uint64
	WrittenKeys  uint64
}

type PeerStorage ¶ 

type PeerStorage struct {
	Engines *Engines

	Tag string
	// contains filtered or unexported fields
}

func NewPeerStorage ¶ 

func NewPeerStorage(engines *Engines, region *metapb.Region, regionSched chan<- task, peerID uint64, tag string) (*PeerStorage, error)

func (*PeerStorage) Append ¶ 

func (ps *PeerStorage) Append(invokeCtx *InvokeContext, entries []eraftpb.Entry, raftWB *WriteBatch) error

Append the given entries to the raft log using previous last index or self.last_index. Return the new last index for later update. After we commit in engine, we can set last_index to the return one.

func (*PeerStorage) AppliedIndex ¶ 

func (ps *PeerStorage) AppliedIndex() uint64

func (*PeerStorage) ApplySnapshot ¶ 

func (ps *PeerStorage) ApplySnapshot(ctx *InvokeContext, snap *eraftpb.Snapshot, kvWB *WriteBatch, raftWB *WriteBatch) error

Apply the peer with given snapshot.

func (*PeerStorage) CancelApplyingSnap ¶ 

func (p *PeerStorage) CancelApplyingSnap() bool

func (*PeerStorage) CheckApplyingSnap ¶ 

func (p *PeerStorage) CheckApplyingSnap() bool

Check if the storage is applying a snapshot.

func (*PeerStorage) ClearData ¶ 

func (ps *PeerStorage) ClearData() error

func (*PeerStorage) CompactTo ¶ 

func (ps *PeerStorage) CompactTo(idx uint64)

func (*PeerStorage) Entries ¶ 

func (ps *PeerStorage) Entries(low, high, maxSize uint64) ([]eraftpb.Entry, error)

func (*PeerStorage) FirstIndex ¶ 

func (ps *PeerStorage) FirstIndex() (uint64, error)

func (*PeerStorage) InitialState ¶ 

func (ps *PeerStorage) InitialState() (eraftpb.HardState, eraftpb.ConfState, error)

func (*PeerStorage) IsApplyingSnapshot ¶ 

func (ps *PeerStorage) IsApplyingSnapshot() bool

func (*PeerStorage) LastIndex ¶ 

func (ps *PeerStorage) LastIndex() (uint64, error)

func (*PeerStorage) MaybeGCCache ¶ 

func (ps *PeerStorage) MaybeGCCache(replicatedIdx, appliedIdx uint64)

func (*PeerStorage) PostReadyPersistent ¶ 

func (ps *PeerStorage) PostReadyPersistent(ctx *InvokeContext) *ApplySnapResult

Update the memory state after ready changes are flushed to disk successfully.

func (*PeerStorage) Region ¶ 

func (ps *PeerStorage) Region() *metapb.Region

func (*PeerStorage) SaveReadyState ¶ 

func (ps *PeerStorage) SaveReadyState(kvWB, raftWB *WriteBatch, ready *raft.Ready) (*InvokeContext, error)

/ Save memory states to disk. / / This function only write data to `ready_ctx`'s `WriteBatch`. It's caller's duty to write / it explicitly to disk. If it's flushed to disk successfully, `post_ready` should be called / to update the memory states properly. / Do not modify ready in this function, this is a requirement to advance the ready object properly later.

func (*PeerStorage) ScheduleApplyingSnapshot ¶ 

func (ps *PeerStorage) ScheduleApplyingSnapshot()

func (*PeerStorage) SetRegion ¶ 

func (ps *PeerStorage) SetRegion(region *metapb.Region)

func (*PeerStorage) Snapshot ¶ 

func (ps *PeerStorage) Snapshot() (eraftpb.Snapshot, error)

func (*PeerStorage) Term ¶ 

func (ps *PeerStorage) Term(idx uint64) (uint64, error)

type PeerTick ¶ 

type PeerTick int
const (
	PeerTickRaft             PeerTick = 0
	PeerTickRaftLogGC        PeerTick = 1
	PeerTickSplitRegionCheck PeerTick = 2
	PeerTickPdHeartbeat      PeerTick = 3
	PeerTickCheckMerge       PeerTick = 4
	PeerTickPeerStaleState   PeerTick = 5
)

type ProposalContext ¶ 

type ProposalContext byte
const (
	ProposalContext_SyncLog      ProposalContext = 1
	ProposalContext_Split        ProposalContext = 1 << 1
	ProposalContext_PrepareMerge ProposalContext = 1 << 2
)

func NewProposalContextFromBytes ¶ 

func NewProposalContextFromBytes(ctx []byte) *ProposalContext

func (ProposalContext) ToBytes ¶ 

func (c ProposalContext) ToBytes() []byte

type ProposalMeta ¶ 

type ProposalMeta struct {
	Index          uint64
	Term           uint64
	RenewLeaseTime *time.Time
}

type ProposalQueue ¶ 

type ProposalQueue struct {
	// contains filtered or unexported fields
}

func (*ProposalQueue) Clear ¶ 

func (q *ProposalQueue) Clear()

func (*ProposalQueue) PopFront ¶ 

func (q *ProposalQueue) PopFront(term uint64) *ProposalMeta

func (*ProposalQueue) Push ¶ 

func (q *ProposalQueue) Push(meta *ProposalMeta)

type RaftClient ¶ 

type RaftClient struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

func (*RaftClient) Flush ¶ 

func (c *RaftClient) Flush()

func (*RaftClient) Send ¶ 

func (c *RaftClient) Send(msg *raft_serverpb.RaftMessage)

func (*RaftClient) Stop ¶ 

func (c *RaftClient) Stop()

type RaftContext ¶ 

type RaftContext struct {
	*GlobalContext

	ReadyRes []*ReadyICPair
	// contains filtered or unexported fields
}

type RaftError ¶ 

type RaftError struct {
	RequestErr *errorpb.Error
}

func (*RaftError) Error ¶ 

func (re *RaftError) Error() string

type RaftInnerServer ¶ 

type RaftInnerServer struct {
	// contains filtered or unexported fields
}

func NewRaftInnerServer ¶ 

func NewRaftInnerServer(globalConfig *config.Config, engines *Engines, raftConfig *Config) *RaftInnerServer

func (*RaftInnerServer) BatchRaft ¶ 

func (ris *RaftInnerServer) BatchRaft(stream tikvpb.Tikv_BatchRaftServer) error

func (*RaftInnerServer) GetRaftstoreRouter ¶ 

func (ris *RaftInnerServer) GetRaftstoreRouter() *RaftstoreRouter

func (*RaftInnerServer) GetStoreMeta ¶ 

func (ris *RaftInnerServer) GetStoreMeta() *metapb.Store

func (*RaftInnerServer) Raft ¶ 

func (ris *RaftInnerServer) Raft(stream tikvpb.Tikv_RaftServer) error

func (*RaftInnerServer) SetPeerEventObserver ¶ 

func (ris *RaftInnerServer) SetPeerEventObserver(ob PeerEventObserver)

func (*RaftInnerServer) Setup ¶ 

func (ris *RaftInnerServer) Setup(pdClient pd.Client)

func (*RaftInnerServer) Snapshot ¶ 

func (ris *RaftInnerServer) Snapshot(stream tikvpb.Tikv_SnapshotServer) error

func (*RaftInnerServer) Start ¶ 

func (ris *RaftInnerServer) Start(pdClient pd.Client) error

func (*RaftInnerServer) Stop ¶ 

func (ris *RaftInnerServer) Stop() error

type RaftstoreRouter ¶ 

type RaftstoreRouter struct {
	// contains filtered or unexported fields
}

RaftstoreRouter exports SendCommand method for other packages.

func (*RaftstoreRouter) SendCommand ¶ 

func (r *RaftstoreRouter) SendCommand(req *raft_cmdpb.RaftCmdRequest, cb *Callback) error

func (*RaftstoreRouter) SplitRegion ¶ 

func (r *RaftstoreRouter) SplitRegion(ctx *kvrpcpb.Context, keys [][]byte) ([]*metapb.Region, error)

type ReadExecutor ¶ 

type ReadExecutor struct {
	// contains filtered or unexported fields
}

func NewReadExecutor ¶ 

func NewReadExecutor(checkEpoch bool) *ReadExecutor

func (*ReadExecutor) Execute ¶ 

func (r *ReadExecutor) Execute(msg *raft_cmdpb.RaftCmdRequest, region *metapb.Region) *raft_cmdpb.RaftCmdResponse

type ReadIndexQueue ¶ 

type ReadIndexQueue struct {
	// contains filtered or unexported fields
}

func (*ReadIndexQueue) ClearUncommitted ¶ 

func (r *ReadIndexQueue) ClearUncommitted(term uint64)

func (*ReadIndexQueue) NextId ¶ 

func (r *ReadIndexQueue) NextId() uint64

func (*ReadIndexQueue) PopFront ¶ 

func (q *ReadIndexQueue) PopFront() *ReadIndexRequest

type ReadIndexRequest ¶ 

type ReadIndexRequest struct {
	// contains filtered or unexported fields
}

func NewReadIndexRequest ¶ 

func NewReadIndexRequest(id uint64, cmds []*ReqCbPair, renewLeaseTime *time.Time) *ReadIndexRequest

type ReadyICPair ¶ 

type ReadyICPair struct {
	Ready raft.Ready
	IC    *InvokeContext
}

type RecentAddedPeer ¶ 

type RecentAddedPeer struct {
	RejectDurationAsSecs uint64
	Id                   uint64
	AddedTime            time.Time
}

func NewRecentAddedPeer ¶ 

func NewRecentAddedPeer(rejectDurationAsSecs uint64) *RecentAddedPeer

func (*RecentAddedPeer) Contains ¶ 

func (r *RecentAddedPeer) Contains(id uint64) bool

func (*RecentAddedPeer) Update ¶ 

func (r *RecentAddedPeer) Update(id uint64, now time.Time)

type RemoteLease ¶ 

type RemoteLease struct {
	// contains filtered or unexported fields
}

/ A remote lease, it can only be derived by `Lease`. It will be sent / to the local read thread, so name it remote. If Lease expires, the remote must / expire too.

func (*RemoteLease) Expire ¶ 

func (r *RemoteLease) Expire()

func (*RemoteLease) Inspect ¶ 

func (r *RemoteLease) Inspect(ts *time.Time) LeaseState

func (*RemoteLease) Renew ¶ 

func (r *RemoteLease) Renew(bound time.Time)

func (*RemoteLease) Term ¶ 

func (r *RemoteLease) Term() uint64

type ReqCbPair ¶ 

type ReqCbPair struct {
	Req *raft_cmdpb.RaftCmdRequest
	Cb  *Callback
}

type RequestInspector ¶ 

type RequestInspector interface {
	// contains filtered or unexported methods
}

type RequestPolicy ¶ 

type RequestPolicy int
const (
	// Handle the read request directly without dispatch.
	RequestPolicy_ReadLocal RequestPolicy = 0 + iota
	// Handle the read request via raft's SafeReadIndex mechanism.
	RequestPolicy_ReadIndex
	RequestPolicy_ProposeNormal
	RequestPolicy_ProposeTransferLeader
	RequestPolicy_ProposeConfChange
	RequestPolicy_Invalid
)

func Inspect ¶ 

func Inspect(i RequestInspector, req *raft_cmdpb.RaftCmdRequest) (RequestPolicy, error)

type ServerTransport ¶ 

type ServerTransport struct {
	// contains filtered or unexported fields
}

func NewServerTransport ¶ 

func NewServerTransport(raftClient *RaftClient, snapScheduler chan<- task, router *router) *ServerTransport

func (*ServerTransport) Flush ¶ 

func (t *ServerTransport) Flush()

func (*ServerTransport) ReportSnapshotStatus ¶ 

func (t *ServerTransport) ReportSnapshotStatus(msg *raft_serverpb.RaftMessage, status raft.SnapshotStatus)

func (*ServerTransport) ReportUnreachable ¶ 

func (t *ServerTransport) ReportUnreachable(msg *raft_serverpb.RaftMessage)

func (*ServerTransport) Send ¶ 

func (t *ServerTransport) Send(msg *raft_serverpb.RaftMessage) error

func (*ServerTransport) SendSnapshotSock ¶ 

func (t *ServerTransport) SendSnapshotSock(msg *raft_serverpb.RaftMessage)

type Snap ¶ 

type Snap struct {
	CFFiles []*CFFile

	MetaFile  *MetaFile
	SizeTrack *int64
	// contains filtered or unexported fields
}

func NewSnap ¶ 

func NewSnap(dir string, key SnapKey, sizeTrack *int64, isSending, toBuild bool,
	deleter SnapshotDeleter, limiter *IOLimiter) (*Snap, error)

func NewSnapForApplying ¶ 

func NewSnapForApplying(dir string, key SnapKey, sizeTrack *int64, deleter SnapshotDeleter) (*Snap, error)

func NewSnapForBuilding ¶ 

func NewSnapForBuilding(dir string, key SnapKey, sizeTrack *int64, deleter SnapshotDeleter, limiter *IOLimiter) (*Snap, error)

func NewSnapForReceiving ¶ 

func NewSnapForReceiving(dir string, key SnapKey, snapshotMeta *rspb.SnapshotMeta,
	sizeTrack *int64, deleter SnapshotDeleter, limiter *IOLimiter) (*Snap, error)

func NewSnapForSending ¶ 

func NewSnapForSending(dir string, key SnapKey, sizeTrack *int64, deleter SnapshotDeleter) (*Snap, error)

func (*Snap) Apply ¶ 

func (s *Snap) Apply(opts ApplyOptions) (ApplyResult, error)

func (*Snap) Build ¶ 

func (s *Snap) Build(dbSnap *regionSnapshot, region *metapb.Region, snapData *rspb.RaftSnapshotData, stat *SnapStatistics, deleter SnapshotDeleter) error

func (*Snap) Delete ¶ 

func (s *Snap) Delete()

func (*Snap) Drop ¶ 

func (s *Snap) Drop()

func (*Snap) Exists ¶ 

func (s *Snap) Exists() bool

func (*Snap) Meta ¶ 

func (s *Snap) Meta() (os.FileInfo, error)

func (*Snap) Path ¶ 

func (s *Snap) Path() string

func (*Snap) Read ¶ 

func (s *Snap) Read(b []byte) (int, error)

func (*Snap) Save ¶ 

func (s *Snap) Save() error

func (*Snap) TotalSize ¶ 

func (s *Snap) TotalSize() (total uint64)

func (*Snap) Write ¶ 

func (s *Snap) Write(b []byte) (int, error)

type SnapEntry ¶ 

type SnapEntry int
const (
	SnapEntryGenerating SnapEntry = 1
	SnapEntrySending    SnapEntry = 2
	SnapEntryReceiving  SnapEntry = 3
	SnapEntryApplying   SnapEntry = 4
)

func (SnapEntry) String ¶ 

func (e SnapEntry) String() string

type SnapKey ¶ 

type SnapKey struct {
	RegionID uint64
	Term     uint64
	Index    uint64
}

func SnapKeyFromRegionSnap ¶ 

func SnapKeyFromRegionSnap(regionID uint64, snap *eraftpb.Snapshot) SnapKey

func SnapKeyFromSnap ¶ 

func SnapKeyFromSnap(snap *eraftpb.Snapshot) (SnapKey, error)

func (SnapKey) String ¶ 

func (k SnapKey) String() string

type SnapKeyWithSending ¶ 

type SnapKeyWithSending struct {
	SnapKey   SnapKey
	IsSending bool
}

type SnapManager ¶ 

type SnapManager struct {
	MaxTotalSize uint64
	// contains filtered or unexported fields
}

func NewSnapManager ¶ 

func NewSnapManager(path string, router *router) *SnapManager

func (*SnapManager) DeleteSnapshot ¶ 

func (sm *SnapManager) DeleteSnapshot(key SnapKey, snapshot Snapshot, checkEntry bool) bool

func (*SnapManager) Deregister ¶ 

func (sm *SnapManager) Deregister(key SnapKey, entry SnapEntry)

func (*SnapManager) GetSnapshotForApplying ¶ 

func (sm *SnapManager) GetSnapshotForApplying(snapKey SnapKey) (Snapshot, error)

func (*SnapManager) GetSnapshotForBuilding ¶ 

func (sm *SnapManager) GetSnapshotForBuilding(key SnapKey) (Snapshot, error)

func (*SnapManager) GetSnapshotForReceiving ¶ 

func (sm *SnapManager) GetSnapshotForReceiving(snapKey SnapKey, data []byte) (Snapshot, error)

func (*SnapManager) GetSnapshotForSending ¶ 

func (sm *SnapManager) GetSnapshotForSending(snapKey SnapKey) (Snapshot, error)

func (*SnapManager) GetTotalSnapSize ¶ 

func (sm *SnapManager) GetTotalSnapSize() uint64

func (*SnapManager) HasRegistered ¶ 

func (sm *SnapManager) HasRegistered(key SnapKey) bool

func (*SnapManager) ListIdleSnap ¶ 

func (sm *SnapManager) ListIdleSnap() ([]SnapKeyWithSending, error)

func (*SnapManager) Register ¶ 

func (sm *SnapManager) Register(key SnapKey, entry SnapEntry)

func (*SnapManager) Stats ¶ 

func (sm *SnapManager) Stats() SnapStats

type SnapManagerBuilder ¶ 

type SnapManagerBuilder struct {
	// contains filtered or unexported fields
}

func (*SnapManagerBuilder) Build ¶ 

func (smb *SnapManagerBuilder) Build(path string, router *router) *SnapManager

func (*SnapManagerBuilder) MaxTotalSize ¶ 

func (smb *SnapManagerBuilder) MaxTotalSize(v uint64) *SnapManagerBuilder

type SnapState ¶ 

type SnapState struct {
	StateType SnapStateType
	Status    *JobStatus
	Receiver  chan *eraftpb.Snapshot
}

type SnapStateType ¶ 

type SnapStateType int
const (
	SnapState_Relax SnapStateType = 0 + iota
	SnapState_Generating
	SnapState_Applying
	SnapState_ApplyAborted
)

type SnapStatistics ¶ 

type SnapStatistics struct {
	Size    uint64
	KVCount int
}

type SnapStats ¶ 

type SnapStats struct {
	ReceivingCount int
	SendingCount   int
}

type Snapshot ¶ 

type Snapshot interface {
	io.Reader
	io.Writer
	Build(dbBundle *regionSnapshot, region *metapb.Region, snapData *rspb.RaftSnapshotData, stat *SnapStatistics, deleter SnapshotDeleter) error
	Path() string
	Exists() bool
	Delete()
	Meta() (os.FileInfo, error)
	TotalSize() uint64
	Save() error
	Apply(option ApplyOptions) (ApplyResult, error)
}

`Snapshot` is an interface for snapshot. It's used in these scenarios:

  1. build local snapshot
  2. read local snapshot and then replicate it to remote raftstores
  3. receive snapshot from remote raftstore and write it to local storage
  4. apply snapshot
  5. snapshot gc

type SnapshotDeleter ¶ 

type SnapshotDeleter interface {
	// DeleteSnapshot returns true if it successfully delete the specified snapshot.
	DeleteSnapshot(key SnapKey, snapshot Snapshot, checkEntry bool) bool
}

`SnapshotDeleter` is a trait for deleting snapshot. It's used to ensure that the snapshot deletion happens under the protection of locking to avoid race case for concurrent read/write.

type StaleState ¶ 

type StaleState int
const (
	StaleStateValid StaleState = 0 + iota
	StaleStateToValidate
	StaleStateLeaderMissing
)

type StoreContext ¶ 

type StoreContext struct {
	*GlobalContext
	// contains filtered or unexported fields
}

type StoreLabel ¶ 

type StoreLabel struct {
	LabelKey, LabelValue string
}

type StoreTick ¶ 

type StoreTick int
const (
	StoreTickCompactCheck     StoreTick = 0
	StoreTickPdStoreHeartbeat StoreTick = 1
	StoreTickSnapGC           StoreTick = 2
	StoreTickConsistencyCheck StoreTick = 3
)

type TestRaftWriter ¶ 

type TestRaftWriter struct {
	// contains filtered or unexported fields
}

TestRaftWriter is used to mock raft write related prewrite and commit operations without sending real raft commands

func (*TestRaftWriter) Close ¶ 

func (w *TestRaftWriter) Close()

func (*TestRaftWriter) DeleteRange ¶ 

func (w *TestRaftWriter) DeleteRange(start, end []byte, latchHandle mvcc.LatchHandle) error

func (*TestRaftWriter) NewWriteBatch ¶ 

func (w *TestRaftWriter) NewWriteBatch(startTS, commitTS uint64, ctx *kvrpcpb.Context) mvcc.WriteBatch

func (*TestRaftWriter) Open ¶ 

func (w *TestRaftWriter) Open()

func (*TestRaftWriter) Write ¶ 

func (w *TestRaftWriter) Write(batch mvcc.WriteBatch) error

type Transport ¶ 

type Transport interface {
	Send(msg *rspb.RaftMessage) error
}

type WaitApplyResultState ¶ 

type WaitApplyResultState struct {
	// contains filtered or unexported fields
}

/ A struct that stores the state to wait for `PrepareMerge` apply result. / / When handling the apply result of a `CommitMerge`, the source peer may have / not handle the apply result of the `PrepareMerge`, so the target peer has / to abort current handle process and wait for it asynchronously.

type WriteBatch ¶ 

type WriteBatch struct {
	// contains filtered or unexported fields
}

func (*WriteBatch) Delete ¶ 

func (wb *WriteBatch) Delete(key y.Key)

func (*WriteBatch) DeleteLock ¶ 

func (wb *WriteBatch) DeleteLock(key []byte)

func (*WriteBatch) Len ¶ 

func (wb *WriteBatch) Len() int

func (*WriteBatch) MustWriteToKV ¶ 

func (wb *WriteBatch) MustWriteToKV(db *mvcc.DBBundle)

func (*WriteBatch) MustWriteToRaft ¶ 

func (wb *WriteBatch) MustWriteToRaft(db *badger.DB)

func (*WriteBatch) Reset ¶ 

func (wb *WriteBatch) Reset()

func (*WriteBatch) Rollback ¶ 

func (wb *WriteBatch) Rollback(key y.Key)

func (*WriteBatch) RollbackToSafePoint ¶ 

func (wb *WriteBatch) RollbackToSafePoint()

func (*WriteBatch) Set ¶ 

func (wb *WriteBatch) Set(key y.Key, val []byte)

func (*WriteBatch) SetLock ¶ 

func (wb *WriteBatch) SetLock(key, val []byte)

func (*WriteBatch) SetMsg ¶ 

func (wb *WriteBatch) SetMsg(key y.Key, msg proto.Message) error

func (*WriteBatch) SetOpLock ¶ 

func (wb *WriteBatch) SetOpLock(key y.Key, userMeta []byte)

func (*WriteBatch) SetSafePoint ¶ 

func (wb *WriteBatch) SetSafePoint()

func (*WriteBatch) SetWithUserMeta ¶ 

func (wb *WriteBatch) SetWithUserMeta(key y.Key, val, userMeta []byte)

func (*WriteBatch) WriteToKV ¶ 

func (wb *WriteBatch) WriteToKV(bundle *mvcc.DBBundle) error

WriteToKV flush WriteBatch to DB by two steps:

  1. Write entries to badger. After save ApplyState to badger, subsequent regionSnapshot will start at new raft index.
  2. Update lockStore, the date in lockStore may be older than the DB, so we need to restore then entries from raft log.

func (*WriteBatch) WriteToRaft ¶ 

func (wb *WriteBatch) WriteToRaft(db *badger.DB) error

Source Files ¶

View all Source files

Directories ¶

Path Synopsis

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.