feat: sync use async send

driftluo · driftluo · commit 74dfc5de24ca · 2025-11-13T11:18:50.000+08:00
diff --git a/network/src/lib.rs b/network/src/lib.rs
@@ -41,7 +41,10 @@ pub use p2p::{
     builder::ServiceBuilder,
     bytes, multiaddr, runtime,
     secio::{self, PeerId, PublicKey},
-    service::{ServiceControl, SessionType as RawSessionType, TargetProtocol, TargetSession},
+    service::{
+        ServiceAsyncControl, ServiceControl, SessionType as RawSessionType, TargetProtocol,
+        TargetSession,
+    },
     traits::ServiceProtocol,
     utils::{extract_peer_id, multiaddr_to_socketaddr},
 };
diff --git a/sync/src/relayer/compact_block_process.rs b/sync/src/relayer/compact_block_process.rs
@@ -269,7 +269,7 @@ fn contextual_check(
             block_hash,
             peer
         );
-        active_chain.send_getheaders_to_peer(nc.as_ref(), peer, (&tip).into());
+        active_chain.send_getheaders_to_peer(nc, peer, (&tip).into());
         return StatusCode::CompactBlockRequiresParent.with_context(format!(
             "{} parent: {}",
             block_hash,
diff --git a/sync/src/relayer/get_block_proposal_process.rs b/sync/src/relayer/get_block_proposal_process.rs
@@ -1,5 +1,5 @@
 use crate::relayer::{MAX_RELAY_TXS_BYTES_PER_BATCH, Relayer};
-use crate::utils::{send_message_to, send_message_to_async};
+use crate::utils::send_message_to_async;
 use crate::{Status, StatusCode, attempt};
 use ckb_logger::debug_target;
 use ckb_network::{CKBProtocolContext, PeerIndex};
diff --git a/sync/src/relayer/get_block_transactions_process.rs b/sync/src/relayer/get_block_transactions_process.rs
@@ -1,6 +1,6 @@
 use crate::relayer::{MAX_RELAY_TXS_NUM_PER_BATCH, Relayer};
-use crate::utils::send_message_to;
-use crate::{Status, StatusCode, attempt};
+use crate::utils::send_message_to_async;
+use crate::{Status, StatusCode};
 use ckb_logger::debug_target;
 use ckb_network::{CKBProtocolContext, PeerIndex};
 use ckb_store::ChainStore;
@@ -10,15 +10,15 @@ use std::sync::Arc;
 pub struct GetBlockTransactionsProcess<'a> {
     message: packed::GetBlockTransactionsReader<'a>,
     relayer: &'a Relayer,
-    nc: Arc<dyn CKBProtocolContext>,
+    nc: Arc<dyn CKBProtocolContext + Sync>,
     peer: PeerIndex,
 }
 
 impl<'a> GetBlockTransactionsProcess<'a> {
     pub fn new(
         message: packed::GetBlockTransactionsReader<'a>,
         relayer: &'a Relayer,
-        nc: Arc<dyn CKBProtocolContext>,
+        nc: Arc<dyn CKBProtocolContext + Sync>,
         peer: PeerIndex,
     ) -> Self {
         GetBlockTransactionsProcess {
@@ -94,7 +94,13 @@ impl<'a> GetBlockTransactionsProcess<'a> {
                 .build();
             let message = packed::RelayMessage::new_builder().set(content).build();
 
-            attempt!(send_message_to(self.nc.as_ref(), self.peer, &message));
+            self.relayer
+                .shared()
+                .shared()
+                .async_handle()
+                .spawn(async move {
+                    let _ignore = send_message_to_async(&self.nc, self.peer, &message).await;
+                });
         }
 
         Status::ok()
diff --git a/sync/src/relayer/get_transactions_process.rs b/sync/src/relayer/get_transactions_process.rs
@@ -1,5 +1,5 @@
 use crate::relayer::{MAX_RELAY_TXS_BYTES_PER_BATCH, MAX_RELAY_TXS_NUM_PER_BATCH, Relayer};
-use crate::utils::{send_message_to, send_message_to_async};
+use crate::utils::send_message_to_async;
 use crate::{Status, StatusCode, attempt};
 use ckb_logger::{debug_target, trace_target};
 use ckb_network::{CKBProtocolContext, PeerIndex};
diff --git a/sync/src/synchronizer/get_blocks_process.rs b/sync/src/synchronizer/get_blocks_process.rs
@@ -1,17 +1,17 @@
 use crate::synchronizer::Synchronizer;
-use crate::utils::send_message_to;
-use crate::{Status, StatusCode, attempt};
+use crate::utils::send_message_to_async;
+use crate::{Status, StatusCode};
 use ckb_constant::sync::{INIT_BLOCKS_IN_TRANSIT_PER_PEER, MAX_HEADERS_LEN};
 use ckb_logger::debug;
 use ckb_network::{CKBProtocolContext, PeerIndex};
 use ckb_shared::block_status::BlockStatus;
 use ckb_types::{packed, prelude::*};
-use std::collections::HashSet;
+use std::{collections::HashSet, sync::Arc};
 
 pub struct GetBlocksProcess<'a> {
     message: packed::GetBlocksReader<'a>,
     synchronizer: &'a Synchronizer,
-    nc: &'a dyn CKBProtocolContext,
+    nc: &'a Arc<dyn CKBProtocolContext + Sync>,
     peer: PeerIndex,
 }
 
@@ -20,7 +20,7 @@ impl<'a> GetBlocksProcess<'a> {
         message: packed::GetBlocksReader<'a>,
         synchronizer: &'a Synchronizer,
         peer: PeerIndex,
-        nc: &'a dyn CKBProtocolContext,
+        nc: &'a Arc<dyn CKBProtocolContext + Sync>,
     ) -> Self {
         GetBlocksProcess {
             peer,
@@ -75,7 +75,12 @@ impl<'a> GetBlocksProcess<'a> {
                 let content = packed::SendBlock::new_builder().block(block.data()).build();
                 let message = packed::SyncMessage::new_builder().set(content).build();
 
-                attempt!(send_message_to(self.nc, self.peer, &message));
+                let nc = Arc::clone(self.nc);
+                self.synchronizer
+                    .shared()
+                    .shared()
+                    .async_handle()
+                    .spawn(async move { send_message_to_async(&nc, self.peer, &message).await });
             } else {
                 // TODO response not found
                 // TODO add timeout check in synchronizer
diff --git a/sync/src/synchronizer/get_headers_process.rs b/sync/src/synchronizer/get_headers_process.rs
@@ -1,6 +1,6 @@
 use crate::synchronizer::Synchronizer;
-use crate::utils::{send_message, send_message_to};
-use crate::{Status, StatusCode, attempt};
+use crate::utils::{send_message_async, send_message_to_async};
+use crate::{Status, StatusCode};
 use ckb_constant::sync::MAX_LOCATOR_SIZE;
 use ckb_logger::{debug, info};
 use ckb_network::{CKBProtocolContext, PeerIndex, SupportProtocols};
@@ -9,20 +9,21 @@ use ckb_types::{
     packed::{self, Byte32},
     prelude::*,
 };
+use std::sync::Arc;
 
 pub struct GetHeadersProcess<'a> {
     message: packed::GetHeadersReader<'a>,
     synchronizer: &'a Synchronizer,
     peer: PeerIndex,
-    nc: &'a dyn CKBProtocolContext,
+    nc: &'a Arc<dyn CKBProtocolContext + Sync>,
 }
 
 impl<'a> GetHeadersProcess<'a> {
     pub fn new(
         message: packed::GetHeadersReader<'a>,
         synchronizer: &'a Synchronizer,
         peer: PeerIndex,
-        nc: &'a dyn CKBProtocolContext,
+        nc: &'a Arc<dyn CKBProtocolContext + Sync>,
     ) -> Self {
         GetHeadersProcess {
             message,
@@ -84,8 +85,12 @@ impl<'a> GetHeadersProcess<'a> {
                 .headers(headers.into_iter().map(|x| x.data()).collect::<Vec<_>>())
                 .build();
             let message = packed::SyncMessage::new_builder().set(content).build();
-
-            attempt!(send_message_to(self.nc, self.peer, &message));
+            let nc = Arc::clone(self.nc);
+            self.synchronizer
+                .shared()
+                .shared()
+                .async_handle()
+                .spawn(async move { send_message_to_async(&nc, self.peer, &message).await });
         } else {
             return StatusCode::GetHeadersMissCommonAncestors
                 .with_context(format!("{block_locator_hashes:#x?}"));
@@ -96,11 +101,16 @@ impl<'a> GetHeadersProcess<'a> {
     fn send_in_ibd(&self) {
         let content = packed::InIBD::new_builder().build();
         let message = packed::SyncMessage::new_builder().set(content).build();
-        let _ignore = send_message(
-            SupportProtocols::Sync.protocol_id(),
-            self.nc,
-            self.peer,
-            &message,
-        );
+        let nc = Arc::clone(self.nc);
+        let peer = self.peer;
+        self.synchronizer
+            .shared()
+            .shared()
+            .async_handle()
+            .spawn(async move {
+                let _ignore =
+                    send_message_async(SupportProtocols::Sync.protocol_id(), &nc, peer, &message)
+                        .await;
+            });
     }
 }
diff --git a/sync/src/synchronizer/headers_process.rs b/sync/src/synchronizer/headers_process.rs
@@ -10,12 +10,13 @@ use ckb_traits::HeaderFieldsProvider;
 use ckb_types::{core, packed, prelude::*};
 use ckb_verification::{HeaderError, HeaderVerifier};
 use ckb_verification_traits::Verifier;
+use std::sync::Arc;
 
 pub struct HeadersProcess<'a> {
     message: packed::SendHeadersReader<'a>,
     synchronizer: &'a Synchronizer,
     peer: PeerIndex,
-    nc: &'a dyn CKBProtocolContext,
+    nc: &'a Arc<dyn CKBProtocolContext + Sync>,
     active_chain: ActiveChain,
 }
 
@@ -24,7 +25,7 @@ impl<'a> HeadersProcess<'a> {
         message: packed::SendHeadersReader<'a>,
         synchronizer: &'a Synchronizer,
         peer: PeerIndex,
-        nc: &'a dyn CKBProtocolContext,
+        nc: &'a Arc<dyn CKBProtocolContext + Sync>,
     ) -> Self {
         let active_chain = synchronizer.shared.active_chain();
         HeadersProcess {
@@ -202,12 +203,16 @@ impl<'a> HeadersProcess<'a> {
             && (!peer_flags.is_protect && !peer_flags.is_whitelist && peer_flags.is_outbound)
         {
             debug!("Disconnect an unprotected outbound peer ({})", self.peer);
-            if let Err(err) = self
-                .nc
-                .disconnect(self.peer, "useless outbound peer in IBD")
-            {
-                return StatusCode::Network.with_context(format!("Disconnect error: {err:?}"));
-            }
+            let nc = Arc::clone(self.nc);
+            self.synchronizer
+                .shared()
+                .shared()
+                .async_handle()
+                .spawn(async move {
+                    let _ignore = nc
+                        .async_disconnect(self.peer, "useless outbound peer in IBD")
+                        .await;
+                });
         }
 
         Status::ok()
diff --git a/sync/src/synchronizer/in_ibd_process.rs b/sync/src/synchronizer/in_ibd_process.rs
@@ -2,18 +2,19 @@ use crate::synchronizer::Synchronizer;
 use crate::{Status, StatusCode};
 use ckb_logger::info;
 use ckb_network::{CKBProtocolContext, PeerIndex};
+use std::sync::Arc;
 
 pub struct InIBDProcess<'a> {
     synchronizer: &'a Synchronizer,
     peer: PeerIndex,
-    nc: &'a dyn CKBProtocolContext,
+    nc: &'a Arc<dyn CKBProtocolContext + Sync>,
 }
 
 impl<'a> InIBDProcess<'a> {
     pub fn new(
         synchronizer: &'a Synchronizer,
         peer: PeerIndex,
-        nc: &'a dyn CKBProtocolContext,
+        nc: &'a Arc<dyn CKBProtocolContext + Sync>,
     ) -> Self {
         InIBDProcess {
             nc,
@@ -22,7 +23,7 @@ impl<'a> InIBDProcess<'a> {
         }
     }
 
-    pub fn execute(self) -> Status {
+    pub async fn execute(self) -> Status {
         info!("getheader with ibd peer {:?}", self.peer);
         if let Some(mut kv_pair) = self.synchronizer.peers().state.get_mut(&self.peer) {
             let state = kv_pair.value_mut();
@@ -34,7 +35,9 @@ impl<'a> InIBDProcess<'a> {
             if state.peer_flags.is_outbound {
                 if state.peer_flags.is_whitelist {
                     self.synchronizer.shared().state().suspend_sync(state);
-                } else if let Err(err) = self.nc.disconnect(self.peer, "outbound in ibd") {
+                } else if let Err(err) =
+                    self.nc.async_disconnect(self.peer, "outbound in ibd").await
+                {
                     return StatusCode::Network.with_context(format!("Disconnect error: {err:?}"));
                 }
             } else {
diff --git a/sync/src/synchronizer/mod.rs b/sync/src/synchronizer/mod.rs
diff --git a/sync/src/tests/sync_shared.rs b/sync/src/tests/sync_shared.rs
diff --git a/sync/src/tests/synchronizer/functions.rs b/sync/src/tests/synchronizer/functions.rs
diff --git a/sync/src/types/mod.rs b/sync/src/types/mod.rs
