diff --git a/src/gtest/test_wallet_notecache.cpp b/src/gtest/test_wallet_notecache.cpp index 9782c0ddb..4cada8aef 100644 --- a/src/gtest/test_wallet_notecache.cpp +++ b/src/gtest/test_wallet_notecache.cpp @@ -815,3 +815,54 @@ TEST_F(WalletNoteCacheUndecryptable, WarmUpSkipsBadNotes) { EXPECT_EQ(total - 3, cached); EXPECT_EQ(7u, Filtered({}, 1, INT_MAX, true, false, true).size()) << "the bad notes are spent or locked"; } + +// Big batches are decrypted on several threads; the result is exactly the serial one. +TEST(WalletNoteCacheParallel, ParallelFillMatchesSerial) { + NoteCacheWallet wallet; + auto skA = TestKey(0), skB = TestKey(1); + auto addrsA = AddressesOf(skA); + LOCK2(cs_main, wallet.cs_wallet); + ASSERT_TRUE(wallet.AddSaplingZKey(skA, addrsA[0])); + ASSERT_TRUE(wallet.AddSaplingIncomingViewingKey(skB.expsk.full_viewing_key().in_viewing_key(), skB.DefaultAddress())); + uint64_t value = 0; + for (int t = 0; t < 30; t++) { + std::vector outs; + for (int i = 0; i < 10; i++) + outs.push_back(MakeOutput(i % 3 == 2 ? skB.DefaultAddress() : addrsA[i % addrsA.size()], ++value)); + CTransaction tx = MakeTx(outs); + CWalletTx wtx(&wallet, tx); + mapSaplingNoteData_t found = wallet.FindMySaplingNotes(tx).first; + ASSERT_EQ(10u, found.size()); + wtx.SetSaplingNoteData(found); + ASSERT_TRUE(wallet.AddToWallet(wtx, true, nullptr)); + } + + auto snapshot = [&]() { + std::vector> out; + for (const auto& p : wallet.mapWallet) + for (const auto& item : p.second.mapSaplingNoteData) + if (item.second.cachedPlaintext) + out.push_back(std::make_pair(item.first, *item.second.cachedPlaintext)); + return out; + }; + auto eager = snapshot(); + ASSERT_EQ(300u, eager.size()); + + for (const char* threads : {"1", "4", "7"}) { + mapArgs["-witnessbuildthreads"] = threads; + wallet.ClearPlaintextCache(); + const uint64_t base = wallet.nSaplingNoteCacheFills; + wallet.WarmSaplingNoteCache(); + EXPECT_EQ(base + 300, wallet.nSaplingNoteCacheFills) << threads << " thread(s)"; + auto filled = snapshot(); + ASSERT_EQ(eager.size(), filled.size()) << threads << " thread(s)"; + for (size_t i = 0; i < eager.size(); i++) { + EXPECT_TRUE(eager[i].first == filled[i].first); + const SaplingNote& a = eager[i].second.note; + const SaplingNote& b = filled[i].second.note; + EXPECT_TRUE(a.d == b.d && a.pk_d == b.pk_d && a.r == b.r && a.value() == b.value()) << threads << " thread(s), note " << i; + EXPECT_TRUE(eager[i].second.memo == filled[i].second.memo); + } + } + mapArgs.erase("-witnessbuildthreads"); +} diff --git a/src/wallet/wallet.cpp b/src/wallet/wallet.cpp index b6e22e492..885387e9f 100644 --- a/src/wallet/wallet.cpp +++ b/src/wallet/wallet.cpp @@ -5730,16 +5730,59 @@ bool CMerkleTx::AcceptToMemoryPool(bool fLimitFree, bool fRejectAbsurdFee) unsigned int CWallet::FillSaplingNoteCache(const std::vector& notes) { AssertLockHeld(cs_wallet); - unsigned int nFilled = 0; - for (const SaplingNoteToDecrypt& n : notes) { - auto result = SaplingNotePlaintext::decrypt_with_note( - n.output->encCiphertext, n.nd->ivk, n.output->ephemeralKey, n.output->cm); - if (result) { - n.nd->cachedPlaintext = SaplingNoteCache{ result->second, result->first.memo() }; - nFilled++; - } + + // Each decryption is independent: it reads its own output and writes only its own note's + // cachedPlaintext. So a big batch (a whole wallet after a restart, ~0.5ms a note) is split across + // threads the way BuildWitnessCache's Phase 2 is, while this thread keeps cs_wallet held. Workers + // must not lock, or even assert a lock: under DEBUG_LOCKORDER a fresh thread has no lock stack. + // Spawning costs ~1ms, so small batches stay serial. + int nPar = 1; + if (notes.size() >= 128) { + nPar = (int)GetArg("-witnessbuildthreads", 0); + if (nPar <= 0) nPar = (int)std::thread::hardware_concurrency(); + nPar = std::max(1, std::min(nPar, (int)(notes.size() / 64))); } + + std::vector filled(nPar, 0); + auto worker = [&](int tid) { + for (size_t k = (size_t)tid; k < notes.size(); k += (size_t)nPar) { + const SaplingNoteToDecrypt& n = notes[k]; + try { + auto result = SaplingNotePlaintext::decrypt_with_note( + n.output->encCiphertext, n.nd->ivk, n.output->ephemeralKey, n.output->cm); + if (result) { + n.nd->cachedPlaintext = SaplingNoteCache{ result->second, result->first.memo() }; + filled[tid]++; + } + } catch (...) { + // Left unfilled, exactly like a note that doesn't decrypt; callers handle that. Nothing may + // escape a std::thread. + } + } + }; + + if (nPar <= 1) { + worker(0); + } else { + // If the OS refuses a thread, run the remaining shards here instead of letting the exception + // unwind past joinable threads (std::terminate). Each shard still runs exactly once. + std::vector threads; + threads.reserve(nPar); + int t = 0; + try { + for (; t < nPar; t++) threads.emplace_back(worker, t); + } catch (const std::exception& e) { + LogPrintf("%s: could only start %d of %d threads (%s); decrypting the rest inline\n", __func__, t, nPar, e.what()); + } + for (int u = t; u < nPar; u++) worker(u); + for (std::thread& th : threads) th.join(); + } + + unsigned int nFilled = 0; + for (unsigned int f : filled) + nFilled += f; nSaplingNoteCacheFills += nFilled; + LogPrint("bench", "%s: decrypted %u of %u notes using %d thread(s)\n", __func__, nFilled, (unsigned int)notes.size(), nPar); return nFilled; }