perf(stages): Adds benchmark to TransactionLookupStage

bin/reth/src/db/mod.rs

@@ -175,6 +175,7 @@ impl Command {
                     BlockTransitionIndex,
                     TxTransitionIndex,
                     SyncStage,
+                    TxHashNumber,
                     Transactions
                 ]);
             }

crates/primitives/src/block.rs

@@ -23,7 +23,10 @@ impl Deref for Block {
 }
 
 /// Sealed Ethereum full block.
-#[derive(Debug, Clone, PartialEq, Eq, Default, RlpEncodable, RlpDecodable)]
+#[derive_arbitrary(rlp, 10)]
+#[derive(
+    Debug, Clone, PartialEq, Eq, Default, RlpEncodable, RlpDecodable, Serialize, Deserialize,
+)]
 pub struct SealedBlock {
     /// Locked block header.
     pub header: SealedHeader,

crates/primitives/src/header.rs

@@ -5,7 +5,7 @@ use crate::{
 };
 use bytes::{BufMut, BytesMut};
 use ethers_core::types::{Block, H256 as EthersH256, H64};
-use reth_codecs::{derive_arbitrary, main_codec, Compact};
+use reth_codecs::{add_arbitrary_tests, derive_arbitrary, main_codec, Compact};
 use reth_rlp::{length_of_length, Decodable, Encodable};
 use serde::{Deserialize, Serialize};
 use std::ops::Deref;
@@ -209,14 +209,41 @@ impl Decodable for Header {
 
 /// A [`Header`] that is sealed at a precalculated hash, use [`SealedHeader::unseal()`] if you want
 /// to modify header.
-#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+#[add_arbitrary_tests(rlp)]
+#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
 pub struct SealedHeader {
     /// Locked Header fields.
     header: Header,
     /// Locked Header hash.
     hash: BlockHash,
 }
 
+#[cfg(any(test, feature = "arbitrary"))]
+impl proptest::arbitrary::Arbitrary for SealedHeader {
+    type Parameters = ();
+    type Strategy = proptest::strategy::BoxedStrategy<SealedHeader>;
+
+    fn arbitrary_with(_: Self::Parameters) -> Self::Strategy {
+        use proptest::prelude::{any, Strategy};
+
+        any::<(Header, BlockHash)>()
+            .prop_map(move |(header, _)| {
+                let hash = header.hash_slow();
+                SealedHeader { header, hash }
+            })
+            .boxed()
+    }
+}
+
+#[cfg(any(test, feature = "arbitrary"))]
+impl<'a> arbitrary::Arbitrary<'a> for SealedHeader {
+    fn arbitrary(u: &mut arbitrary::Unstructured<'a>) -> arbitrary::Result<Self> {
+        let header = Header::arbitrary(u)?;
+        let hash = header.hash_slow();
+        Ok(SealedHeader { header, hash })
+    }
+}
+
 impl From<Block<EthersH256>> for SealedHeader {
     fn from(block: Block<EthersH256>) -> Self {
         let header = Header {

crates/stages/benches/criterion.rs

@@ -16,8 +16,7 @@ criterion_main!(benches);
 
 fn senders(c: &mut Criterion) {
     let mut group = c.benchmark_group("Stages");
-    group.measurement_time(std::time::Duration::from_millis(2000));
-    group.warm_up_time(std::time::Duration::from_millis(2000));
+
     // don't need to run each stage for that many times
     group.sample_size(10);
 
@@ -33,8 +32,7 @@ fn senders(c: &mut Criterion) {
 
 fn tx_lookup(c: &mut Criterion) {
     let mut group = c.benchmark_group("Stages");
-    group.measurement_time(std::time::Duration::from_millis(2000));
-    group.warm_up_time(std::time::Duration::from_millis(2000));
+
     // don't need to run each stage for that many times
     group.sample_size(10);
 

crates/stages/src/stages/tx_lookup.rs

@@ -66,14 +66,37 @@ impl<DB: Database> Stage<DB> for TransactionLookupStage {
                 .unwrap_or_default()
         });
 
-        // Collect tranasctions for each body and insert the reverse lookup for hash -> tx_id.
+        // Collect transactions for each body
+        let mut tx_list = vec![];
         for body_entry in bodies {
             let (_, body) = body_entry?;
             let transactions = tx_cursor.walk(body.start_tx_id)?.take(body.tx_count as usize);
 
             for tx_entry in transactions {
                 let (id, transaction) = tx_entry?;
-                tx.put::<tables::TxHashNumber>(transaction.hash(), id)?;
+                tx_list.push((transaction.hash(), id));
+            }
+        }
+
+        // Sort before inserting the reverse lookup for hash -> tx_id.
+        tx_list.sort_by(|txa, txb| txa.0.cmp(&txb.0));
+
+        let mut txhash_cursor = tx.cursor_write::<tables::TxHashNumber>()?;
+
+        // If the last inserted element in the database is smaller than the first in our set, then
+        // we can just append into the DB. This probably only ever happens during sync, on
+        // the first table insertion.
+        let append = tx_list
+            .first()
+            .zip(txhash_cursor.last()?)
+            .map(|((first, _), (last, _))| &last < first)
+            .unwrap_or_default();
+
+        for (tx_hash, id) in tx_list {
+            if append {
+                txhash_cursor.append(tx_hash, id)?;
+            } else {
+                txhash_cursor.insert(tx_hash, id)?;
             }
         }
 

crates/storage/db/Cargo.toml

@@ -82,6 +82,10 @@ arbitrary = [
     "dep:proptest-derive",
 ]
 
+[[bench]]
+name = "hash_keys"
+harness = false
+
 [[bench]]
 name = "criterion"
 harness = false