Interrupt orphan processing after every transaction

This makes orphan processing work like handling getdata messages:
After every actual transaction validation attempt, interrupt
processing to deal with messages arriving from other peers.

Author: sipa

src/net.h

@@ -739,6 +739,8 @@ class CNode
     CAmount lastSentFeeFilter{0};
     int64_t nextSendTimeFeeFilter{0};
 
+    std::set<uint256> orphan_work_set;
+
     CNode(NodeId id, ServiceFlags nLocalServicesIn, int nMyStartingHeightIn, SOCKET hSocketIn, const CAddress &addrIn, uint64_t nKeyedNetGroupIn, uint64_t nLocalHostNonceIn, const CAddress &addrBindIn, const std::string &addrNameIn = "", bool fInboundIn = false);
     ~CNode();
     CNode(const CNode&) = delete;

src/net_processing.cpp

@@ -1718,7 +1718,8 @@ void static ProcessOrphanTx(CConnman* connman, std::set<uint256>& orphan_work_se
     AssertLockHeld(cs_main);
     AssertLockHeld(g_cs_orphans);
     std::set<NodeId> setMisbehaving;
-    while (!orphan_work_set.empty()) {
+    bool done = false;
+    while (!done && !orphan_work_set.empty()) {
         const uint256 orphanHash = *orphan_work_set.begin();
         orphan_work_set.erase(orphan_work_set.begin());
 
@@ -1747,6 +1748,7 @@ void static ProcessOrphanTx(CConnman* connman, std::set<uint256>& orphan_work_se
                 }
             }
             EraseOrphanTx(orphanHash);
+            done = true;
         } else if (!fMissingInputs2) {
             int nDos = 0;
             if (stateDummy.IsInvalid(nDos) && nDos > 0) {
@@ -1766,6 +1768,7 @@ void static ProcessOrphanTx(CConnman* connman, std::set<uint256>& orphan_work_se
                 recentRejects->insert(orphanHash);
             }
             EraseOrphanTx(orphanHash);
+            done = true;
         }
         mempool.check(pcoinsTip.get());
     }
@@ -2400,8 +2403,6 @@ bool static ProcessMessage(CNode* pfrom, const std::string& strCommand, CDataStr
             return true;
         }
 
-        std::set<uint256> orphan_work_set;
-
         CTransactionRef ptx;
         vRecv >> ptx;
         const CTransaction& tx = *ptx;
@@ -2429,7 +2430,7 @@ bool static ProcessMessage(CNode* pfrom, const std::string& strCommand, CDataStr
                 auto it_by_prev = mapOrphanTransactionsByPrev.find(COutPoint(inv.hash, i));
                 if (it_by_prev != mapOrphanTransactionsByPrev.end()) {
                     for (const auto& elem : it_by_prev->second) {
-                        orphan_work_set.insert(elem->first);
+                        pfrom->orphan_work_set.insert(elem->first);
                     }
                 }
             }
@@ -2442,7 +2443,7 @@ bool static ProcessMessage(CNode* pfrom, const std::string& strCommand, CDataStr
                 mempool.size(), mempool.DynamicMemoryUsage() / 1000);
 
             // Recursively process any orphan transactions that depended on this one
-            ProcessOrphanTx(connman, orphan_work_set, lRemovedTxn);
+            ProcessOrphanTx(connman, pfrom->orphan_work_set, lRemovedTxn);
         }
         else if (fMissingInputs)
         {
@@ -3174,11 +3175,21 @@ bool PeerLogicValidation::ProcessMessages(CNode* pfrom, std::atomic<bool>& inter
     if (!pfrom->vRecvGetData.empty())
         ProcessGetData(pfrom, chainparams, connman, interruptMsgProc);
 
+    if (!pfrom->orphan_work_set.empty()) {
+        std::list<CTransactionRef> removed_txn;
+        LOCK2(cs_main, g_cs_orphans);
+        ProcessOrphanTx(connman, pfrom->orphan_work_set, removed_txn);
+        for (const CTransactionRef& removedTx : removed_txn) {
+            AddToCompactExtraTransactions(removedTx);
+        }
+    }
+
     if (pfrom->fDisconnect)
         return false;
 
     // this maintains the order of responses
     if (!pfrom->vRecvGetData.empty()) return true;
+    if (!pfrom->orphan_work_set.empty()) return true;
 
     // Don't bother if send buffer is too full to respond anyway
     if (pfrom->fPauseSend)