Merge branch 'master' into xlu/sync-api

Author: Xuhui Lu

.github/workflows/ci.yml

@@ -20,6 +20,7 @@ jobs:
       - uses: actions-rs/cargo@v1
         with:
           command: check
+          args: --all-targets --all-features
 
   fmt:
     name: rustfmt
@@ -49,7 +50,7 @@ jobs:
       - uses: actions-rs/clippy-check@v1
         with:
           token: ${{ secrets.GITHUB_TOKEN }}
-          args: --all-features
+          args: --all-targets --all-features -- -D clippy::all
           name: Clippy Output
   unit-test:
     name: unit test
@@ -98,7 +99,6 @@ jobs:
           path: |
             ~/.cargo/.crates.toml
             ~/.cargo/.crates2.json
-            ~/.cargo/bin
             ~/.cargo/registry/index
             ~/.cargo/registry/cache
           key: ${{ runner.os }}-build-${{ env.cache-name }}-${{ hashFiles('Cargo.lock') }}

Cargo.toml

@@ -34,7 +34,8 @@ serde_derive = "1.0"
 slog = { version = "2.3", features = ["max_level_trace", "release_max_level_debug"] }
 slog-term = { version = "2.4" }
 thiserror = "1"
-tokio = { version = "1.0", features = [ "sync", "time" ] }
+tokio = { version = "1", features = [ "sync", "rt-multi-thread", "macros" ] }
+async-recursion = "0.3"
 
 tikv-client-common = { version = "0.1.0", path = "tikv-client-common" }
 tikv-client-pd = { version = "0.1.0", path = "tikv-client-pd" }
@@ -50,7 +51,7 @@ proptest = "1"
 proptest-derive = "0.3"
 serial_test = "0.5.0"
 simple_logger = "1"
-tokio = { version = "1.0", features = [ "sync", "rt-multi-thread", "macros" ] }
+tokio = { version = "1", features = [ "sync", "rt-multi-thread", "macros" ] }
 reqwest = {version = "0.11", default-features = false, features = ["native-tls-vendored"]}
 serde_json = "1"
 

Makefile

@@ -3,9 +3,9 @@
 default: check
 
 check:
-	cargo check --all
+	cargo check --all --all-targets --all-features
 	cargo fmt -- --check
-	cargo clippy -- -D clippy::all
+	cargo clippy --all-targets --all-features -- -D clippy::all
 
 unit-test:
 	cargo test --all

src/backoff.rs

@@ -76,6 +76,11 @@ impl Backoff {
         self.kind == BackoffKind::None
     }
 
+    /// Returns the number of attempts
+    pub fn current_attempts(&self) -> u32 {
+        self.current_attempts
+    }
+
     /// Don't wait. Usually indicates that we should not retry a request.
     pub const fn no_backoff() -> Backoff {
         Backoff {

src/lib.rs

@@ -104,6 +104,7 @@ mod pd;
 #[doc(hidden)]
 pub mod raw;
 mod region;
+mod region_cache;
 mod stats;
 mod store;
 mod timestamp;

src/mock.rs

@@ -7,12 +7,13 @@
 
 use crate::{
     pd::{PdClient, PdRpcClient, RetryClient},
-    region::{Region, RegionId},
-    store::Store,
+    region::{RegionId, RegionWithLeader},
+    store::RegionStore,
     Config, Error, Key, Result, Timestamp,
 };
 use async_trait::async_trait;
 use derive_new::new;
+use slog::{Drain, Logger};
 use std::{any::Any, sync::Arc};
 use tikv_client_proto::metapb;
 use tikv_client_store::{KvClient, KvConnect, Request};
@@ -21,8 +22,16 @@ use tikv_client_store::{KvClient, KvConnect, Request};
 /// client can be tested without doing any RPC calls.
 pub async fn pd_rpc_client() -> PdRpcClient<MockKvConnect, MockCluster> {
     let config = Config::default();
+    let plain = slog_term::PlainSyncDecorator::new(std::io::stdout());
+    let logger = Logger::root(
+        slog_term::FullFormat::new(plain)
+            .build()
+            .filter_level(slog::Level::Info)
+            .fuse(),
+        o!(),
+    );
     PdRpcClient::new(
-        &config,
+        config.clone(),
         |_, _| MockKvConnect,
         |e, sm| {
             futures::future::ok(RetryClient::new_with_cluster(
@@ -33,11 +42,13 @@ pub async fn pd_rpc_client() -> PdRpcClient<MockKvConnect, MockCluster> {
             ))
         },
         false,
+        logger,
     )
     .await
     .unwrap()
 }
 
+#[allow(clippy::type_complexity)]
 #[derive(new, Default, Clone)]
 pub struct MockKvClient {
     pub addr: String,
@@ -93,27 +104,31 @@ impl MockPdClient {
         }
     }
 
-    pub fn region1() -> Region {
-        let mut region = Region::default();
+    pub fn region1() -> RegionWithLeader {
+        let mut region = RegionWithLeader::default();
         region.region.id = 1;
         region.region.set_start_key(vec![0]);
         region.region.set_end_key(vec![10]);
 
-        let mut leader = metapb::Peer::default();
-        leader.store_id = 41;
+        let leader = metapb::Peer {
+            store_id: 41,
+            ..Default::default()
+        };
         region.leader = Some(leader);
 
         region
     }
 
-    pub fn region2() -> Region {
-        let mut region = Region::default();
+    pub fn region2() -> RegionWithLeader {
+        let mut region = RegionWithLeader::default();
         region.region.id = 2;
         region.region.set_start_key(vec![10]);
         region.region.set_end_key(vec![250, 250]);
 
-        let mut leader = metapb::Peer::default();
-        leader.store_id = 42;
+        let leader = metapb::Peer {
+            store_id: 42,
+            ..Default::default()
+        };
         region.leader = Some(leader);
 
         region
@@ -124,11 +139,11 @@ impl MockPdClient {
 impl PdClient for MockPdClient {
     type KvClient = MockKvClient;
 
-    async fn map_region_to_store(self: Arc<Self>, region: Region) -> Result<Store> {
-        Ok(Store::new(region, Arc::new(self.client.clone())))
+    async fn map_region_to_store(self: Arc<Self>, region: RegionWithLeader) -> Result<RegionStore> {
+        Ok(RegionStore::new(region, Arc::new(self.client.clone())))
     }
 
-    async fn region_for_key(&self, key: &Key) -> Result<Region> {
+    async fn region_for_key(&self, key: &Key) -> Result<RegionWithLeader> {
         let bytes: &[_] = key.into();
         let region = if bytes.is_empty() || bytes[0] < 10 {
             Self::region1()
@@ -139,11 +154,11 @@ impl PdClient for MockPdClient {
         Ok(region)
     }
 
-    async fn region_for_id(&self, id: RegionId) -> Result<Region> {
+    async fn region_for_id(&self, id: RegionId) -> Result<RegionWithLeader> {
         match id {
             1 => Ok(Self::region1()),
             2 => Ok(Self::region2()),
-            _ => Err(Error::RegionNotFound { region_id: id }),
+            _ => Err(Error::RegionNotFoundInResponse { region_id: id }),
         }
     }
 
@@ -154,11 +169,21 @@ impl PdClient for MockPdClient {
     async fn update_safepoint(self: Arc<Self>, _safepoint: u64) -> Result<bool> {
         unimplemented!()
     }
+
+    async fn update_leader(
+        &self,
+        _ver_id: crate::region::RegionVerId,
+        _leader: metapb::Peer,
+    ) -> Result<()> {
+        todo!()
+    }
+
+    async fn invalidate_region_cache(&self, _ver_id: crate::region::RegionVerId) {}
 }
 
-pub fn mock_store() -> Store {
-    Store {
-        region: Region::default(),
+pub fn mock_store() -> RegionStore {
+    RegionStore {
+        region_with_leader: RegionWithLeader::default(),
         client: Arc::new(MockKvClient::new("foo".to_owned(), None)),
     }
 }