Struct raftstore::store::fsm::apply::ApplyDelegate

pub struct ApplyDelegate<EK> where
    EK: KvEngine,  {
    id: u64,
    term: u64,
    region: Region,
    tag: String,
    stopped: bool,
    handle_start: Option<Instant>,
    pending_remove: bool,
    pending_cmds: PendingCmdQueue<EK::Snapshot>,
    pending_request_snapshot_count: Arc<AtomicUsize>,
    is_merging: bool,
    last_merge_version: u64,
    yield_state: Option<YieldState<EK>>,
    wait_merge_state: Option<WaitSourceMergeState>,
    ready_source_region_id: u64,
    apply_state: RaftApplyState,
    applied_index_term: u64,
    last_sync_apply_index: u64,
    observe_info: CmdObserveInfo,
    metrics: ApplyMetrics,
    priority: Priority,
}

The apply delegate of a Region which is responsible for handling committed raft log entries of a Region.

Apply is a term of Raft, which means executing the actual commands. In Raft, once some log entries are committed, for every peer of the Raft group will apply the logs one by one. For write commands, it does write or delete to local engine; for admin commands, it does some meta change of the Raft group.

Delegate is just a structure to congregate all apply related fields of a Region. The apply worker receives all the apply tasks of different Regions located at this store, and it will get the corresponding apply delegate to handle the apply task to make the code logic more clear.

Fields

id: u64

The ID of the peer.

term: u64

The term of the Region.

region: Region

The Region information of the peer.

tag: String

Peer_tag, “[region region_id] peer_id”.

stopped: bool

If the delegate should be stopped from polling. A delegate can be stopped in conf change, merge or requested by destroy message.

handle_start: Option<Instant>

The start time of the current round to execute commands.

pending_remove: bool

Set to true when removing itself because of ConfChangeType::RemoveNode, and then any following committed logs in same Ready should be applied failed.

pending_cmds: PendingCmdQueue<EK::Snapshot>

The commands waiting to be committed and applied

pending_request_snapshot_count: Arc<AtomicUsize>

The counter of pending request snapshots. See more in Peer.

is_merging: bool

Indicates the peer is in merging, if that compact log won’t be performed.

last_merge_version: u64

Records the epoch version after the last merge.

yield_state: Option<YieldState<EK>>

wait_merge_state: Option<WaitSourceMergeState>

A temporary state that keeps track of the progress of the source peer state when CommitMerge is unable to be executed.

ready_source_region_id: u64

apply_state: RaftApplyState

TiKV writes apply_state to KV RocksDB, in one write batch together with kv data.

If we write it to Raft RocksDB, apply_state and kv data (Put, Delete) are in separate WAL file. When power failure, for current raft log, apply_index may synced to file, but KV data may not synced to file, so we will lose data.

applied_index_term: u64

The term of the raft log at applied index.

last_sync_apply_index: u64

The latest synced apply index.

observe_info: CmdObserveInfo

Info about cmd observer.

metrics: ApplyMetrics

The local metrics, and it will be flushed periodically.

priority: Priority

Priority in batch system. When applying some commands which have high latency, we decrease the priority of current fsm to reduce the impact on other normal commands.

Implementations

impl<EK> ApplyDelegate<EK> where
    EK: KvEngine, 

fn from_registration(reg: Registration) -> ApplyDelegate<EK>

pub fn region_id(&self) -> u64

pub fn id(&self) -> u64

fn handle_raft_committed_entries<W: WriteBatch<EK>>(
    &mut self,
    apply_ctx: &mut ApplyContext<EK, W>,
    committed_entries_drainer: Drain<'_, Entry>
)

Handles all the committed_entries, namely, applies the committed entries.

fn update_metrics<W: WriteBatch<EK>>(&mut self, apply_ctx: &ApplyContext<EK, W>)

fn write_apply_state<W: WriteBatch<EK>>(&self, wb: &mut W)

fn handle_raft_entry_normal<W: WriteBatch<EK>>(
    &mut self,
    apply_ctx: &mut ApplyContext<EK, W>,
    entry: &Entry
) -> ApplyResult<EK::Snapshot>

fn handle_raft_entry_conf_change<W: WriteBatch<EK>>(
    &mut self,
    apply_ctx: &mut ApplyContext<EK, W>,
    entry: &Entry
) -> ApplyResult<EK::Snapshot>

fn find_pending(
    &mut self,
    index: u64,
    term: u64,
    is_conf_change: bool
) -> Option<Callback<EK::Snapshot>>

fn process_raft_cmd<W: WriteBatch<EK>>(
    &mut self,
    apply_ctx: &mut ApplyContext<EK, W>,
    index: u64,
    term: u64,
    cmd: RaftCmdRequest
) -> ApplyResult<EK::Snapshot>

fn apply_raft_cmd<W: WriteBatch<EK>>(
    &mut self,
    ctx: &mut ApplyContext<EK, W>,
    index: u64,
    term: u64,
    req: &RaftCmdRequest
) -> (RaftCmdResponse, ApplyResult<EK::Snapshot>)

Applies raft command.

An apply operation can fail in the following situations:

1. it encounters an error that will occur on all stores, it can continue applying next entry safely, like epoch not match for example;
2. it encounters an error that may not occur on all stores, in this case we should try to apply the entry again or panic. Considering that this usually due to disk operation fail, which is rare, so just panic is ok.

fn destroy<W: WriteBatch<EK>>(&mut self, apply_ctx: &mut ApplyContext<EK, W>)

fn clear_all_commands_as_stale(&mut self)

fn new_ctx(&self, index: u64, term: u64) -> ExecContext

impl<EK> ApplyDelegate<EK> where
    EK: KvEngine, 

fn exec_raft_cmd<W: WriteBatch<EK>>(
    &mut self,
    ctx: &mut ApplyContext<EK, W>,
    req: &RaftCmdRequest
) -> Result<(RaftCmdResponse, ApplyResult<EK::Snapshot>)>

fn exec_admin_cmd<W: WriteBatch<EK>>(
    &mut self,
    ctx: &mut ApplyContext<EK, W>,
    req: &RaftCmdRequest
) -> Result<(RaftCmdResponse, ApplyResult<EK::Snapshot>)>

fn exec_write_cmd<W: WriteBatch<EK>>(
    &mut self,
    ctx: &mut ApplyContext<EK, W>,
    req: &RaftCmdRequest
) -> Result<(RaftCmdResponse, ApplyResult<EK::Snapshot>)>

impl<EK> ApplyDelegate<EK> where
    EK: KvEngine, 

fn handle_put<W: WriteBatch<EK>>(
    &mut self,
    wb: &mut W,
    req: &Request
) -> Result<Response>

fn handle_delete<W: WriteBatch<EK>>(
    &mut self,
    wb: &mut W,
    req: &Request
) -> Result<Response>

fn handle_delete_range(
    &mut self,
    engine: &EK,
    req: &Request,
    ranges: &mut Vec<Range>,
    use_delete_range: bool
) -> Result<Response>

fn handle_ingest_sst(
    &mut self,
    importer: &Arc<SSTImporter>,
    engine: &EK,
    req: &Request,
    ssts: &mut Vec<SSTMetaInfo>
) -> Result<Response>

impl<EK> ApplyDelegate<EK> where
    EK: KvEngine, 

fn exec_change_peer<W: WriteBatch<EK>>(
    &mut self,
    ctx: &mut ApplyContext<EK, W>,
    request: &AdminRequest
) -> Result<(AdminResponse, ApplyResult<EK::Snapshot>)>

fn exec_change_peer_v2<W: WriteBatch<EK>>(
    &mut self,
    ctx: &mut ApplyContext<EK, W>,
    request: &AdminRequest
) -> Result<(AdminResponse, ApplyResult<EK::Snapshot>)>

fn apply_conf_change(
    &mut self,
    kind: ConfChangeKind,
    changes: &[ChangePeerRequest]
) -> Result<Region>

fn apply_leave_joint(&self) -> Result<Region>

fn exec_split<W: WriteBatch<EK>>(
    &mut self,
    ctx: &mut ApplyContext<EK, W>,
    req: &AdminRequest
) -> Result<(AdminResponse, ApplyResult<EK::Snapshot>)>

fn exec_batch_split<W: WriteBatch<EK>>(
    &mut self,
    ctx: &mut ApplyContext<EK, W>,
    req: &AdminRequest
) -> Result<(AdminResponse, ApplyResult<EK::Snapshot>)>

fn exec_prepare_merge<W: WriteBatch<EK>>(
    &mut self,
    ctx: &mut ApplyContext<EK, W>,
    req: &AdminRequest
) -> Result<(AdminResponse, ApplyResult<EK::Snapshot>)>

fn exec_commit_merge<W: WriteBatch<EK>>(
    &mut self,
    ctx: &mut ApplyContext<EK, W>,
    req: &AdminRequest
) -> Result<(AdminResponse, ApplyResult<EK::Snapshot>)>

fn exec_rollback_merge<W: WriteBatch<EK>>(
    &mut self,
    ctx: &mut ApplyContext<EK, W>,
    req: &AdminRequest
) -> Result<(AdminResponse, ApplyResult<EK::Snapshot>)>

fn exec_compact_log<W: WriteBatch<EK>>(
    &mut self,
    ctx: &mut ApplyContext<EK, W>,
    req: &AdminRequest
) -> Result<(AdminResponse, ApplyResult<EK::Snapshot>)>

fn exec_compute_hash<W: WriteBatch<EK>>(
    &self,
    ctx: &ApplyContext<EK, W>,
    req: &AdminRequest
) -> Result<(AdminResponse, ApplyResult<EK::Snapshot>)>

fn exec_verify_hash<W: WriteBatch<EK>>(
    &self,
    _: &ApplyContext<EK, W>,
    req: &AdminRequest
) -> Result<(AdminResponse, ApplyResult<EK::Snapshot>)>

Trait Implementations

impl<EK: Debug> Debug for ApplyDelegate<EK> where
    EK: KvEngine,
    EK::Snapshot: Debug, 

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.