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.

Cherry-picked from: 9f2ab9e
Github Pull Request: dogecoin#3575

Author: sipa

src/net.h

@@ -696,6 +696,7 @@ class CNode
 
     // Counts getheaders requests sent to this peer
     std::atomic<int64_t> nPendingHeaderRequests;
+    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 std::string &addrNameIn = "", bool fInboundIn = false);
     ~CNode();

src/net_processing.cpp

@@ -1346,7 +1346,8 @@ void static ProcessOrphanTx(CConnman* connman, std::set<uint256>& orphan_work_se
 {
     AssertLockHeld(cs_main);
     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());
@@ -1381,6 +1382,7 @@ void static ProcessOrphanTx(CConnman* connman, std::set<uint256>& orphan_work_se
             }
 
             EraseOrphanTx(orphanHash);
+            done = true;
 
         } else if (!fMissingInputs2) {
 
@@ -1403,6 +1405,7 @@ void static ProcessOrphanTx(CConnman* connman, std::set<uint256>& orphan_work_se
             }
 
             EraseOrphanTx(orphanHash);
+            done = true;
 
         }
 
@@ -2073,7 +2076,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;
@@ -2100,7 +2102,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);
                     }
                 }
             }
@@ -2113,7 +2115,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)
         {
@@ -2926,11 +2928,21 @@ bool ProcessMessages(CNode* pfrom, CConnman& connman, const std::atomic<bool>& i
     if (!pfrom->vRecvGetData.empty())
         ProcessGetData(pfrom, chainparams.GetConsensus(chainActive.Height()), connman, interruptMsgProc);
 
+    if (!pfrom->orphan_work_set.empty()) {
+        std::list<CTransactionRef> removed_txn;
+        LOCK(cs_main);
+        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)