Optimize and refactor lookup_txos()

- Avoid unnecessary copying of prev outpoints

- When looking for both mempool and on-chain txos, accumulate the set of
  outpoints that remain to be looked up to avoid re-checking for them
  later again in the found set

- Refactored lookup_txos() to use lookup_txo() internally rather than
  the other way around, which was less efficient

- Lookup txos in mempool first, then on-chain

- ChainQuery::lookup_txos() now returns a Result instead of panicking
  when outpoints are missing

- Removed ChainQuery::lookup_avail_txos(), which is now longer neceesary

Author: shesek

src/bin/tx-fingerprint-stats.rs

@@ -83,12 +83,12 @@ fn main() {
         //info!("{:?},{:?}", txid, blockid);
 
         let prevouts = chain.lookup_txos(
-            &tx.input
+            tx.input
                 .iter()
                 .filter(|txin| has_prevout(txin))
                 .map(|txin| txin.previous_output)
                 .collect(),
-        );
+        ).unwrap();
 
         let total_out: u64 = tx.output.iter().map(|out| out.value.to_sat()).sum();
         let small_out = tx

src/new_index/mempool.rs

@@ -1,5 +1,5 @@
 use arraydeque::{ArrayDeque, Wrapping};
-use itertools::Itertools;
+use itertools::{Either, Itertools};
 
 #[cfg(not(feature = "liquid"))]
 use bitcoin::consensus::encode::serialize;
@@ -340,7 +340,7 @@ impl Mempool {
             self.txstore.insert(txid, tx);
         }
         // Phase 2: index history and spend edges (can fail if some txos cannot be found)
-        let txos = match self.lookup_txos(&self.get_prevouts(&txids)) {
+        let txos = match self.lookup_txos(self.get_prevouts(&txids)) {
             Ok(txos) => txos,
             Err(err) => {
                 warn!("lookup txouts failed: {}", err);
@@ -427,34 +427,29 @@ impl Mempool {
         }
     }
 
-    pub fn lookup_txo(&self, outpoint: &OutPoint) -> Result<TxOut> {
-        let mut outpoints = BTreeSet::new();
-        outpoints.insert(*outpoint);
-        Ok(self.lookup_txos(&outpoints)?.remove(outpoint).unwrap())
+    fn lookup_txo(&self, outpoint: &OutPoint) -> Option<TxOut> {
+        self.txstore
+            .get(&outpoint.txid)
+            .and_then(|tx| tx.output.get(outpoint.vout as usize).cloned())
     }
 
-    pub fn lookup_txos(&self, outpoints: &BTreeSet<OutPoint>) -> Result<HashMap<OutPoint, TxOut>> {
+    pub fn lookup_txos(&self, outpoints: BTreeSet<OutPoint>) -> Result<HashMap<OutPoint, TxOut>> {
         let _timer = self
             .latency
             .with_label_values(&["lookup_txos"])
             .start_timer();
 
-        let confirmed_txos = self.chain.lookup_avail_txos(outpoints);
+        // Get the txos available in the mempool, skipping over (and collecting) missing ones
+        let (mut txos, remain_outpoints): (HashMap<_, _>, _) = outpoints
+            .into_iter()
+            .partition_map(|outpoint| match self.lookup_txo(&outpoint) {
+                Some(txout) => Either::Left((outpoint, txout)),
+                None => Either::Right(outpoint),
+            });
 
-        let mempool_txos = outpoints
-            .iter()
-            .filter(|outpoint| !confirmed_txos.contains_key(outpoint))
-            .map(|outpoint| {
-                self.txstore
-                    .get(&outpoint.txid)
-                    .and_then(|tx| tx.output.get(outpoint.vout as usize).cloned())
-                    .map(|txout| (*outpoint, txout))
-                    .chain_err(|| format!("missing outpoint {:?}", outpoint))
-            })
-            .collect::<Result<HashMap<OutPoint, TxOut>>>()?;
+        // Get the remaining txos from the chain (fails if any are missing)
+        txos.extend(self.chain.lookup_txos(remain_outpoints)?);
 
-        let mut txos = confirmed_txos;
-        txos.extend(mempool_txos);
         Ok(txos)
     }
 

src/new_index/query.rs

@@ -118,7 +118,7 @@ impl Query {
             .or_else(|| self.mempool().lookup_raw_txn(txid))
     }
 
-    pub fn lookup_txos(&self, outpoints: &BTreeSet<OutPoint>) -> HashMap<OutPoint, TxOut> {
+    pub fn lookup_txos(&self, outpoints: BTreeSet<OutPoint>) -> HashMap<OutPoint, TxOut> {
         // the mempool lookup_txos() internally looks up confirmed txos as well
         self.mempool()
             .lookup_txos(outpoints)

src/new_index/schema.rs

@@ -327,7 +327,7 @@ impl Indexer {
     fn index(&self, blocks: &[BlockEntry]) {
         let previous_txos_map = {
             let _timer = self.start_timer("index_lookup");
-            lookup_txos(&self.store.txstore_db, &get_previous_txos(blocks), false)
+            lookup_txos(&self.store.txstore_db, get_previous_txos(blocks)).unwrap()
         };
         let rows = {
             let _timer = self.start_timer("index_process");
@@ -859,14 +859,9 @@ impl ChainQuery {
         lookup_txo(&self.store.txstore_db, outpoint)
     }
 
-    pub fn lookup_txos(&self, outpoints: &BTreeSet<OutPoint>) -> HashMap<OutPoint, TxOut> {
+    pub fn lookup_txos(&self, outpoints: BTreeSet<OutPoint>) -> Result<HashMap<OutPoint, TxOut>> {
         let _timer = self.start_timer("lookup_txos");
-        lookup_txos(&self.store.txstore_db, outpoints, false)
-    }
-
-    pub fn lookup_avail_txos(&self, outpoints: &BTreeSet<OutPoint>) -> HashMap<OutPoint, TxOut> {
-        let _timer = self.start_timer("lookup_available_txos");
-        lookup_txos(&self.store.txstore_db, outpoints, true)
+        lookup_txos(&self.store.txstore_db, outpoints)
     }
 
     pub fn lookup_spend(&self, outpoint: &OutPoint) -> Option<SpendingInput> {
@@ -1033,21 +1028,16 @@ fn get_previous_txos(block_entries: &[BlockEntry]) -> BTreeSet<OutPoint> {
         .collect()
 }
 
-fn lookup_txos(
-    txstore_db: &DB,
-    outpoints: &BTreeSet<OutPoint>,
-    allow_missing: bool,
-) -> HashMap<OutPoint, TxOut> {
+fn lookup_txos(txstore_db: &DB, outpoints: BTreeSet<OutPoint>) -> Result<HashMap<OutPoint, TxOut>> {
     let mut remain_outpoints = outpoints.iter();
     txstore_db
         .multi_get(outpoints.iter().map(TxOutRow::key))
         .into_iter()
-        .filter_map(|res| {
+        .map(|res| {
             let outpoint = remain_outpoints.next().unwrap();
             match res.unwrap() {
-                Some(txo) => Some((*outpoint, deserialize(&txo).expect("failed to parse TxOut"))),
-                None if allow_missing => None,
-                None => panic!("missing txo {}", outpoint),
+                Some(txo) => Ok((*outpoint, deserialize(&txo).expect("failed to parse TxOut"))),
+                None => Err(format!("missing txo {}", outpoint).into()),
             }
         })
         .collect()

src/rest.rs

@@ -468,7 +468,7 @@ fn prepare_txs(
         })
         .collect();
 
-    let prevouts = query.lookup_txos(&outpoints);
+    let prevouts = query.lookup_txos(outpoints);
 
     txs.into_iter()
         .map(|(tx, blockid)| TransactionValue::new(tx, blockid, &prevouts, config))