Merge #6952: feat: rework message processing in CNode to prioritize quorum messages

627a8a5 fix: whitepace (pasta)
d0d42c7 refactor: prefix add (PastaPastaPasta)
1d16451 fix: ensure that flooding quorum messages cannot starve the normal queue of processing (pasta)
6c955b7 fix: whitespace (pasta)
df6cc76 feat: rework message processing in CNode to prioritize quorum messages (pasta)

Pull request description:

  ## Issue being fixed or feature implemented
  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.

  ## What was done?
    _Describe your changes in detail_

  ## How Has This Been Tested?
    _Please describe in detail how you tested your changes._

    _Include details of your testing environment, and the tests you ran
  to see how your change affects other areas of the code, etc._

  ## Breaking Changes
    _Please describe any breaking changes your code introduces_

  ## Checklist:
    _Go over all the following points, and put an `x` in all the boxes that apply._
  - [ ] I have performed a self-review of my own code
  - [ ] I have commented my code, particularly in hard-to-understand areas
  - [ ] I have added or updated relevant unit/integration/functional/e2e tests
  - [ ] I have made corresponding changes to the documentation
  - [x] I have assigned this pull request to a milestone _(for repository code-owners and collaborators only)_

ACKs for top commit:
  UdjinM6:
    utACK 627a8a5
  kwvg:
    utACK 627a8a5

Tree-SHA512: d8eac0934c377daa36e12ed75a7d944ec0025f021c5d36a609a2967bdc41464e66ba8395abc4559a949f41c136ca496bda6276b032ec9011213563b0134764d4

Author: PastaPastaPasta

src/net.cpp

@@ -4652,31 +4652,79 @@ 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);
+
+    // Ratio-based processing: process N quorum messages for every 1 normal message
+    // This ensures forward progress for both queues while strongly prioritizing quorum messages
+    // However, if normal queue is empty, process quorum messages in bursts (like old algorithm)
+    constexpr size_t QUORUM_TO_NORMAL_RATIO = 100;
+
+    // Check if we should process normal queue for forward progress
+    // Only apply ratio when both queues have messages to allow burst processing when normal queue is empty
+    bool skip_quorum_processing = !m_msg_process_queue.empty() &&
+                                  m_quorum_msg_count_since_normal >= QUORUM_TO_NORMAL_RATIO;
+
+    // Prioritize quorum queue: pop from it first if non-empty and ratio not reached
+    // If normal queue is empty, process quorum messages without ratio limit (burst mode)
+    if (!m_msg_quorum_queue.empty() && !skip_quorum_processing) {
+        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;
+        // Only increment counter if normal queue has messages (to track ratio)
+        // If normal queue is empty, don't increment so we can process bursts quickly
+        if (!m_msg_process_queue.empty()) {
+            ++m_quorum_msg_count_since_normal;
+        }
+        // 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());
+    }
+
+    // Process normal queue (either because quorum queue is empty or ratio reached)
     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;
+    m_quorum_msg_count_since_normal = 0; // Reset counter after processing normal message
+    // 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,9 @@ 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};
+    size_t m_quorum_msg_count_since_normal 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 +2005,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