raftstore

package
v0.0.0-...-76a7b18 Latest Latest
Warning

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

 Go to latest Published: Sep 29, 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.