Merge #1839

1839: perf: after exiting the ibd mode, the invalid notify should be removed r=driftluo a=driftluo

After exiting the `ibd` mode, the invalid notify should be removed. Frequent switching of the context is also expensive in the Tokio runtime.

on tokio 0.1, `notify` submit to global queue will cause a  thundering herd on all task thread, it will cause all thread to catch the same mutex lock


Co-authored-by: driftluo <driftluo@foxmail.com>

chain/src/tests/basic.rs

@@ -127,7 +127,7 @@ fn test_transaction_spend_in_same_block() {
     let mut chain = MockChain::new(parent.clone(), shared.consensus());
     chain.gen_empty_block(&mock_store);
 
-    let last_cellbase = &shared.consensus().genesis_block().transactions()[1];;
+    let last_cellbase = &shared.consensus().genesis_block().transactions()[1];
     let last_cellbase_hash = last_cellbase.hash().to_owned();
     let tx1 = create_multi_outputs_transaction(&last_cellbase, vec![0], 2, vec![1]);
     let tx1_hash = tx1.hash().to_owned();
@@ -216,7 +216,7 @@ fn test_transaction_conflict_in_same_block() {
     let mut chain = MockChain::new(parent.clone(), shared.consensus());
     chain.gen_empty_block(&mock_store);
 
-    let last_cellbase = &shared.consensus().genesis_block().transactions()[1];;
+    let last_cellbase = &shared.consensus().genesis_block().transactions()[1];
     let tx1 = create_transaction(&last_cellbase.hash(), 1);
     let tx1_hash = tx1.hash().to_owned();
     let tx2 = create_transaction(&tx1_hash, 2);
@@ -250,7 +250,7 @@ fn test_transaction_conflict_in_different_blocks() {
     let mut chain = MockChain::new(parent.clone(), shared.consensus());
     chain.gen_empty_block(&mock_store);
 
-    let last_cellbase = &shared.consensus().genesis_block().transactions()[1];;
+    let last_cellbase = &shared.consensus().genesis_block().transactions()[1];
     let tx1 = create_multi_outputs_transaction(&last_cellbase, vec![0], 2, vec![1]);
     let tx1_hash = tx1.hash();
     let tx2 = create_multi_outputs_transaction(&tx1, vec![0], 2, vec![1]);
@@ -287,7 +287,7 @@ fn test_invalid_out_point_index_in_same_block() {
     let mut chain = MockChain::new(parent.clone(), shared.consensus());
     chain.gen_empty_block(&mock_store);
 
-    let last_cellbase = &shared.consensus().genesis_block().transactions()[1];;
+    let last_cellbase = &shared.consensus().genesis_block().transactions()[1];
     let tx1 = create_transaction(&last_cellbase.hash(), 1);
     let tx1_hash = tx1.hash().to_owned();
     let tx2 = create_transaction(&tx1_hash, 2);

network/src/protocols/mod.rs

@@ -32,6 +32,7 @@ use crate::{
 pub trait CKBProtocolContext: Send {
     // Interact with underlying p2p service
     fn set_notify(&self, interval: Duration, token: u64) -> Result<(), Error>;
+    fn remove_notify(&self, token: u64) -> Result<(), Error>;
     fn quick_send_message(
         &self,
         proto_id: ProtocolId,
@@ -269,6 +270,11 @@ impl CKBProtocolContext for DefaultCKBProtocolContext {
             .set_service_notify(self.proto_id, interval, token)?;
         Ok(())
     }
+    fn remove_notify(&self, token: u64) -> Result<(), Error> {
+        self.p2p_control
+            .remove_service_notify(self.proto_id, token)?;
+        Ok(())
+    }
     fn quick_send_message(
         &self,
         proto_id: ProtocolId,

sync/src/relayer/tests/helper.rs

@@ -174,6 +174,9 @@ impl CKBProtocolContext for MockProtocalContext {
     fn set_notify(&self, _interval: Duration, _token: u64) -> Result<(), Error> {
         unimplemented!()
     }
+    fn remove_notify(&self, _token: u64) -> Result<(), Error> {
+        unimplemented!()
+    }
     fn quick_send_message(
         &self,
         _proto_id: ProtocolId,

sync/src/synchronizer/mod.rs

@@ -482,6 +482,8 @@ impl CKBProtocolHandler for Synchronizer {
                 IBD_BLOCK_FETCH_TOKEN => {
                     if snapshot.is_initial_block_download() {
                         self.find_blocks_to_fetch(nc.as_ref());
+                    } else if nc.remove_notify(IBD_BLOCK_FETCH_TOKEN).is_err() {
+                        trace!("remove ibd block fetch fail");
                     }
                 }
                 NOT_IBD_BLOCK_FETCH_TOKEN => {
@@ -941,6 +943,10 @@ mod tests {
             unimplemented!();
         }
 
+        fn remove_notify(&self, _token: u64) -> Result<(), ckb_network::Error> {
+            unimplemented!()
+        }
+
         fn future_task(
             &self,
             task: Box<

sync/src/tests/mod.rs

@@ -154,6 +154,10 @@ impl CKBProtocolContext for TestNetworkContext {
         Ok(())
     }
 
+    fn remove_notify(&self, _token: u64) -> Result<(), ckb_network::Error> {
+        Ok(())
+    }
+
     fn future_task(
         &self,
         task: Box<