Merge pull request #232 from nrc/plan

Refactor the command abstraction

Author: nrc

.travis.yml

@@ -12,8 +12,8 @@ rust:
 env:
   global:
     - RUST_BACKTRACE=1
-    - RUSTFLAGS="-D warnings"
-    - RUSTDOCFLAGS="-D warnings" 
+#    - RUSTFLAGS="-D warnings"
+#    - RUSTDOCFLAGS="-D warnings" 
 
 addons:
   apt:
@@ -43,7 +43,7 @@ script:
   - if [[ $TRAVIS_OS_NAME == "linux" && $TRAVIS_RUST_VERSION == "stable" ]]; then cargo clippy -- -D clippy::all; fi
   - cargo build --all
   - cargo build --examples
-  - if [[ $TRAVIS_OS_NAME == "linux" && $TRAVIS_RUST_VERSION == "nightly" ]]; then cargo doc --exclude tikv-client-proto --document-private-items; fi
+  - if [[ $TRAVIS_OS_NAME == "linux" && $TRAVIS_RUST_VERSION == "nightly" ]]; then cargo doc --workspace --exclude tikv-client-proto --document-private-items --no-deps; fi
   - if [[ $TRAVIS_RUST_VERSION == "nightly" ]]; then cargo test --all -- --nocapture; fi
   # For now we only run full integration tests on Linux. Here's why:
   # * Docker on OS X is not supported by Travis.
@@ -53,7 +53,7 @@ script:
   - if [[ $TRAVIS_OS_NAME == "linux" && $TRAVIS_RUST_VERSION == "nightly" ]]; then docker ps; fi
   - if [[ $TRAVIS_OS_NAME == "linux" && $TRAVIS_RUST_VERSION == "nightly" ]]; then docker logs pd; fi
   - if [[ $TRAVIS_OS_NAME == "linux" && $TRAVIS_RUST_VERSION == "nightly" ]]; then docker logs kv; fi
-  - if [[ $TRAVIS_OS_NAME == "linux" && $TRAVIS_RUST_VERSION == "nightly" ]]; then sleep 2; fi
+  - if [[ $TRAVIS_OS_NAME == "linux" && $TRAVIS_RUST_VERSION == "nightly" ]]; then sleep 60; fi
   - if [[ $TRAVIS_OS_NAME == "linux" && $TRAVIS_RUST_VERSION == "nightly" ]]; then PD_ADDRS="127.0.0.1:2379" cargo test --all --features integration-tests -- --nocapture; fi
   - if [[ $TRAVIS_OS_NAME == "linux" && $TRAVIS_RUST_VERSION == "nightly" ]]; then cargo run --example raw -- --pd="127.0.0.1:2379"; fi
   - if [[ $TRAVIS_OS_NAME == "linux" && $TRAVIS_RUST_VERSION == "nightly" ]]; then cargo run --example transaction -- --pd="127.0.0.1:2379"; fi

Cargo.toml

@@ -22,7 +22,7 @@ async-trait = "0.1"
 derive-new = "0.5"
 futures = { version = "0.3.5", features = ["async-await", "thread-pool"] }
 futures-timer = "3.0"
-grpcio = { version = "0.7", features = [ "secure", "prost-codec" ], default-features = false }
+grpcio = { version = "0.8", features = [ "secure", "prost-codec", "use-bindgen" ], default-features = false }
 lazy_static = "1"
 log = "0.4"
 prometheus = { version = "0.11", features = [ "push", "process" ], default-features = false } 

examples/pessimistic.rs

@@ -3,7 +3,7 @@
 mod common;
 
 use crate::common::parse_args;
-use tikv_client::{Config, Key, TransactionClient as Client, Value};
+use tikv_client::{Config, Key, TransactionClient as Client, TransactionOptions, Value};
 
 #[tokio::main]
 async fn main() {
@@ -50,7 +50,7 @@ async fn main() {
     {
         // another txn cannot write to the locked key
         let mut txn2 = client
-            .begin_optimistic()
+            .begin_with_options(TransactionOptions::new_optimistic().no_resolve_locks())
             .await
             .expect("Could not begin a transaction");
         let value2: Value = b"value2".to_vec();

mock-tikv/Cargo.toml

@@ -6,6 +6,6 @@ edition = "2018"
 [dependencies]
 derive-new = "0.5.8"
 futures = "0.3.5"
-grpcio = { version = "0.7", features = [ "secure", "prost-codec" ], default-features = false }
+grpcio = { version = "0.8", features = [ "secure", "prost-codec", "use-bindgen" ], default-features = false }
 log = "0.4"
 tikv-client-proto = { path = "../tikv-client-proto"}

rustfmt.toml

@@ -1,2 +1,2 @@
-merge_imports = true
+imports_granularity="Crate"
 format_code_in_doc_comments = true

src/backoff.rs

@@ -5,6 +5,10 @@
 use rand::{thread_rng, Rng};
 use std::time::Duration;
 
+pub const DEFAULT_REGION_BACKOFF: Backoff = Backoff::no_jitter_backoff(2, 500, 10);
+pub const OPTIMISTIC_BACKOFF: Backoff = Backoff::no_jitter_backoff(2, 500, 10);
+pub const PESSIMISTIC_BACKOFF: Backoff = Backoff::no_backoff();
+
 /// When a request is retried, we can backoff for some time to avoid saturating the network.
 ///
 /// `Backoff` is an object which determines how long to wait for.

src/compat.rs

@@ -18,6 +18,7 @@ use std::pin::Pin;
 pub struct LoopFn<A, F> {
     future: A,
     func: F,
+    errored: bool,
 }
 
 pub fn stream_fn<S, T, A, F, E>(initial_state: S, mut func: F) -> LoopFn<A, F>
@@ -28,6 +29,7 @@ where
     LoopFn {
         future: func(initial_state),
         func,
+        errored: false,
     }
 }
 
@@ -38,10 +40,16 @@ where
 {
     type Item = Result<T, E>;
     fn poll_next(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Option<Self::Item>> {
+        if self.errored {
+            return Poll::Ready(None);
+        }
         unsafe {
             let this = Pin::get_unchecked_mut(self);
             match ready!(Pin::new_unchecked(&mut this.future).poll(cx)) {
-                Err(e) => Poll::Ready(Some(Err(e))),
+                Err(e) => {
+                    this.errored = true;
+                    Poll::Ready(Some(Err(e)))
+                }
                 Ok(None) => Poll::Ready(None),
                 Ok(Some((s, t))) => {
                     this.future = (this.func)(s);

src/kv/bound_range.rs

@@ -75,6 +75,14 @@ impl BoundRange {
         BoundRange { from, to }
     }
 
+    /// Create a new BoundRange bounded below by `from` and unbounded above.
+    pub fn range_from(from: Key) -> BoundRange {
+        BoundRange {
+            from: Bound::Included(from),
+            to: Bound::Unbounded,
+        }
+    }
+
     /// Ranges used in scanning TiKV have a particularity to them.
     ///
     /// The **start** of a scan is inclusive, unless appended with an '\0', then it is exclusive.

src/kv/key.rs

@@ -152,6 +152,11 @@ impl<'a> From<&'a Key> for &'a [u8] {
     }
 }
 
+impl<'a> From<&'a Vec<u8>> for &'a Key {
+    fn from(key: &'a Vec<u8>) -> Self {
+        unsafe { &*(key as *const Vec<u8> as *const Key) }
+    }
+}
 impl AsRef<Key> for Key {
     fn as_ref(&self) -> &Key {
         self

src/kv/kvpair.rs

@@ -96,6 +96,12 @@ impl From<KvPair> for (Key, Value) {
     }
 }
 
+impl From<KvPair> for Key {
+    fn from(pair: KvPair) -> Self {
+        pair.0
+    }
+}
+
 impl From<kvrpcpb::KvPair> for KvPair {
     fn from(mut pair: kvrpcpb::KvPair) -> Self {
         KvPair(Key::from(pair.take_key()), pair.take_value())