raftstore-v2: implement local read for raftstore-v2

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

@@ -1,10 +1,8 @@
 // Copyright 2022 TiKV Project Authors. Licensed under Apache-2.0.
 
 use std::{
-    cell::Cell,
-    mem,
     ops::{Deref, DerefMut},
-    sync::{atomic::AtomicUsize, Arc, Mutex},
+    sync::{Arc, Mutex},
     time::Duration,
 };
 
@@ -23,7 +21,7 @@ use kvproto::{
 use raft::INVALID_ID;
 use raftstore::store::{
     fsm::store::PeerTickBatch, local_metrics::RaftMetrics, Config, RaftlogFetchRunner,
-    RaftlogFetchTask, StoreWriters, Transport, WriteMsg, WriteSenders,
+    RaftlogFetchTask, StoreWriters, Transport, WriteSenders,
 };
 use slog::Logger;
 use tikv_util::{
@@ -42,8 +40,8 @@ use time::Timespec;
 
 use crate::{
     fsm::{PeerFsm, PeerFsmDelegate, SenderFsmPair, StoreFsm, StoreFsmDelegate, StoreMeta},
-    raft::{Peer, Storage},
-    router::{PeerMsg, PeerTick, QueryResChannel, StoreMsg},
+    raft::Storage,
+    router::{PeerMsg, PeerTick, StoreMsg},
     Error, Result,
 };
 
@@ -390,7 +388,7 @@ impl<EK: KvEngine, ER: RaftEngine> StoreSystem<EK, ER> {
             log_fetch_scheduler,
             &mut workers.store_writers,
             self.logger.clone(),
-            store_meta,
+            store_meta.clone(),
         );
         self.workers = Some(workers);
         let peers = builder.init()?;
@@ -401,12 +399,20 @@ impl<EK: KvEngine, ER: RaftEngine> StoreSystem<EK, ER> {
 
         let mut mailboxes = Vec::with_capacity(peers.len());
         let mut address = Vec::with_capacity(peers.len());
-        for (region_id, (tx, fsm)) in peers {
-            address.push(region_id);
-            mailboxes.push((
-                region_id,
-                BasicMailbox::new(tx, fsm, router.state_cnt().clone()),
-            ));
+        {
+            let mut meta = store_meta.as_ref().lock().unwrap();
+            for (region_id, (tx, fsm)) in peers {
+                meta.readers
+                    .insert(region_id, fsm.peer().generate_read_delegate());
+                meta.tablet_caches
+                    .insert(region_id, fsm.peer().tablet().clone());
+
+                address.push(region_id);
+                mailboxes.push((
+                    region_id,
+                    BasicMailbox::new(tx, fsm, router.state_cnt().clone()),
+                ));
+            }
         }
         router.register_all(mailboxes);
 

components/raftstore-v2/src/bootstrap.rs

@@ -7,7 +7,7 @@ use error_code::ErrorCodeExt;
 use fail::fail_point;
 use kvproto::{
     metapb::{Region, Store},
-    raft_serverpb::{RaftLocalState, RegionLocalState, StoreIdent},
+    raft_serverpb::{RaftLocalState, StoreIdent},
 };
 use pd_client::PdClient;
 use raft::INVALID_ID;

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

@@ -7,12 +7,11 @@ use std::borrow::Cow;
 use batch_system::{BasicMailbox, Fsm};
 use crossbeam::channel::TryRecvError;
 use engine_traits::{KvEngine, RaftEngine, TabletFactory};
-use kvproto::metapb;
 use raftstore::store::{Config, Transport};
 use slog::{debug, error, info, trace, Logger};
 use tikv_util::{
     is_zero_duration,
-    mpsc::{self, LooseBoundedSender, Receiver, Sender},
+    mpsc::{self, LooseBoundedSender, Receiver},
     time::{duration_to_sec, Instant},
     yatp_pool::FuturePool,
 };

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

@@ -4,7 +4,6 @@ use std::time::SystemTime;
 
 use batch_system::Fsm;
 use collections::HashMap;
-use crossbeam::channel::TryRecvError;
 use engine_traits::{KvEngine, RaftEngine};
 use raftstore::store::{Config, ReadDelegate};
 use slog::{o, Logger};

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

@@ -11,7 +11,6 @@ use raftstore::{
     },
     Result,
 };
-use tikv_util::Either;
 
 use crate::{
     batch::StoreContext,

components/raftstore-v2/src/operation/command/write/simple_write.rs

@@ -1,11 +1,8 @@
 // Copyright 2022 TiKV Project Authors. Licensed under Apache-2.0.
 
-use std::borrow::Cow;
-
 use engine_traits::{CF_DEFAULT, CF_LOCK, CF_WRITE};
 use kvproto::raft_cmdpb::{CmdType, RaftCmdRequest, RaftRequestHeader, Request};
 use protobuf::{CodedInputStream, Message, SingularPtrField};
-use tikv_util::Either;
 
 use crate::router::CmdResChannel;
 

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

@@ -25,7 +25,7 @@ use tikv_util::store::find_peer;
 
 use crate::{
     batch::StoreContext,
-    fsm::{PeerFsm, Store, StoreFsmDelegate},
+    fsm::{PeerFsm, Store},
     raft::{Peer, Storage},
     router::PeerMsg,
 };

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

@@ -8,3 +8,5 @@ mod ready;
 pub use command::{CommittedEntries, SimpleWriteDecoder, SimpleWriteEncoder};
 pub use life::DestroyProgress;
 pub use ready::AsyncWriter;
+
+pub(crate) use self::query::LocalReader;

components/raftstore-v2/src/operation/query/lease.rs

@@ -3,33 +3,22 @@
 use std::sync::{Arc, Mutex};
 
 use engine_traits::{KvEngine, RaftEngine};
-use kvproto::{
-    kvrpcpb::ExtraOp as TxnExtraOp,
-    raft_cmdpb::{self, RaftCmdRequest, RaftCmdResponse},
+use kvproto::raft_cmdpb::RaftCmdRequest;
+use raftstore::store::{
+    can_amend_read, fsm::apply::notify_stale_req, metrics::RAFT_READ_INDEX_PENDING_COUNT,
+    msg::ReadCallback, propose_read_index, should_renew_lease, util::LeaseState, ReadDelegate,
+    ReadIndexRequest, ReadProgress, TrackVer, Transport,
 };
-use raftstore::{
-    store::{
-        can_amend_read, cmd_resp,
-        fsm::{apply::notify_stale_req, Proposal},
-        metrics::RAFT_READ_INDEX_PENDING_COUNT,
-        msg::{ErrorCallback, ReadCallback},
-        propose_read_index, should_renew_lease,
-        util::{check_region_epoch, LeaseState},
-        ReadDelegate, ReadIndexRequest, ReadProgress, TrackVer, Transport,
-    },
-    Error,
-};
-use slog::{debug, error, info, o, Logger};
-use tikv_util::{box_err, time::monotonic_raw_now, Either};
+use slog::debug;
+use tikv_util::time::monotonic_raw_now;
 use time::Timespec;
 use tracker::GLOBAL_TRACKERS;
 
 use crate::{
     batch::StoreContext,
     fsm::StoreMeta,
     raft::Peer,
-    router::{CmdResChannel, QueryResChannel, QueryResult, ReadResponse},
-    Result,
+    router::{QueryResChannel, QueryResult, ReadResponse},
 };
 
 impl<EK: KvEngine, ER: RaftEngine> Peer<EK, ER> {