coinjoin: protect m_wallet_manager_map with cs_wallet_manager_map

Avoid TSan-reported data race

```
WARNING: ThreadSanitizer: data race (pid=374820)
  Read of size 8 at 0x7b140002ce10 by thread T12:
    #0 _M_ptr /usr/bin/../lib/gcc/x86_64-linux-gnu/9/../../../../include/c++/9/bits/unique_ptr.h:154:42 (dashd+0xb58e08) (BuildId: c3fdce9f7e778985a4fb0968ff4506d9ad24d408)
    dashpay#1 get /usr/bin/../lib/gcc/x86_64-linux-gnu/9/../../../../include/c++/9/bits/unique_ptr.h:361:21 (dashd+0xb58e08)
    dashpay#2 operator-> /usr/bin/../lib/gcc/x86_64-linux-gnu/9/../../../../include/c++/9/bits/unique_ptr.h:355:9 (dashd+0xb58e08)
    dashpay#3 CoinJoinWalletManager::DoMaintenance() /src/dash/src/coinjoin/client.cpp:1907:9 (dashd+0xb58e08)
    [...]
  Previous write of size 8 at 0x7b140002ce10 by thread T17 (mutexes: write M0):
    #0 operator new(unsigned long) <null> (dashd+0x162657) (BuildId: c3fdce9f7e778985a4fb0968ff4506d9ad24d408)
    dashpay#1 allocate /usr/bin/../lib/gcc/x86_64-linux-gnu/9/../../../../include/c++/9/ext/new_allocator.h:114:27 (dashd+0xb772b4) (BuildId: c3fdce9f7e778985a4fb0968ff4506d9ad24d408)
    dashpay#2 allocate /usr/bin/../lib/gcc/x86_64-linux-gnu/9/../../../../include/c++/9/bits/alloc_traits.h:443:20 (dashd+0xb772b4)
    dashpay#3 _M_get_node /usr/bin/../lib/gcc/x86_64-linux-gnu/9/../../../../include/c++/9/bits/stl_tree.h:580:16 (dashd+0xb772b4)
    [...]
    dashpay#8 CoinJoinWalletManager::Add(CWallet&) /src/dash/src/coinjoin/client.cpp:1898:26 (dashd+0xb58c73) (BuildId: c3fdce9f7e778985a4fb0968ff4506d9ad24d408)
[...]
SUMMARY: ThreadSanitizer: data race [...]
```

Author: UdjinM6

src/coinjoin/client.cpp

@@ -96,11 +96,9 @@ PeerMsgRet CCoinJoinClientQueueManager::ProcessDSQueue(const CNode& peer, CDataS
         }
 
         // if the queue is ready, submit if we can
-        if (dsq.fReady && ranges::any_of(m_walletman.raw(),
-                                         [this, &dmn](const auto &pair) {
-                                             return pair.second->TrySubmitDenominate(dmn->pdmnState->addr,
-                                                                                     this->connman);
-                                         })) {
+        if (dsq.fReady && m_walletman.ForAnyCJClientMan([this, &dmn](std::unique_ptr<CCoinJoinClientManager>& clientman) {
+                return clientman->TrySubmitDenominate(dmn->pdmnState->addr, this->connman);
+            })) {
             LogPrint(BCLog::COINJOIN, "DSQUEUE -- CoinJoin queue (%s) is ready on masternode %s\n", dsq.ToString(),
                      dmn->pdmnState->addr.ToString());
             return {};
@@ -121,8 +119,9 @@ PeerMsgRet CCoinJoinClientQueueManager::ProcessDSQueue(const CNode& peer, CDataS
             LogPrint(BCLog::COINJOIN, "DSQUEUE -- new CoinJoin queue (%s) from masternode %s\n", dsq.ToString(),
                      dmn->pdmnState->addr.ToString());
 
-            ranges::any_of(m_walletman.raw(),
-                           [&dsq](const auto &pair) { return pair.second->MarkAlreadyJoinedQueueAsTried(dsq); });
+            m_walletman.ForAnyCJClientMan([&dsq](const std::unique_ptr<CCoinJoinClientManager>& clientman) {
+                return clientman->MarkAlreadyJoinedQueueAsTried(dsq);
+            });
 
             WITH_LOCK(cs_vecqueue, vecCoinJoinQueue.push_back(dsq));
         }
@@ -155,11 +154,14 @@ void CCoinJoinClientManager::ProcessMessage(CNode& peer, CChainState& active_cha
     }
 }
 
-CCoinJoinClientSession::CCoinJoinClientSession(CWallet& wallet, CoinJoinWalletManager& walletman, CDeterministicMNManager& dmnman, CMasternodeMetaMan& mn_metaman,
-                                               const CMasternodeSync& mn_sync, const std::unique_ptr<CCoinJoinClientQueueManager>& queueman, bool is_masternode) :
+CCoinJoinClientSession::CCoinJoinClientSession(CWallet& wallet, CoinJoinWalletManager& walletman,
+                                               CCoinJoinClientManager& clientman, CDeterministicMNManager& dmnman,
+                                               CMasternodeMetaMan& mn_metaman, const CMasternodeSync& mn_sync,
+                                               const std::unique_ptr<CCoinJoinClientQueueManager>& queueman,
+                                               bool is_masternode) :
     m_wallet(wallet),
     m_walletman(walletman),
-    m_manager(*Assert(walletman.Get(wallet.GetName()))),
+    m_clientman(clientman),
     m_dmnman(dmnman),
     m_mn_metaman(mn_metaman),
     m_mn_sync(mn_sync),
@@ -684,7 +686,7 @@ void CCoinJoinClientSession::CompletedTransaction(PoolMessage nMessageID)
     if (m_is_masternode) return;
 
     if (nMessageID == MSG_SUCCESS) {
-        m_manager.UpdatedSuccessBlock();
+        m_clientman.UpdatedSuccessBlock();
         keyHolderStorage.KeepAll();
         WalletCJLogPrint(m_wallet, "CompletedTransaction -- success\n");
     } else {
@@ -995,7 +997,8 @@ bool CCoinJoinClientManager::DoAutomaticDenominating(CChainState& active_chainst
     AssertLockNotHeld(cs_deqsessions);
     LOCK(cs_deqsessions);
     if (int(deqSessions.size()) < CCoinJoinClientOptions::GetSessions()) {
-        deqSessions.emplace_back(m_wallet, m_walletman, m_dmnman, m_mn_metaman, m_mn_sync, m_queueman, m_is_masternode);
+        deqSessions.emplace_back(m_wallet, m_walletman, *this, m_dmnman, m_mn_metaman, m_mn_sync, m_queueman,
+                                 m_is_masternode);
     }
     for (auto& session : deqSessions) {
         if (!CheckAutomaticBackup()) return false;
@@ -1100,7 +1103,7 @@ bool CCoinJoinClientSession::JoinExistingQueue(CAmount nBalanceNeedsAnonymized,
             continue;
         }
 
-        m_manager.AddUsedMasternode(dsq.masternodeOutpoint);
+        m_clientman.AddUsedMasternode(dsq.masternodeOutpoint);
 
         if (connman.IsMasternodeOrDisconnectRequested(dmn->pdmnState->addr)) {
             WalletCJLogPrint(m_wallet, "CCoinJoinClientSession::JoinExistingQueue -- skipping masternode connection, addr=%s\n", dmn->pdmnState->addr.ToString());
@@ -1145,14 +1148,14 @@ bool CCoinJoinClientSession::StartNewQueue(CAmount nBalanceNeedsAnonymized, CCon
 
     // otherwise, try one randomly
     while (nTries < 10) {
-        auto dmn = m_manager.GetRandomNotUsedMasternode();
+        auto dmn = m_clientman.GetRandomNotUsedMasternode();
         if (!dmn) {
             strAutoDenomResult = _("Can't find random Masternode.");
             WalletCJLogPrint(m_wallet, "CCoinJoinClientSession::StartNewQueue -- %s\n", strAutoDenomResult.original);
             return false;
         }
 
-        m_manager.AddUsedMasternode(dmn->collateralOutpoint);
+        m_clientman.AddUsedMasternode(dmn->collateralOutpoint);
 
         // skip next mn payments winners
         if (dmn->pdmnState->nLastPaidHeight + nWeightedMnCount < mnList.GetHeight() + WinnersToSkip()) {
@@ -1526,7 +1529,7 @@ bool CCoinJoinClientSession::MakeCollateralAmounts(const CompactTallyItem& tally
         return false;
     }
 
-    m_manager.UpdatedSuccessBlock();
+    m_clientman.UpdatedSuccessBlock();
 
     WalletCJLogPrint(m_wallet, "CCoinJoinClientSession::%s -- txid: %s\n", __func__, strResult.original);
 
@@ -1803,7 +1806,7 @@ bool CCoinJoinClientSession::CreateDenominated(CAmount nBalanceToDenominate, con
     }
 
     // use the same nCachedLastSuccessBlock as for DS mixing to prevent race
-    m_manager.UpdatedSuccessBlock();
+    m_clientman.UpdatedSuccessBlock();
 
     WalletCJLogPrint(m_wallet, "CCoinJoinClientSession::%s -- txid: %s\n", __func__, strResult.original);
 
@@ -1894,35 +1897,43 @@ void CCoinJoinClientManager::GetJsonInfo(UniValue& obj) const
     obj.pushKV("sessions",  arrSessions);
 }
 
-void CoinJoinWalletManager::Add(CWallet& wallet) {
-    m_wallet_manager_map.try_emplace(
-        wallet.GetName(),
-        std::make_unique<CCoinJoinClientManager>(wallet, *this, m_dmnman, m_mn_metaman, m_mn_sync, m_queueman, m_is_masternode)
-    );
+void CoinJoinWalletManager::Add(CWallet& wallet)
+{
+    {
+        LOCK(cs_wallet_manager_map);
+        m_wallet_manager_map.try_emplace(wallet.GetName(),
+                                         std::make_unique<CCoinJoinClientManager>(wallet, *this, m_dmnman, m_mn_metaman,
+                                                                                  m_mn_sync, m_queueman, m_is_masternode));
+    }
     g_wallet_init_interface.InitCoinJoinSettings(*this);
 }
 
-void CoinJoinWalletManager::DoMaintenance() {
-    for (auto& [wallet_str, walletman] : m_wallet_manager_map) {
-        walletman->DoMaintenance(m_chainstate, m_connman, m_mempool);
+void CoinJoinWalletManager::DoMaintenance()
+{
+    LOCK(cs_wallet_manager_map);
+    for (auto& [_, clientman] : m_wallet_manager_map) {
+        clientman->DoMaintenance(m_chainstate, m_connman, m_mempool);
     }
 }
 
 void CoinJoinWalletManager::Remove(const std::string& name) {
-    m_wallet_manager_map.erase(name);
+    {
+        LOCK(cs_wallet_manager_map);
+        m_wallet_manager_map.erase(name);
+    }
     g_wallet_init_interface.InitCoinJoinSettings(*this);
 }
 
 void CoinJoinWalletManager::Flush(const std::string& name)
 {
-    auto clientman = Get(name);
-    assert(clientman != nullptr);
+    auto clientman = Assert(Get(name));
     clientman->ResetPool();
     clientman->StopMixing();
 }
 
 CCoinJoinClientManager* CoinJoinWalletManager::Get(const std::string& name) const
 {
+    LOCK(cs_wallet_manager_map);
     auto it = m_wallet_manager_map.find(name);
     return (it != m_wallet_manager_map.end()) ? it->second.get() : nullptr;
 }

src/coinjoin/client.h

@@ -80,6 +80,7 @@ class CoinJoinWalletManager {
     {}
 
     ~CoinJoinWalletManager() {
+        LOCK(cs_wallet_manager_map);
         for (auto& [wallet_name, cj_man] : m_wallet_manager_map) {
             cj_man.reset();
         }
@@ -93,7 +94,24 @@ class CoinJoinWalletManager {
 
     CCoinJoinClientManager* Get(const std::string& name) const;
 
-    const wallet_name_cjman_map& raw() const { return m_wallet_manager_map; }
+    template <typename Callable>
+    void ForEachCJClientMan(Callable&& func)
+    {
+        LOCK(cs_wallet_manager_map);
+        for (auto&& [_, clientman] : m_wallet_manager_map) {
+            func(clientman);
+        }
+    };
+
+    template <typename Callable>
+    bool ForAnyCJClientMan(Callable&& func)
+    {
+        LOCK(cs_wallet_manager_map);
+        for (auto&& [_, clientman] : m_wallet_manager_map) {
+            if (func(clientman)) return true;
+        }
+        return false;
+    };
 
 private:
     CChainState& m_chainstate;
@@ -105,15 +123,17 @@ class CoinJoinWalletManager {
     const std::unique_ptr<CCoinJoinClientQueueManager>& m_queueman;
 
     const bool m_is_masternode;
-    wallet_name_cjman_map m_wallet_manager_map;
+
+    mutable Mutex cs_wallet_manager_map;
+    wallet_name_cjman_map m_wallet_manager_map GUARDED_BY(cs_wallet_manager_map);
 };
 
 class CCoinJoinClientSession : public CCoinJoinBaseSession
 {
 private:
     CWallet& m_wallet;
     CoinJoinWalletManager& m_walletman;
-    CCoinJoinClientManager& m_manager;
+    CCoinJoinClientManager& m_clientman;
     CDeterministicMNManager& m_dmnman;
     CMasternodeMetaMan& m_mn_metaman;
     const CMasternodeSync& m_mn_sync;
@@ -168,8 +188,10 @@ class CCoinJoinClientSession : public CCoinJoinBaseSession
     void SetNull() override EXCLUSIVE_LOCKS_REQUIRED(cs_coinjoin);
 
 public:
-    explicit CCoinJoinClientSession(CWallet& wallet, CoinJoinWalletManager& walletman, CDeterministicMNManager& dmnman, CMasternodeMetaMan& mn_metaman,
-                                    const CMasternodeSync& mn_sync, const std::unique_ptr<CCoinJoinClientQueueManager>& queueman, bool is_masternode);
+    explicit CCoinJoinClientSession(CWallet& wallet, CoinJoinWalletManager& walletman,
+                                    CCoinJoinClientManager& clientman, CDeterministicMNManager& dmnman,
+                                    CMasternodeMetaMan& mn_metaman, const CMasternodeSync& mn_sync,
+                                    const std::unique_ptr<CCoinJoinClientQueueManager>& queueman, bool is_masternode);
 
     void ProcessMessage(CNode& peer, CChainState& active_chainstate, CConnman& connman, const CTxMemPool& mempool, std::string_view msg_type, CDataStream& vRecv);
 

src/dsnotificationinterface.cpp

@@ -86,9 +86,8 @@ void CDSNotificationInterface::UpdatedBlockTip(const CBlockIndex *pindexNew, con
 
     m_cj_ctx->dstxman->UpdatedBlockTip(pindexNew, *m_llmq_ctx->clhandler, m_mn_sync);
 #ifdef ENABLE_WALLET
-    for (auto& pair : m_cj_ctx->walletman->raw()) {
-        pair.second->UpdatedBlockTip(pindexNew);
-    }
+    m_cj_ctx->walletman->ForEachCJClientMan(
+        [&pindexNew](std::unique_ptr<CCoinJoinClientManager>& clientman) { clientman->UpdatedBlockTip(pindexNew); });
 #endif // ENABLE_WALLET
 
     m_llmq_ctx->isman->UpdatedBlockTip(pindexNew);

src/masternode/utils.cpp

@@ -23,9 +23,9 @@ void CMasternodeUtils::DoMaintenance(CConnman& connman, CDeterministicMNManager&
 
     std::vector<CDeterministicMNCPtr> vecDmns; // will be empty when no wallet
 #ifdef ENABLE_WALLET
-    for (auto& pair : cj_ctx.walletman->raw()) {
-        pair.second->GetMixingMasternodesInfo(vecDmns);
-    }
+    cj_ctx.walletman->ForEachCJClientMan([&vecDmns](const std::unique_ptr<CCoinJoinClientManager>& clientman) {
+        clientman->GetMixingMasternodesInfo(vecDmns);
+    });
 #endif // ENABLE_WALLET
 
     // Don't disconnect masternode connections when we have less then the desired amount of outbound nodes

src/net_processing.cpp

@@ -5095,9 +5095,9 @@ void PeerManagerImpl::ProcessMessage(
         //probably one the extensions
 #ifdef ENABLE_WALLET
         ProcessPeerMsgRet(m_cj_ctx->queueman->ProcessMessage(pfrom, msg_type, vRecv), pfrom);
-        for (auto& pair : m_cj_ctx->walletman->raw()) {
-            pair.second->ProcessMessage(pfrom, m_chainman.ActiveChainstate(), m_connman, m_mempool, msg_type, vRecv);
-        }
+        m_cj_ctx->walletman->ForEachCJClientMan([this, &pfrom, &msg_type, &vRecv](std::unique_ptr<CCoinJoinClientManager>& clientman) {
+            clientman->ProcessMessage(pfrom, m_chainman.ActiveChainstate(), m_connman, m_mempool, msg_type, vRecv);
+        });
 #endif // ENABLE_WALLET
         ProcessPeerMsgRet(m_cj_ctx->server->ProcessMessage(pfrom, msg_type, vRecv), pfrom);
         ProcessPeerMsgRet(m_sporkman.ProcessMessage(pfrom, m_connman, *this, msg_type, vRecv), pfrom);

src/wallet/test/coinjoin_tests.cpp

@@ -208,8 +208,8 @@ class CTransactionBuilderTestSetup : public TestChain100Setup
 
 BOOST_FIXTURE_TEST_CASE(coinjoin_manager_start_stop_tests, CTransactionBuilderTestSetup)
 {
-    BOOST_CHECK_EQUAL(m_node.cj_ctx->walletman->raw().size(), 1);
-    auto& cj_man = m_node.cj_ctx->walletman->raw().begin()->second;
+    CCoinJoinClientManager* cj_man = m_node.cj_ctx->walletman->Get("");
+    BOOST_ASSERT(cj_man != nullptr);
     BOOST_CHECK_EQUAL(cj_man->IsMixing(), false);
     BOOST_CHECK_EQUAL(cj_man->StartMixing(), true);
     BOOST_CHECK_EQUAL(cj_man->IsMixing(), true);