wallet: decrypt big note-cache batches on worker threads
The startup warm-up, or a lazy fill after -notecachewarmup=0, decrypts the whole wallet: ~0.5ms per note, ~6s serial for 12k notes. Each decryption is independent: it reads its own output and writes only its own note's cachedPlaintext. So FillSaplingNoteCache now splits batches of 128 or more notes round-robin across -witnessbuildthreads (default: hardware_concurrency) workers, at least 64 notes per thread. It follows BuildWitnessCache's Phase 2 pattern: - the caller keeps cs_wallet held throughout; - workers take no lock and assert none; under DEBUG_LOCKORDER a fresh thread has no lock stack; - per-shard fill counters are summed after join; - nothing escapes a worker: a throwing decryption leaves that note unfilled, like one that doesn't decrypt; - if the OS refuses a thread, the remaining shards run inline. librustzcash is already called concurrently by BuildWitnessCache's workers, and libsodium is thread-safe after sodium_init. ParallelFillMatchesSerial: 300 notes across two keys and diversified addresses decrypt to exactly the same cache with 1, 4 and 7 threads. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -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<OutputDescription> 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<std::pair<SaplingOutPoint, SaplingNoteCache>> 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");
|
||||
}
|
||||
|
||||
+46
-3
@@ -5730,16 +5730,59 @@ bool CMerkleTx::AcceptToMemoryPool(bool fLimitFree, bool fRejectAbsurdFee)
|
||||
unsigned int CWallet::FillSaplingNoteCache(const std::vector<SaplingNoteToDecrypt>& notes)
|
||||
{
|
||||
AssertLockHeld(cs_wallet);
|
||||
unsigned int nFilled = 0;
|
||||
for (const SaplingNoteToDecrypt& n : notes) {
|
||||
|
||||
// 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<unsigned int> 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() };
|
||||
nFilled++;
|
||||
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<std::thread> 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;
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user