feat(mempool): implement cached batch txs broadcast (nervosnetwork#20)

* feat(mempool): implement cached batch txs broadcast

* feat(mempool): split interval broadcaster into standalone task future

Remove lock on whole transaction cache

* fix(mempool): clippy warnings

* chore(mempool): split interval timer into standalone task

* fix(mempool): error on empty receiver channel in default adapter

It's ok when err_rx channel is empty

* fix(mempool): default adapter broadcast on None timer signal

Should check whether there is an indeed interval signal

* fix(mempool): panic in default adapter select macro

Add complete branch

* test(mempool): add interval broadcast unit tests

* chore(config): add mempool interval broadcast entries

core/mempool/Cargo.toml

@@ -11,7 +11,7 @@ protocol = { path = "../../protocol" }
 common-crypto = { path = "../../common/crypto" }
 core-network = { path = "../network" }
 
-futures-preview = "0.3.0-alpha.19"
+futures-preview = { version = "0.3.0-alpha.19", features = [ "async-await" ] }
 runtime-tokio = "0.3.0-alpha.6"
 runtime = "0.3.0-alpha.7"
 crossbeam-queue = "0.1"
@@ -25,6 +25,8 @@ rand = "0.6"
 hex = "0.3"
 serde_derive = "1.0"
 serde = "1.0"
+futures-timer = "1.0"
+log = "0.4"
 
 [dev-dependencies]
 chashmap = "2.2"

core/mempool/src/adapter/message.rs

@@ -1,6 +1,7 @@
 use std::sync::Arc;
 
 use async_trait::async_trait;
+use futures::future::try_join_all;
 use protocol::{
     traits::{Context, MemPool, MessageHandler, Priority, Rpc},
     types::{Hash, SignedTransaction},
@@ -16,8 +17,8 @@ pub const END_RESP_PULL_TXS: &str = "/rpc_resp/mempool/pull_txs";
 
 #[derive(Debug, Serialize, Deserialize)]
 pub struct MsgNewTxs {
-    #[serde(with = "core_network::serde")]
-    pub stx: SignedTransaction,
+    #[serde(with = "core_network::serde_multi")]
+    pub batch_stxs: Vec<SignedTransaction>,
 }
 
 pub struct NewTxsHandler<M> {
@@ -43,7 +44,22 @@ where
     async fn process(&self, ctx: Context, msg: Self::Message) -> ProtocolResult<()> {
         let ctx = ctx.mark_network_origin_new_txs();
 
-        self.mem_pool.insert(ctx, msg.stx).await
+        let insert_stx = |stx| -> _ {
+            let mem_pool = Arc::clone(&self.mem_pool);
+            let ctx = ctx.clone();
+
+            runtime::spawn(async move { mem_pool.insert(ctx, stx).await })
+        };
+
+        // Concurrently insert them
+        try_join_all(
+            msg.batch_stxs
+                .into_iter()
+                .map(insert_stx)
+                .collect::<Vec<_>>(),
+        )
+        .await
+        .map(|_| ())
     }
 }