fix: 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)
    #1 get /usr/bin/../lib/gcc/x86_64-linux-gnu/9/../../../../include/c++/9/bits/unique_ptr.h:361:21 (dashd+0xb58e08)
    #2 operator-> /usr/bin/../lib/gcc/x86_64-linux-gnu/9/../../../../include/c++/9/bits/unique_ptr.h:355:9 (dashd+0xb58e08)
    #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)
    #1 allocate /usr/bin/../lib/gcc/x86_64-linux-gnu/9/../../../../include/c++/9/ext/new_allocator.h:114:27 (dashd+0xb772b4) (BuildId: c3fdce9f7e778985a4fb0968ff4506d9ad24d408)
    #2 allocate /usr/bin/../lib/gcc/x86_64-linux-gnu/9/../../../../include/c++/9/bits/alloc_traits.h:443:20 (dashd+0xb772b4)
    #3 _M_get_node /usr/bin/../lib/gcc/x86_64-linux-gnu/9/../../../../include/c++/9/bits/stl_tree.h:580:16 (dashd+0xb772b4)
    [...]
    #8 CoinJoinWalletManager::Add(CWallet&) /src/dash/src/coinjoin/client.cpp:1898:26 (dashd+0xb58c73) (BuildId: c3fdce9f7e778985a4fb0968ff4506d9ad24d408)
[...]
SUMMARY: ThreadSanitizer: data race [...]
```

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](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,22 +1897,28 @@ 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));
+ } // cs_wallet_manager_map
  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);
+void CoinJoinWalletManager::Remove(const std::string& name)
+{
+ WITH_LOCK(cs_wallet_manager_map, m_wallet_manager_map.erase(name));
  g_wallet_init_interface.InitCoinJoinSettings(*this);
 }
 
@@ -1923,6 +1932,7 @@ void CoinJoinWalletManager::Flush(const std::string& name)
 
 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,23 @@ 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 +122,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 +187,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

@@ -89,9 +89,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,8 @@ 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](std::unique_ptr<CCoinJoinClientManager>& cj_man) { cj_man->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

@@ -209,8 +209,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);