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

theStack · 2022-01-21T12:38:08Z

This PR is related to #19303 and gets rid of the RecursiveMutex cs_vProcessMsg. Both of the critical sections only directly access the guarded elements, i.e. it is not possible that within one section another one is called, and we can use a regular Mutex:

bitcoin/src/net.cpp

Lines 1597 to 1602 in e3ce019

    
           { 
        
               LOCK(pnode->cs_vProcessMsg); 
        
               pnode->vProcessMsg.splice(pnode->vProcessMsg.end(), pnode->vRecvMsg, pnode->vRecvMsg.begin(), it); 
        
               pnode->nProcessQueueSize += nSizeAdded; 
        
               pnode->fPauseRecv = pnode->nProcessQueueSize > nReceiveFloodSize; 
        
           }

bitcoin/src/net_processing.cpp

Lines 4142 to 4150 in e3ce019

    
           { 
        
               LOCK(pfrom->cs_vProcessMsg); 
        
               if (pfrom->vProcessMsg.empty()) return false; 
        
               // Just take one message 
        
               msgs.splice(msgs.begin(), pfrom->vProcessMsg, pfrom->vProcessMsg.begin()); 
        
               pfrom->nProcessQueueSize -= msgs.front().m_raw_message_size; 
        
               pfrom->fPauseRecv = pfrom->nProcessQueueSize > m_connman.GetReceiveFloodSize(); 
        
               fMoreWork = !pfrom->vProcessMsg.empty(); 
        
           }

Also, it is renamed to adapt to the (unwritten) naming convention to use the _mutex suffix rather than the cs_ prefix.

-BEGIN VERIFY SCRIPT- sed -i 's/cs_vProcessMsg/m_process_msgs_mutex/g' $(git grep -l cs_vProcessMsg) sed -i 's/vProcessMsg/m_process_msgs/g' $(git grep -l vProcessMsg) -END VERIFY SCRIPT-

In each of the critical sections, only the the m_process_msgs member is accessed, without any chance that within one section another one is called. Hence, we can use an ordinary Mutex instead of RecursiveMutex.

shaavan

Concept ACK

I was thinking, should we use AssertLockNotHeld(m_process_msgs_mutex) before locking the mutex and add thread safety annotation LOCKS_EXCLUDED(m_process_msgs_mutex), with the function declaration of where this mutex is used?

Doing so will prevent any risk of error due to simultaneous locking of this mutex.

w0xlt

crACK 9a355e1

I agree with @shaavan .

It might be nice to add AssertLockNotHeld(mutex) macros combined with LOCKS_EXCLUDED(mutex) thread safety annotations. This avoids any risk of recursive locking.

theStack · 2022-01-22T20:54:40Z

@shaavan @w0xlt: Thanks for your feedback, I agree that adding thread safety assertions and annotations is a good idea. Added a commit doing that -- note though that the LOCKS_EXCLUDED annotation is only possible for one method (PeerManagerImpl::ProcessMessages), as for the other (CConnman::SocketHandlerConnected) the elements that cointain the loop are in a list rather than a direct parameter that can be accessed.

DrahtBot · 2022-01-23T09:11:00Z

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

Conflicts

Reviewers, this pull request conflicts with the following ones:

#25109 (Strengthen AssertLockNotHeld assertions by ajtowns)
#24931 (Strengthen thread safety assertions by ajtowns)
#24474 (Additional thread safety annotations for CNode/Peer by ajtowns)
#21527 (net_processing: lock clean up by ajtowns)

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

as for the other (CConnman::SocketHandlerConnected) the elements that cointain the loop are in a list rather than a direct parameter that can be accessed.

Sounds interesting. Let me take another look at this before I can be 100% sure. In the meantime, I had one observation that I would like to discuss.

shaavan · 2022-01-23T14:21:41Z

src/test/util/net.cpp

@@ -22,8 +22,8 @@ void ConnmanTestMsg::NodeReceiveMsgBytes(CNode& node, Span<const uint8_t> msg_by
            nSizeAdded += it->m_raw_message_size;
        }
        {
-            LOCK(node.cs_vProcessMsg);
-            node.vProcessMsg.splice(node.vProcessMsg.end(), node.vRecvMsg, node.vRecvMsg.begin(), it);
+            LOCK(node.m_process_msgs_mutex);


Can we add AssertLockNotHeld above this line?

Suggested change

LOCK(node.m_process_msgs_mutex);

AssertLockNotHeld(node.m_process_msgs_mutex);

LOCK(node.m_process_msgs_mutex);

Good catch, done! Also added the thread safety annotation to the corresponding method ConnmanTestMsg::NodeReceiveMsgBytes.

Co-authored-by: Shashwat <svangani239@gmail.com>

hebasto

Concept ACK.

Following best practices, mutexes should synchronize access to internal class data to prevent violation of class invariants.

The current implementation has a significant design drawback -- external synchronizing, i.e., acquiring CNode::m_process_msgs_mutex from outside of the CNode instance.

Therefore, I'm suggesting to consider to encapsulate access to the private members m_process_msgs and nProcessQueueSize with internal synchronizing using m_process_msgs_mutex.

One new CNode member function could implement

bitcoin/src/net.cpp

Lines 1592 to 1604 in e3de7cb

    
           size_t nSizeAdded = 0; 
        
           auto it(pnode->vRecvMsg.begin()); 
        
           for (; it != pnode->vRecvMsg.end(); ++it) { 
        
               // vRecvMsg contains only completed CNetMessage 
        
               // the single possible partially deserialized message are held by TransportDeserializer 
        
               nSizeAdded += it->m_raw_message_size; 
        
           } 
        
           { 
        
               LOCK(pnode->cs_vProcessMsg); 
        
               pnode->vProcessMsg.splice(pnode->vProcessMsg.end(), pnode->vRecvMsg, pnode->vRecvMsg.begin(), it); 
        
               pnode->nProcessQueueSize += nSizeAdded; 
        
               pnode->fPauseRecv = pnode->nProcessQueueSize > nReceiveFloodSize; 
        
           }

Another one

bitcoin/src/net_processing.cpp

Lines 4165 to 4173 in e3de7cb

    
           { 
        
               LOCK(pfrom->cs_vProcessMsg); 
        
               if (pfrom->vProcessMsg.empty()) return false; 
        
               // Just take one message 
        
               msgs.splice(msgs.begin(), pfrom->vProcessMsg, pfrom->vProcessMsg.begin()); 
        
               pfrom->nProcessQueueSize -= msgs.front().m_raw_message_size; 
        
               pfrom->fPauseRecv = pfrom->nProcessQueueSize > m_connman.GetReceiveFloodSize(); 
        
               fMoreWork = !pfrom->vProcessMsg.empty(); 
        
           }

hebasto · 2022-01-24T14:57:32Z

src/net.h

-    RecursiveMutex cs_vProcessMsg;
-    std::list<CNetMessage> vProcessMsg GUARDED_BY(cs_vProcessMsg);
+    Mutex m_process_msgs_mutex;
+    std::list<CNetMessage> m_process_msgs GUARDED_BY(m_process_msgs_mutex);
    size_t nProcessQueueSize{0};


Reading the code makes me strongly believe that the value of nProcessQueueSize must correspond to the m_process_msgs content. To preserve this invariant of the CNode class I'm suggesting:

Suggested change

size_t nProcessQueueSize{0};

size_t nProcessQueueSize GUARDED_BY(m_process_msgs_mutex){0};

DrahtBot · 2022-05-16T13:05:57Z

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

_{Want to unsubscribe from rebase notifications on this pull request? Just convert this pull request to a "draft".}

DrahtBot · 2022-08-01T10:51:04Z

There hasn't been much activity lately and the patch still needs rebase. What is the status here?

Is it still relevant? ➡️ Please solve the conflicts to make it ready for review and to ensure the CI passes.
Is it no longer relevant? ➡️ Please close.
Did the author lose interest or time to work on this? ➡️ Please close it and mark it 'Up for grabs' with the label, so that it can be picked up in the future.

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

Are you still working on this?

achow101 · 2022-11-10T17:12:57Z

Closing this as it has not had any activity in a while. If you are interested in continuing work on this, please leave a comment so that it can be reopened.

theStack added 2 commits January 21, 2022 12:40

scripted-diff: rename vProcessMsg list/mutex members in CNode

d55f8a5

-BEGIN VERIFY SCRIPT- sed -i 's/cs_vProcessMsg/m_process_msgs_mutex/g' $(git grep -l cs_vProcessMsg) sed -i 's/vProcessMsg/m_process_msgs/g' $(git grep -l vProcessMsg) -END VERIFY SCRIPT-

refactor: replace RecursiveMutex m_process_msgs_mutex with Mutex

9a355e1

In each of the critical sections, only the the m_process_msgs member is accessed, without any chance that within one section another one is called. Hence, we can use an ordinary Mutex instead of RecursiveMutex.

DrahtBot added P2P Refactoring labels Jan 21, 2022

shaavan reviewed Jan 21, 2022

View reviewed changes

w0xlt approved these changes Jan 22, 2022

View reviewed changes

DrahtBot mentioned this pull request Jan 23, 2022

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

Closed

shaavan reviewed Jan 23, 2022

View reviewed changes

DrahtBot mentioned this pull request Jan 23, 2022

net_processing: lock clean up #21527

Closed

theStack force-pushed the 202201-refactor_replace_RecursiveMutex_cs_vProcess branch from 9add3f9 to b819ae4 Compare January 24, 2022 13:44

refactor: add thread safety lock assertions for m_process_msgs_mutex

9c0b268

Co-authored-by: Shashwat <svangani239@gmail.com>

theStack force-pushed the 202201-refactor_replace_RecursiveMutex_cs_vProcess branch from b819ae4 to 9c0b268 Compare January 24, 2022 13:45

hebasto mentioned this pull request Jan 24, 2022

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

Open

36 tasks

hebasto reviewed Jan 24, 2022

View reviewed changes

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 mentioned this pull request Apr 20, 2022

Strengthen thread safety assertions #24931

Merged

DrahtBot mentioned this pull request May 12, 2022

Strengthen AssertLockNotHeld assertions #25109

Merged

DrahtBot added the Needs rebase label May 16, 2022

achow101 closed this Nov 10, 2022

bitcoin locked and limited conversation to collaborators Nov 10, 2023

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

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

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

Uh oh!

theStack commented Jan 21, 2022

Uh oh!

shaavan left a comment

Uh oh!

w0xlt left a comment

Uh oh!

theStack commented Jan 22, 2022

Uh oh!

DrahtBot commented Jan 23, 2022 •

edited

Loading

Uh oh!

shaavan left a comment

Uh oh!

shaavan Jan 23, 2022

Uh oh!

theStack Jan 24, 2022

Uh oh!

hebasto left a comment

Uh oh!

hebasto Jan 24, 2022

Uh oh!

DrahtBot commented May 16, 2022

Uh oh!

DrahtBot commented Aug 1, 2022

Uh oh!

achow101 commented Oct 12, 2022

Uh oh!

achow101 commented Nov 10, 2022

Uh oh!

Uh oh!

	{
	LOCK(pnode->cs_vProcessMsg);
	pnode->vProcessMsg.splice(pnode->vProcessMsg.end(), pnode->vRecvMsg, pnode->vRecvMsg.begin(), it);
	pnode->nProcessQueueSize += nSizeAdded;
	pnode->fPauseRecv = pnode->nProcessQueueSize > nReceiveFloodSize;
	}

	{
	LOCK(pfrom->cs_vProcessMsg);
	if (pfrom->vProcessMsg.empty()) return false;
	// Just take one message
	msgs.splice(msgs.begin(), pfrom->vProcessMsg, pfrom->vProcessMsg.begin());
	pfrom->nProcessQueueSize -= msgs.front().m_raw_message_size;
	pfrom->fPauseRecv = pfrom->nProcessQueueSize > m_connman.GetReceiveFloodSize();
	fMoreWork = !pfrom->vProcessMsg.empty();
	}

	LOCK(node.m_process_msgs_mutex);
	AssertLockNotHeld(node.m_process_msgs_mutex);
	LOCK(node.m_process_msgs_mutex);

	size_t nSizeAdded = 0;
	auto it(pnode->vRecvMsg.begin());
	for (; it != pnode->vRecvMsg.end(); ++it) {
	// vRecvMsg contains only completed CNetMessage
	// the single possible partially deserialized message are held by TransportDeserializer
	nSizeAdded += it->m_raw_message_size;
	}
	{
	LOCK(pnode->cs_vProcessMsg);
	pnode->vProcessMsg.splice(pnode->vProcessMsg.end(), pnode->vRecvMsg, pnode->vRecvMsg.begin(), it);
	pnode->nProcessQueueSize += nSizeAdded;
	pnode->fPauseRecv = pnode->nProcessQueueSize > nReceiveFloodSize;
	}

	size_t nProcessQueueSize{0};
	size_t nProcessQueueSize GUARDED_BY(m_process_msgs_mutex){0};

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

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

Uh oh!

Conversation

theStack commented Jan 21, 2022

Uh oh!

shaavan left a comment

Choose a reason for hiding this comment

Uh oh!

w0xlt left a comment

Choose a reason for hiding this comment

Uh oh!

theStack commented Jan 22, 2022

Uh oh!

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

Uh oh!

Conflicts

Uh oh!

shaavan left a comment

Choose a reason for hiding this comment

Uh oh!

shaavan Jan 23, 2022

Choose a reason for hiding this comment

Uh oh!

theStack Jan 24, 2022

Choose a reason for hiding this comment

Uh oh!

hebasto left a comment

Choose a reason for hiding this comment

Uh oh!

hebasto Jan 24, 2022

Choose a reason for hiding this comment

Uh oh!

DrahtBot commented May 16, 2022

Uh oh!

DrahtBot commented Aug 1, 2022

Uh oh!

achow101 commented Oct 12, 2022

Uh oh!

achow101 commented Nov 10, 2022

Uh oh!

Uh oh!

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

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

DrahtBot commented Jan 23, 2022 •

edited

Loading