Make queued chache to decrease importing time

Author: HoOngEe

core/src/verification/queue/mod.rs

@@ -37,7 +37,7 @@ const MIN_MEM_LIMIT: usize = 16384;
 const MIN_QUEUE_LIMIT: usize = 512;
 
 // maximum possible number of verification threads.
-const MAX_VERIFIERS: usize = 8;
+const MAX_VERIFIERS: usize = 2;
 
 /// Type alias for block queue convenience.
 pub type BlockQueue = VerificationQueue<kind::Blocks>;

sync/src/block/downloader/header.rs

@@ -45,6 +45,7 @@ pub struct HeaderDownloader {
     pivot: Pivot,
     request_time: Option<Instant>,
     downloaded: HashMap<H256, Header>,
+    queued: HashMap<H256, Header>,
     trial: usize,
 }
 
@@ -69,6 +70,7 @@ impl HeaderDownloader {
             },
             request_time: None,
             downloaded: HashMap::new(),
+            queued: HashMap::new(),
             trial: 0,
         }
     }
@@ -100,12 +102,17 @@ impl HeaderDownloader {
         self.request_time.map_or(false, |time| (Instant::now() - time).as_secs() > MAX_WAIT)
     }
 
-    /// Find header from download cache, and then from blockchain
+    /// Find header from queued headers, downloaded cache and then from blockchain
     /// Panics if header dosn't exist
     fn pivot_header(&self) -> Header {
-        match self.downloaded.get(&self.pivot.hash) {
+        match self.queued.get(&self.pivot.hash) {
             Some(header) => header.clone(),
-            None => self.client.block_header(&BlockId::Hash(self.pivot.hash)).unwrap(),
+            None => self
+                .downloaded
+                .get(&self.pivot.hash)
+                .map(Clone::clone)
+                .or_else(|| self.client.block_header(&BlockId::Hash(self.pivot.hash)))
+                .unwrap(),
         }
     }
 
@@ -143,7 +150,7 @@ impl HeaderDownloader {
         if self.best_hash == self.pivot.hash {
             ctrace!(SYNC, "Ignore received headers, pivot already reached the best hash");
         } else if first_header_hash == self.pivot.hash {
-            for header in headers.iter() {
+            for header in headers[1..].iter() {
                 self.downloaded.insert(header.hash(), header.clone());
             }
 
@@ -173,7 +180,7 @@ impl HeaderDownloader {
 
     pub fn mark_as_imported(&mut self, hashes: Vec<H256>) {
         for hash in hashes {
-            self.downloaded.remove(&hash);
+            self.queued.remove(&hash);
 
             if self.best_hash == hash {
                 self.pivot = Pivot {
@@ -183,4 +190,12 @@ impl HeaderDownloader {
             }
         }
     }
+
+    pub fn mark_as_queued(&mut self, hashes: Vec<H256>) {
+        for hash in hashes {
+            if let Some(header) = self.downloaded.remove(&hash) {
+                self.queued.insert(hash, header);
+            }
+        }
+    }
 }

sync/src/block/extension.rs

@@ -680,19 +680,23 @@ impl Extension {
         completed.sort_unstable_by_key(EncodedHeader::number);
 
         let mut exists = Vec::new();
+        let mut queued = Vec::new();
+
         for header in completed {
+            let hash = header.hash();
             match self.client.import_header(header.clone().into_inner()) {
-                Err(BlockImportError::Import(ImportError::AlreadyInChain)) => exists.push(header.hash()),
+                Err(BlockImportError::Import(ImportError::AlreadyInChain)) => exists.push(hash),
                 // FIXME: handle import errors
                 Err(err) => {
                     cwarn!(SYNC, "Cannot import header({}): {:?}", header.hash(), err);
                     break
                 }
-                _ => {}
+                _ => queued.push(hash),
             }
         }
 
         let request = self.header_downloaders.get_mut(from).and_then(|peer| {
+            peer.mark_as_queued(queued);
             peer.mark_as_imported(exists);
             peer.create_request()
         });