wallet: setup wallet threadpool

All wallets will use the same ThreadPool owned by the WalletContext.
ThreadPool::Submit() is threadsafe so there's no need for the use of
external synchronization primitives when submiting tasks.
All threads started by the ThreadPool will be destroyed with the
ThreadPool and WalletContext.

Author: Eunovo

src/dummywallet.cpp

@@ -47,6 +47,7 @@ void DummyWalletInit::AddWalletOptions(ArgsManager& argsman) const
         "-walletdir=<dir>",
         "-walletnotify=<cmd>",
         "-walletrbf",
+        "-walletpar",
         "-walletrejectlongchains",
         "-walletcrosschain",
         "-unsafesqlitesync",

src/wallet/context.h

@@ -6,6 +6,7 @@
 #define BITCOIN_WALLET_CONTEXT_H
 
 #include <sync.h>
+#include <util/threadpool.h>
 
 #include <functional>
 #include <list>
@@ -39,6 +40,7 @@ struct WalletContext {
     ArgsManager* args{nullptr}; // Currently a raw pointer because the memory is not managed by this struct
     // It is unsafe to lock this after locking a CWallet::cs_wallet mutex because
     // this could introduce inconsistent lock ordering and cause deadlocks.
+    ThreadPool thread_pool{"wallet"}; //!< Thread pool for wallet operations
     Mutex wallets_mutex;
     std::vector<std::shared_ptr<CWallet>> wallets GUARDED_BY(wallets_mutex);
     std::list<LoadWalletFn> wallet_load_fns GUARDED_BY(wallets_mutex);

src/wallet/dump.cpp

@@ -6,6 +6,7 @@
 
 #include <common/args.h>
 #include <util/fs.h>
+#include <util/threadpool.h>
 #include <util/translation.h>
 #include <wallet/wallet.h>
 #include <wallet/walletdb.h>

src/wallet/init.cpp

@@ -77,6 +77,7 @@ void WalletInit::AddWalletOptions(ArgsManager& argsman) const
     argsman.AddArg("-walletnotify=<cmd>", "Execute command when a wallet transaction changes. %s in cmd is replaced by TxID, %w is replaced by wallet name, %b is replaced by the hash of the block including the transaction (set to 'unconfirmed' if the transaction is not included) and %h is replaced by the block height (-1 if not included). %w is not currently implemented on windows. On systems where %w is supported, it should NOT be quoted because this would break shell escaping used to invoke the command.", ArgsManager::ALLOW_ANY, OptionsCategory::WALLET);
 #endif
     argsman.AddArg("-walletrbf", strprintf("Send transactions with full-RBF opt-in enabled (RPC only, default: %u)", DEFAULT_WALLET_RBF), ArgsManager::ALLOW_ANY, OptionsCategory::WALLET);
+    argsman.AddArg("-walletpar=<n>", strprintf("Number of threads to use for wallet operations (0 = auto, up to %d, <0 = leave that many cores free, default: %d)", MAX_WALLETPAR, DEFAULT_WALLETPAR), ArgsManager::ALLOW_ANY, OptionsCategory::WALLET);
 
     argsman.AddArg("-unsafesqlitesync", "Set SQLite synchronous=OFF to disable waiting for the database to sync to disk. This is unsafe and can cause data loss and corruption. This option is only used by tests to improve their performance (default: false)", ArgsManager::ALLOW_ANY | ArgsManager::DEBUG_ONLY, OptionsCategory::WALLET_DEBUG_TEST);
 

src/wallet/interfaces.cpp

@@ -5,6 +5,7 @@
 #include <interfaces/wallet.h>
 
 #include <common/args.h>
+#include <common/system.h>
 #include <consensus/amount.h>
 #include <interfaces/chain.h>
 #include <interfaces/handler.h>
@@ -29,6 +30,7 @@
 #include <wallet/spend.h>
 #include <wallet/wallet.h>
 
+#include <algorithm>
 #include <memory>
 #include <string>
 #include <utility>
@@ -544,6 +546,14 @@ class WalletLoaderImpl : public WalletLoader
     {
         m_context.chain = &chain;
         m_context.args = &args;
+        int num_threads = args.GetIntArg("-walletpar", DEFAULT_WALLETPAR);
+        if (num_threads <= 0) {
+            // '0' means 'autodetect'
+            // '<0' means leave that many cores free
+            num_threads += GetNumCores();
+        }
+        num_threads = std::clamp(num_threads, 1, MAX_WALLETPAR);
+        m_context.thread_pool.Start(num_threads);
     }
     ~WalletLoaderImpl() override { stop(); }
 

src/wallet/test/util.cpp

@@ -9,6 +9,7 @@
 #include <key_io.h>
 #include <streams.h>
 #include <test/util/setup_common.h>
+#include <util/threadpool.h>
 #include <validationinterface.h>
 #include <wallet/context.h>
 #include <wallet/wallet.h>
@@ -19,7 +20,9 @@
 namespace wallet {
 std::unique_ptr<CWallet> CreateSyncedWallet(interfaces::Chain& chain, CChain& cchain, const CKey& key)
 {
-    auto wallet = std::make_unique<CWallet>(&chain, "", CreateMockableWalletDatabase());
+    ThreadPool thread_pool{"wallet_test"};
+    thread_pool.Start(2);
+    auto wallet = std::make_unique<CWallet>(&chain, "", CreateMockableWalletDatabase(), &thread_pool);
     {
         LOCK2(wallet->cs_wallet, ::cs_main);
         wallet->SetLastBlockProcessed(cchain.Height(), cchain.Tip()->GetBlockHash());

src/wallet/wallet.cpp

@@ -3206,7 +3206,7 @@ std::shared_ptr<CWallet> CWallet::CreateNew(WalletContext& context, const std::s
     const auto start{SteadyClock::now()};
     // TODO: Can't use std::make_shared because we need a custom deleter but
     // should be possible to use std::allocate_shared.
-    std::shared_ptr<CWallet> walletInstance(new CWallet(chain, name, std::move(database)), FlushAndDeleteWallet);
+    std::shared_ptr<CWallet> walletInstance(new CWallet(chain, name, std::move(database), &context.thread_pool), FlushAndDeleteWallet);
 
     if (!LoadWalletArgs(walletInstance, context, error, warnings)) {
         return nullptr;
@@ -3258,7 +3258,7 @@ std::shared_ptr<CWallet> CWallet::LoadExisting(WalletContext& context, const std
     const std::string& walletFile = database->Filename();
 
     const auto start{SteadyClock::now()};
-    std::shared_ptr<CWallet> walletInstance(new CWallet(chain, name, std::move(database)), FlushAndDeleteWallet);
+    std::shared_ptr<CWallet> walletInstance(new CWallet(chain, name, std::move(database), &context.thread_pool), FlushAndDeleteWallet);
 
     if (!LoadWalletArgs(walletInstance, context, error, warnings)) {
         return nullptr;

src/wallet/wallet.h

@@ -27,6 +27,7 @@
 #include <util/hasher.h>
 #include <util/result.h>
 #include <util/string.h>
+#include <util/threadpool.h>
 #include <util/time.h>
 #include <util/ui_change_type.h>
 #include <wallet/crypter.h>
@@ -135,6 +136,9 @@ static const bool DEFAULT_WALLET_RBF = true;
 static const bool DEFAULT_WALLETBROADCAST = true;
 static const bool DEFAULT_DISABLE_WALLET = false;
 static const bool DEFAULT_WALLETCROSSCHAIN = false;
+//! Default for -walletpar
+static const int DEFAULT_WALLETPAR = 0;
+static const int MAX_WALLETPAR = 16;
 //! -maxtxfee default
 constexpr CAmount DEFAULT_TRANSACTION_MAXFEE{COIN / 10};
 //! Discourage users to set fees higher than this amount (in satoshis) per kB
@@ -403,6 +407,9 @@ class CWallet final : public WalletStorage, public interfaces::Chain::Notificati
     /** Internal database handle. */
     std::unique_ptr<WalletDatabase> m_database;
 
+    /** Thread pool for wallet operations. */
+    [[maybe_unused]] ThreadPool* m_thread_pool;
+
     /**
      * The following is used to keep track of how far behind the wallet is
      * from the chain sync, and to allow clients to block on us being caught up.
@@ -479,10 +486,11 @@ class CWallet final : public WalletStorage, public interfaces::Chain::Notificati
     unsigned int nMasterKeyMaxID = 0;
 
     /** Construct wallet with specified name and database implementation. */
-    CWallet(interfaces::Chain* chain, const std::string& name, std::unique_ptr<WalletDatabase> database)
+    CWallet(interfaces::Chain* chain, const std::string& name, std::unique_ptr<WalletDatabase> database, ThreadPool* thread_pool = nullptr)
         : m_chain(chain),
           m_name(name),
-          m_database(std::move(database))
+          m_database(std::move(database)),
+          m_thread_pool(thread_pool)
     {
     }