feat: rework message processing in CNode to prioritize quorum messages

This update introduces a new queue for quorum-priority messages, allowing for better handling of critical network messages. The `MarkReceivedMsgsForProcessing` function now classifies incoming messages into quorum and normal queues, while the `PollMessage` function prioritizes messages from the quorum queue when available. This change enhances the efficiency of message processing and ensures that important messages are handled promptly.

Author: PastaPastaPasta

src/net.cpp

@@ -4652,31 +4652,62 @@ void CNode::MarkReceivedMsgsForProcessing()
 {
     AssertLockNotHeld(m_msg_process_queue_mutex);
 
-    size_t nSizeAdded = 0;
-    for (const auto& msg : vRecvMsg) {
+    size_t nQuorumSizeAdded = 0;
+    size_t nNormalSizeAdded = 0;
+    std::list<CNetMessage> quorumMsgs;
+    std::list<CNetMessage> normalMsgs;
+
+    // Classify messages into quorum-priority and normal queues
+    for (auto it = vRecvMsg.begin(); it != vRecvMsg.end();) {
+        auto& msg = *it;
         // vRecvMsg contains only completed CNetMessage
         // the single possible partially deserialized message are held by TransportDeserializer
-        nSizeAdded += msg.m_raw_message_size;
+        if (IsQuorumPriorityMessage(msg.m_type)) {
+            quorumMsgs.splice(quorumMsgs.end(), vRecvMsg, it++);
+            nQuorumSizeAdded += msg.m_raw_message_size;
+        } else {
+            normalMsgs.splice(normalMsgs.end(), vRecvMsg, it++);
+            nNormalSizeAdded += msg.m_raw_message_size;
+        }
     }
 
     LOCK(m_msg_process_queue_mutex);
-    m_msg_process_queue.splice(m_msg_process_queue.end(), vRecvMsg);
-    m_msg_process_queue_size += nSizeAdded;
-    fPauseRecv = m_msg_process_queue_size > m_recv_flood_size;
+    // Splice classified messages into appropriate queues
+    m_msg_quorum_queue.splice(m_msg_quorum_queue.end(), quorumMsgs);
+    m_msg_quorum_queue_size += nQuorumSizeAdded;
+    m_msg_process_queue.splice(m_msg_process_queue.end(), normalMsgs);
+    m_msg_process_queue_size += nNormalSizeAdded;
+    // Compute backpressure over combined size of both queues
+    fPauseRecv = (m_msg_quorum_queue_size + m_msg_process_queue_size) > m_recv_flood_size;
 }
 
 std::optional<std::pair<CNetMessage, bool>> CNode::PollMessage()
 {
     LOCK(m_msg_process_queue_mutex);
+    
+    // Prioritize quorum queue: pop from it first if non-empty
+    if (!m_msg_quorum_queue.empty()) {
+        std::list<CNetMessage> msgs;
+        // Just take one message from quorum queue
+        msgs.splice(msgs.begin(), m_msg_quorum_queue, m_msg_quorum_queue.begin());
+        m_msg_quorum_queue_size -= msgs.front().m_raw_message_size;
+        // Compute backpressure over combined size of both queues
+        fPauseRecv = (m_msg_quorum_queue_size + m_msg_process_queue_size) > m_recv_flood_size;
+        // Return true for 'more' if either queue has remaining messages
+        return std::make_pair(std::move(msgs.front()), !m_msg_quorum_queue.empty() || !m_msg_process_queue.empty());
+    }
+    
+    // Fall back to normal queue if quorum queue is empty
     if (m_msg_process_queue.empty()) return std::nullopt;
 
     std::list<CNetMessage> msgs;
     // Just take one message
     msgs.splice(msgs.begin(), m_msg_process_queue, m_msg_process_queue.begin());
     m_msg_process_queue_size -= msgs.front().m_raw_message_size;
-    fPauseRecv = m_msg_process_queue_size > m_recv_flood_size;
+    // Compute backpressure over combined size of both queues
+    fPauseRecv = (m_msg_quorum_queue_size + m_msg_process_queue_size) > m_recv_flood_size;
 
-    return std::make_pair(std::move(msgs.front()), !m_msg_process_queue.empty());
+    return std::make_pair(std::move(msgs.front()), !m_msg_quorum_queue.empty() || !m_msg_process_queue.empty());
 }
 
 bool CConnman::NodeFullyConnected(const CNode* pnode)

src/net.h

@@ -1094,6 +1094,8 @@ class CNode
     Mutex m_msg_process_queue_mutex;
     std::list<CNetMessage> m_msg_process_queue GUARDED_BY(m_msg_process_queue_mutex);
     size_t m_msg_process_queue_size GUARDED_BY(m_msg_process_queue_mutex){0};
+    std::list<CNetMessage> m_msg_quorum_queue GUARDED_BY(m_msg_process_queue_mutex);
+    size_t m_msg_quorum_queue_size GUARDED_BY(m_msg_process_queue_mutex){0};
 
     // Our address, as reported by the peer
     CService addrLocal GUARDED_BY(m_addr_local_mutex);
@@ -2002,4 +2004,24 @@ class CExplicitNetCleanup
     static void callCleanup();
 };
 
+// Helper function to determine if a message type should be prioritized in the quorum queue
+inline bool IsQuorumPriorityMessage(const std::string& msg_type)
+{
+    // LLMQ signing/data messages
+    if (msg_type == NetMsgType::QSIGSHARE ||
+        msg_type == NetMsgType::QBSIGSHARES ||
+        msg_type == NetMsgType::QSIGSHARESINV ||
+        msg_type == NetMsgType::QGETSIGSHARES ||
+        msg_type == NetMsgType::QSIGSESANN ||
+        msg_type == NetMsgType::QSIGREC) {
+        return true;
+    }
+    // High-level lock messages (ChainLocks, InstantSend locks)
+    if (msg_type == NetMsgType::CLSIG ||
+        msg_type == NetMsgType::ISDLOCK) {
+        return true;
+    }
+    return false;
+}
+
 #endif // BITCOIN_NET_H