diff --git a/third_party/silentdragonxlite/silentdragonxlite-cli/lib/src/lightwallet.rs b/third_party/silentdragonxlite/silentdragonxlite-cli/lib/src/lightwallet.rs index 5d99c71..f92499e 100644 --- a/third_party/silentdragonxlite/silentdragonxlite-cli/lib/src/lightwallet.rs +++ b/third_party/silentdragonxlite/silentdragonxlite-cli/lib/src/lightwallet.rs @@ -1891,6 +1891,50 @@ pub fn scan_full_mempool_tx(&self, tx: &Transaction, height: i32, _datetime: u64 num_invalidated as u64 } + // Below these sizes the old inline path is kept / threads are not worth their start-up cost. + const PAR_WITNESS_MIN_NOTES: usize = 16; + const PAR_WITNESS_MIN_APPENDS: usize = 128; + + /// Append `nodes` (in order) to every witness. Equivalent to + /// `for w in witnesses { for n in nodes { w.append(*n).unwrap() } }`; large inputs are split over + /// scoped threads, each owning a disjoint chunk of witnesses. + fn append_nodes_to_witnesses( + witnesses: &mut [&mut IncrementalWitness], + nodes: &[Node], + min_parallel_appends: usize, + ) { + let max_threads_by_work = witnesses.len() / 8; + let total = witnesses.len().saturating_mul(nodes.len()); + let threads = if max_threads_by_work < 2 || total < min_parallel_appends { + 1 + } else { + let cores = std::thread::available_parallelism().map(|n| n.get()).unwrap_or(1); + cmp::min(cmp::min(cores, 8), max_threads_by_work) + }; + + if threads < 2 { + for w in witnesses.iter_mut() { + for n in nodes { + w.append(*n).expect("sapling tree full"); + } + } + return; + } + + let chunk_size = (witnesses.len() + threads - 1) / threads; + std::thread::scope(|s| { + for chunk in witnesses.chunks_mut(chunk_size) { + s.spawn(move || { + for w in chunk.iter_mut() { + for n in nodes { + w.append(*n).expect("sapling tree full"); + } + } + }); + } + }); + } + /// Scans a [`CompactOutput`] with a set of [`ExtendedFullViewingKey`]s. /// /// Returns a [`WalletShieldedOutput`] and corresponding [`IncrementalWitness`] if this @@ -1904,6 +1948,8 @@ pub fn scan_full_mempool_tx(&self, tx: &Transaction, height: i32, _datetime: u64 ivks: &[Fs], tree: &mut CommitmentTree, existing_witnesses: &mut [&mut IncrementalWitness], + defer_existing: bool, + deferred_nodes: &mut Vec, block_witnesses: &mut [&mut IncrementalWitness], new_witnesses: &mut [&mut IncrementalWitness], pool: &ThreadPool @@ -1930,8 +1976,12 @@ pub fn scan_full_mempool_tx(&self, tx: &Transaction, height: i32, _datetime: u64 Err(_) => { error!("Failed to parse epk for output index {} — committing cmu to tree, skipping decryption", index); let node = Node::new(cmu.into()); - for witness in existing_witnesses { - witness.append(node).unwrap(); + if defer_existing { + deferred_nodes.push(node); + } else { + for witness in existing_witnesses { + witness.append(node).unwrap(); + } } for witness in block_witnesses { witness.append(node).unwrap(); @@ -1973,8 +2023,12 @@ pub fn scan_full_mempool_tx(&self, tx: &Transaction, height: i32, _datetime: u64 // Increment tree and witnesses let node = Node::new(cmu.into()); - for witness in existing_witnesses { - witness.append(node).unwrap(); + if defer_existing { + deferred_nodes.push(node); + } else { + for witness in existing_witnesses { + witness.append(node).unwrap(); + } } for witness in block_witnesses { witness.append(node).unwrap(); @@ -2036,6 +2090,11 @@ pub fn scan_full_mempool_tx(&self, tx: &Transaction, height: i32, _datetime: u64 let mut wtxs: Vec = vec![]; let ivks = extfvks.iter().map(|extfvk| extfvk.fvk.vk.ivk()).collect::>(); + // Large wallets: collect this block's leaves and append them to the existing notes' witnesses + // once, at the end, in output order (in parallel when worthwhile). Small wallets keep the inline path. + let defer_existing = existing_witnesses.len() >= LightWallet::PAR_WITNESS_MIN_NOTES; + let mut deferred_nodes: Vec = Vec::new(); + for tx in block.vtx.into_iter() { let num_spends = tx.spends.len(); let num_outputs = tx.outputs.len(); @@ -2111,6 +2170,8 @@ pub fn scan_full_mempool_tx(&self, tx: &Transaction, height: i32, _datetime: u64 &ivks, tree, existing_witnesses, + defer_existing, + &mut deferred_nodes, &mut block_witnesses, &mut new_witnesses, pool @@ -2144,6 +2205,10 @@ pub fn scan_full_mempool_tx(&self, tx: &Transaction, height: i32, _datetime: u64 } } + if !deferred_nodes.is_empty() { + LightWallet::append_nodes_to_witnesses(existing_witnesses, &deferred_nodes, LightWallet::PAR_WITNESS_MIN_APPENDS); + } + wtxs } pub fn scan_block(&self, block_bytes: &[u8]) -> Result, i32> { diff --git a/third_party/silentdragonxlite/silentdragonxlite-cli/lib/src/lightwallet/witness_tests.rs b/third_party/silentdragonxlite/silentdragonxlite-cli/lib/src/lightwallet/witness_tests.rs index c68f2de..668caf6 100644 --- a/third_party/silentdragonxlite/silentdragonxlite-cli/lib/src/lightwallet/witness_tests.rs +++ b/third_party/silentdragonxlite/silentdragonxlite-cli/lib/src/lightwallet/witness_tests.rs @@ -105,7 +105,6 @@ fn random_cmu_bytes() -> Vec { } /// A valid cmu with a 31-byte epk: the leaf is committed, decryption is skipped. -#[allow(dead_code)] fn bad_epk() -> CompactOutput { let mut o = CompactOutput::new(); o.set_cmu(random_cmu_bytes()); @@ -115,7 +114,6 @@ fn bad_epk() -> CompactOutput { } /// A 31-byte cmu: nothing is appended anywhere. -#[allow(dead_code)] fn bad_cmu() -> CompactOutput { let mut o = CompactOutput::new(); o.set_cmu(vec![1u8; 31]); @@ -696,3 +694,102 @@ fn witness_serialization_compat() { assert_eq!(c.wallet.clear_old_witnesses(), 0); assert!(before(&c, &x) == bx && before(&c, &y) == by, "pending / unknown-spender notes were touched"); } + +fn random_node() -> Node { + Node::new(Fr::random(&mut OsRng).into_repr()) +} + +#[test] +fn append_nodes_parallel_matches_serial() { + for &k in &[0usize, 1, 7, 8, 16, 17, 37, 64] { + // K witnesses at different positions of one tree, all following it to its tip. + let mut tree = CommitmentTree::::new(); + let mut base: Vec> = vec![]; + for i in 0..k { + for _ in 0..1 + i % 3 { + let n = random_node(); + tree.append(n).unwrap(); + for w in base.iter_mut() { + w.append(n).unwrap(); + } + } + base.push(IncrementalWitness::from_tree(&tree)); + } + for &n_nodes in &[0usize, 1, 5] { + let ctx = format!("K={} nodes={}", k, n_nodes); + let nodes: Vec = (0..n_nodes).map(|_| random_node()).collect(); + let mut tip = tree.clone(); + for n in &nodes { + tip.append(*n).unwrap(); + } + // (i) a literal double loop + let mut serial = base.clone(); + for w in serial.iter_mut() { + for n in &nodes { + w.append(*n).unwrap(); + } + } + // (ii) the helper, forced inline + let mut inline = base.clone(); + { + let mut refs: Vec<&mut IncrementalWitness> = inline.iter_mut().collect(); + LightWallet::append_nodes_to_witnesses(&mut refs[..], &nodes, usize::MAX); + } + // (iii) the helper with no size threshold (threaded for K >= 16 on >= 2 cores) + let mut par = base.clone(); + { + let mut refs: Vec<&mut IncrementalWitness> = par.iter_mut().collect(); + LightWallet::append_nodes_to_witnesses(&mut refs[..], &nodes, 0); + } + let root = tip.root(); + for i in 0..k { + let s = wbytes(&serial[i]); + assert!(s == wbytes(&inline[i]), "{}: witness {} differs (inline)", ctx, i); + assert!(s == wbytes(&par[i]), "{}: witness {} differs (parallel)", ctx, i); + assert!(par[i].root() == root, "{}: witness {} is not at the tree's root", ctx, i); + } + } + } +} + +/// K notes, one per block; then one block whose first tx pays us and carries an output with an +/// unparseable epk (its leaf is committed) and one with an unparseable cmu (nothing is appended), +/// and whose second tx carries foreign outputs: 8 leaves. With K = 17 that is 17 x 8 = 136 +/// existing-witness appends, which takes the deferred, threaded path; K = 3 takes the inline one. +fn deferred_case(k: usize) { + let mut c = Chain::new(); + for _ in 0..k { + let g = c.pay_us(10_000); + c.mine(vec![g]); + } + let ctx = format!("deferred K={}", k); + c.compare(true, &ctx); + assert_eq!(c.model.notes.iter().filter(|n| n.spent_at.is_none()).count(), k); + + let f = foreign(); + let ours = c.ours(); + let (t1, id1) = tx_with(vec![out_to(&ours, 777), bad_epk(), bad_cmu(), out_to(&f, 1), out_to(&f, 2), out_to(&f, 3)], vec![]); + let (t2, id2) = tx_with(vec![out_to(&f, 4), out_to(&f, 5), out_to(&f, 6)], vec![]); + let size_before = c.model.tree.size(); + c.mine(vec![ + GenTx { ctx: t1, txid: id1, ours: Some(0), spends: vec![] }, + GenTx { ctx: t2, txid: id2, ours: None, spends: vec![] }, + ]); + assert_eq!(c.model.tree.size() - size_before, 8, "{}: leaves of the bad-output block", ctx); + c.compare(true, &ctx); + assert!(check_anchor_alignment(&c.wallet, &ctx) >= k + 1, "{}: every live note is checked", ctx); + let root = c.wallet.blocks.read().unwrap().last().unwrap().tree.root(); + assert!(root == c.model.tree.root(), "{}: tip root differs from the model", ctx); + + // One more block: every note, including the one created in the bad-output block, advances. + let g = c.pay_foreign(2); + c.mine(vec![g]); + c.compare(true, &ctx); + check_anchor_alignment(&c.wallet, &ctx); +} + +#[test] +fn deferred_append_path_with_bad_outputs() { + deferred_case(17); + deferred_case(3); +}