Added txids to 'BlockEntry' and optimized block parsing thread usage.

Author: Randy808

src/new_index/fetch.rs

@@ -16,7 +16,7 @@ use std::thread;
 
 use electrs_macros::trace;
 
-use crate::chain::{Block, BlockHash};
+use crate::chain::{Block, BlockHash, Txid};
 use crate::daemon::Daemon;
 use crate::errors::*;
 use crate::util::{spawn_thread, HeaderEntry, SyncChannel};
@@ -45,6 +45,8 @@ pub struct BlockEntry {
     pub block: Block,
     pub entry: HeaderEntry,
     pub size: u32,
+    /// Pre-computed txids, must always correspond 1:1 with block.txdata
+    pub txids: Vec<Txid>,
 }
 
 type SizedBlock = (Block, u32);
@@ -106,10 +108,14 @@ fn bitcoind_fetcher(
                 let block_entries: Vec<BlockEntry> = blocks
                     .into_iter()
                     .zip(entries)
-                    .map(|(block, entry)| BlockEntry {
-                        entry: entry.clone(), // TODO: remove this clone()
-                        size: block.total_size() as u32,
-                        block,
+                    .map(|(block, entry)| {
+                        let txids = block.txdata.iter().map(|tx| tx.compute_txid()).collect();
+                        BlockEntry {
+                            entry: entry.clone(), // TODO: remove this clone()
+                            size: block.total_size() as u32,
+                            txids,
+                            block,
+                        }
                     })
                     .collect();
                 assert_eq!(block_entries.len(), entries.len());
@@ -156,7 +162,10 @@ fn blkfiles_fetcher(
                         let blockhash = block.block_hash();
                         entry_map
                             .remove(&blockhash)
-                            .map(|entry| BlockEntry { block, entry, size })
+                            .map(|entry| {
+                                let txids = block.txdata.iter().map(|tx| tx.compute_txid()).collect();
+                                BlockEntry { block, entry, size, txids }
+                            })
                             .or_else(|| {
                                 trace!("skipping block {}", blockhash);
                                 None
@@ -224,9 +233,14 @@ fn blkfiles_parser(blobs: Fetcher<Vec<u8>>, magic: u32) -> Fetcher<Vec<SizedBloc
     Fetcher::from(
         chan.into_receiver(),
         spawn_thread("blkfiles_parser", move || {
+            let pool = rayon::ThreadPoolBuilder::new()
+                .num_threads(0) // CPU-bound
+                .thread_name(|i| format!("parse-blocks-{}", i))
+                .build()
+                .unwrap();
             blobs.map(|blob| {
                 trace!("parsing {} bytes", blob.len());
-                let blocks = parse_blocks(blob, magic).expect("failed to parse blk*.dat file");
+                let blocks = parse_blocks(&pool, blob, magic).expect("failed to parse blk*.dat file");
                 sender
                     .send(blocks)
                     .expect("failed to send blocks from blk*.dat file");
@@ -236,7 +250,7 @@ fn blkfiles_parser(blobs: Fetcher<Vec<u8>>, magic: u32) -> Fetcher<Vec<SizedBloc
 }
 
 #[trace]
-fn parse_blocks(blob: Vec<u8>, magic: u32) -> Result<Vec<SizedBlock>> {
+fn parse_blocks(pool: &rayon::ThreadPool, blob: Vec<u8>, magic: u32) -> Result<Vec<SizedBlock>> {
     let mut cursor = Cursor::new(&blob);
     let mut slices = vec![];
     let max_pos = blob.len() as u64;
@@ -273,11 +287,6 @@ fn parse_blocks(blob: Vec<u8>, magic: u32) -> Result<Vec<SizedBlock>> {
         cursor.set_position(end as u64);
     }
 
-    let pool = rayon::ThreadPoolBuilder::new()
-        .num_threads(0) // CPU-bound
-        .thread_name(|i| format!("parse-blocks-{}", i))
-        .build()
-        .unwrap();
     Ok(pool.install(|| {
         slices
             .into_par_iter()

src/new_index/schema.rs

@@ -1193,13 +1193,12 @@ fn add_blocks(block_entries: &[BlockEntry], iconfig: &IndexerConfig) -> Vec<DBRo
         .map(|b| {
             let mut rows = vec![];
             let blockhash = full_hash(&b.entry.hash()[..]);
-            let txids: Vec<Txid> = b.block.txdata.iter().map(|tx| tx.compute_txid()).collect();
-            for (tx, txid) in b.block.txdata.iter().zip(txids.iter()) {
+            for (tx, txid) in b.block.txdata.iter().zip(b.txids.iter()) {
                 add_transaction(*txid, tx, &mut rows, iconfig);
             }
 
             if !iconfig.light_mode {
-                rows.push(BlockRow::new_txids(blockhash, &txids).into_row());
+                rows.push(BlockRow::new_txids(blockhash, &b.txids).into_row());
                 rows.push(BlockRow::new_meta(blockhash, &BlockMeta::from(b)).into_row());
             }
 
@@ -1290,9 +1289,10 @@ fn index_blocks(
         .par_iter() // serialization is CPU-intensive
         .map(|b| {
             let mut rows = vec![];
-            for tx in &b.block.txdata {
-                let height = b.entry.height() as u32;
-                index_transaction(tx, height, previous_txos_map, &mut rows, iconfig);
+            let height = b.entry.height() as u32;
+            for (tx, txid) in b.block.txdata.iter().zip(b.txids.iter()) {
+                let txid_hash = full_hash(&txid[..]);
+                index_transaction(tx, txid_hash, height, previous_txos_map, &mut rows, iconfig);
             }
             rows.push(BlockRow::new_done(full_hash(&b.entry.hash()[..])).into_row()); // mark block as "indexed"
             rows
@@ -1304,12 +1304,12 @@ fn index_blocks(
 // TODO: return an iterator?
 fn index_transaction(
     tx: &Transaction,
+    txid: FullHash,
     confirmed_height: u32,
     previous_txos_map: &HashMap<OutPoint, TxOut>,
     rows: &mut Vec<DBRow>,
     iconfig: &IndexerConfig,
 ) {
-    let txid = full_hash(&tx.compute_txid()[..]);
 
     // persist tx confirmation row:
     //      C{txid} → "{block_height}"
@@ -1911,7 +1911,9 @@ pub mod bench {
             let height = 702861;
             let hash = block.block_hash();
             let header = block.header.clone();
+            let txids = block.txdata.iter().map(|tx| tx.compute_txid()).collect();
             let block_entry = BlockEntry {
+                txids,
                 block,
                 entry: HeaderEntry::new(height, hash, header),
                 size: 0u32, // wrong but not needed for benching