From 44874bc557567174d18ca02f3e7ae141a3de492a Mon Sep 17 00:00:00 2001 From: joshieDo <93316087+joshieDo@users.noreply.github.com> Date: Tue, 15 Aug 2023 18:40:09 +0100 Subject: [PATCH] feat: add `TransactionSigned::recover_signers` with the same order (#4120) --- Cargo.lock | 1 + crates/consensus/auto-seal/src/lib.rs | 6 +- crates/primitives/Cargo.toml | 1 + crates/primitives/src/block.rs | 2 +- crates/primitives/src/transaction/mod.rs | 61 ++++++++++++++++++- crates/revm/src/executor.rs | 7 +-- .../src/providers/database/provider.rs | 18 +++--- 7 files changed, 76 insertions(+), 20 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index ceac36ad04..4a8d706e77 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5817,6 +5817,7 @@ dependencies = [ "proptest", "proptest-derive", "rand 0.8.5", + "rayon", "reth-codecs", "reth-rlp", "reth-rlp-derive", diff --git a/crates/consensus/auto-seal/src/lib.rs b/crates/consensus/auto-seal/src/lib.rs index 7b14f544a4..cadfb5593b 100644 --- a/crates/consensus/auto-seal/src/lib.rs +++ b/crates/consensus/auto-seal/src/lib.rs @@ -349,10 +349,8 @@ impl StorageInner { let block = Block { header, body: transactions, ommers: vec![], withdrawals: None }; - let senders = - block.body.iter().map(|tx| tx.recover_signer()).collect::>>().ok_or( - BlockExecutionError::Validation(BlockValidationError::SenderRecoveryError), - )?; + let senders = TransactionSigned::recover_signers(&block.body, block.body.len()) + .ok_or(BlockExecutionError::Validation(BlockValidationError::SenderRecoveryError))?; trace!(target: "consensus::auto", transactions=?&block.body, "executing transactions"); diff --git a/crates/primitives/Cargo.toml b/crates/primitives/Cargo.toml index 8716a1e228..da697b8b25 100644 --- a/crates/primitives/Cargo.toml +++ b/crates/primitives/Cargo.toml @@ -60,6 +60,7 @@ impl-serde = "0.4.0" once_cell = "1.17.0" zstd = { version = "0.12", features = ["experimental"] } paste = "1.0" +rayon = "1.7" tempfile = "3.3" sha2 = "0.10.7" diff --git a/crates/primitives/src/block.rs b/crates/primitives/src/block.rs index 0c9866ba28..5248787c15 100644 --- a/crates/primitives/src/block.rs +++ b/crates/primitives/src/block.rs @@ -163,7 +163,7 @@ impl SealedBlock { /// Expensive operation that recovers transaction signer. See [SealedBlockWithSenders]. pub fn senders(&self) -> Option> { - self.body.iter().map(|tx| tx.recover_signer()).collect::>>() + TransactionSigned::recover_signers(&self.body, self.body.len()) } /// Seal sealed block with recovered transaction senders. diff --git a/crates/primitives/src/transaction/mod.rs b/crates/primitives/src/transaction/mod.rs index f63c6b9e1e..ed347ce3f4 100644 --- a/crates/primitives/src/transaction/mod.rs +++ b/crates/primitives/src/transaction/mod.rs @@ -7,6 +7,8 @@ use bytes::{Buf, BytesMut}; use derive_more::{AsRef, Deref}; pub use error::InvalidTransactionError; pub use meta::TransactionMeta; +use once_cell::sync::Lazy; +use rayon::prelude::{IntoParallelIterator, ParallelIterator}; use reth_codecs::{add_arbitrary_tests, derive_arbitrary, Compact}; use reth_rlp::{ length_of_length, Decodable, DecodeError, Encodable, Header, EMPTY_LIST_CODE, EMPTY_STRING_CODE, @@ -34,6 +36,15 @@ mod signature; mod tx_type; pub(crate) mod util; +// Expected number of transactions where we can expect a speed-up by recovering the senders in +// parallel. +pub(crate) static PARALLEL_SENDER_RECOVERY_THRESHOLD: Lazy = + Lazy::new(|| match rayon::current_num_threads() { + 0..=1 => usize::MAX, + 2..=8 => 10, + _ => 5, + }); + /// A raw transaction. /// /// Transaction types were introduced in [EIP-2718](https://eips.ethereum.org/EIPS/eip-2718). @@ -939,6 +950,21 @@ impl TransactionSigned { self.signature.recover_signer(signature_hash) } + /// Recovers a list of signers from a transaction list iterator + /// + /// Returns `None`, if some transaction's signature is invalid, see also + /// [Self::recover_signer]. + pub fn recover_signers<'a, T>(txes: T, num_txes: usize) -> Option> + where + T: IntoParallelIterator + IntoIterator + Send, + { + if num_txes < *PARALLEL_SENDER_RECOVERY_THRESHOLD { + txes.into_iter().map(|tx| tx.recover_signer()).collect() + } else { + txes.into_par_iter().map(|tx| tx.recover_signer()).collect() + } + } + /// Consumes the type, recover signer and return [`TransactionSignedEcRecovered`] /// /// Returns `None` if the transaction's signature is invalid, see also [Self::recover_signer]. @@ -1323,12 +1349,17 @@ impl IntoRecoveredTransaction for TransactionSignedEcRecovered { #[cfg(test)] mod tests { use crate::{ - transaction::{signature::Signature, TransactionKind, TxEip1559, TxLegacy}, + sign_message, + transaction::{ + signature::Signature, TransactionKind, TxEip1559, TxLegacy, + PARALLEL_SENDER_RECOVERY_THRESHOLD, + }, Address, Bytes, Transaction, TransactionSigned, TransactionSignedEcRecovered, H256, U256, }; use bytes::BytesMut; use ethers_core::utils::hex; use reth_rlp::{Decodable, DecodeError, Encodable}; + use secp256k1::{KeyPair, Secp256k1}; use std::str::FromStr; #[test] @@ -1564,4 +1595,32 @@ mod tests { tx.encode(&mut b); assert_eq!(s, hex::encode(&b)); } + + proptest::proptest! { + #![proptest_config(proptest::prelude::ProptestConfig::with_cases(1))] + + #[test] + fn test_parallel_recovery_order(txes in proptest::collection::vec(proptest::prelude::any::(), *PARALLEL_SENDER_RECOVERY_THRESHOLD * 5)) { + let mut rng =rand::thread_rng(); + let secp = Secp256k1::new(); + let txes: Vec = txes.into_iter().map(|mut tx| { + if let Some(chain_id) = tx.chain_id() { + // Otherwise we might overflow when calculating `v` on `recalculate_hash` + tx.set_chain_id(chain_id % (u64::MAX / 2 - 36)); + } + + let key_pair = KeyPair::new(&secp, &mut rng); + + let signature = + sign_message(H256::from_slice(&key_pair.secret_bytes()[..]), tx.signature_hash()).unwrap(); + + TransactionSigned::from_transaction_and_signature(tx, signature) + }).collect(); + + let parallel_senders = TransactionSigned::recover_signers(&txes, txes.len()).unwrap(); + let seq_senders = txes.iter().map(|tx| tx.recover_signer()).collect::>>().unwrap(); + + assert_eq!(parallel_senders, seq_senders); + } + } } diff --git a/crates/revm/src/executor.rs b/crates/revm/src/executor.rs index 5ce714521f..990e674bb7 100644 --- a/crates/revm/src/executor.rs +++ b/crates/revm/src/executor.rs @@ -84,11 +84,8 @@ where Err(BlockValidationError::SenderRecoveryError.into()) } } else { - body.iter() - .map(|tx| { - tx.recover_signer().ok_or(BlockValidationError::SenderRecoveryError.into()) - }) - .collect() + TransactionSigned::recover_signers(body, body.len()) + .ok_or(BlockValidationError::SenderRecoveryError.into()) } } diff --git a/crates/storage/provider/src/providers/database/provider.rs b/crates/storage/provider/src/providers/database/provider.rs index ba1474afd6..1d80bf4662 100644 --- a/crates/storage/provider/src/providers/database/provider.rs +++ b/crates/storage/provider/src/providers/database/provider.rs @@ -22,7 +22,10 @@ use reth_db::{ transaction::{DbTx, DbTxMut}, BlockNumberList, DatabaseError, }; -use reth_interfaces::Result; +use reth_interfaces::{ + executor::{BlockExecutionError, BlockValidationError}, + Result, +}; use reth_primitives::{ keccak256, stage::{StageCheckpoint, StageId}, @@ -1908,14 +1911,11 @@ impl<'this, TX: DbTxMut<'this> + DbTx<'this>> BlockWriter for DatabaseProvider<' let tx_iter = if Some(block.body.len()) == senders_len { block.body.into_iter().zip(senders.unwrap()).collect::>() } else { - block - .body - .into_iter() - .map(|tx| { - let signer = tx.recover_signer(); - (tx, signer.unwrap_or_default()) - }) - .collect::>() + let senders = TransactionSigned::recover_signers(&block.body, block.body.len()).ok_or( + BlockExecutionError::Validation(BlockValidationError::SenderRecoveryError), + )?; + debug_assert_eq!(senders.len(), block.body.len(), "missing one or more senders"); + block.body.into_iter().zip(senders).collect() }; for (transaction, sender) in tx_iter {