p2p: Replace RecursiveMutex `m_tx_inventory_mutex` with Mutex and rename it #24125

w0xlt · 2022-01-22T01:28:10Z

This PR is related to #19303 and gets rid of the RecursiveMutex m_tx_inventory_mutex and also adds AssertLockNotHeld macros combined with LOCKS_EXCLUDED thread safety annotations to avoid recursive locking.

DrahtBot · 2022-01-22T08:17:34Z

The following sections might be updated with supplementary metadata relevant to reviewers and maintainers.

Reviews

See the guideline for information on the review process.

Type	Reviewers
Concept ACK	hebasto
Approach ACK	shaavan

If your review is incorrectly listed, please react with 👎 to this comment and the bot will ignore it on the next update.

Conflicts

Reviewers, this pull request conflicts with the following ones:

#26593 (tracing: Only prepare tracepoint arguments when actually tracing by 0xB10C)
#26569 (p2p: Ensure transaction announcements are only queued for fully connected peers by dergoegge)
#26283 (p2p: Fill reconciliation sets and request reconciliation (Erlay) by naumenkogs)
#26140 (refactor: Move CNodeState members guarded by g_msgproc_mutex to Peer by dergoegge)

If you consider this pull request important, please also help to review the conflicting pull requests. Ideally, start with the one that should be merged first.

shaavan

Approach ACK d097e5d

I agree with the name change of cs_tx_inventory -> m_tx_inventory_mutex.
Checked that all the locking instances of m_tx_inventory_mutex are protected by a preceding AssertLockNotHeld() statement.
LOCKS_EXCLUDED macro is appropriately used with all the function definitions where AssertLockNotHeld() is used.

There is a minor nit that you should address before merging this PR. 🥂

src/net_processing.cpp

hebasto · 2022-01-24T17:12:55Z

Concept ACK.

But reviewing changes in PeerManagerImpl::SendMessages() is not trivial because of a huge critical section size. Also there are concerns about external synchronizing (similar to #24122 (review)).

w0xlt · 2022-01-25T13:49:20Z

To address the external synchronizing, the code snippets below should be encapsulated as CNode::TxRelay methods?

bitcoin/src/net_processing.cpp

Lines 2038 to 2041 in bd482b3

    
           if (WITH_LOCK(pfrom.m_tx_relay->cs_tx_inventory, return !pfrom.m_tx_relay->filterInventoryKnown.contains(parent_txid))) { 
        
               LOCK(cs_main); 
        
               State(pfrom.GetId())->m_recently_announced_invs.insert(parent_txid); 
        
           }

bitcoin/src/net_processing.cpp

Lines 1904 to 1910 in bd482b3

    
           if (pfrom.m_tx_relay != nullptr) { 
        
               LOCK(pfrom.m_tx_relay->cs_filter); 
        
               if (pfrom.m_tx_relay->pfilter) { 
        
                   sendMerkleBlock = true; 
        
                   merkleBlock = CMerkleBlock(*pblock, *pfrom.m_tx_relay->pfilter); 
        
               } 
        
           }

bitcoin/src/net_processing.cpp

Lines 4807 to 4935 in bd482b3

    
           if (pto->m_tx_relay != nullptr) { 
        
                   LOCK(pto->m_tx_relay->cs_tx_inventory); 
        
                   // Check whether periodic sends should happen 
        
                   bool fSendTrickle = pto->HasPermission(NetPermissionFlags::NoBan); 
        
                   if (pto->m_tx_relay->nNextInvSend < current_time) { 
        
                       fSendTrickle = true; 
        
                       if (pto->IsInboundConn()) { 
        
                           pto->m_tx_relay->nNextInvSend = NextInvToInbounds(current_time, INBOUND_INVENTORY_BROADCAST_INTERVAL); 
        
                       } else { 
        
                           pto->m_tx_relay->nNextInvSend = GetExponentialRand(current_time, OUTBOUND_INVENTORY_BROADCAST_INTERVAL); 
        
                       } 
        
                   } 
        
                   // Time to send but the peer has requested we not relay transactions. 
        
                   if (fSendTrickle) { 
        
                       LOCK(pto->m_tx_relay->cs_filter); 
        
                       if (!pto->m_tx_relay->fRelayTxes) pto->m_tx_relay->setInventoryTxToSend.clear(); 
        
                   } 
        
                   // Respond to BIP35 mempool requests 
        
                   if (fSendTrickle && pto->m_tx_relay->fSendMempool) { 
        
                       auto vtxinfo = m_mempool.infoAll(); 
        
                       pto->m_tx_relay->fSendMempool = false; 
        
                       const CFeeRate filterrate{pto->m_tx_relay->minFeeFilter.load()}; 
        
                       LOCK(pto->m_tx_relay->cs_filter); 
        
                       for (const auto& txinfo : vtxinfo) { 
        
                           const uint256& hash = state.m_wtxid_relay ? txinfo.tx->GetWitnessHash() : txinfo.tx->GetHash(); 
        
                           CInv inv(state.m_wtxid_relay ? MSG_WTX : MSG_TX, hash); 
        
                           pto->m_tx_relay->setInventoryTxToSend.erase(hash); 
        
                           // Don't send transactions that peers will not put into their mempool 
        
                           if (txinfo.fee < filterrate.GetFee(txinfo.vsize)) { 
        
                               continue; 
        
                           } 
        
                           if (pto->m_tx_relay->pfilter) { 
        
                               if (!pto->m_tx_relay->pfilter->IsRelevantAndUpdate(*txinfo.tx)) continue; 
        
                           } 
        
                           pto->m_tx_relay->filterInventoryKnown.insert(hash); 
        
                           // Responses to MEMPOOL requests bypass the m_recently_announced_invs filter. 
        
                           vInv.push_back(inv); 
        
                           if (vInv.size() == MAX_INV_SZ) { 
        
                               m_connman.PushMessage(pto, msgMaker.Make(NetMsgType::INV, vInv)); 
        
                               vInv.clear(); 
        
                           } 
        
                       } 
        
                       pto->m_tx_relay->m_last_mempool_req = std::chrono::duration_cast<std::chrono::seconds>(current_time); 
        
                   } 
        
                   // Determine transactions to relay 
        
                   if (fSendTrickle) { 
        
                       // Produce a vector with all candidates for sending 
        
                       std::vector<std::set<uint256>::iterator> vInvTx; 
        
                       vInvTx.reserve(pto->m_tx_relay->setInventoryTxToSend.size()); 
        
                       for (std::set<uint256>::iterator it = pto->m_tx_relay->setInventoryTxToSend.begin(); it != pto->m_tx_relay->setInventoryTxToSend.end(); it++) { 
        
                           vInvTx.push_back(it); 
        
                       } 
        
                       const CFeeRate filterrate{pto->m_tx_relay->minFeeFilter.load()}; 
        
                       // Topologically and fee-rate sort the inventory we send for privacy and priority reasons. 
        
                       // A heap is used so that not all items need sorting if only a few are being sent. 
        
                       CompareInvMempoolOrder compareInvMempoolOrder(&m_mempool, state.m_wtxid_relay); 
        
                       std::make_heap(vInvTx.begin(), vInvTx.end(), compareInvMempoolOrder); 
        
                       // No reason to drain out at many times the network's capacity, 
        
                       // especially since we have many peers and some will draw much shorter delays. 
        
                       unsigned int nRelayedTransactions = 0; 
        
                       LOCK(pto->m_tx_relay->cs_filter); 
        
                       while (!vInvTx.empty() && nRelayedTransactions < INVENTORY_BROADCAST_MAX) { 
        
                           // Fetch the top element from the heap 
        
                           std::pop_heap(vInvTx.begin(), vInvTx.end(), compareInvMempoolOrder); 
        
                           std::set<uint256>::iterator it = vInvTx.back(); 
        
                           vInvTx.pop_back(); 
        
                           uint256 hash = *it; 
        
                           CInv inv(state.m_wtxid_relay ? MSG_WTX : MSG_TX, hash); 
        
                           // Remove it from the to-be-sent set 
        
                           pto->m_tx_relay->setInventoryTxToSend.erase(it); 
        
                           // Check if not in the filter already 
        
                           if (pto->m_tx_relay->filterInventoryKnown.contains(hash)) { 
        
                               continue; 
        
                           } 
        
                           // Not in the mempool anymore? don't bother sending it. 
        
                           auto txinfo = m_mempool.info(ToGenTxid(inv)); 
        
                           if (!txinfo.tx) { 
        
                               continue; 
        
                           } 
        
                           auto txid = txinfo.tx->GetHash(); 
        
                           auto wtxid = txinfo.tx->GetWitnessHash(); 
        
                           // Peer told you to not send transactions at that feerate? Don't bother sending it. 
        
                           if (txinfo.fee < filterrate.GetFee(txinfo.vsize)) { 
        
                               continue; 
        
                           } 
        
                           if (pto->m_tx_relay->pfilter && !pto->m_tx_relay->pfilter->IsRelevantAndUpdate(*txinfo.tx)) continue; 
        
                           // Send 
        
                           State(pto->GetId())->m_recently_announced_invs.insert(hash); 
        
                           vInv.push_back(inv); 
        
                           nRelayedTransactions++; 
        
                           { 
        
                               // Expire old relay messages 
        
                               while (!g_relay_expiration.empty() && g_relay_expiration.front().first < current_time) 
        
                               { 
        
                                   mapRelay.erase(g_relay_expiration.front().second); 
        
                                   g_relay_expiration.pop_front(); 
        
                               } 
        
                               auto ret = mapRelay.emplace(txid, std::move(txinfo.tx)); 
        
                               if (ret.second) { 
        
                                   g_relay_expiration.emplace_back(current_time + RELAY_TX_CACHE_TIME, ret.first); 
        
                               } 
        
                               // Add wtxid-based lookup into mapRelay as well, so that peers can request by wtxid 
        
                               auto ret2 = mapRelay.emplace(wtxid, ret.first->second); 
        
                               if (ret2.second) { 
        
                                   g_relay_expiration.emplace_back(current_time + RELAY_TX_CACHE_TIME, ret2.first); 
        
                               } 
        
                           } 
        
                           if (vInv.size() == MAX_INV_SZ) { 
        
                               m_connman.PushMessage(pto, msgMaker.Make(NetMsgType::INV, vInv)); 
        
                               vInv.clear(); 
        
                           } 
        
                           pto->m_tx_relay->filterInventoryKnown.insert(hash); 
        
                           if (hash != txid) { 
        
                               // Insert txid into filterInventoryKnown, even for 
        
                               // wtxidrelay peers. This prevents re-adding of 
        
                               // unconfirmed parents to the recently_announced 
        
                               // filter, when a child tx is requested. See 
        
                               // ProcessGetData(). 
        
                               pto->m_tx_relay->filterInventoryKnown.insert(txid); 
        
                           } 
        
                       } 
        
                   } 
        
           }

hebasto · 2022-01-25T13:53:53Z

To address the external synchronizing, the code snippets below should be encapsulated as CNode::TxRelay methods?

That is my guess. Of course, it would be nice to hear other devs opinion.

w0xlt · 2022-02-12T02:32:45Z

The new commit addresses most of #24125 (comment).

All code snippets protected by cs_tx_inventory\m_tx_inventory_mutex (except for the one in PeerManagerImpl::SendMessages(CNode* pto)) were encapsulated in struct CNode::TxRelay.

Annotations that use negative capabilities were also added.

It doesn't seem trivial to bring the code in PeerManagerImpl::SendMessages(CNode* pto) into net.cpp, as it uses a significant amount of net_processing objects and references. I will try to find a way to do this.

hebasto · 2022-05-16T20:28:20Z

@w0xlt

Still working on this?

w0xlt · 2022-06-24T13:33:46Z

It doesn't seem trivial to bring the code in PeerManagerImpl::SendMessages(CNode* pto) into net.cpp, as it uses a significant amount of net_processing objects and references. I will try to find a way to do this.

@hebasto I hadn't seen the message before. I will try to solve the issue described in the comment above.

w0xlt · 2022-06-24T20:10:42Z

It doesn't seem trivial to bring the code in PeerManagerImpl::SendMessages(CNode* pto) into net.cpp, as it uses a significant amount of net_processing objects and references. I will try to find a way to do this.

@hebasto 88777c4 does the trick, apparently with no change in behavior..
However, I'm not sure it's an elegant solution due to the amount of parameters.
I'm pushing it anyway so that reviewers can eventually suggest some improvement.

This commit also makes it easier to change RecursiveMutex m_bloom_filter_mutex to Mutex

achow101 · 2022-10-12T18:57:33Z

Are you still working on this?

hebasto · 2022-10-17T19:20:05Z

After 1066d10, the PR title and description should be updated, no?

w0xlt · 2022-10-17T20:14:25Z

Are you still working on this?

The PR is ready for reviews.

PR title and description should be updated

Done.

maflcko · 2022-10-18T06:35:16Z

src/net_processing.cpp

+            m_send_mempool = value;
+        }
+
+        void Relay(


I haven't reviewed this pull request, but this would be the first time that a net processing logic is put in the peers struct, instead of the net processing implementation.

The idea is to prevent external synchronizing (acquiring m_tx_inventory_mutex from outside of the CNode::TxRelay instance).
Perhaps this requires a more complex refactoring.

@dergoegge you might have thoughts here?

DrahtBot · 2022-12-02T16:29:34Z

🐙 This pull request conflicts with the target branch and needs rebase.

w0xlt marked this pull request as draft January 22, 2022 02:36

DrahtBot added the P2P label Jan 22, 2022

w0xlt force-pushed the cs_tx_inventory_mutex branch 2 times, most recently from 77c522b to 50f1534 Compare January 22, 2022 08:11

w0xlt force-pushed the cs_tx_inventory_mutex branch from 50f1534 to d097e5d Compare January 22, 2022 08:27

w0xlt marked this pull request as ready for review January 22, 2022 09:54

shaavan reviewed Jan 22, 2022

View reviewed changes

src/net_processing.cpp Outdated Show resolved Hide resolved

w0xlt force-pushed the cs_tx_inventory_mutex branch from d097e5d to f10db76 Compare January 22, 2022 11:45

This was referenced Jan 22, 2022

refactor: Release cs_main before MaybeSendFeefilter #22053

Closed

net_processing: lock clean up #21527

Closed

refactor: replace RecursiveMutex cs_vProcessMsg with Mutex (and rename) #24122

Closed

hebasto mentioned this pull request Jan 24, 2022

Replace all of the RecursiveMutex instances with the Mutex ones #19303

Open

36 tasks

This was referenced Feb 4, 2022

net processing: Reduce resource usage for inbound block-relay-only connections #22778

Merged

net/net processing: Move tx inventory into net_processing #21160

Merged

w0xlt force-pushed the cs_tx_inventory_mutex branch 2 times, most recently from e587cdf to a3b7192 Compare February 12, 2022 01:34

DrahtBot mentioned this pull request Mar 4, 2022

Additional thread safety annotations for CNode/Peer members accessed via the message processing thread #24474

Closed

DrahtBot added the Needs rebase label Mar 25, 2022

w0xlt added 4 commits June 24, 2022 11:03

net: add AddKnownTx() method to struct TxRelay

12e7276

net: add RelayTransaction() method to struct TxRelay

6b2b9fc

net: add VerifyUnconfirmedParent() method to struct TxRelay

da7b59a

net: add SetSendMempool() method to struct TxRelay

7a44768

w0xlt force-pushed the cs_tx_inventory_mutex branch from a3b7192 to 7a44768 Compare June 24, 2022 15:24

DrahtBot removed the Needs rebase label Jun 24, 2022

w0xlt added 2 commits June 24, 2022 15:27

net: add Relay() method to struct TxRelay

88777c4

net: replace RecursiveMutex m_tx_inventory_mutex with Mutex

60519a8

DrahtBot mentioned this pull request Sep 21, 2022

refactor: Move CNodeState members guarded by g_msgproc_mutex to Peer #26140

Merged

This was referenced Oct 4, 2022

refactor: Make m_mempool optional in PeerManager #26247

Closed

p2p: Fill reconciliation sets and request reconciliation (Erlay) #26283

Closed

w0xlt changed the title ~~p2p: Replace RecursiveMutex cs_tx_inventory with Mutex and rename it~~ p2p: Replace RecursiveMutex m_tx_inventory_mutex with Mutex and rename it Oct 17, 2022

maflcko reviewed Oct 18, 2022

View reviewed changes

DrahtBot mentioned this pull request Nov 16, 2022

Erlay: bandwidth-efficient transaction relay protocol #21515

Closed

This was referenced Nov 25, 2022

p2p: Ensure transaction announcements are only queued for fully connected peers #26569

Merged

tracing: Only prepare tracepoint arguments when actually tracing #26593

Merged

DrahtBot added the Needs rebase label Dec 2, 2022

achow101 added the Up for grabs label Apr 25, 2023

achow101 closed this Apr 25, 2023

bitcoin locked and limited conversation to collaborators Apr 24, 2024

p2p: Replace RecursiveMutex m_tx_inventory_mutex with Mutex and rename it #24125

p2p: Replace RecursiveMutex m_tx_inventory_mutex with Mutex and rename it #24125

Uh oh!

Conversation

w0xlt commented Jan 22, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

DrahtBot commented Jan 22, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Reviews

Conflicts

Uh oh!

shaavan left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

hebasto commented Jan 24, 2022

Uh oh!

w0xlt commented Jan 25, 2022

Uh oh!

hebasto commented Jan 25, 2022

Uh oh!

w0xlt commented Feb 12, 2022

Uh oh!

hebasto commented May 16, 2022

Uh oh!

w0xlt commented Jun 24, 2022

Uh oh!

w0xlt commented Jun 24, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

achow101 commented Oct 12, 2022

Uh oh!

hebasto commented Oct 17, 2022

Uh oh!

w0xlt commented Oct 17, 2022 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

maflcko Oct 18, 2022

Choose a reason for hiding this comment

Uh oh!

w0xlt Oct 18, 2022

Choose a reason for hiding this comment

Uh oh!

fanquake Feb 6, 2023

Choose a reason for hiding this comment

Uh oh!

DrahtBot commented Dec 2, 2022

Uh oh!

Uh oh!

p2p: Replace RecursiveMutex `m_tx_inventory_mutex` with Mutex and rename it #24125

p2p: Replace RecursiveMutex `m_tx_inventory_mutex` with Mutex and rename it #24125

w0xlt commented Jan 22, 2022 •

edited

Loading

DrahtBot commented Jan 22, 2022 •

edited

Loading

w0xlt commented Jun 24, 2022 •

edited

Loading

w0xlt commented Oct 17, 2022 •

edited

Loading