TiKV Raft 快照全流程丨 TiKV 源码解读（二十二）

raft-rs: src/raft.rsimpl RaftCore {  fn maybe_send_append(..., pr: &mut Progress, ) {      ...      let term = self.raft_log.term(pr.next_idx - 1);      match (term, ents) {          ...          _ => {               // send snapshot if we failed to get term or entries.               if !self.prepare_send_snapshot(&mut m, pr, to) {                    return false;                }          }      }  }     fn prepare_send_snapshot(&mut self, m: &mut Message, pr: &mut Progress, to: u64) -> bool {      let snapshot_r = self.raft_log.snapshot(pr.pending_request_snapshot, to);      // ...   }
}
raft-rs: src/raft_log.rsimpl RaftLog {  pub fn snapshot(&self, request_index: u64, to: u64) -> Result<Snapshot> {      // ...      self.store.snapshot(request_index, to)  }
}

components/raftstore/src/store/peer_storage.rsimpl PeerStorage {  pub fn snapshot(&self, request_index: u64, to: u64) -> raft::Result<Snapshot> {      // ...      let task = GenSnapTask::new(...);      let mut gen_snap_task = self.gen_snap_task.borrow_mut();      *gen_snap_task = Some(task);      Err(raft::Error::Store(          raft::StorageError::SnapshotTemporarilyUnavailable,      ))  }}

components/raftstore/src/store/peer.rsimpl Peer {  pub fn handle_raft_ready_append(...){      // ...      if !ready.committed_entries().is_empty() {          self.handle_raft_committed_entries(ctx, ready.take_committed_entries());      }            if let Some(mut gen_task) = self.mut_store().take_gen_snap_task() {          ctx.apply_router.schedule_task(self.region_id, ApplyTask::Snapshot(gen_task));      }  }    fn handle_raft_committed_entries<T>() {      let mut apply = Apply::new(...)      ctx.apply_router.schedule_task(self.region_id, ApplyTask::apply(apply));  }}

components/raftstore/src/store/fsm/apply.rsimpl ApplyFsm {  fn handle_tasks() {      loop {          match msg {              Msg::Snapshot(snap_task) => self.handle_snapshot(..., snap_task),          }      }  }
  fn handle_snapshot(..., snap_task: GenSnapTask) {      snap_task.generate_and_schedule_snapshot()  }}
impl GenSnapTask {  pub fn generate_and_schedule_snapshot(){      let snapshot = RegionTask::Gen {...}      region_sched.schedule(snapshot)  }}

components/raftstore/src/store/worker/region.rsimpl Runnable for Runner {  fn run(&mut self, task: Task<EK::Snapshot>) {      match task {          Task::Gen {...} => {               let ctx = SnapGenContext {...}              self.pool.spawn(async move {                  ctx.handle_gen(...)              }          }      }  }}
impl SnapGenContext {  fn generate_snap( ... ) {       let snap = box_try!(store::do_snapshot::<EK>(...));      notifier.try_send(snap)  }  fn handle_gen( ... ) {      self.generate_snap(...)  }}

src/server/raft_client.rsimpl AsyncRaftSender {  fn fill_msg(&mut self, ctx: &Context<'_>) {      // ...      if msg.get_message().has_snapshot() {          self.send_snapshot_sock(msg);          continue;      }      // ...  }    fn send_snapshot_sock(&self, msg: RaftMessage) {      if let Err(e) = self.snap_scheduler.schedule(SnapTask::Send {...}) {          // ...      }  }    }

src/server/snap.rsimpl Runnable for Runner {  fn run(&mut self, task: Task) {      match task {          Task::Send { addr, msg, cb } => {              let send_task = send_snap(...);              let task = async move {                  let res = match send_task {                      Err(e) => Err(e),                      Ok(f) => f.await,                  };                  // ...              }              self.pool.spawn(task);          }      }  }}
pub fn send_snap(...) {  // ... }

src/server/service/kv.rsimpl Tikv for Service {  fn snapshot(...) {      let task = SnapTask::Recv { stream, sink };      if let Err(e) = self.snap_scheduler.schedule(task) {...}  }}

src/server/snap.rsimpl Runnable for Runner {  fn run(&mut self, task: Task) {      match task {          Task::Recv { ... } => {              let task = async move {                  let result = recv_snap(...).await;              }              self.pool.spawn(task);          }      }  }}
fn recv_snap(...) {  let mut context = RecvSnapContext::new(head, &snap_mgr)?;  while let Some(item) = stream.next().await {      // ...  }  context.finish(raft_router)}

raft-rs: src/raft.rsimpl Raft {  fn handle_snapshot(&mut self, mut m: Message) {      if self.restore(m.take_snapshot()) {...}  }    pub fn restore(&mut self, snap: Snapshot) -> bool {      ...      self.raft_log.restore(snap);  }}

components/raftstore/src/store/peer_storage.rsimpl PeerStorage {  pub fn apply_snapshot() {...}    pub fn persist_snapshot(&mut self, res: &PersistSnapshotResult) {      self.schedule_applying_snapshot();  }    pub fn schedule_applying_snapshot(&mut self) {      let task = RegionTask::Apply {}      if let Err(e) = self.region_scheduler.schedule(task) {...}  }    }

components/raftstore/src/store/worker/region.rsimpl Runnable for Runner {  fn run(&mut self, task: Task<EK::Snapshot>) {      match task {          Task::Apply { .. } => {              self.pending_applies.push_back(task);              self.handle_pending_applies(false);              }      }  }    fn handle_pending_applies(&mut self, is_timeout: bool) {      while !self.pending_applies.is_empty() {          if let Some(Task::Apply { region_id, .. }) = self.pending_applies.front() {              if let Some(Task::Apply {}) = self.pending_applies.pop_front() {                  self.handle_apply(region_id, peer_id, status);              }          }      }  }    fn handle_apply(&mut self, region_id: u64, peer_id: u64, status: Arc<AtomicUsize>) {      match self.apply_snap(region_id, peer_id, Arc::clone(&status)) {...}  }    fn apply_snap(&mut self, region_id: u64, peer_id: u64, abort: Arc<AtomicUsize>) -> Result<()> {      // ...      }}