Add batch command for performance

Cargo.toml

@@ -26,7 +26,7 @@ slog-term = { version = "2.4" }
 tokio = { version = "1", features = ["full"] }
 tokio-stream = "0.1"
 tokio-util = { version = "0.7.1", features = ["rt"] }
-tikv-client = { git = "https://github.com/yongman/client-rust.git", branch = "perfomance-optimize" }
+tikv-client = { git = "https://github.com/yongman/client-rust.git", branch = "super-batch" }
 #tikv-client = { path = "../client-rust" }
 lazy_static = "1.4.0"
 thiserror = "1"

config.toml

@@ -18,10 +18,18 @@ log_file = "tikv-service.log"
 use_txn_api = true
 use_async_commit = true
 try_one_pc_commit = true
-use_pessimistic_txn = true
-local_pool_number = 4
+use_pessimistic_txn = false
+local_pool_number = 10
 txn_retry_count = 10
 txn_region_backoff_delay_ms = 2
 txn_region_backoff_delay_attemps = 5
 txn_lock_backoff_delay_ms = 2
 txn_lock_backoff_delay_attemps = 5
+
+completion_queue_size = 1
+grpc_keepalive_time = 10000
+grpc_keepalive_timeout = 2000
+allow_batch = true
+max_batch_wait_time = 10
+max_batch_size = 20
+max_inflight_requests = 10000

rust-toolchain

@@ -1 +1 @@
-nightly-2022-05-01
+nightly-2022-07-27

src/bin/server.rs

@@ -24,21 +24,18 @@ pub async fn main() -> tikv_service::Result<()> {
 
     let mut config: Option<Config> = None;
 
-    match cli.config {
-        Some(config_file_name) => {
-            let config_content =
-                fs::read_to_string(config_file_name).expect("Failed to read config file");
-
-            // deserialize toml config
-            config = match toml::from_str(&config_content) {
-                Ok(d) => Some(d),
-                Err(e) => {
-                    println!("Unable to load config file {}", e);
-                    exit(1);
-                }
-            };
-        }
-        None => (),
+    if let Some(config_file_name) = cli.config {
+        let config_content =
+            fs::read_to_string(config_file_name).expect("Failed to read config file");
+
+        // deserialize toml config
+        config = match toml::from_str(&config_content) {
+            Ok(d) => Some(d),
+            Err(e) => {
+                println!("Unable to load config file {}", e);
+                exit(1);
+            }
+        };
     };
 
     match &config {

src/client.rs

@@ -35,8 +35,8 @@ impl Client {
             name: "".to_owned(),
             fd: socket.as_raw_fd(),
             cmd: "".to_owned(),
-            local_addr: (&socket).local_addr().unwrap().to_string(),
-            peer_addr: (&socket).peer_addr().unwrap().to_string(),
+            local_addr: socket.local_addr().unwrap().to_string(),
+            peer_addr: socket.peer_addr().unwrap().to_string(),
             create_time: now,
             last_interaction: now,
             kill_tx,

src/cmd/fake.rs

@@ -9,11 +9,10 @@ use crate::tikv::errors::{
     REDIS_INVALID_CLIENT_ID_ERR, REDIS_NOT_SUPPORTED_ERR, REDIS_NO_SUCH_CLIENT_ERR,
     REDIS_VALUE_IS_NOT_INTEGER_ERR,
 };
-use crate::utils::resp_int;
 use crate::{
     config::LOGGER,
     tikv::errors::REDIS_UNKNOWN_SUBCOMMAND,
-    utils::{resp_bulk, resp_err, resp_invalid_arguments, resp_nil, resp_ok},
+    utils::{resp_bulk, resp_err, resp_int, resp_invalid_arguments, resp_nil, resp_ok},
     Connection, Frame, Parse,
 };
 
@@ -42,7 +41,7 @@ impl Fake {
         self,
         command: &str,
         dst: &mut Connection,
-        cur_client: u64,
+        cur_client: Arc<Mutex<Client>>,
         clients: Arc<Mutex<HashMap<u64, Arc<Mutex<Client>>>>>,
     ) -> crate::Result<()> {
         let response = self.do_apply(command, cur_client, clients).await;
@@ -63,7 +62,7 @@ impl Fake {
     async fn do_apply(
         self,
         command: &str,
-        cur_client: u64,
+        cur_client: Arc<Mutex<Client>>,
         clients: Arc<Mutex<HashMap<u64, Arc<Mutex<Client>>>>>,
     ) -> Frame {
         if !self.valid {
@@ -73,14 +72,16 @@ impl Fake {
             "READWRITE" => resp_ok(),
             "READONLY" => resp_ok(),
             "CLIENT" => {
+                // TODO client more management will be added later
                 match self.args[0].clone().to_uppercase().as_str() {
-                    "ID" => resp_int(cur_client as i64),
+                    "ID" => resp_int(cur_client.lock().await.id() as i64),
                     "LIST" => {
                         if self.args.len() == 1 {
-                            let clients_guard = clients.lock().await;
                             return resp_bulk(
-                                encode_clients_info(clients_guard.clone().into_values().collect())
-                                    .await,
+                                encode_clients_info(
+                                    clients.lock().await.clone().into_values().collect(),
+                                )
+                                .await,
                             );
                         }
 
@@ -113,19 +114,21 @@ impl Fake {
                         // three arguments format (old format)
                         if self.args.len() == 2 {
                             let mut target_client = None;
-                            let clients_guard = clients.lock().await;
-                            for client in clients_guard.values() {
-                                // make sure get the client guard with clients aguard obtained
-                                let client = client.lock().await;
-                                if client.peer_addr() == self.args[1] {
-                                    target_client = Some(client.clone());
-                                    break;
+                            {
+                                let lk_clients = clients.lock().await;
+                                for client in lk_clients.values() {
+                                    let lk_client = client.lock().await;
+                                    if lk_client.peer_addr() == self.args[1] {
+                                        target_client = Some(client.clone());
+                                        break;
+                                    }
                                 }
                             }
 
                             return match target_client {
                                 Some(client) => {
-                                    client.kill().await;
+                                    let lk_client = client.lock().await;
+                                    lk_client.kill().await;
                                     resp_ok()
                                 }
                                 None => resp_err(REDIS_NO_SUCH_CLIENT_ERR),
@@ -163,34 +166,37 @@ impl Fake {
                         }
 
                         // retrieve current client id in advance for preventing dead lock during clients traverse
+                        let cur_client_id = cur_client.lock().await.id();
                         let mut eligible_clients: Vec<Arc<Mutex<Client>>> = vec![];
-                        let clients_guard = clients.lock().await;
-                        for client in clients_guard.values() {
-                            let client_guard = client.lock().await;
-                            if !filter_peer_addr.is_empty()
-                                && client_guard.peer_addr() != filter_peer_addr
-                            {
-                                continue;
-                            }
-                            if !filter_local_addr.is_empty()
-                                && client_guard.local_addr() != filter_local_addr
-                            {
-                                continue;
-                            }
-                            if filter_id != 0 && client_guard.id() != filter_id {
-                                continue;
-                            }
-                            if cur_client == client_guard.id() && filter_skipme {
-                                continue;
-                            }
+                        {
+                            let lk_clients = clients.lock().await;
+                            for client in lk_clients.values() {
+                                let lk_client = client.lock().await;
+                                if !filter_peer_addr.is_empty()
+                                    && lk_client.peer_addr() != filter_peer_addr
+                                {
+                                    continue;
+                                }
+                                if !filter_local_addr.is_empty()
+                                    && lk_client.local_addr() != filter_local_addr
+                                {
+                                    continue;
+                                }
+                                if filter_id != 0 && lk_client.id() != filter_id {
+                                    continue;
+                                }
+                                if cur_client_id == lk_client.id() && filter_skipme {
+                                    continue;
+                                }
 
-                            eligible_clients.push(client.clone());
+                                eligible_clients.push(client.clone());
+                            }
                         }
 
                         let killed = eligible_clients.len() as i64;
                         for eligible_client in eligible_clients {
-                            // make sure get the client guard with clients aguard obtained
-                            eligible_client.lock().await.kill().await;
+                            let lk_eligible_client = eligible_client.lock().await;
+                            lk_eligible_client.kill().await;
                         }
 
                         resp_int(killed)
@@ -200,25 +206,13 @@ impl Fake {
                             return resp_invalid_arguments();
                         }
 
-                        let clients_guard = clients.lock().await;
-                        clients_guard
-                            .get(&cur_client)
-                            .unwrap()
-                            .lock()
-                            .await
-                            .set_name(&self.args[1]);
-
+                        let mut w_cur_client = cur_client.lock().await;
+                        w_cur_client.set_name(&self.args[1]);
                         resp_ok()
                     }
                     "GETNAME" => {
-                        let clients_guard = clients.lock().await;
-                        let name = clients_guard
-                            .get(&cur_client)
-                            .unwrap()
-                            .lock()
-                            .await
-                            .name()
-                            .to_owned();
+                        let r_cur_client = cur_client.lock().await;
+                        let name = r_cur_client.name().to_owned();
                         if name.is_empty() {
                             return resp_nil();
                         }
@@ -249,8 +243,8 @@ impl Fake {
 async fn encode_clients_info(clients: Vec<Arc<Mutex<Client>>>) -> Vec<u8> {
     let mut resp_list = String::new();
     for client in clients {
-        let client = client.lock().await;
-        resp_list.push_str(&client.to_string());
+        let r_client = client.lock().await;
+        resp_list.push_str(&r_client.to_string());
         resp_list.push('\n');
     }