Merge remote-tracking branch 'upstream/master' into batch-get-for-update

Signed-off-by: ekexium <ekexium@gmail.com>

Author: ekexium

Cargo.toml

@@ -20,6 +20,7 @@ name = "tikv_client"
 [dependencies]
 async-trait = "0.1"
 derive-new = "0.5"
+fail = "0.4"
 futures = { version = "0.3", features = ["async-await", "thread-pool"] }
 futures-timer = "3.0"
 grpcio = { version = "0.8", features = [ "secure", "prost-codec", "use-bindgen" ], default-features = false }
@@ -31,7 +32,7 @@ regex = "1"
 serde = "1.0"
 serde_derive = "1.0"
 thiserror = "1"
-tokio = { version = "1.0", features = [ "sync" ] }
+tokio = { version = "1.0", features = [ "sync", "time" ] }
 
 tikv-client-common = { path = "tikv-client-common" }
 tikv-client-pd = { path = "tikv-client-pd" }
@@ -57,3 +58,8 @@ members = [
     "tikv-client-store",
     "mock-tikv"
 ]
+
+[[test]]
+name = "failpoint_tests"
+path = "tests/failpoint_tests.rs"
+required-features = ["fail/failpoints"]

rustfmt.toml

@@ -1,2 +1,3 @@
 imports_granularity="Crate"
 format_code_in_doc_comments = true
+edition = "2018"

src/mock.rs

@@ -147,7 +147,7 @@ impl PdClient for MockPdClient {
     }
 
     async fn get_timestamp(self: Arc<Self>) -> Result<Timestamp> {
-        unimplemented!()
+        Ok(Timestamp::default())
     }
 
     async fn update_safepoint(self: Arc<Self>, _safepoint: u64) -> Result<bool> {

src/transaction/client.rs

@@ -10,7 +10,6 @@ use crate::{
     transaction::{Snapshot, Transaction, TransactionOptions},
     Result,
 };
-use futures::executor::ThreadPool;
 use std::{mem, sync::Arc};
 use tikv_client_proto::{kvrpcpb, pdpb::Timestamp};
 
@@ -35,9 +34,6 @@ const SCAN_LOCK_BATCH_SIZE: u32 = 1024; // FIXME: cargo-culted value
 /// The returned results of transactional requests are [`Future`](std::future::Future)s that must be awaited to execute.
 pub struct Client {
     pd: Arc<PdRpcClient>,
-    /// The thread pool for background tasks including committing secondary keys and failed
-    /// transaction cleanups.
-    bg_worker: ThreadPool,
 }
 
 impl Client {
@@ -76,9 +72,8 @@ impl Client {
         config: Config,
     ) -> Result<Client> {
         let pd_endpoints: Vec<String> = pd_endpoints.into_iter().map(Into::into).collect();
-        let bg_worker = ThreadPool::new()?;
         let pd = Arc::new(PdRpcClient::connect(&pd_endpoints, &config, true).await?);
-        Ok(Client { pd, bg_worker })
+        Ok(Client { pd })
     }
 
     /// Creates a new [`Transaction`](Transaction) in optimistic mode.
@@ -223,6 +218,6 @@ impl Client {
     }
 
     fn new_transaction(&self, timestamp: Timestamp, options: TransactionOptions) -> Transaction {
-        Transaction::new(timestamp, self.bg_worker.clone(), self.pd.clone(), options)
+        Transaction::new(timestamp, self.pd.clone(), options)
     }
 }