Avoid recomputing txids when possible

Following Blockstream#89 (comment) and Blockstream#89 (comment)
shesek · Aug 13, 2024 · f3a13e7 · f3a13e7
1 parent 1e799d4
commit f3a13e7
Show file tree

Hide file tree

Showing 3 changed files with 13 additions and 14 deletions.
diff --git a/src/daemon.rs b/src/daemon.rs
@@ -11,7 +11,7 @@ use std::time::Duration;
 use base64::prelude::{Engine, BASE64_STANDARD};
 use error_chain::ChainedError;
 use hex::FromHex;
-use rayon::iter::{IntoParallelIterator, ParallelIterator};
+use rayon::iter::{IndexedParallelIterator, IntoParallelIterator, ParallelIterator};
 use serde_json::{from_str, from_value, Value};
 
 #[cfg(not(feature = "liquid"))]
@@ -461,7 +461,7 @@ impl Daemon {
  &'a self,
  method: &'a str,
  params_list: Vec<Value>,
- ) -> impl ParallelIterator<Item = Result<Value>> + 'a {
+ ) -> impl ParallelIterator<Item = Result<Value>> + IndexedParallelIterator + 'a {
  self.rpc_threads.install(move || {
  params_list.into_par_iter().map(move |params| {
  // Store a local per-thread Daemon, each with its own TCP connection. These will
@@ -537,7 +537,7 @@ impl Daemon {
 
  /// Fetch the given transactions in parallel over multiple threads and RPC connections,
  /// ignoring any missing ones and returning whatever is available.
- pub fn gettransactions_available(&self, txids: &[&Txid]) -> Result<Vec<Transaction>> {
+ pub fn gettransactions_available(&self, txids: &[&Txid]) -> Result<Vec<(Txid, Transaction)>> {
  const RPC_INVALID_ADDRESS_OR_KEY: i64 = -5;
 
  let params_list: Vec<Value> = txids
@@ -546,8 +546,9 @@ impl Daemon {
  .collect();
 
  self.requests_iter("getrawtransaction", params_list)
- .filter_map(|res| match res {
- Ok(val) => Some(tx_from_value(val)),
+ .zip(txids)
+ .filter_map(|(res, txid)| match res {
+ Ok(val) => Some(tx_from_value(val).map(|tx| (**txid, tx))),
  // Ignore 'tx not found' errors
  Err(Error(ErrorKind::RpcError(code, _, _), _))
  if code == RPC_INVALID_ADDRESS_OR_KEY =>

diff --git a/src/new_index/mempool.rs b/src/new_index/mempool.rs
@@ -288,14 +288,12 @@ impl Mempool {
  self.backlog_stats = (BacklogStats::new(&self.feeinfo), Instant::now());
  }
 
- pub fn add_by_txid(&mut self, daemon: &Daemon, txid: &Txid) -> Result<()> {
- if self.txstore.get(txid).is_none() {
+ pub fn add_by_txid(&mut self, daemon: &Daemon, txid: Txid) -> Result<()> {
+ if self.txstore.get(&txid).is_none() {
  if let Ok(tx) = daemon.getmempooltx(&txid) {
- self.add({
- let mut txs_map = HashMap::new();
- txs_map.insert(tx.txid(), tx);
- txs_map
- })
+ let mut txs_map = HashMap::new();
+ txs_map.insert(txid, tx);
+ self.add(txs_map)
  } else {
  bail!("add_by_txid cannot find {}", txid);
  }
@@ -528,7 +526,7 @@ impl Mempool {
  }
 
  let fetched_count = new_txs.len();
- fetched_txs.extend(&mut new_txs.into_iter().map(|tx| (tx.txid(), tx)));
+ fetched_txs.extend(new_txs);
 
  // Retry if any transactions were evicted form the mempool before we managed to get them
  if fetched_count != new_txids.len() {

diff --git a/src/new_index/query.rs b/src/new_index/query.rs
@@ -75,7 +75,7 @@ impl Query {
  .mempool
  .write()
  .unwrap()
- .add_by_txid(&self.daemon, &txid);
+ .add_by_txid(&self.daemon, txid);
  Ok(txid)
  }