grpc: multi conns to send raft messages.

etc/config-template.toml

@@ -25,6 +25,8 @@ messages-per-tick = 4096
 # grpc-concurrency = 2
 # The number of max concurrent streams/requests on a client connection.
 # grpc-concurrent-stream = 1024
+# The number of connections with each tikv server to send raft messages.
+# grpc-raft-conn-num = 10
 # Amount to read ahead on individual grpc streams.
 # grpc-stream-initial-window-size = "2MB"
 

src/bin/tikv-server.rs

@@ -631,6 +631,9 @@ fn build_cfg(matches: &ArgMatches,
     cfg_usize(&mut cfg.grpc_concurrent_stream,
               config,
               "server.grpc-concurrent-stream");
+    cfg_usize(&mut cfg.grpc_raft_conn_num,
+              config,
+              "server.grpc-raft-conn-num");
     cfg_usize(&mut cfg.grpc_stream_initial_window_size,
               config,
               "server.grpc-stream-initial-window-size");

src/server/config.rs

@@ -25,6 +25,7 @@ const DEFAULT_NOTIFY_CAPACITY: usize = 40960;
 const DEFAULT_END_POINT_CONCURRENCY: usize = 8;
 const DEFAULT_GRPC_CONCURRENCY: usize = 2;
 const DEFAULT_GRPC_CONCURRENT_STREAM: usize = 1024;
+const DEFAULT_GRPC_RAFT_CONN_NUM: usize = 10;
 const DEFAULT_GRPC_STREAM_INITIAL_WINDOW_SIZE: usize = 2 * 1024 * 1024;
 const DEFAULT_END_POINT_TXN_CONCURRENCY_RATIO: f64 = 0.25;
 const DEFAULT_END_POINT_SMALL_TXN_TASKS_LIMIT: usize = 2;
@@ -47,6 +48,7 @@ pub struct Config {
     pub messages_per_tick: usize,
     pub grpc_concurrency: usize,
     pub grpc_concurrent_stream: usize,
+    pub grpc_raft_conn_num: usize,
     pub grpc_stream_initial_window_size: usize,
     pub storage: StorageConfig,
     pub raft_store: RaftStoreConfig,
@@ -66,6 +68,7 @@ impl Default for Config {
             messages_per_tick: DEFAULT_MESSAGES_PER_TICK,
             grpc_concurrency: DEFAULT_GRPC_CONCURRENCY,
             grpc_concurrent_stream: DEFAULT_GRPC_CONCURRENT_STREAM,
+            grpc_raft_conn_num: DEFAULT_GRPC_RAFT_CONN_NUM,
             grpc_stream_initial_window_size: DEFAULT_GRPC_STREAM_INITIAL_WINDOW_SIZE,
             end_point_concurrency: DEFAULT_END_POINT_CONCURRENCY,
             end_point_txn_concurrency_on_busy: usize::default(),

src/server/raft_client.rs

@@ -64,7 +64,7 @@ impl Conn {
 /// `RaftClient` is used for sending raft messages to other stores.
 pub struct RaftClient {
     env: Arc<Environment>,
-    conns: HashMap<SocketAddr, Conn>,
+    conns: HashMap<(SocketAddr, usize), Conn>,
     cfg: Config,
 }
 
@@ -77,24 +77,25 @@ impl RaftClient {
         }
     }
 
-    fn get_conn(&mut self, addr: SocketAddr) -> &Conn {
+    fn get_conn(&mut self, addr: SocketAddr, index: usize) -> &Conn {
         let env = self.env.clone();
         let cfg = self.cfg.clone();
         self.conns
-            .entry(addr)
+            .entry((addr, index))
             .or_insert_with(|| Conn::new(env, addr, &cfg))
     }
 
     pub fn send(&mut self, addr: SocketAddr, msg: RaftMessage) -> Result<()> {
+        let index = msg.get_region_id() as usize % self.cfg.grpc_raft_conn_num;
         let res = {
-            let conn = self.get_conn(addr);
+            let conn = self.get_conn(addr, index);
             UnboundedSender::send(&conn.stream, (msg, WriteFlags::default()))
         };
         if let Err(e) = res {
             warn!("server: drop conn with tikv endpoint {} error: {:?}",
                   addr,
                   e);
-            self.conns.remove(&addr);
+            self.conns.remove(&(addr, index));
             return Err(box_err!(e));
         }
         Ok(())