Skip to content

fix: protect CoinJoinWalletManager::m_wallet_manager_map with Mutex, avoid data race, partial bitcoin#22868 #6192

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
PastaPastaPasta merged 2 commits into from
Aug 29, 2024
Merged

fix: protect CoinJoinWalletManager::m_wallet_manager_map with Mutex, avoid data race, partial bitcoin#22868 #6192

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
c602ca1
coinjoin: protect `m_wallet_manager_map` with `cs_wallet_manager_map`
kwvg Aug 26, 2024
87d775d
partial bitcoin#22868: Call load handlers without cs_wallet locked
kwvg Aug 11, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
67 changes: 39 additions & 28 deletions src/coinjoin/client.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {};
Expand All @@ -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));
}
Expand Down Expand Up @@ -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),
Expand Down Expand Up @@ -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 {
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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());
Expand Down Expand Up @@ -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()) {
Expand Down Expand Up @@ -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);

Expand Down Expand Up @@ -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);

Expand Down Expand Up @@ -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;
}
30 changes: 25 additions & 5 deletions src/coinjoin/client.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@
#include <coinjoin/coinjoin.h>

#include <net_types.h>
#include <util/ranges.h>
#include <util/translation.h>

#include <atomic>
Expand Down Expand Up @@ -80,6 +81,7 @@ class CoinJoinWalletManager {
{}

~CoinJoinWalletManager() {
LOCK(cs_wallet_manager_map);
for (auto& [wallet_name, cj_man] : m_wallet_manager_map) {
cj_man.reset();
}
Expand All @@ -93,7 +95,21 @@ 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);
}
Comment on lines +102 to +104
Copy link
Member

@PastaPastaPasta PastaPastaPasta Aug 20, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

should we use std::for_each here?

Copy link
Collaborator Author

@kwvg kwvg Aug 23, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We would not be able to do the below snippet because the function isn't taking in a pair but one only one half of it

std::for_each(m_wallet_manager_map.begin(), m_wallet_manager_map.end(), func);

Copy link
Member

@PastaPastaPasta PastaPastaPasta Aug 25, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

could do something ```suggestion
ranges::for_each(m_wallet_manager_map, [&](auto [_, clientman]){func(clientman)});

Copy link
Collaborator Author

@kwvg kwvg Aug 25, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@UdjinM6, thoughts?

Copy link

@UdjinM6 UdjinM6 Aug 25, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

can't use structured bindings there, would be more like 019a780 instead

Copy link
Collaborator

@knst knst Aug 26, 2024 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't like to use std::for_each if it's not super trivial.

ranges::for_each(m_wallet_manager_map, [&](auto [_, clientman]){func(clientman)});

it looks like extra more difficult to read, I disagree with this suggestment.

For-each has been a great option when there was no range-loop in C++! But in this particular case it looks difficult to read, difficult to understand. That's a case when 2 lines better than one!

using any_of below makes sense, but not here.

};

template <typename Callable>
bool ForAnyCJClientMan(Callable&& func)
{
LOCK(cs_wallet_manager_map);
return ranges::any_of(m_wallet_manager_map, [&](auto& pair) { return func(pair.second); });
};

private:
CChainState& m_chainstate;
Expand All @@ -105,15 +121,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;
Expand Down Expand Up @@ -168,8 +186,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);

Expand Down
5 changes: 2 additions & 3 deletions src/dsnotificationinterface.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down
6 changes: 3 additions & 3 deletions src/masternode/utils.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
6 changes: 3 additions & 3 deletions src/net_processing.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down
4 changes: 2 additions & 2 deletions src/wallet/test/coinjoin_tests.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down
6 changes: 1 addition & 5 deletions src/wallet/test/wallet_tests.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1298,18 +1298,14 @@ BOOST_FIXTURE_TEST_CASE(CreateWallet, TestChain100Setup)
// deadlock during the sync and simulates a new block notification happening
// as soon as possible.
addtx_count = 0;
auto handler = HandleLoadWallet([&](std::unique_ptr<interfaces::Wallet> wallet) EXCLUSIVE_LOCKS_REQUIRED(wallet->wallet()->cs_wallet, cs_wallets) {
auto handler = HandleLoadWallet([&](std::unique_ptr<interfaces::Wallet> wallet) {
BOOST_CHECK(rescan_completed);
m_coinbase_txns.push_back(CreateAndProcessBlock({}, GetScriptForRawPubKey(coinbaseKey.GetPubKey())).vtx[0]);
block_tx = TestSimpleSpend(*m_coinbase_txns[2], 0, coinbaseKey, GetScriptForRawPubKey(key.GetPubKey()));
m_coinbase_txns.push_back(CreateAndProcessBlock({block_tx}, GetScriptForRawPubKey(coinbaseKey.GetPubKey())).vtx[0]);
mempool_tx = TestSimpleSpend(*m_coinbase_txns[3], 0, coinbaseKey, GetScriptForRawPubKey(key.GetPubKey()));
BOOST_CHECK(m_node.chain->broadcastTransaction(MakeTransactionRef(mempool_tx), DEFAULT_TRANSACTION_MAXFEE, false, error));
LEAVE_CRITICAL_SECTION(cs_wallets);
LEAVE_CRITICAL_SECTION(wallet->wallet()->cs_wallet);
SyncWithValidationInterfaceQueue();
ENTER_CRITICAL_SECTION(wallet->wallet()->cs_wallet);
ENTER_CRITICAL_SECTION(cs_wallets);
});
wallet = TestLoadWallet(m_node.chain.get(), m_node.coinjoin_loader.get());
BOOST_CHECK_EQUAL(addtx_count, 4);
Expand Down
5 changes: 2 additions & 3 deletions src/wallet/wallet.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4949,8 +4949,6 @@ std::shared_ptr<CWallet> CWallet::Create(interfaces::Chain* chain, interfaces::C
// Try to top up keypool. No-op if the wallet is locked.
walletInstance->TopUpKeyPool();

LOCK(walletInstance->cs_wallet);

if (chain && !AttachChain(walletInstance, *chain, error, warnings)) {
return nullptr;
}
Expand All @@ -4966,9 +4964,10 @@ std::shared_ptr<CWallet> CWallet::Create(interfaces::Chain* chain, interfaces::C
}
}

walletInstance->SetBroadcastTransactions(gArgs.GetBoolArg("-walletbroadcast", DEFAULT_WALLETBROADCAST));

{
LOCK(walletInstance->cs_wallet);
walletInstance->SetBroadcastTransactions(gArgs.GetBoolArg("-walletbroadcast", DEFAULT_WALLETBROADCAST));
walletInstance->WalletLogPrintf("setExternalKeyPool.size() = %u\n", walletInstance->KeypoolCountExternalKeys());
walletInstance->WalletLogPrintf("GetKeyPoolSize() = %u\n", walletInstance->GetKeyPoolSize());
walletInstance->WalletLogPrintf("mapWallet.size() = %u\n", walletInstance->mapWallet.size());
Expand Down