tikv
/
client-rust
mirror of https://github.com/tikv/client-rust.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Improve rustfmt config (#401) 

Signed-off-by: Andy Lok <andylokandy@hotmail.com>
This commit is contained in:
Andy Lok 2023-07-09 17:07:13 +08:00 committed by GitHub
parent 0a1de19450
commit 797960edfb
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
55 changed files with 960 additions and 594 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
examples/common/mod.rs
View File

@ -1,8 +1,11 @@
// Copyright 2019 TiKV Project Authors. Licensed under Apache-2.0.
use clap::{crate_version, App, Arg};
use std::path::PathBuf;
use clap::crate_version;
use clap::App;
use clap::Arg;
pub struct CommandArgs {
pub pd: Vec<String>,
pub ca: Option<PathBuf>,

7
examples/pessimistic.rs
View File

@ -2,8 +2,13 @@
mod common;
use tikv_client::Config;
use tikv_client::Key;
use tikv_client::TransactionClient as Client;
use tikv_client::TransactionOptions;
use tikv_client::Value;
use crate::common::parse_args;
use tikv_client::{Config, Key, TransactionClient as Client, TransactionOptions, Value};
#[tokio::main]
async fn main() {

38
examples/raw.rs
View File

@ -4,8 +4,15 @@
mod common;
use tikv_client::Config;
use tikv_client::IntoOwnedRange;
use tikv_client::Key;
use tikv_client::KvPair;
use tikv_client::RawClient as Client;
use tikv_client::Result;
use tikv_client::Value;
use crate::common::parse_args;
use tikv_client::{Config, IntoOwnedRange, Key, KvPair, RawClient as Client, Result, Value};
const KEY: &str = "TiKV";
const VALUE: &str = "Rust";
@ -93,10 +100,10 @@ async fn main() -> Result<()> {
.expect("Could not scan");
let keys: Vec<_> = pairs.into_iter().map(|p| p.key().clone()).collect();
assert_eq!(
&keys,
&[Key::from("k1".to_owned()), Key::from("k2".to_owned()),]
);
assert_eq!(&keys, &[
Key::from("k1".to_owned()),
Key::from("k2".to_owned()),
]);
println!("Scanning from {start:?} to {end:?} gives: {keys:?}");
let k1 = "k1";
@ -115,18 +122,15 @@ async fn main() -> Result<()> {
.into_iter()
.map(|p| String::from_utf8(p.1).unwrap())
.collect();
assert_eq!(
&vals,
&[
"v1".to_owned(),
"v2".to_owned(),
"v2".to_owned(),
"v3".to_owned(),
"v1".to_owned(),
"v2".to_owned(),
"v3".to_owned()
]
);
assert_eq!(&vals, &[
"v1".to_owned(),
"v2".to_owned(),
"v2".to_owned(),
"v3".to_owned(),
"v1".to_owned(),
"v2".to_owned(),
"v3".to_owned()
]);
println!("Scanning batch scan from {batch_scan_keys:?} gives: {vals:?}");
// Cleanly exit.

16
examples/transaction.rs
View File

@ -2,8 +2,14 @@
mod common;
use tikv_client::BoundRange;
use tikv_client::Config;
use tikv_client::Key;
use tikv_client::KvPair;
use tikv_client::TransactionClient as Client;
use tikv_client::Value;
use crate::common::parse_args;
use tikv_client::{BoundRange, Config, Key, KvPair, TransactionClient as Client, Value};
async fn puts(client: &Client, pairs: impl IntoIterator<Item = impl Into<KvPair>>) {
let mut txn = client
@ -104,10 +110,10 @@ async fn main() {
let key1_exists = key_exists(&txn, key1.clone()).await;
let key2: Key = b"key_not_exist".to_vec().into();
let key2_exists = key_exists(&txn, key2.clone()).await;
println!(
"check exists {:?}",
vec![(key1, key1_exists), (key2, key2_exists)]
);
println!("check exists {:?}", vec![
(key1, key1_exists),
(key2, key2_exists)
]);
// scan
let key1: Key = b"key1".to_vec().into();

11
rustfmt.toml
View File

@ -1,3 +1,10 @@
imports_granularity="Crate"
edition = "2021"
version = "Two"
reorder_imports = true
imports_granularity = "Item"
group_imports = "StdExternalCrate"
where_single_line = true
trailing_comma = "Vertical"
overflow_delimited_expr = true
format_code_in_doc_comments = true
edition = "2018"
normalize_comments = true

7
src/backoff.rs
View File

@ -2,9 +2,11 @@
// https://aws.amazon.com/blogs/architecture/exponential-backoff-and-jitter/
use rand::{thread_rng, Rng};
use std::time::Duration;
use rand::thread_rng;
use rand::Rng;
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_jitter_backoff(2, 500, 10);
@ -198,9 +200,10 @@ enum BackoffKind {
#[cfg(test)]
mod test {
use super::*;
use std::convert::TryInto;
use super::*;
#[test]
fn test_no_jitter_backoff() {
// Tests for zero attempts.

10
src/compat.rs
View File

@ -3,13 +3,13 @@
//! This module contains utility types and functions for making the transition
//! from futures 0.1 to 1.0 easier.
use futures::{
prelude::*,
ready,
task::{Context, Poll},
};
use std::pin::Pin;
use futures::prelude::*;
use futures::ready;
use futures::task::Context;
use futures::task::Poll;
/// A future implementing a tail-recursive loop.
///
/// Created by the `loop_fn` function.

7
src/config.rs
View File

@ -1,7 +1,10 @@
// Copyright 2019 TiKV Project Authors. Licensed under Apache-2.0.
use serde_derive::{Deserialize, Serialize};
use std::{path::PathBuf, time::Duration};
use std::path::PathBuf;
use std::time::Duration;
use serde_derive::Deserialize;
use serde_derive::Serialize;
/// The configuration for either a [`RawClient`](crate::RawClient) or a
/// [`TransactionClient`](crate::TransactionClient).

18
src/kv/bound_range.rs
View File

@ -1,12 +1,16 @@
// Copyright 2019 TiKV Project Authors. Licensed under Apache-2.0.
use std::{
borrow::Borrow,
cmp::{Eq, PartialEq},
ops::{
Bound, Range, RangeBounds, RangeFrom, RangeFull, RangeInclusive, RangeTo, RangeToInclusive,
},
};
use std::borrow::Borrow;
use std::cmp::Eq;
use std::cmp::PartialEq;
use std::ops::Bound;
use std::ops::Range;
use std::ops::RangeBounds;
use std::ops::RangeFrom;
use std::ops::RangeFull;
use std::ops::RangeInclusive;
use std::ops::RangeTo;
use std::ops::RangeToInclusive;
#[cfg(test)]
use proptest_derive::Arbitrary;

36
src/kv/codec.rs
View File

@ -1,4 +1,6 @@
use std::{io::Write, ptr};
use std::io::Write;
use std::ptr;
use tikv_client_common::internal_err;
use crate::Result;
@ -153,26 +155,18 @@ pub mod test {
#[test]
fn test_enc_dec_bytes() {
let pairs = vec![
(
vec![],
vec![0, 0, 0, 0, 0, 0, 0, 0, 247],
vec![255, 255, 255, 255, 255, 255, 255, 255, 8],
),
(
vec![0],
vec![0, 0, 0, 0, 0, 0, 0, 0, 248],
vec![255, 255, 255, 255, 255, 255, 255, 255, 7],
),
(
vec![1, 2, 3],
vec![1, 2, 3, 0, 0, 0, 0, 0, 250],
vec![254, 253, 252, 255, 255, 255, 255, 255, 5],
),
(
vec![1, 2, 3, 0],
vec![1, 2, 3, 0, 0, 0, 0, 0, 251],
vec![254, 253, 252, 255, 255, 255, 255, 255, 4],
),
(vec![], vec![0, 0, 0, 0, 0, 0, 0, 0, 247], vec![
255, 255, 255, 255, 255, 255, 255, 255, 8,
]),
(vec![0], vec![0, 0, 0, 0, 0, 0, 0, 0, 248], vec![
255, 255, 255, 255, 255, 255, 255, 255, 7,
]),
(vec![1, 2, 3], vec![1, 2, 3, 0, 0, 0, 0, 0, 250], vec![
254, 253, 252, 255, 255, 255, 255, 255, 5,
]),
(vec![1, 2, 3, 0], vec![1, 2, 3, 0, 0, 0, 0, 0, 251], vec![
254, 253, 252, 255, 255, 255, 255, 255, 4,
]),
(
vec![1, 2, 3, 4, 5, 6, 7],
vec![1, 2, 3, 4, 5, 6, 7, 0, 254],

16
src/kv/key.rs
View File

@ -1,15 +1,23 @@
// Copyright 2019 TiKV Project Authors. Licensed under Apache-2.0.
use super::HexRepr;
use crate::kv::codec::{self, BytesEncoder};
use std::fmt;
use std::ops::Bound;
use std::u8;
#[allow(unused_imports)]
#[cfg(test)]
use proptest::{arbitrary::any_with, collection::size_range};
use proptest::arbitrary::any_with;
#[allow(unused_imports)]
#[cfg(test)]
use proptest::collection::size_range;
#[cfg(test)]
use proptest_derive::Arbitrary;
use std::{fmt, ops::Bound, u8};
use tikv_client_proto::kvrpcpb;
use super::HexRepr;
use crate::kv::codec::BytesEncoder;
use crate::kv::codec::{self};
const _PROPTEST_KEY_MAX: usize = 1024 * 2; // 2 KB
/// The key part of a key/value pair.

9
src/kv/kvpair.rs
View File

@ -1,11 +1,16 @@
// Copyright 2019 TiKV Project Authors. Licensed under Apache-2.0.
use super::{HexRepr, Key, Value};
use std::fmt;
use std::str;
#[cfg(test)]
use proptest_derive::Arbitrary;
use std::{fmt, str};
use tikv_client_proto::kvrpcpb;
use super::HexRepr;
use super::Key;
use super::Value;
/// A key/value pair.
///
/// # Examples

6
src/kv/mod.rs
View File

@ -1,5 +1,6 @@
// Copyright 2019 TiKV Project Authors. Licensed under Apache-2.0.
use std::{fmt, u8};
use std::fmt;
use std::u8;
mod bound_range;
pub mod codec;
@ -7,7 +8,8 @@ mod key;
mod kvpair;
mod value;
pub use bound_range::{BoundRange, IntoOwnedRange};
pub use bound_range::BoundRange;
pub use bound_range::IntoOwnedRange;
pub use key::Key;
pub use kvpair::KvPair;
pub use value::Value;

44
src/lib.rs
View File

@ -124,22 +124,48 @@ mod proptests;
extern crate slog;
extern crate slog_term;
#[doc(inline)]
pub use config::Config;
#[doc(inline)]
pub use tikv_client_common::security::SecurityManager;
#[doc(inline)]
pub use tikv_client_common::Error;
#[doc(inline)]
pub use tikv_client_common::Result;
#[doc(inline)]
pub use crate::backoff::Backoff;
#[doc(inline)]
pub use crate::kv::{BoundRange, IntoOwnedRange, Key, KvPair, Value};
pub use crate::kv::BoundRange;
#[doc(inline)]
pub use crate::raw::{lowering as raw_lowering, Client as RawClient, ColumnFamily};
pub use crate::kv::IntoOwnedRange;
#[doc(inline)]
pub use crate::kv::Key;
#[doc(inline)]
pub use crate::kv::KvPair;
#[doc(inline)]
pub use crate::kv::Value;
#[doc(inline)]
pub use crate::raw::lowering as raw_lowering;
#[doc(inline)]
pub use crate::raw::Client as RawClient;
#[doc(inline)]
pub use crate::raw::ColumnFamily;
#[doc(inline)]
pub use crate::request::RetryOptions;
#[doc(inline)]
pub use crate::timestamp::{Timestamp, TimestampExt};
pub use crate::timestamp::Timestamp;
#[doc(inline)]
pub use crate::transaction::{
lowering as transaction_lowering, CheckLevel, Client as TransactionClient, Snapshot,
Transaction, TransactionOptions,
};
pub use crate::timestamp::TimestampExt;
#[doc(inline)]
pub use config::Config;
pub use crate::transaction::lowering as transaction_lowering;
#[doc(inline)]
pub use tikv_client_common::{security::SecurityManager, Error, Result};
pub use crate::transaction::CheckLevel;
#[doc(inline)]
pub use crate::transaction::Client as TransactionClient;
#[doc(inline)]
pub use crate::transaction::Snapshot;
#[doc(inline)]
pub use crate::transaction::Transaction;
#[doc(inline)]
pub use crate::transaction::TransactionOptions;

36
src/mock.rs
View File

@ -5,18 +5,30 @@
//! The goal is to be able to test functionality independently of the rest of
//! the system, in particular without requiring a TiKV or PD server, or RPC layer.
use crate::{
pd::{PdClient, PdRpcClient, RetryClient},
region::{RegionId, RegionWithLeader},
store::RegionStore,
Config, Error, Key, Result, Timestamp,
};
use std::any::Any;
use std::sync::Arc;
use async_trait::async_trait;
use derive_new::new;
use slog::{Drain, Logger};
use std::{any::Any, sync::Arc};
use tikv_client_proto::metapb::{self, RegionEpoch};
use tikv_client_store::{KvClient, KvConnect, Request};
use slog::Drain;
use slog::Logger;
use tikv_client_proto::metapb::RegionEpoch;
use tikv_client_proto::metapb::{self};
use tikv_client_store::KvClient;
use tikv_client_store::KvConnect;
use tikv_client_store::Request;
use crate::pd::PdClient;
use crate::pd::PdRpcClient;
use crate::pd::RetryClient;
use crate::region::RegionId;
use crate::region::RegionWithLeader;
use crate::store::RegionStore;
use crate::Config;
use crate::Error;
use crate::Key;
use crate::Result;
use crate::Timestamp;
/// Create a `PdRpcClient` with it's internals replaced with mocks so that the
/// client can be tested without doing any RPC calls.
@ -56,9 +68,7 @@ pub struct MockKvClient {
impl MockKvClient {
pub fn with_dispatch_hook<F>(dispatch: F) -> MockKvClient
where
F: Fn(&dyn Any) -> Result<Box<dyn Any>> + Send + Sync + 'static,
{
where F: Fn(&dyn Any) -> Result<Box<dyn Any>> + Send + Sync + 'static {
MockKvClient {
addr: String::new(),
dispatch: Some(Arc::new(dispatch)),

78
src/pd/client.rs
View File

@ -1,23 +1,36 @@
// Copyright 2018 TiKV Project Authors. Licensed under Apache-2.0.
use crate::{
compat::stream_fn,
kv::codec,
pd::{retry::RetryClientTrait, RetryClient},
region::{RegionId, RegionVerId, RegionWithLeader},
region_cache::RegionCache,
store::RegionStore,
BoundRange, Config, Key, Result, SecurityManager, Timestamp,
};
use std::collections::HashMap;
use std::sync::Arc;
use async_trait::async_trait;
use futures::{prelude::*, stream::BoxStream};
use futures::prelude::*;
use futures::stream::BoxStream;
use slog::Logger;
use std::{collections::HashMap, sync::Arc};
use tikv_client_pd::Cluster;
use tikv_client_proto::{kvrpcpb, metapb};
use tikv_client_store::{KvClient, KvConnect, TikvConnect};
use tikv_client_proto::kvrpcpb;
use tikv_client_proto::metapb;
use tikv_client_store::KvClient;
use tikv_client_store::KvConnect;
use tikv_client_store::TikvConnect;
use tokio::sync::RwLock;
use crate::compat::stream_fn;
use crate::kv::codec;
use crate::pd::retry::RetryClientTrait;
use crate::pd::RetryClient;
use crate::region::RegionId;
use crate::region::RegionVerId;
use crate::region::RegionWithLeader;
use crate::region_cache::RegionCache;
use crate::store::RegionStore;
use crate::BoundRange;
use crate::Config;
use crate::Key;
use crate::Result;
use crate::SecurityManager;
use crate::Timestamp;
/// The PdClient handles all the encoding stuff.
///
/// Raw APIs does not require encoding/decoding at all.
@ -325,11 +338,12 @@ fn make_key_range(start_key: Vec<u8>, end_key: Vec<u8>) -> kvrpcpb::KeyRange {
#[cfg(test)]
pub mod test {
use futures::executor;
use futures::executor::block_on;
use super::*;
use crate::mock::*;
use futures::{executor, executor::block_on};
#[tokio::test]
async fn test_kv_client_caching() {
let client = block_on(pd_rpc_client());
@ -363,19 +377,16 @@ pub mod test {
let mut stream = executor::block_on_stream(stream);
let result: Vec<Key> = stream.next().unwrap().unwrap().1;
assert_eq!(
result,
vec![
vec![1].into(),
vec![2].into(),
vec![3].into(),
vec![5, 2].into()
]
);
assert_eq!(
stream.next().unwrap().unwrap().1,
vec![vec![12].into(), vec![11, 4].into()]
);
assert_eq!(result, vec![
vec![1].into(),
vec![2].into(),
vec![3].into(),
vec![5, 2].into()
]);
assert_eq!(stream.next().unwrap().unwrap().1, vec![
vec![12].into(),
vec![11, 4].into()
]);
assert!(stream.next().is_none());
}
@ -417,13 +428,10 @@ pub mod test {
let ranges4 = stream.next().unwrap().unwrap();
assert_eq!(ranges1.0.id(), 1);
assert_eq!(
ranges1.1,
vec![
make_key_range(k1.clone(), k2.clone()),
make_key_range(k1, k_split.clone()),
]
);
assert_eq!(ranges1.1, vec![
make_key_range(k1.clone(), k2.clone()),
make_key_range(k1, k_split.clone()),
]);
assert_eq!(ranges2.0.id(), 2);
assert_eq!(ranges2.1, vec![make_key_range(k_split.clone(), k3)]);
assert_eq!(ranges3.0.id(), 1);

6
src/pd/mod.rs
View File

@ -1,5 +1,7 @@
mod client;
mod retry;
pub use client::{PdClient, PdRpcClient};
pub use retry::{RetryClient, RetryClientTrait};
pub use client::PdClient;
pub use client::PdRpcClient;
pub use retry::RetryClient;
pub use retry::RetryClientTrait;

50
src/pd/retry.rs
View File

@ -2,23 +2,27 @@
//! A utility module for managing and retrying PD requests.
use crate::{
region::{RegionId, RegionWithLeader, StoreId},
stats::pd_stats,
Error, Result, SecurityManager,
};
use std::fmt;
use std::sync::Arc;
use std::time::Duration;
use std::time::Instant;
use async_trait::async_trait;
use std::{
fmt,
sync::Arc,
time::{Duration, Instant},
};
use tikv_client_pd::{Cluster, Connection};
use tikv_client_proto::{
metapb,
pdpb::{self, Timestamp},
};
use tokio::{sync::RwLock, time::sleep};
use tikv_client_pd::Cluster;
use tikv_client_pd::Connection;
use tikv_client_proto::metapb;
use tikv_client_proto::pdpb::Timestamp;
use tikv_client_proto::pdpb::{self};
use tokio::sync::RwLock;
use tokio::time::sleep;
use crate::region::RegionId;
use crate::region::RegionWithLeader;
use crate::region::StoreId;
use crate::stats::pd_stats;
use crate::Error;
use crate::Result;
use crate::SecurityManager;
// FIXME: these numbers and how they are used are all just cargo-culted in, there
// may be more optimal values.
@ -224,14 +228,16 @@ impl Reconnect for RetryClient<Cluster> {
#[cfg(test)]
mod test {
use super::*;
use futures::{executor, future::ready};
use std::sync::{
atomic::{AtomicUsize, Ordering},
Mutex,
};
use std::sync::atomic::AtomicUsize;
use std::sync::atomic::Ordering;
use std::sync::Mutex;
use futures::executor;
use futures::future::ready;
use tikv_client_common::internal_err;
use super::*;
#[tokio::test(flavor = "multi_thread")]
async fn test_reconnect() {
struct MockClient {

50
src/proptests/mod.rs
View File

@ -4,29 +4,27 @@
// directory tests don't have access to `cfg(tests)` functions and we don't want to force
// users to depend on proptest or manually enable features to test.
/*
* Temporarily disabled
use proptest::strategy::Strategy;
use std::env::var;
mod raw;
pub(crate) const ENV_PD_ADDRS: &str = "PD_ADDRS";
pub(crate) const PROPTEST_BATCH_SIZE_MAX: usize = 16;
pub fn arb_batch<T: core::fmt::Debug>(
single_strategy: impl Strategy<Value = T>,
max_batch_size: impl Into<Option<usize>>,
) -> impl Strategy<Value = Vec<T>> {
let max_batch_size = max_batch_size.into().unwrap_or(PROPTEST_BATCH_SIZE_MAX);
proptest::collection::vec(single_strategy, 0..max_batch_size)
}
pub fn pd_addrs() -> Vec<String> {
var(ENV_PD_ADDRS)
.expect(&format!("Expected {}:", ENV_PD_ADDRS))
.split(",")
.map(From::from)
.collect()
}
*/
// Temporarily disabled
//
// use proptest::strategy::Strategy;
// use std::env::var;
//
// mod raw;
// pub(crate) const ENV_PD_ADDRS: &str = "PD_ADDRS";
// pub(crate) const PROPTEST_BATCH_SIZE_MAX: usize = 16;
//
// pub fn arb_batch<T: core::fmt::Debug>(
// single_strategy: impl Strategy<Value = T>,
// max_batch_size: impl Into<Option<usize>>,
// ) -> impl Strategy<Value = Vec<T>> {
// let max_batch_size = max_batch_size.into().unwrap_or(PROPTEST_BATCH_SIZE_MAX);
// proptest::collection::vec(single_strategy, 0..max_batch_size)
// }
//
// pub fn pd_addrs() -> Vec<String> {
// var(ENV_PD_ADDRS)
// .expect(&format!("Expected {}:", ENV_PD_ADDRS))
// .split(",")
// .map(From::from)
// .collect()
// }

79
src/raw/client.rs
View File

@ -1,20 +1,30 @@
// Copyright 2019 TiKV Project Authors. Licensed under Apache-2.0.
use core::ops::Range;
use std::{str::FromStr, sync::Arc, u32};
use std::str::FromStr;
use std::sync::Arc;
use std::u32;
use slog::{Drain, Logger};
use slog::Drain;
use slog::Logger;
use tikv_client_common::Error;
use tikv_client_proto::metapb;
use crate::{
backoff::DEFAULT_REGION_BACKOFF,
config::Config,
pd::{PdClient, PdRpcClient},
raw::lowering::*,
request::{Collect, CollectSingle, Plan},
Backoff, BoundRange, ColumnFamily, Key, KvPair, Result, Value,
};
use crate::backoff::DEFAULT_REGION_BACKOFF;
use crate::config::Config;
use crate::pd::PdClient;
use crate::pd::PdRpcClient;
use crate::raw::lowering::*;
use crate::request::Collect;
use crate::request::CollectSingle;
use crate::request::Plan;
use crate::Backoff;
use crate::BoundRange;
use crate::ColumnFamily;
use crate::Key;
use crate::KvPair;
use crate::Result;
use crate::Value;
const MAX_RAW_KV_SCAN_LIMIT: u32 = 10240;
@ -762,14 +772,16 @@ impl<PdC: PdClient> Client<PdC> {
#[cfg(test)]
mod tests {
use super::*;
use crate::{
mock::{MockKvClient, MockPdClient},
Result,
};
use std::{any::Any, sync::Arc};
use std::any::Any;
use std::sync::Arc;
use tikv_client_proto::kvrpcpb;
use super::*;
use crate::mock::MockKvClient;
use crate::mock::MockPdClient;
use crate::Result;
#[tokio::test]
async fn test_raw_coprocessor() -> Result<()> {
let plain = slog_term::PlainSyncDecorator::new(std::io::stdout());
@ -813,26 +825,21 @@ mod tests {
.into_iter()
.map(|(data, ranges)| (String::from_utf8(data).unwrap(), ranges))
.collect();
assert_eq!(
resps,
vec![
(
"1:[Key(05)..Key(0A)]".to_string(),
vec![Key::from(vec![5])..Key::from(vec![10])]
),
(
"2:[Key(0A)..Key(0F), Key(14)..Key(FAFA)]".to_string(),
vec![
Key::from(vec![10])..Key::from(vec![15]),
Key::from(vec![20])..Key::from(vec![250, 250])
]
),
(
"3:[Key(FAFA)..Key()]".to_string(),
vec![Key::from(vec![250, 250])..Key::from(vec![])]
)
]
);
assert_eq!(resps, vec![
("1:[Key(05)..Key(0A)]".to_string(), vec![
Key::from(vec![5])..Key::from(vec![10])
]),
(
"2:[Key(0A)..Key(0F), Key(14)..Key(FAFA)]".to_string(),
vec![
Key::from(vec![10])..Key::from(vec![15]),
Key::from(vec![20])..Key::from(vec![250, 250])
]
),
("3:[Key(FAFA)..Key()]".to_string(), vec![
Key::from(vec![250, 250])..Key::from(vec![])
])
]);
Ok(())
}
}

14
src/raw/lowering.rs
View File

@ -4,11 +4,19 @@
//! types (i.e., the types from the client crate) and converts these to the types used in the
//! generated protobuf code, then calls the low-level ctor functions in the requests module.
use std::{iter::Iterator, ops::Range, sync::Arc};
use std::iter::Iterator;
use std::ops::Range;
use std::sync::Arc;
use tikv_client_proto::{kvrpcpb, metapb};
use tikv_client_proto::kvrpcpb;
use tikv_client_proto::metapb;
use crate::{raw::requests, BoundRange, ColumnFamily, Key, KvPair, Value};
use crate::raw::requests;
use crate::BoundRange;
use crate::ColumnFamily;
use crate::Key;
use crate::KvPair;
use crate::Value;
pub fn new_raw_get_request(key: Key, cf: Option<ColumnFamily>) -> kvrpcpb::RawGetRequest {
requests::new_raw_get_request(key.into(), cf)

4
src/raw/mod.rs
View File

@ -9,9 +9,11 @@
//!
//! **Warning:** It is not advisable to use both raw and transactional functionality in the same keyspace.
use std::convert::TryFrom;
use std::fmt;
pub use self::client::Client;
use crate::Error;
use std::{convert::TryFrom, fmt};
mod client;
pub mod lowering;

60
src/raw/requests.rs
View File

@ -1,26 +1,40 @@
// Copyright 2019 TiKV Project Authors. Licensed under Apache-2.0.
use std::{any::Any, ops::Range, sync::Arc, time::Duration};
use std::any::Any;
use std::ops::Range;
use std::sync::Arc;
use std::time::Duration;
use async_trait::async_trait;
use futures::stream::BoxStream;
use tikv_client_proto::{kvrpcpb, metapb, tikvpb::tikv_client::TikvClient};
use tikv_client_proto::kvrpcpb;
use tikv_client_proto::metapb;
use tikv_client_proto::tikvpb::tikv_client::TikvClient;
use tikv_client_store::Request;
use tonic::transport::Channel;
use super::RawRpcRequest;
use crate::{
collect_first,
pd::PdClient,
request::{
plan::ResponseWithShard, Collect, CollectSingle, DefaultProcessor, KvRequest, Merge,
Process, Shardable, SingleKey,
},
store::{store_stream_for_keys, store_stream_for_ranges, RegionStore},
transaction::HasLocks,
util::iter::FlatMapOkIterExt,
ColumnFamily, Key, KvPair, Result, Value,
};
use crate::collect_first;
use crate::pd::PdClient;
use crate::request::plan::ResponseWithShard;
use crate::request::Collect;
use crate::request::CollectSingle;
use crate::request::DefaultProcessor;
use crate::request::KvRequest;
use crate::request::Merge;
use crate::request::Process;
use crate::request::Shardable;
use crate::request::SingleKey;
use crate::store::store_stream_for_keys;
use crate::store::store_stream_for_ranges;
use crate::store::RegionStore;
use crate::transaction::HasLocks;
use crate::util::iter::FlatMapOkIterExt;
use crate::ColumnFamily;
use crate::Key;
use crate::KvPair;
use crate::Result;
use crate::Value;
pub fn new_raw_get_request(key: Vec<u8>, cf: Option<ColumnFamily>) -> kvrpcpb::RawGetRequest {
let mut req = kvrpcpb::RawGetRequest::default();
@ -469,17 +483,19 @@ impl HasLocks for kvrpcpb::RawCoprocessorResponse {}
#[cfg(test)]
mod test {
use super::*;
use crate::{
backoff::{DEFAULT_REGION_BACKOFF, OPTIMISTIC_BACKOFF},
mock::{MockKvClient, MockPdClient},
request::Plan,
Key,
};
use futures::executor;
use std::any::Any;
use futures::executor;
use tikv_client_proto::kvrpcpb;
use super::*;
use crate::backoff::DEFAULT_REGION_BACKOFF;
use crate::backoff::OPTIMISTIC_BACKOFF;
use crate::mock::MockKvClient;
use crate::mock::MockPdClient;
use crate::request::Plan;
use crate::Key;
#[test]
#[ignore]
fn test_raw_scan() {

8
src/region.rs
View File

@ -1,8 +1,12 @@
// Copyright 2021 TiKV Project Authors. Licensed under Apache-2.0.
use crate::{Error, Key, Result};
use derive_new::new;
use tikv_client_proto::{kvrpcpb, metapb};
use tikv_client_proto::kvrpcpb;
use tikv_client_proto::metapb;
use crate::Error;
use crate::Key;
use crate::Result;
/// The ID of a region
pub type RegionId = u64;

92
src/region_cache.rs
View File

@ -1,18 +1,25 @@
// Copyright 2021 TiKV Project Authors. Licensed under Apache-2.0.
use crate::{
pd::{RetryClient, RetryClientTrait},
region::{RegionId, RegionVerId, RegionWithLeader, StoreId},
Key, Result,
};
use std::{
collections::{BTreeMap, HashMap, HashSet},
sync::Arc,
};
use std::collections::BTreeMap;
use std::collections::HashMap;
use std::collections::HashSet;
use std::sync::Arc;
use tikv_client_common::Error;
use tikv_client_pd::Cluster;
use tikv_client_proto::metapb::{self, Store};
use tokio::sync::{Notify, RwLock};
use tikv_client_proto::metapb::Store;
use tikv_client_proto::metapb::{self};
use tokio::sync::Notify;
use tokio::sync::RwLock;
use crate::pd::RetryClient;
use crate::pd::RetryClientTrait;
use crate::region::RegionId;
use crate::region::RegionVerId;
use crate::region::RegionWithLeader;
use crate::region::StoreId;
use crate::Key;
use crate::Result;
const MAX_RETRY_WAITING_CONCURRENT_REQUEST: usize = 4;
@ -229,24 +236,26 @@ impl<C: RetryClientTrait> RegionCache<C> {
#[cfg(test)]
mod test {
use super::RegionCache;
use crate::{
pd::RetryClientTrait,
region::{RegionId, RegionWithLeader},
Key, Result,
};
use std::collections::BTreeMap;
use std::collections::HashMap;
use std::collections::HashSet;
use std::sync::atomic::AtomicU64;
use std::sync::atomic::Ordering::SeqCst;
use std::sync::Arc;
use async_trait::async_trait;
use std::{
collections::{BTreeMap, HashMap, HashSet},
sync::{
atomic::{AtomicU64, Ordering::SeqCst},
Arc,
},
};
use tikv_client_common::Error;
use tikv_client_proto::metapb::{self, RegionEpoch};
use tikv_client_proto::metapb::RegionEpoch;
use tikv_client_proto::metapb::{self};
use tokio::sync::Mutex;
use super::RegionCache;
use crate::pd::RetryClientTrait;
use crate::region::RegionId;
use crate::region::RegionWithLeader;
use crate::Key;
use crate::Result;
#[derive(Default)]
struct MockRetryClient {
pub regions: Mutex<HashMap<RegionId, RegionWithLeader>>,
@ -309,9 +318,11 @@ mod test {
async fn cache_is_used() -> Result<()> {
let retry_client = Arc::new(MockRetryClient::default());
let cache = RegionCache::new(retry_client.clone());
retry_client.regions.lock().await.insert(
1,
RegionWithLeader {
retry_client
.regions
.lock()
.await
.insert(1, RegionWithLeader {
region: metapb::Region {
id: 1,
start_key: vec![],
@ -326,11 +337,12 @@ mod test {
store_id: 1,
..Default::default()
}),
},
);
retry_client.regions.lock().await.insert(
2,
RegionWithLeader {
});
retry_client
.regions
.lock()
.await
.insert(2, RegionWithLeader {
region: metapb::Region {
id: 2,
start_key: vec![101],
@ -345,8 +357,7 @@ mod test {
store_id: 2,
..Default::default()
}),
},
);
});
assert_eq!(retry_client.get_region_count.load(SeqCst), 0);
@ -367,13 +378,10 @@ mod test {
// update leader should work
cache
.update_leader(
cache.get_region_by_id(2).await?.ver_id(),
metapb::Peer {
store_id: 102,
..Default::default()
},
)
.update_leader(cache.get_region_by_id(2).await?.ver_id(), metapb::Peer {
store_id: 102,
..Default::default()
})
.await?;
assert_eq!(
cache.get_region_by_id(2).await?.leader.unwrap().store_id,

74
src/request/mod.rs
View File

@ -1,22 +1,36 @@
// Copyright 2019 TiKV Project Authors. Licensed under Apache-2.0.
use crate::{
backoff::{Backoff, DEFAULT_REGION_BACKOFF, OPTIMISTIC_BACKOFF, PESSIMISTIC_BACKOFF},
transaction::HasLocks,
};
use async_trait::async_trait;
use derive_new::new;
use tikv_client_store::{HasKeyErrors, Request};
use tikv_client_store::HasKeyErrors;
use tikv_client_store::Request;
pub use self::{
plan::{
Collect, CollectError, CollectSingle, CollectWithShard, DefaultProcessor, Dispatch,
ExtractError, Merge, MergeResponse, Plan, Process, ProcessResponse, ResolveLock,
ResponseWithShard, RetryableMultiRegion,
},
plan_builder::{PlanBuilder, SingleKey},
shard::{Batchable, HasNextBatch, NextBatch, Shardable},
};
pub use self::plan::Collect;
pub use self::plan::CollectError;
pub use self::plan::CollectSingle;
pub use self::plan::CollectWithShard;
pub use self::plan::DefaultProcessor;
pub use self::plan::Dispatch;
pub use self::plan::ExtractError;
pub use self::plan::Merge;
pub use self::plan::MergeResponse;
pub use self::plan::Plan;
pub use self::plan::Process;
pub use self::plan::ProcessResponse;
pub use self::plan::ResolveLock;
pub use self::plan::ResponseWithShard;
pub use self::plan::RetryableMultiRegion;
pub use self::plan_builder::PlanBuilder;
pub use self::plan_builder::SingleKey;
pub use self::shard::Batchable;
pub use self::shard::HasNextBatch;
pub use self::shard::NextBatch;
pub use self::shard::Shardable;
use crate::backoff::Backoff;
use crate::backoff::DEFAULT_REGION_BACKOFF;
use crate::backoff::OPTIMISTIC_BACKOFF;
use crate::backoff::PESSIMISTIC_BACKOFF;
use crate::transaction::HasLocks;
pub mod plan;
mod plan_builder;
@ -63,23 +77,27 @@ impl RetryOptions {
#[cfg(test)]
mod test {
use super::*;
use crate::{
mock::{MockKvClient, MockPdClient},
store::store_stream_for_keys,
transaction::lowering::new_commit_request,
Error, Key, Result,
};
use std::{
any::Any,
iter,
sync::{atomic::AtomicUsize, Arc},
time::Duration,
};
use tikv_client_proto::{kvrpcpb, pdpb::Timestamp, tikvpb::tikv_client::TikvClient};
use std::any::Any;
use std::iter;
use std::sync::atomic::AtomicUsize;
use std::sync::Arc;
use std::time::Duration;
use tikv_client_proto::kvrpcpb;
use tikv_client_proto::pdpb::Timestamp;
use tikv_client_proto::tikvpb::tikv_client::TikvClient;
use tikv_client_store::HasRegionError;
use tonic::transport::Channel;
use super::*;
use crate::mock::MockKvClient;
use crate::mock::MockPdClient;
use crate::store::store_stream_for_keys;
use crate::transaction::lowering::new_commit_request;
use crate::Error;
use crate::Key;
use crate::Result;
#[tokio::test]
async fn test_region_retry() {
#[derive(Clone)]

67
src/request/plan.rs
View File

@ -1,24 +1,37 @@
// Copyright 2021 TiKV Project Authors. Licensed under Apache-2.0.
use std::{marker::PhantomData, sync::Arc};
use std::marker::PhantomData;
use std::sync::Arc;
use async_recursion::async_recursion;
use async_trait::async_trait;
use futures::{future::try_join_all, prelude::*};
use tikv_client_proto::{errorpb, errorpb::EpochNotMatch, kvrpcpb};
use tikv_client_store::{HasKeyErrors, HasRegionError, HasRegionErrors, KvClient};
use tokio::{sync::Semaphore, time::sleep};
use futures::future::try_join_all;
use futures::prelude::*;
use tikv_client_proto::errorpb;
use tikv_client_proto::errorpb::EpochNotMatch;
use tikv_client_proto::kvrpcpb;
use tikv_client_store::HasKeyErrors;
use tikv_client_store::HasRegionError;
use tikv_client_store::HasRegionErrors;
use tikv_client_store::KvClient;
use tokio::sync::Semaphore;
use tokio::time::sleep;
use crate::{
backoff::Backoff,
pd::PdClient,
request::{shard::HasNextBatch, KvRequest, NextBatch, Shardable},
stats::tikv_stats,
store::RegionStore,
transaction::{resolve_locks, HasLocks, ResolveLocksContext, ResolveLocksOptions},
util::iter::FlatMapOkIterExt,
Error, Result,
};
use crate::backoff::Backoff;
use crate::pd::PdClient;
use crate::request::shard::HasNextBatch;
use crate::request::KvRequest;
use crate::request::NextBatch;
use crate::request::Shardable;
use crate::stats::tikv_stats;
use crate::store::RegionStore;
use crate::transaction::resolve_locks;
use crate::transaction::HasLocks;
use crate::transaction::ResolveLocksContext;
use crate::transaction::ResolveLocksOptions;
use crate::util::iter::FlatMapOkIterExt;
use crate::Error;
use crate::Result;
/// A plan for how to execute a request. A user builds up a plan with various
/// options, then exectutes it.
@ -72,8 +85,7 @@ pub struct RetryableMultiRegion<P: Plan, PdC: PdClient> {
}
impl<P: Plan + Shardable, PdC: PdClient> RetryableMultiRegion<P, PdC>
where
P::Result: HasKeyErrors + HasRegionError,
where P::Result: HasKeyErrors + HasRegionError
{
// A plan may involve multiple shards
#[async_recursion]
@ -272,8 +284,7 @@ impl<P: Plan, PdC: PdClient> Clone for RetryableMultiRegion<P, PdC> {
#[async_trait]
impl<P: Plan + Shardable, PdC: PdClient> Plan for RetryableMultiRegion<P, PdC>
where
P::Result: HasKeyErrors + HasRegionError,
where P::Result: HasKeyErrors + HasRegionError
{
type Result = Vec<Result<P::Result>>;
@ -403,8 +414,7 @@ impl<P: Plan, PdC: PdClient> Clone for ResolveLock<P, PdC> {
#[async_trait]
impl<P: Plan, PdC: PdClient> Plan for ResolveLock<P, PdC>
where
P::Result: HasLocks,
where P::Result: HasLocks
{
type Result = P::Result;
@ -506,8 +516,7 @@ impl<P: Plan, PdC: PdClient> Clone for CleanupLocks<P, PdC> {
#[async_trait]
impl<P: Plan + Shardable + NextBatch, PdC: PdClient> Plan for CleanupLocks<P, PdC>
where
P::Result: HasLocks + HasNextBatch + HasKeyErrors + HasRegionError,
where P::Result: HasLocks + HasNextBatch + HasKeyErrors + HasRegionError
{
type Result = CleanupLocksResult;
@ -622,8 +631,7 @@ impl<P: Plan> Clone for ExtractError<P> {
#[async_trait]
impl<P: Plan> Plan for ExtractError<P>
where
P::Result: HasKeyErrors + HasRegionErrors,
where P::Result: HasKeyErrors + HasRegionErrors
{
type Result = P::Result;
@ -665,8 +673,7 @@ impl<P: Plan + Shardable> Clone for PreserveShard<P> {
#[async_trait]
impl<P> Plan for PreserveShard<P>
where
P: Plan + Shardable,
where P: Plan + Shardable
{
type Result = ResponseWithShard<P::Result, P::Shard>;
@ -705,10 +712,12 @@ impl<Resp: HasRegionError, Shard> HasRegionError for ResponseWithShard<Resp, Sha
#[cfg(test)]
mod test {
use futures::stream::BoxStream;
use futures::stream::{self};
use tikv_client_proto::kvrpcpb::BatchGetResponse;
use super::*;
use crate::mock::MockPdClient;
use futures::stream::{self, BoxStream};
use tikv_client_proto::kvrpcpb::BatchGetResponse;
#[derive(Clone)]
struct ErrPlan;

56
src/request/plan_builder.rs
View File

@ -1,20 +1,35 @@
// Copyright 2021 TiKV Project Authors. Licensed under Apache-2.0.
use std::marker::PhantomData;
use std::sync::Arc;
use tikv_client_store::HasKeyErrors;
use tikv_client_store::HasRegionError;
use tikv_client_store::HasRegionErrors;
use super::plan::PreserveShard;
use crate::{
backoff::Backoff,
pd::PdClient,
request::{
plan::CleanupLocks, shard::HasNextBatch, DefaultProcessor, Dispatch, ExtractError,
KvRequest, Merge, MergeResponse, NextBatch, Plan, Process, ProcessResponse, ResolveLock,
RetryableMultiRegion, Shardable,
},
store::RegionStore,
transaction::{HasLocks, ResolveLocksContext, ResolveLocksOptions},
Result,
};
use std::{marker::PhantomData, sync::Arc};
use tikv_client_store::{HasKeyErrors, HasRegionError, HasRegionErrors};
use crate::backoff::Backoff;
use crate::pd::PdClient;
use crate::request::plan::CleanupLocks;
use crate::request::shard::HasNextBatch;
use crate::request::DefaultProcessor;
use crate::request::Dispatch;
use crate::request::ExtractError;
use crate::request::KvRequest;
use crate::request::Merge;
use crate::request::MergeResponse;
use crate::request::NextBatch;
use crate::request::Plan;
use crate::request::Process;
use crate::request::ProcessResponse;
use crate::request::ResolveLock;
use crate::request::RetryableMultiRegion;
use crate::request::Shardable;
use crate::store::RegionStore;
use crate::transaction::HasLocks;
use crate::transaction::ResolveLocksContext;
use crate::transaction::ResolveLocksOptions;
use crate::Result;
/// Builder type for plans (see that module for more).
pub struct PlanBuilder<PdC: PdClient, P: Plan, Ph: PlanBuilderPhase> {
@ -55,9 +70,7 @@ impl<PdC: PdClient, P: Plan> PlanBuilder<PdC, P, Targetted> {
impl<PdC: PdClient, P: Plan, Ph: PlanBuilderPhase> PlanBuilder<PdC, P, Ph> {
/// If there is a lock error, then resolve the lock and retry the request.
pub fn resolve_lock(self, backoff: Backoff) -> PlanBuilder<PdC, ResolveLock<P, PdC>, Ph>
where
P::Result: HasLocks,
{
where P::Result: HasLocks {
PlanBuilder {
pd_client: self.pd_client.clone(),
plan: ResolveLock {
@ -133,8 +146,7 @@ impl<PdC: PdClient, P: Plan, Ph: PlanBuilderPhase> PlanBuilder<PdC, P, Ph> {
}
impl<PdC: PdClient, P: Plan + Shardable> PlanBuilder<PdC, P, NoTarget>
where
P::Result: HasKeyErrors + HasRegionError,
where P::Result: HasKeyErrors + HasRegionError
{
/// Split the request into shards sending a request to the region of each shard.
pub fn retry_multi_region(
@ -194,8 +206,7 @@ impl<PdC: PdClient, R: KvRequest> PlanBuilder<PdC, Dispatch<R>, NoTarget> {
}
impl<PdC: PdClient, P: Plan + Shardable> PlanBuilder<PdC, P, NoTarget>
where
P::Result: HasKeyErrors,
where P::Result: HasKeyErrors
{
pub fn preserve_shard(self) -> PlanBuilder<PdC, PreserveShard<P>, NoTarget> {
PlanBuilder {
@ -210,8 +221,7 @@ where
}
impl<PdC: PdClient, P: Plan> PlanBuilder<PdC, P, Targetted>
where
P::Result: HasKeyErrors + HasRegionErrors,
where P::Result: HasKeyErrors + HasRegionErrors
{
pub fn extract_error(self) -> PlanBuilder<PdC, ExtractError<P>, Targetted> {
PlanBuilder {

23
src/request/shard.rs
View File

@ -1,15 +1,19 @@
// Copyright 2021 TiKV Project Authors. Licensed under Apache-2.0.
use super::plan::PreserveShard;
use crate::{
pd::PdClient,
request::{plan::CleanupLocks, Dispatch, KvRequest, Plan, ResolveLock},
store::RegionStore,
Result,
};
use futures::stream::BoxStream;
use std::sync::Arc;
use futures::stream::BoxStream;
use super::plan::PreserveShard;
use crate::pd::PdClient;
use crate::request::plan::CleanupLocks;
use crate::request::Dispatch;
use crate::request::KvRequest;
use crate::request::Plan;
use crate::request::ResolveLock;
use crate::store::RegionStore;
use crate::Result;
macro_rules! impl_inner_shardable {
() => {
type Shard = P::Shard;
@ -230,7 +234,8 @@ macro_rules! shardable_range {
#[cfg(test)]
mod test {
use rand::{thread_rng, Rng};
use rand::thread_rng;
use rand::Rng;
use super::Batchable;

15
src/stats.rs
View File

@ -1,11 +1,16 @@
// Copyright 2018 TiKV Project Authors. Licensed under Apache-2.0.
use std::time::Duration;
use std::time::Instant;
use prometheus::register_histogram;
use prometheus::register_histogram_vec;
use prometheus::register_int_counter_vec;
use prometheus::Histogram;
use prometheus::HistogramVec;
use prometheus::IntCounterVec;
use crate::Result;
use prometheus::{
register_histogram, register_histogram_vec, register_int_counter_vec, Histogram, HistogramVec,
IntCounterVec,
};
use std::time::{Duration, Instant};
pub struct RequestStats {
start: Instant,

22
src/store.rs
View File

@ -1,15 +1,23 @@
// Copyright 2019 TiKV Project Authors. Licensed under Apache-2.0.
use crate::{pd::PdClient, region::RegionWithLeader, BoundRange, Key, Result};
use std::cmp::max;
use std::cmp::min;
use std::sync::Arc;
use async_trait::async_trait;
use derive_new::new;
use futures::{prelude::*, stream::BoxStream};
use std::{
cmp::{max, min},
sync::Arc,
};
use futures::prelude::*;
use futures::stream::BoxStream;
use tikv_client_proto::kvrpcpb;
use tikv_client_store::{KvClient, KvConnect, TikvConnect};
use tikv_client_store::KvClient;
use tikv_client_store::KvConnect;
use tikv_client_store::TikvConnect;
use crate::pd::PdClient;
use crate::region::RegionWithLeader;
use crate::BoundRange;
use crate::Key;
use crate::Result;
#[derive(new, Clone)]
pub struct RegionStore {

1
src/timestamp.rs
View File

@ -5,6 +5,7 @@
//! The higher bits of the version are the physical part of the timestamp.
use std::convert::TryInto;
pub use tikv_client_proto::pdpb::Timestamp;
const PHYSICAL_SHIFT_BITS: i64 = 18;

41
src/transaction/buffer.rs
View File

@ -1,12 +1,18 @@
// Copyright 2019 TiKV Project Authors. Licensed under Apache-2.0.
use crate::{BoundRange, Key, KvPair, Result, Value};
use std::{
collections::{btree_map::Entry, BTreeMap, HashMap},
future::Future,
};
use std::collections::btree_map::Entry;
use std::collections::BTreeMap;
use std::collections::HashMap;
use std::future::Future;
use tikv_client_proto::kvrpcpb;
use crate::BoundRange;
use crate::Key;
use crate::KvPair;
use crate::Result;
use crate::Value;
/// A caching layer which buffers reads and writes in a transaction.
pub struct Buffer {
primary_key: Option<Key>,
@ -391,10 +397,12 @@ impl MutationValue {
#[cfg(test)]
mod tests {
use super::*;
use futures::{executor::block_on, future::ready};
use futures::executor::block_on;
use futures::future::ready;
use tikv_client_common::internal_err;
use super::*;
#[test]
fn set_and_get_from_buffer() {
let mut buffer = Buffer::new(false);
@ -483,18 +491,15 @@ mod tests {
ready(Ok(vec![]))
}),
);
assert_eq!(
r1.unwrap().collect::<Vec<_>>(),
vec![
KvPair(k1.clone(), v1.clone()),
KvPair(k2.clone(), v2.clone())
]
);
assert_eq!(r1.unwrap().collect::<Vec<_>>(), vec![
KvPair(k1.clone(), v1.clone()),
KvPair(k2.clone(), v2.clone())
]);
assert_eq!(r2.unwrap().unwrap(), v2);
assert_eq!(
r3.unwrap().collect::<Vec<_>>(),
vec![KvPair(k1, v1), KvPair(k2, v2)]
);
assert_eq!(r3.unwrap().collect::<Vec<_>>(), vec![
KvPair(k1, v1),
KvPair(k2, v2)
]);
}
// Check that multiple writes to the same key combine in the correct way.

33
src/transaction/client.rs
View File

@ -1,21 +1,28 @@
// Copyright 2019 TiKV Project Authors. Licensed under Apache-2.0.
use crate::{
backoff::DEFAULT_REGION_BACKOFF,
config::Config,
pd::{PdClient, PdRpcClient},
request::{plan::CleanupLocksResult, Plan},
timestamp::TimestampExt,
transaction::{
lock::ResolveLocksOptions, ResolveLocksContext, Snapshot, Transaction, TransactionOptions,
},
transaction_lowering::new_scan_lock_request,
Backoff, BoundRange, Result,
};
use slog::{Drain, Logger};
use std::sync::Arc;
use slog::Drain;
use slog::Logger;
use tikv_client_proto::pdpb::Timestamp;
use crate::backoff::DEFAULT_REGION_BACKOFF;
use crate::config::Config;
use crate::pd::PdClient;
use crate::pd::PdRpcClient;
use crate::request::plan::CleanupLocksResult;
use crate::request::Plan;
use crate::timestamp::TimestampExt;
use crate::transaction::lock::ResolveLocksOptions;
use crate::transaction::ResolveLocksContext;
use crate::transaction::Snapshot;
use crate::transaction::Transaction;
use crate::transaction::TransactionOptions;
use crate::transaction_lowering::new_scan_lock_request;
use crate::Backoff;
use crate::BoundRange;
use crate::Result;
// FIXME: cargo-culted value
const SCAN_LOCK_BATCH_SIZE: u32 = 1024;

59
src/transaction/lock.rs
View File

@ -1,31 +1,36 @@
// Copyright 2020 TiKV Project Authors. Licensed under Apache-2.0.
use crate::{
backoff::{Backoff, DEFAULT_REGION_BACKOFF, OPTIMISTIC_BACKOFF},
pd::PdClient,
region::RegionVerId,
request::{Collect, CollectSingle, Plan},
store::RegionStore,
timestamp::TimestampExt,
transaction::{
requests,
requests::{
new_check_secondary_locks_request, new_check_txn_status_request, SecondaryLocksStatus,
TransactionStatus, TransactionStatusKind,
},
},
Error, Result,
};
use std::collections::HashMap;
use std::collections::HashSet;
use std::sync::Arc;
use fail::fail_point;
use log::debug;
use slog::Logger;
use std::{
collections::{HashMap, HashSet},
sync::Arc,
};
use tikv_client_proto::{kvrpcpb, kvrpcpb::TxnInfo, pdpb::Timestamp};
use tikv_client_proto::kvrpcpb;
use tikv_client_proto::kvrpcpb::TxnInfo;
use tikv_client_proto::pdpb::Timestamp;
use tokio::sync::RwLock;
use crate::backoff::Backoff;
use crate::backoff::DEFAULT_REGION_BACKOFF;
use crate::backoff::OPTIMISTIC_BACKOFF;
use crate::pd::PdClient;
use crate::region::RegionVerId;
use crate::request::Collect;
use crate::request::CollectSingle;
use crate::request::Plan;
use crate::store::RegionStore;
use crate::timestamp::TimestampExt;
use crate::transaction::requests;
use crate::transaction::requests::new_check_secondary_locks_request;
use crate::transaction::requests::new_check_txn_status_request;
use crate::transaction::requests::SecondaryLocksStatus;
use crate::transaction::requests::TransactionStatus;
use crate::transaction::requests::TransactionStatusKind;
use crate::Error;
use crate::Result;
const RESOLVE_LOCK_RETRY_LIMIT: usize = 10;
/// _Resolves_ the given locks. Returns whether all the given locks are resolved.
@ -251,7 +256,10 @@ impl LockResolver {
self.logger,
"secondary status, txn_id:{}, commit_ts:{:?}, min_commit_version:{}, fallback_2pc:{}",
txn_id,
secondary_status.commit_ts.as_ref().map_or(0, |ts| ts.version()),
secondary_status
.commit_ts
.as_ref()
.map_or(0, |ts| ts.version()),
secondary_status.min_commit_ts,
secondary_status.fallback_2pc,
);
@ -417,11 +425,14 @@ pub trait HasLocks {
#[cfg(test)]
mod tests {
use super::*;
use crate::mock::{MockKvClient, MockPdClient};
use std::any::Any;
use tikv_client_proto::errorpb;
use super::*;
use crate::mock::MockKvClient;
use crate::mock::MockPdClient;
#[tokio::test]
async fn test_resolve_lock_with_retry() {
// Test resolve lock within retry limit

21
src/transaction/lowering.rs
View File

@ -1,11 +1,26 @@
// Copyright 2021 TiKV Project Authors. Licensed under Apache-2.0.
use std::iter::Iterator;
use tikv_client_proto::kvrpcpb;
use tikv_client_proto::pdpb::Timestamp;
/// This module provides constructor functions for requests which take arguments as high-level
/// types (i.e., the types from the client crate) and converts these to the types used in the
/// generated protobuf code, then calls the low-level ctor functions in the requests module.
use crate::{timestamp::TimestampExt, transaction::requests, BoundRange, Key};
use std::iter::Iterator;
use tikv_client_proto::{kvrpcpb, pdpb::Timestamp};
use crate::timestamp::TimestampExt;
/// This module provides constructor functions for requests which take arguments as high-level
/// types (i.e., the types from the client crate) and converts these to the types used in the
/// generated protobuf code, then calls the low-level ctor functions in the requests module.
use crate::transaction::requests;
/// This module provides constructor functions for requests which take arguments as high-level
/// types (i.e., the types from the client crate) and converts these to the types used in the
/// generated protobuf code, then calls the low-level ctor functions in the requests module.
use crate::BoundRange;
/// This module provides constructor functions for requests which take arguments as high-level
/// types (i.e., the types from the client crate) and converts these to the types used in the
/// generated protobuf code, then calls the low-level ctor functions in the requests module.
use crate::Key;
pub fn new_get_request(key: Key, timestamp: Timestamp) -> kvrpcpb::GetRequest {
requests::new_get_request(key.into(), timestamp.version())

11
src/transaction/mod.rs
View File

@ -9,11 +9,14 @@
//! **Warning:** It is not advisable to use both raw and transactional functionality in the same keyspace.
pub use client::Client;
pub(crate) use lock::{resolve_locks, HasLocks};
pub(crate) use lock::resolve_locks;
pub(crate) use lock::HasLocks;
pub use snapshot::Snapshot;
pub use transaction::CheckLevel;
#[doc(hidden)]
pub use transaction::HeartbeatOption;
pub use transaction::{CheckLevel, Transaction, TransactionOptions};
pub use transaction::Transaction;
pub use transaction::TransactionOptions;
mod buffer;
mod client;
@ -21,7 +24,9 @@ pub mod lowering;
#[macro_use]
mod requests;
mod lock;
pub use lock::{LockResolver, ResolveLocksContext, ResolveLocksOptions};
pub use lock::LockResolver;
pub use lock::ResolveLocksContext;
pub use lock::ResolveLocksOptions;
mod snapshot;
#[allow(clippy::module_inception)]
mod transaction;

98
src/transaction/requests.rs
View File

@ -1,31 +1,46 @@
// Copyright 2020 TiKV Project Authors. Licensed under Apache-2.0.
use crate::{
collect_first,
pd::PdClient,
request::{
Batchable, Collect, CollectSingle, CollectWithShard, DefaultProcessor, HasNextBatch,
KvRequest, Merge, NextBatch, Process, ResponseWithShard, Shardable, SingleKey,
},
store::{store_stream_for_keys, store_stream_for_range, RegionStore},
timestamp::TimestampExt,
transaction::HasLocks,
util::iter::FlatMapOkIterExt,
KvPair, Result, Value,
};
use std::cmp;
use std::iter;
use std::sync::Arc;
use either::Either;
use futures::{
stream::{self, BoxStream},
StreamExt,
};
use std::{cmp, iter, sync::Arc};
use futures::stream::BoxStream;
use futures::stream::{self};
use futures::StreamExt;
use tikv_client_common::Error::PessimisticLockError;
use tikv_client_proto::{
kvrpcpb::{self, Action, LockInfo, TxnHeartBeatResponse, TxnInfo},
pdpb::Timestamp,
};
use tikv_client_proto::kvrpcpb::Action;
use tikv_client_proto::kvrpcpb::LockInfo;
use tikv_client_proto::kvrpcpb::TxnHeartBeatResponse;
use tikv_client_proto::kvrpcpb::TxnInfo;
use tikv_client_proto::kvrpcpb::{self};
use tikv_client_proto::pdpb::Timestamp;
use super::transaction::TXN_COMMIT_BATCH_SIZE;
use crate::collect_first;
use crate::pd::PdClient;
use crate::request::Batchable;
use crate::request::Collect;
use crate::request::CollectSingle;
use crate::request::CollectWithShard;
use crate::request::DefaultProcessor;
use crate::request::HasNextBatch;
use crate::request::KvRequest;
use crate::request::Merge;
use crate::request::NextBatch;
use crate::request::Process;
use crate::request::ResponseWithShard;
use crate::request::Shardable;
use crate::request::SingleKey;
use crate::store::store_stream_for_keys;
use crate::store::store_stream_for_range;
use crate::store::RegionStore;
use crate::timestamp::TimestampExt;
use crate::transaction::HasLocks;
use crate::util::iter::FlatMapOkIterExt;
use crate::KvPair;
use crate::Result;
use crate::Value;
// implement HasLocks for a response type that has a `pairs` field,
// where locks can be extracted from both the `pairs` and `error` fields
@ -490,11 +505,7 @@ impl Merge<ResponseWithShard<kvrpcpb::PessimisticLockResponse, Vec<kvrpcpb::Muta
} else {
assert_eq!(kvpairs.len(), not_founds.len());
Either::Right(kvpairs.zip(not_founds).filter_map(|(kvpair, not_found)| {
if not_found {
None
} else {
Some(kvpair)
}
if not_found { None } else { Some(kvpair) }
}))
}
})
@ -851,13 +862,15 @@ impl HasLocks for kvrpcpb::PrewriteResponse {
#[cfg(test)]
#[cfg_attr(feature = "protobuf-codec", allow(clippy::useless_conversion))]
mod tests {
use crate::{
request::{plan::Merge, CollectWithShard, ResponseWithShard},
KvPair,
};
use tikv_client_common::Error::{PessimisticLockError, ResolveLockError};
use tikv_client_common::Error::PessimisticLockError;
use tikv_client_common::Error::ResolveLockError;
use tikv_client_proto::kvrpcpb;
use crate::request::plan::Merge;
use crate::request::CollectWithShard;
use crate::request::ResponseWithShard;
use crate::KvPair;
#[tokio::test]
async fn test_merge_pessimistic_lock_response() {
let (key1, key2, key3, key4) = (b"key1", b"key2", b"key3", b"key4");
@ -918,13 +931,10 @@ mod tests {
];
let result = merger.merge(input);
assert_eq!(
result.unwrap(),
vec![
KvPair::new(key1.to_vec(), value1.to_vec()),
KvPair::new(key4.to_vec(), value4.to_vec()),
]
);
assert_eq!(result.unwrap(), vec![
KvPair::new(key1.to_vec(), value1.to_vec()),
KvPair::new(key4.to_vec(), value4.to_vec()),
]);
}
{
let input = vec![
@ -941,10 +951,12 @@ mod tests {
} = result.unwrap_err()
{
assert!(matches!(*inner, ResolveLockError));
assert_eq!(
success_keys,
vec![key1.to_vec(), key2.to_vec(), key3.to_vec(), key4.to_vec()]
);
assert_eq!(success_keys, vec![
key1.to_vec(),
key2.to_vec(),
key3.to_vec(),
key4.to_vec()
]);
} else {
panic!();
}

8
src/transaction/snapshot.rs
View File

@ -1,9 +1,15 @@
// Copyright 2019 TiKV Project Authors. Licensed under Apache-2.0.
use crate::{BoundRange, Key, KvPair, Result, Transaction, Value};
use derive_new::new;
use slog::Logger;
use crate::BoundRange;
use crate::Key;
use crate::KvPair;
use crate::Result;
use crate::Transaction;
use crate::Value;
/// A read-only transaction which reads at the given timestamp.
///
/// It behaves as if the snapshot was taken at the given timestamp,

75
src/transaction/transaction.rs
View File

@ -1,22 +1,38 @@
// Copyright 2019 TiKV Project Authors. Licensed under Apache-2.0.
use crate::{
backoff::{Backoff, DEFAULT_REGION_BACKOFF},
pd::{PdClient, PdRpcClient},
request::{
Collect, CollectError, CollectSingle, CollectWithShard, Plan, PlanBuilder, RetryOptions,
},
timestamp::TimestampExt,
transaction::{buffer::Buffer, lowering::*},
BoundRange, Error, Key, KvPair, Result, Value,
};
use std::iter;
use std::sync::Arc;
use std::time::Instant;
use derive_new::new;
use fail::fail_point;
use futures::prelude::*;
use slog::Logger;
use std::{iter, sync::Arc, time::Instant};
use tikv_client_proto::{kvrpcpb, pdpb::Timestamp};
use tokio::{sync::RwLock, time::Duration};
use tikv_client_proto::kvrpcpb;
use tikv_client_proto::pdpb::Timestamp;
use tokio::sync::RwLock;
use tokio::time::Duration;
use crate::backoff::Backoff;
use crate::backoff::DEFAULT_REGION_BACKOFF;
use crate::pd::PdClient;
use crate::pd::PdRpcClient;
use crate::request::Collect;
use crate::request::CollectError;
use crate::request::CollectSingle;
use crate::request::CollectWithShard;
use crate::request::Plan;
use crate::request::PlanBuilder;
use crate::request::RetryOptions;
use crate::timestamp::TimestampExt;
use crate::transaction::buffer::Buffer;
use crate::transaction::lowering::*;
use crate::BoundRange;
use crate::Error;
use crate::Key;
use crate::KvPair;
use crate::Result;
use crate::Value;
/// An undo-able set of actions on the dataset.
///
@ -1386,23 +1402,24 @@ enum TransactionStatus {
#[cfg(test)]
mod tests {
use crate::{
mock::{MockKvClient, MockPdClient},
transaction::HeartbeatOption,
Transaction, TransactionOptions,
};
use std::any::Any;
use std::io;
use std::sync::atomic::AtomicUsize;
use std::sync::atomic::Ordering;
use std::sync::Arc;
use std::time::Duration;
use fail::FailScenario;
use slog::{Drain, Logger};
use std::{
any::Any,
io,
sync::{
atomic::{AtomicUsize, Ordering},
Arc,
},
time::Duration,
};
use tikv_client_proto::{kvrpcpb, pdpb::Timestamp};
use slog::Drain;
use slog::Logger;
use tikv_client_proto::kvrpcpb;
use tikv_client_proto::pdpb::Timestamp;
use crate::mock::MockKvClient;
use crate::mock::MockPdClient;
use crate::transaction::HeartbeatOption;
use crate::Transaction;
use crate::TransactionOptions;
#[tokio::test]
async fn test_optimistic_heartbeat() -> Result<(), io::Error> {

27
src/util/iter.rs
View File

@ -32,13 +32,13 @@ pub struct FlatMapOk<U, F, Ti, E> {
// FIXME: implement other iterator methods like size_hint, etc.
impl<
T: IntoIterator<IntoIter = Ti>,
U: Iterator<Item = std::result::Result<I, E>>,
F: FnMut(I) -> T,
Ti: Iterator<Item = T::Item>,
I,
E,
> Iterator for FlatMapOk<U, F, Ti, E>
T: IntoIterator<IntoIter = Ti>,
U: Iterator<Item = std::result::Result<I, E>>,
F: FnMut(I) -> T,
Ti: Iterator<Item = T::Item>,
I,
E,
> Iterator for FlatMapOk<U, F, Ti, E>
{
type Item = std::result::Result<T::Item, E>;
@ -92,9 +92,14 @@ mod test {
.into_iter()
.flat_map_ok(|i| vec![i, i, i].into_iter())
.collect();
assert_eq!(
result,
vec![Ok(0), Ok(0), Ok(0), Err(()), Ok(2), Ok(2), Ok(2)]
);
assert_eq!(result, vec![
Ok(0),
Ok(0),
Ok(0),
Err(()),
Ok(2),
Ok(2),
Ok(2)
]);
}
}

4
tests/common/ctl.rs
View File

@ -2,8 +2,10 @@
//! The module provides some utility functions to control and get information
//! from PD, using its HTTP API.
use tikv_client_common::Error;
use tikv_client_common::Result;
use super::pd_addrs;
use tikv_client_common::{Error, Result};
pub async fn get_region_count() -> Result<u64> {
let res = reqwest::get(format!("http://{}/pd/api/v1/regions", pd_addrs()[0]))

16
tests/common/mod.rs
View File

@ -2,11 +2,21 @@
mod ctl;
use log::{info, warn};
use std::collections::HashSet;
use std::convert::TryInto;
use std::env;
use std::time::Duration;
use log::info;
use log::warn;
use rand::Rng;
use slog::Drain;
use std::{collections::HashSet, convert::TryInto, env, time::Duration};
use tikv_client::{ColumnFamily, Key, RawClient, Result, Transaction, TransactionClient};
use tikv_client::ColumnFamily;
use tikv_client::Key;
use tikv_client::RawClient;
use tikv_client::Result;
use tikv_client::Transaction;
use tikv_client::TransactionClient;
use tokio::time::sleep;
const ENV_PD_ADDRS: &str = "PD_ADDRS";

19
tests/failpoint_tests.rs
View File

@ -2,16 +2,25 @@
mod common;
use std::collections::HashSet;
use std::iter::FromIterator;
use std::thread;
use std::time::Duration;
use common::*;
use fail::FailScenario;
use rand::thread_rng;
use serial_test::serial;
use slog::info;
use std::{collections::HashSet, iter::FromIterator, thread, time::Duration};
use tikv_client::{
transaction::{Client, HeartbeatOption, ResolveLocksOptions},
Backoff, CheckLevel, Result, RetryOptions, TransactionClient, TransactionOptions,
};
use tikv_client::transaction::Client;
use tikv_client::transaction::HeartbeatOption;
use tikv_client::transaction::ResolveLocksOptions;
use tikv_client::Backoff;
use tikv_client::CheckLevel;
use tikv_client::Result;
use tikv_client::RetryOptions;
use tikv_client::TransactionClient;
use tikv_client::TransactionOptions;
#[tokio::test]
#[serial]

31
tests/integration_tests.rs
View File

@ -12,15 +12,25 @@
//! requirements on the region boundaries.
mod common;
use std::collections::HashMap;
use std::iter;
use common::*;
use futures::prelude::*;
use rand::{seq::IteratorRandom, thread_rng, Rng};
use rand::seq::IteratorRandom;
use rand::thread_rng;
use rand::Rng;
use serial_test::serial;
use std::{collections::HashMap, iter};
use tikv_client::{
transaction::HeartbeatOption, BoundRange, Error, Key, KvPair, RawClient, Result,
TransactionClient, TransactionOptions, Value,
};
use tikv_client::transaction::HeartbeatOption;
use tikv_client::BoundRange;
use tikv_client::Error;
use tikv_client::Key;
use tikv_client::KvPair;
use tikv_client::RawClient;
use tikv_client::Result;
use tikv_client::TransactionClient;
use tikv_client::TransactionOptions;
use tikv_client::Value;
// Parameters used in test
const NUM_PEOPLE: u32 = 100;
@ -737,10 +747,11 @@ async fn txn_lock_keys_error_handle() -> Result<()> {
let mut t3 = client.begin_pessimistic().await?;
t1.lock_keys(vec![k[0].clone(), k[1].clone()]).await?;
assert!(t2
.lock_keys(vec![k[0].clone(), k[2].clone()])
.await
.is_err());
assert!(
t2.lock_keys(vec![k[0].clone(), k[2].clone()])
.await
.is_err()
);
t3.lock_keys(vec![k[2].clone(), k[3].clone()]).await?;
t1.rollback().await?;

1
tikv-client-common/src/errors.rs
View File

@ -1,6 +1,7 @@
// Copyright 2018 TiKV Project Authors. Licensed under Apache-2.0.
use std::result;
use thiserror::Error;
/// An error originating from the TiKV client or dependencies.

4
tikv-client-common/src/lib.rs
View File

@ -6,4 +6,6 @@ pub mod security;
extern crate log;
#[doc(inline)]
pub use crate::errors::{Error, Result};
pub use crate::errors::Error;
#[doc(inline)]
pub use crate::errors::Result;

27
tikv-client-common/src/security.rs
View File

@ -1,15 +1,19 @@
// Copyright 2018 TiKV Project Authors. Licensed under Apache-2.0.
use crate::Result;
use std::fs::File;
use std::io::Read;
use std::path::Path;
use std::path::PathBuf;
use std::time::Duration;
// use grpcio::{Channel, ChannelBuilder, ChannelCredentialsBuilder, Environment};
use regex::Regex;
use std::{
fs::File,
io::Read,
path::{Path, PathBuf},
time::Duration,
};
use tonic::transport::{Certificate, Channel, ClientTlsConfig, Identity};
use tonic::transport::Certificate;
use tonic::transport::Channel;
use tonic::transport::ClientTlsConfig;
use tonic::transport::Identity;
use crate::Result;
lazy_static::lazy_static! {
static ref SCHEME_REG: Regex = Regex::new(r"^\s*(https?://)").unwrap();
@ -98,11 +102,14 @@ impl SecurityManager {
#[cfg(test)]
mod tests {
use super::*;
use std::fs::File;
use std::io::Write;
use std::path::PathBuf;
use std::{fs::File, io::Write, path::PathBuf};
use tempfile;
use super::*;
#[test]
fn test_security() {
let temp = tempfile::tempdir().unwrap();

22
tikv-client-pd/src/cluster.rs
View File

@ -1,15 +1,21 @@
// Copyright 2018 TiKV Project Authors. Licensed under Apache-2.0.
use crate::{timestamp::TimestampOracle, Result, SecurityManager};
use std::collections::HashSet;
use std::sync::Arc;
use std::time::Duration;
use std::time::Instant;
use async_trait::async_trait;
use std::{
collections::HashSet,
sync::Arc,
time::{Duration, Instant},
};
use tikv_client_common::internal_err;
use tikv_client_proto::pdpb::{self, Timestamp};
use tonic::{transport::Channel, IntoRequest, Request};
use tikv_client_proto::pdpb::Timestamp;
use tikv_client_proto::pdpb::{self};
use tonic::transport::Channel;
use tonic::IntoRequest;
use tonic::Request;
use crate::timestamp::TimestampOracle;
use crate::Result;
use crate::SecurityManager;
/// A PD cluster.
pub struct Cluster {

10
tikv-client-pd/src/lib.rs
View File

@ -1,9 +1,15 @@
// Copyright 2018 TiKV Project Authors. Licensed under Apache-2.0.
#[doc(inline)]
pub use cluster::{Cluster, Connection};
pub use cluster::Cluster;
#[doc(inline)]
pub use tikv_client_common::{security::SecurityManager, Error, Result};
pub use cluster::Connection;
#[doc(inline)]
pub use tikv_client_common::security::SecurityManager;
#[doc(inline)]
pub use tikv_client_common::Error;
#[doc(inline)]
pub use tikv_client_common::Result;
#[macro_use]
extern crate log;

25
tikv-client-pd/src/timestamp.rs
View File

@ -11,20 +11,27 @@
//! single `TsoRequest` to the PD server. The other future receives `TsoResponse`s from the PD
//! server and allocates timestamps for the requests.
use crate::Result;
use futures::{
pin_mut,
prelude::*,
task::{AtomicWaker, Context, Poll},
};
use std::collections::VecDeque;
use std::pin::Pin;
use std::sync::Arc;
use futures::pin_mut;
use futures::prelude::*;
use futures::task::AtomicWaker;
use futures::task::Context;
use futures::task::Poll;
use log::debug;
use pin_project::pin_project;
use std::{collections::VecDeque, pin::Pin, sync::Arc};
use tikv_client_common::internal_err;
use tikv_client_proto::pdpb::{pd_client::PdClient, *};
use tokio::sync::{mpsc, oneshot, Mutex};
use tikv_client_proto::pdpb::pd_client::PdClient;
use tikv_client_proto::pdpb::*;
use tokio::sync::mpsc;
use tokio::sync::oneshot;
use tokio::sync::Mutex;
use tonic::transport::Channel;
use crate::Result;
/// It is an empirical value.
const MAX_BATCH_SIZE: usize = 64;

10
tikv-client-store/src/client.rs
View File

@ -1,12 +1,18 @@
// Copyright 2020 TiKV Project Authors. Licensed under Apache-2.0.
use crate::{request::Request, Result, SecurityManager};
use std::any::Any;
use std::sync::Arc;
use std::time::Duration;
use async_trait::async_trait;
use derive_new::new;
use std::{any::Any, sync::Arc, time::Duration};
use tikv_client_proto::tikvpb::tikv_client::TikvClient;
use tonic::transport::Channel;
use crate::request::Request;
use crate::Result;
use crate::SecurityManager;
/// A trait for connecting to TiKV stores.
#[async_trait]
pub trait KvConnect: Sized + Send + Sync + 'static {

10
tikv-client-store/src/errors.rs
View File

@ -1,9 +1,11 @@
// Copyright 2019 TiKV Project Authors. Licensed under Apache-2.0.
use crate::Error;
use std::fmt::Display;
use tikv_client_proto::kvrpcpb;
use crate::Error;
// Those that can have a single region error
pub trait HasRegionError {
fn region_error(&mut self) -> Option<tikv_client_proto::errorpb::Error>;
@ -210,9 +212,11 @@ fn extract_errors(
#[cfg(test)]
mod test {
use super::HasKeyErrors;
use tikv_client_common::{internal_err, Error};
use tikv_client_common::internal_err;
use tikv_client_common::Error;
use tikv_client_proto::kvrpcpb;
use super::HasKeyErrors;
#[test]
fn result_haslocks() {
let mut resp: Result<_, Error> = Ok(kvrpcpb::CommitResponse::default());

23
tikv-client-store/src/lib.rs
View File

@ -4,10 +4,21 @@ mod client;
mod errors;
mod request;
pub use tikv_client_common::security::SecurityManager;
pub use tikv_client_common::Error;
pub use tikv_client_common::Result;
#[doc(inline)]
pub use crate::{
client::{KvClient, KvConnect, TikvConnect},
errors::{HasKeyErrors, HasRegionError, HasRegionErrors},
request::Request,
};
pub use tikv_client_common::{security::SecurityManager, Error, Result};
pub use crate::client::KvClient;
#[doc(inline)]
pub use crate::client::KvConnect;
#[doc(inline)]
pub use crate::client::TikvConnect;
#[doc(inline)]
pub use crate::errors::HasKeyErrors;
#[doc(inline)]
pub use crate::errors::HasRegionError;
#[doc(inline)]
pub use crate::errors::HasRegionErrors;
#[doc(inline)]
pub use crate::request::Request;

14
tikv-client-store/src/request.rs
View File

@ -1,10 +1,16 @@
// Copyright 2020 TiKV Project Authors. Licensed under Apache-2.0.
use crate::{Error, Result};
use std::any::Any;
use std::time::Duration;
use async_trait::async_trait;
use std::{any::Any, time::Duration};
use tikv_client_proto::{kvrpcpb, tikvpb::tikv_client::TikvClient};
use tonic::{transport::Channel, IntoRequest};
use tikv_client_proto::kvrpcpb;
use tikv_client_proto::tikvpb::tikv_client::TikvClient;
use tonic::transport::Channel;
use tonic::IntoRequest;
use crate::Error;
use crate::Result;
#[async_trait]
pub trait Request: Any + Sync + Send + 'static {