node: Request blocks in an uneven distribution

0-200k blocks are essentially empty. Abuse this fact

Author: rustaceanrob

node/config_spec.toml

@@ -19,7 +19,7 @@ doc = "The bitcoin network to operate on. Default `bitcoin`. Options are `bitcoi
 [[param]]
 name = "ping_timeout"
 type = "u64"
-default = "15"
+default = "60"
 doc = "The time (seconds) a peer has to respond to a `ping` message. Pings are sent aggressively throughout IBD to find slow peers."
 
 [[param]]
@@ -43,11 +43,11 @@ doc = "The maximum time (seconds) to write to a TCP stream until the connection
 [[param]]
 name = "min_blocks_per_sec"
 type = "f64"
-default = "1."
+default = "0.20"
 doc = "The minimum rate a peer has to respond to block requests."
 
 [[param]]
 name = "tasks"
 type = "usize"
-default = "64"
+default = "128"
 doc = "The number of tasks to download blocks. Default is 64. Each task uses two OS threads."

node/src/bin/ibd.rs

@@ -74,8 +74,7 @@ fn main() {
     let acc_task = std::thread::spawn(move || accumulator_state.verify());
     let peers = Arc::new(Mutex::new(peers));
     let mut tasks = Vec::new();
-    let chunk_size = chain.best_header().height() as usize / task_num;
-    let hashes = hashes_from_chain(Arc::clone(&chain), chunk_size);
+    let hashes = hashes_from_chain(Arc::clone(&chain), task_num);
     for (task_id, chunk) in hashes.into_iter().enumerate() {
         let chain = Arc::clone(&chain);
         let tx = tx.clone();

node/src/lib.rs

@@ -34,6 +34,7 @@ use p2p::{
 };
 
 const PROTOCOL_VERSION: ProtocolVersion = ProtocolVersion::WTXID_RELAY_VERSION;
+const MAX_GETDATA: usize = 50_000;
 
 pub fn elapsed_time(then: Instant) {
     let duration_sec = then.elapsed().as_secs_f64();
@@ -79,11 +80,12 @@ pub fn sync_block_headers(
     hosts: &[SocketAddr],
     chainman: Arc<ChainstateManager>,
     network: Network,
-    timeout_params: TimeoutParams,
+    mut timeout_params: TimeoutParams,
 ) {
     let mut rng = thread_rng();
     let then = Instant::now();
     tracing::info!("Syncing block headers to assume valid hash");
+    timeout_params.ping_interval(Duration::from_secs(30));
     loop {
         let random = hosts
             .choose(&mut rng)
@@ -162,6 +164,7 @@ pub fn get_blocks_for_range(
     updater: Sender<AccumulatorUpdate>,
     mut batch: Vec<BlockHash>,
 ) {
+    tracing::info!("{task_id} assigned {} blocks", batch.len());
     let mut rng = thread_rng();
     loop {
         let peer = {
@@ -207,7 +210,15 @@ pub fn get_blocks_for_range(
                         hints.get_block_offsets(block_height).into_iter().collect();
                     // tracing::info!("{task_id} -> {block_height}:{hash}");
                     let file_path = block_dir.join(format!("{hash}.block"));
-                    let mut file = File::create_new(file_path).expect("duplicate block file");
+                    let file = File::create_new(file_path);
+                    let mut file = match file {
+                        Ok(file) => file,
+                        Err(e) => {
+                            tracing::warn!("Conflicting open files at: {}", block_height);
+                            tracing::warn!("{e}");
+                            panic!("files cannot conflict");
+                        }
+                    };
                     let block_bytes = consensus::serialize(&block);
                     file.write_all(&block_bytes)
                         .expect("failed to write block file");
@@ -278,7 +289,7 @@ pub fn get_blocks_for_range(
                 }
             }
             if metrics.ping_timed_out(ping_timeout) {
-                tracing::info!("{task_id} failed to respond to a ping");
+                tracing::warn!("{task_id} failed to respond to a ping");
                 break;
             }
         }
@@ -289,21 +300,50 @@ pub fn get_blocks_for_range(
     tracing::info!("All block ranges fetched: {task_id}");
 }
 
-pub fn hashes_from_chain(chain: Arc<ChainstateManager>, chunks: usize) -> Vec<Vec<BlockHash>> {
+pub fn hashes_from_chain(chain: Arc<ChainstateManager>, jobs: usize) -> Vec<Vec<BlockHash>> {
     let height = chain.best_header().height();
     let mut hashes = Vec::with_capacity(height as usize);
     let mut curr = chain.best_header();
     let tip_hash = BlockHash::from_byte_array(curr.block_hash().hash);
     hashes.push(tip_hash);
+    let mut out = Vec::new();
     while let Ok(next) = curr.prev() {
         if next.height() == 0 {
-            return hashes.chunks(chunks).map(|slice| slice.to_vec()).collect();
+            break;
         }
         let hash = BlockHash::from_byte_array(next.block_hash().hash);
         hashes.push(hash);
         curr = next;
     }
-    hashes.chunks(chunks).map(|slice| slice.to_vec()).collect()
+    // These blocks are empty. Fetch the maximum amount of blocks.
+    let first_epoch = hashes.split_off(hashes.len() - 200_000);
+    let first_chunks: Vec<Vec<BlockHash>> = first_epoch
+        .chunks(MAX_GETDATA)
+        .map(|slice| slice.to_vec())
+        .collect();
+    out.extend(first_chunks);
+    // These start to get larger, but are still small
+    let next_epoch = hashes.split_off(hashes.len() - 100_000);
+    let next_chunks: Vec<Vec<BlockHash>> = next_epoch
+        .chunks(MAX_GETDATA / 2)
+        .map(|slice| slice.to_vec())
+        .collect();
+    out.extend(next_chunks);
+    // Still not entirely full, but almost there
+    let to_segwit = hashes.split_off(hashes.len() - 100_000);
+    let to_segwit_chunks: Vec<Vec<BlockHash>> = to_segwit
+        .chunks(MAX_GETDATA / 4)
+        .map(|slice| slice.to_vec())
+        .collect();
+    out.extend(to_segwit_chunks);
+    // Now divide the rest among jobs
+    let chunk_size = hashes.len() / jobs;
+    let rest: Vec<Vec<BlockHash>> = hashes
+        .chunks(chunk_size)
+        .map(|slice| slice.to_vec())
+        .collect();
+    out.extend(rest);
+    out
 }
 
 pub trait ChainExt {