Test Response Multiple Headers at ONce

sync/src/synchronizer/get_headers_process.rs

@@ -74,18 +74,34 @@ impl<'a> GetHeadersProcess<'a> {
             );
 
             self.synchronizer.peers().getheaders_received(self.peer);
-            let headers: Vec<core::HeaderView> =
-                active_chain.get_locator_response(block_number, &hash_stop);
-            // response headers
 
-            debug!("headers len={}", headers.len());
+            let hash_size: packed::Uint32 = 20_u32.pack();
+            let length_20_for_test = packed::Byte32::new_unchecked(hash_size.as_bytes());
+            if hash_stop.eq(&length_20_for_test) {
+                let headers: Vec<core::HeaderView> =
+                    active_chain.get_locator_response(block_number, &hash_stop);
+                // response headers
 
-            let content = packed::SendHeaders::new_builder()
-                .headers(headers.into_iter().map(|x| x.data()).pack())
-                .build();
-            let message = packed::SyncMessage::new_builder().set(content).build();
+                debug!("headers len={}", headers.len());
+                let content = packed::SendHeaders::new_builder()
+                    .headers(headers.into_iter().map(|x| x.data()).pack())
+                    .build();
+                let message = packed::SyncMessage::new_builder().set(content).build();
+                attempt!(send_message_to(self.nc, self.peer, &message));
+            } else {
+                let headers_vec: Vec<Vec<core::HeaderView>> =
+                    active_chain.get_locator_responses(block_number, &hash_stop);
+                // response headers
 
-            attempt!(send_message_to(self.nc, self.peer, &message));
+                debug!("headers vec len={}", headers_vec.len());
+                for headers in headers_vec {
+                    let content = packed::SendHeaders::new_builder()
+                        .headers(headers.into_iter().map(|x| x.data()).pack())
+                        .build();
+                    let message = packed::SyncMessage::new_builder().set(content).build();
+                    attempt!(send_message_to(self.nc, self.peer, &message));
+                }
+            }
         } else {
             return StatusCode::GetHeadersMissCommonAncestors
                 .with_context(format!("{block_locator_hashes:#x?}"));

sync/src/synchronizer/headers_process.rs

@@ -52,6 +52,13 @@ impl<'a> HeadersProcess<'a> {
         true
     }
 
+    fn is_parent_exists(&self, first_header: &core::HeaderView) -> bool {
+        let shared: &SyncShared = self.synchronizer.shared();
+        shared
+            .get_header_fields(&first_header.parent_hash())
+            .is_some()
+    }
+
     pub fn accept_first(&self, first: &core::HeaderView) -> ValidationResult {
         let shared: &SyncShared = self.synchronizer.shared();
         let verifier = HeaderVerifier::new(shared, shared.consensus());
@@ -92,15 +99,19 @@ impl<'a> HeadersProcess<'a> {
 
     pub fn execute(self) -> Status {
         debug!("HeadersProcess begins");
-        let shared: &SyncShared = self.synchronizer.shared();
-        let consensus = shared.consensus();
         let headers = self
             .message
             .headers()
             .to_entity()
             .into_iter()
             .map(packed::Header::into_view)
             .collect::<Vec<_>>();
+        self.execute_inner(headers)
+    }
+
+    fn execute_inner(self, headers: Vec<core::HeaderView>) -> Status {
+        let shared: &SyncShared = self.synchronizer.shared();
+        let consensus = shared.consensus();
 
         if headers.len() > MAX_HEADERS_LEN {
             warn!("HeadersProcess is oversized");
@@ -128,6 +139,15 @@ impl<'a> HeadersProcess<'a> {
             return StatusCode::HeadersIsInvalid.with_context("not continuous");
         }
 
+        if !self.is_parent_exists(&headers[0]) {
+            // put the headers into a memory cache
+            self.synchronizer
+                .header_cache
+                .insert(headers[0].parent_hash(), headers);
+            // verify them later
+            return Status::ok();
+        }
+
         let result = self.accept_first(&headers[0]);
         match result.state {
             ValidationState::Invalid => {
@@ -210,6 +230,18 @@ impl<'a> HeadersProcess<'a> {
             }
         }
 
+        {
+            // these headers verify success
+            // may the headers's tail header_hash exist in headers_cahce?
+            if let Some(headers) = self
+                .synchronizer
+                .header_cache
+                .get(&headers.last().expect("last header must exist").hash())
+            {
+                return self.execute_inner(headers.to_owned());
+            }
+        }
+
         Status::ok()
     }
 }

sync/src/synchronizer/mod.rs

@@ -44,11 +44,13 @@ use ckb_systemtime::unix_time_as_millis;
 
 #[cfg(test)]
 use ckb_types::core;
+use ckb_types::core::HeaderView;
 use ckb_types::{
     core::BlockNumber,
     packed::{self, Byte32},
     prelude::*,
 };
+use dashmap::DashMap;
 use std::{
     collections::HashSet,
     sync::{atomic::Ordering, Arc},
@@ -301,6 +303,9 @@ pub struct Synchronizer {
     pub(crate) chain: ChainController,
     /// Sync shared state
     pub shared: Arc<SyncShared>,
+
+    // First Headers's parent_hash -> Headers
+    pub(crate) header_cache: DashMap<Byte32, Vec<HeaderView>>,
     fetch_channel: Option<channel::Sender<FetchCMD>>,
 }
 
@@ -309,10 +314,12 @@ impl Synchronizer {
     ///
     /// This is a runtime sync protocol shared state, and any Sync protocol messages will be processed and forwarded by it
     pub fn new(chain: ChainController, shared: Arc<SyncShared>) -> Synchronizer {
+        let header_cache = DashMap::new();
         Synchronizer {
             chain,
             shared,
             fetch_channel: None,
+            header_cache,
         }
     }
 

sync/src/types/mod.rs

@@ -1909,6 +1909,21 @@ impl ActiveChain {
             .collect()
     }
 
+    pub fn get_locator_responses(
+        &self,
+        block_number: BlockNumber,
+        _hash_stop: &Byte32,
+    ) -> Vec<Vec<core::HeaderView>> {
+        (0..32)
+            .map(|index| {
+                self.get_locator_response(
+                    block_number + (index as u64 * MAX_HEADERS_LEN as u64),
+                    &Byte32::default(),
+                )
+            })
+            .collect()
+    }
+
     pub fn send_getheaders_to_peer(
         &self,
         nc: &dyn CKBProtocolContext,
@@ -1945,9 +1960,11 @@ impl ActiveChain {
             block_number_and_hash.hash()
         );
         let locator_hash = self.get_locator(block_number_and_hash);
+        let hash_size: packed::Uint32 = 20_u32.pack();
+        let length_20_for_test = packed::Byte32::new_unchecked(hash_size.as_bytes());
         let content = packed::GetHeaders::new_builder()
             .block_locator_hashes(locator_hash.pack())
-            .hash_stop(packed::Byte32::zero())
+            .hash_stop(length_20_for_test)
             .build();
         let message = packed::SyncMessage::new_builder().set(content).build();
         let _status = send_message(SupportProtocols::Sync.protocol_id(), nc, peer, &message);