raftstore-v2: support tablet split

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

@@ -14,6 +14,7 @@ use crossbeam::channel::{Sender, TrySendError};
 use engine_traits::{Engines, KvEngine, RaftEngine, TabletFactory};
 use file_system::{set_io_type, IoType};
 use futures::{compat::Future01CompatExt, FutureExt};
+use keys::enc_end_key;
 use kvproto::{
     metapb::Store,
     raft_serverpb::{PeerState, RaftMessage},
@@ -264,6 +265,7 @@ impl<EK: KvEngine, ER: RaftEngine, T> StorePollerBuilder<EK, ER, T> {
     fn init(&self) -> Result<HashMap<u64, SenderFsmPair<EK, ER>>> {
         let mut regions = HashMap::default();
         let cfg = self.cfg.value();
+        let mut meta = self.store_meta.lock().unwrap();
         self.engine
             .for_each_raft_group::<Error, _>(&mut |region_id| {
                 assert_ne!(region_id, INVALID_ID);
@@ -277,8 +279,12 @@ impl<EK: KvEngine, ER: RaftEngine, T> StorePollerBuilder<EK, ER, T> {
                     Some(p) => p,
                     None => return Ok(()),
                 };
-                let pair = PeerFsm::new(&cfg, &*self.tablet_factory, storage)?;
-                let prev = regions.insert(region_id, pair);
+                let (sender, peer_fsm) = PeerFsm::new(&cfg, &*self.tablet_factory, storage)?;
+                let region = peer_fsm.as_ref().peer().region().clone();
+                meta.region_read_progress
+                    .insert(region_id, peer_fsm.as_ref().peer().read_progress().clone());
+
+                let prev = regions.insert(region_id, (sender, peer_fsm));
                 if let Some((_, p)) = prev {
                     return Err(box_err!(
                         "duplicate region {:?} vs {:?}",

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

@@ -220,6 +220,7 @@ impl<'a, EK: KvEngine, ER: RaftEngine, T: Transport> PeerFsmDelegate<'a, EK, ER,
                 }
                 PeerMsg::Tick(tick) => self.on_tick(tick),
                 PeerMsg::ApplyRes(res) => self.fsm.peer.on_apply_res(self.store_ctx, res),
+                PeerMsg::SplitInit(msg) => self.fsm.peer.on_split_init(self.store_ctx, msg),
                 PeerMsg::Start => self.on_start(),
                 PeerMsg::Noop => unimplemented!(),
                 PeerMsg::Persisted {

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

@@ -1,16 +1,21 @@
 // Copyright 2022 TiKV Project Authors. Licensed under Apache-2.0.
 
-use std::time::SystemTime;
+use std::{collections::BTreeMap, time::SystemTime};
 
 use batch_system::Fsm;
 use collections::HashMap;
 use engine_traits::{KvEngine, RaftEngine};
-use raftstore::store::{Config, ReadDelegate};
+use kvproto::{metapb::Region, raft_serverpb::RaftMessage};
+use raftstore::{
+    coprocessor::RegionChangeReason,
+    store::{Config, ReadDelegate, RegionReadProgressRegistry},
+};
 use slog::{o, Logger};
 use tikv_util::mpsc::{self, LooseBoundedSender, Receiver};
 
 use crate::{
     batch::StoreContext,
+    raft::Peer,
     router::{StoreMsg, StoreTick},
     tablet::CachedTablet,
 };
@@ -24,6 +29,8 @@ where
     pub readers: HashMap<u64, ReadDelegate>,
     /// region_id -> tablet cache
     pub tablet_caches: HashMap<u64, CachedTablet<E>>,
+    /// region_id -> `RegionReadProgress`
+    pub region_read_progress: RegionReadProgressRegistry,
 }
 
 impl<E> StoreMeta<E>
@@ -35,8 +42,21 @@ where
             store_id: None,
             readers: HashMap::default(),
             tablet_caches: HashMap::default(),
+            region_read_progress: RegionReadProgressRegistry::new(),
         }
     }
+
+    #[inline]
+    pub fn set_region<ER: RaftEngine>(
+        &mut self,
+        region: Region,
+        peer: &mut Peer<E, ER>,
+        reason: RegionChangeReason,
+        tablet_index: u64,
+    ) {
+        let reader = self.readers.get_mut(&region.get_id()).unwrap();
+        peer.set_region(reader, region, reason, tablet_index);
+    }
 }
 
 impl<E: KvEngine> Default for StoreMeta<E> {
@@ -149,6 +169,7 @@ impl<'a, EK: KvEngine, ER: RaftEngine, T> StoreFsmDelegate<'a, EK, ER, T> {
                 StoreMsg::Start => self.on_start(),
                 StoreMsg::Tick(tick) => self.on_tick(tick),
                 StoreMsg::RaftMessage(msg) => self.fsm.store.on_raft_message(self.store_ctx, msg),
+                StoreMsg::SplitInit(msg) => self.fsm.store.on_split_init(self.store_ctx, msg),
             }
         }
     }

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

@@ -4,10 +4,7 @@ mod conf_change;
 mod split;
 
 use engine_traits::{KvEngine, RaftEngine};
-use kvproto::{
-    raft_cmdpb::{AdminCmdType, AdminRequest, RaftCmdRequest},
-    raft_serverpb::PeerState,
-};
+use kvproto::raft_cmdpb::{AdminCmdType, AdminRequest, RaftCmdRequest};
 use protobuf::Message;
 use raft::prelude::ConfChangeV2;
 use raftstore::{
@@ -20,7 +17,7 @@ use raftstore::{
     Result,
 };
 use slog::info;
-pub use split::SplitResult;
+pub use split::{SplitInit, SplitResult};
 use tikv_util::box_err;
 
 use self::conf_change::ConfChangeResult;