raftstore-v2: add replay watch

components/raftstore-v2/src/batch/store.rs

@@ -54,7 +54,9 @@ use time::Timespec;
 
 use crate::{
     fsm::{PeerFsm, PeerFsmDelegate, SenderFsmPair, StoreFsm, StoreFsmDelegate, StoreMeta},
-    operation::{SharedReadTablet, MERGE_IN_PROGRESS_PREFIX, MERGE_SOURCE_PREFIX, SPLIT_PREFIX},
+    operation::{
+        ReplayWatch, SharedReadTablet, MERGE_IN_PROGRESS_PREFIX, MERGE_SOURCE_PREFIX, SPLIT_PREFIX,
+    },
     raft::Storage,
     router::{PeerMsg, PeerTick, StoreMsg},
     worker::{checkpoint, pd, tablet},
@@ -728,8 +730,11 @@ impl<EK: KvEngine, ER: RaftEngine> StoreSystem<EK, ER> {
         router.register_all(mailboxes);
 
         // Make sure Msg::Start is the first message each FSM received.
+        let watch = Arc::new(ReplayWatch::new(self.logger.clone()));
         for addr in address {
-            router.force_send(addr, PeerMsg::Start).unwrap();
+            router
+                .force_send(addr, PeerMsg::Start(Some(watch.clone())))
+                .unwrap();
         }
         router.send_control(StoreMsg::Start).unwrap();
         Ok(())

components/raftstore-v2/src/fsm/peer.rs

@@ -2,7 +2,7 @@
 
 //! This module contains the peer implementation for batch system.
 
-use std::borrow::Cow;
+use std::{borrow::Cow, sync::Arc};
 
 use batch_system::{BasicMailbox, Fsm};
 use crossbeam::channel::TryRecvError;
@@ -20,6 +20,7 @@ use tikv_util::{
 
 use crate::{
     batch::StoreContext,
+    operation::ReplayWatch,
     raft::{Peer, Storage},
     router::{PeerMsg, PeerTick, QueryResult},
     Result,
@@ -187,8 +188,12 @@ impl<'a, EK: KvEngine, ER: RaftEngine, T: Transport> PeerFsmDelegate<'a, EK, ER,
         self.store_ctx.tick_batch[idx].ticks.push(cb);
     }
 
-    fn on_start(&mut self) {
-        if !self.fsm.peer.maybe_pause_for_recovery(self.store_ctx) {
+    fn on_start(&mut self, watch: Option<Arc<ReplayWatch>>) {
+        if !self
+            .fsm
+            .peer
+            .maybe_pause_for_recovery(self.store_ctx, watch)
+        {
             self.schedule_tick(PeerTick::Raft);
         }
         self.schedule_tick(PeerTick::SplitRegionCheck);
@@ -269,7 +274,7 @@ impl<'a, EK: KvEngine, ER: RaftEngine, T: Transport> PeerFsmDelegate<'a, EK, ER,
                 PeerMsg::SplitInitFinish(region_id) => {
                     self.fsm.peer.on_split_init_finish(region_id)
                 }
-                PeerMsg::Start => self.on_start(),
+                PeerMsg::Start(w) => self.on_start(w),
                 PeerMsg::Noop => unimplemented!(),
                 PeerMsg::Persisted {
                     peer_id,

components/raftstore-v2/src/operation/command/mod.rs

@@ -405,7 +405,7 @@ impl<EK: KvEngine, ER: RaftEngine> Peer<EK, ER> {
             progress_to_be_updated,
         );
         self.try_compelete_recovery();
-        if !self.pause_for_recovery() && self.storage_mut().apply_trace_mut().should_flush() {
+        if !self.pause_for_replay() && self.storage_mut().apply_trace_mut().should_flush() {
             if let Some(scheduler) = self.apply_scheduler() {
                 scheduler.send(ApplyTask::ManualFlush);
             }

components/raftstore-v2/src/operation/life.rs

@@ -455,7 +455,7 @@ impl Store {
         let mailbox = BasicMailbox::new(tx, fsm, ctx.router.state_cnt().clone());
         if ctx
             .router
-            .send_and_register(region_id, mailbox, PeerMsg::Start)
+            .send_and_register(region_id, mailbox, PeerMsg::Start(None))
             .is_err()
         {
             panic!(

components/raftstore-v2/src/operation/mod.rs

@@ -16,7 +16,8 @@ pub use command::{
 };
 pub use life::{AbnormalPeerContext, DestroyProgress, GcPeerContext};
 pub use ready::{
-    write_initial_states, ApplyTrace, AsyncWriter, DataTrace, GenSnapTask, SnapState, StateStorage,
+    write_initial_states, ApplyTrace, AsyncWriter, DataTrace, GenSnapTask, ReplayWatch, SnapState,
+    StateStorage,
 };
 
 pub(crate) use self::{

components/raftstore-v2/src/operation/ready/mod.rs

@@ -21,7 +21,15 @@ mod apply_trace;
 mod async_writer;
 mod snapshot;
 
-use std::{cmp, time::Instant};
+use std::{
+    cmp,
+    fmt::{self, Debug, Formatter},
+    sync::{
+        atomic::{AtomicUsize, Ordering},
+        Arc,
+    },
+    time::Instant,
+};
 
 use engine_traits::{KvEngine, RaftEngine};
 use error_code::ErrorCodeExt;
@@ -41,7 +49,7 @@ use raftstore::{
         FetchedLogs, ReadProgress, Transport, WriteCallback, WriteTask,
     },
 };
-use slog::{debug, error, info, trace, warn};
+use slog::{debug, error, info, trace, warn, Logger};
 use tikv_util::{
     log::SlogFormat,
     slog_panic,
@@ -65,6 +73,53 @@ use crate::{
 
 const PAUSE_FOR_RECOVERY_GAP: u64 = 128;
 
+pub struct ReplayWatch {
+    skipped: AtomicUsize,
+    paused: AtomicUsize,
+    logger: Logger,
+    timer: Instant,
+}
+
+impl Debug for ReplayWatch {
+    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
+        f.debug_struct("ReplayWatch")
+            .field("skipped", &self.skipped)
+            .field("paused", &self.paused)
+            .field("logger", &self.logger)
+            .field("timer", &self.timer)
+            .finish()
+    }
+}
+
+impl ReplayWatch {
+    pub fn new(logger: Logger) -> Self {
+        Self {
+            skipped: AtomicUsize::new(0),
+            paused: AtomicUsize::new(0),
+            logger,
+            timer: Instant::now(),
+        }
+    }
+
+    pub fn record_skipped(&self) {
+        self.skipped.fetch_add(1, Ordering::Relaxed);
+    }
+
+    pub fn record_paused(&self) {
+        self.paused.fetch_add(1, Ordering::Relaxed);
+    }
+}
+
+impl Drop for ReplayWatch {
+    fn drop(&mut self) {
+        info!(self.logger, 
+            "The raft log replay completed"; 
+            "skipped" => self.skipped.load(Ordering::Relaxed), 
+            "paused" => self.paused.load(Ordering::Relaxed), 
+            "elapsed" => ?self.timer.elapsed());
+    }
+}
+
 impl Store {
     pub fn on_store_unreachable<EK, ER, T>(
         &mut self,
@@ -115,7 +170,11 @@ impl<'a, EK: KvEngine, ER: RaftEngine, T: Transport> PeerFsmDelegate<'a, EK, ER,
 }
 
 impl<EK: KvEngine, ER: RaftEngine> Peer<EK, ER> {
-    pub fn maybe_pause_for_recovery<T>(&mut self, store_ctx: &mut StoreContext<EK, ER, T>) -> bool {
+    pub fn maybe_pause_for_recovery<T>(
+        &mut self,
+        store_ctx: &mut StoreContext<EK, ER, T>,
+        watch: Option<Arc<ReplayWatch>>,
+    ) -> bool {
         // The task needs to be scheduled even if the tablet may be replaced during
         // recovery. Otherwise if there are merges during recovery, the FSM may
         // be paused forever.
@@ -139,14 +198,18 @@ impl<EK: KvEngine, ER: RaftEngine> Peer<EK, ER> {
             // it may block for ever when there is unapplied conf change.
             self.set_has_ready();
         }
-        if committed_index > applied_index + PAUSE_FOR_RECOVERY_GAP {
+        if committed_index > applied_index + PAUSE_FOR_RECOVERY_GAP && let Some(w) = watch {
             // If there are too many the missing logs, we need to skip ticking otherwise
             // it may block the raftstore thread for a long time in reading logs for
             // election timeout.
             info!(self.logger, "pause for recovery"; "applied" => applied_index, "committed" => committed_index);
-            self.set_pause_for_recovery(true);
+            w.record_paused();
+            self.set_replay_watch(Some(w));
             true
         } else {
+            if let Some(w) = watch {
+                w.record_skipped();
+            }
             false
         }
     }
@@ -189,7 +252,7 @@ impl<EK: KvEngine, ER: RaftEngine> Peer<EK, ER> {
             "from_peer_id" => msg.get_from_peer().get_id(),
             "to_peer_id" => msg.get_to_peer().get_id(),
         );
-        if self.pause_for_recovery() && msg.get_message().get_msg_type() == MessageType::MsgAppend {
+        if self.pause_for_replay() && msg.get_message().get_msg_type() == MessageType::MsgAppend {
             ctx.raft_metrics.message_dropped.recovery.inc();
             return;
         }

components/raftstore-v2/src/raft/peer.rs

@@ -35,8 +35,8 @@ use crate::{
     fsm::ApplyScheduler,
     operation::{
         AbnormalPeerContext, AsyncWriter, BucketStatsInfo, CompactLogContext, DestroyProgress,
-        GcPeerContext, MergeContext, ProposalControl, SimpleWriteReqEncoder, SplitFlowControl,
-        TxnContext,
+        GcPeerContext, MergeContext, ProposalControl, ReplayWatch, SimpleWriteReqEncoder,
+        SplitFlowControl, TxnContext,
     },
     router::{ApplyTask, CmdResChannel, PeerTick, QueryResChannel},
     Result,
@@ -76,7 +76,7 @@ pub struct Peer<EK: KvEngine, ER: RaftEngine> {
     has_ready: bool,
     /// Sometimes there is no ready at all, but we need to trigger async write.
     has_extra_write: bool,
-    pause_for_recovery: bool,
+    replay_watch: Option<Arc<ReplayWatch>>,
     /// Writer for persisting side effects asynchronously.
     pub(crate) async_writer: AsyncWriter<EK, ER>,
 
@@ -179,7 +179,7 @@ impl<EK: KvEngine, ER: RaftEngine> Peer<EK, ER> {
             apply_scheduler: None,
             has_ready: false,
             has_extra_write: false,
-            pause_for_recovery: false,
+            replay_watch: None,
             destroy_progress: DestroyProgress::None,
             raft_group,
             logger,
@@ -468,21 +468,21 @@ impl<EK: KvEngine, ER: RaftEngine> Peer<EK, ER> {
     }
 
     #[inline]
-    pub fn set_pause_for_recovery(&mut self, pause: bool) {
-        self.pause_for_recovery = pause;
+    pub fn set_replay_watch(&mut self, watch: Option<Arc<ReplayWatch>>) {
+        self.replay_watch = watch;
     }
 
     #[inline]
-    pub fn pause_for_recovery(&self) -> bool {
-        self.pause_for_recovery
+    pub fn pause_for_replay(&self) -> bool {
+        self.replay_watch.is_some()
     }
 
     #[inline]
     // we may have skipped scheduling raft tick when start due to noticable gap
     // between commit index and apply index. We should scheduling it when raft log
     // apply catches up.
     pub fn try_compelete_recovery(&mut self) {
-        if self.pause_for_recovery()
+        if self.pause_for_replay()
             && self.storage().entry_storage().commit_index()
                 <= self.storage().entry_storage().applied_index()
         {
@@ -491,7 +491,7 @@ impl<EK: KvEngine, ER: RaftEngine> Peer<EK, ER> {
                 "recovery completed";
                 "apply_index" =>  self.storage().entry_storage().applied_index()
             );
-            self.set_pause_for_recovery(false);
+            self.set_replay_watch(None);
             // Flush to avoid recover again and again.
             if let Some(scheduler) = self.apply_scheduler() {
                 scheduler.send(ApplyTask::ManualFlush);

components/raftstore-v2/src/router/message.rs

@@ -1,6 +1,7 @@
 // Copyright 2022 TiKV Project Authors. Licensed under Apache-2.0.
 
 // #[PerformanceCriticalPath]
+use std::sync::Arc;
 
 use kvproto::{
     import_sstpb::SstMeta,
@@ -21,7 +22,7 @@ use super::response_channel::{
     QueryResSubscriber,
 };
 use crate::{
-    operation::{CatchUpLogs, RequestHalfSplit, RequestSplit, SplitInit},
+    operation::{CatchUpLogs, ReplayWatch, RequestHalfSplit, RequestSplit, SplitInit},
     router::ApplyRes,
 };
 
@@ -167,7 +168,7 @@ pub enum PeerMsg {
     LogsFetched(FetchedLogs),
     SnapshotGenerated(GenSnapRes),
     /// Start the FSM.
-    Start,
+    Start(Option<Arc<ReplayWatch>>),
     /// Messages from peer to peer in the same store
     SplitInit(Box<SplitInit>),
     SplitInitFinish(u64),