Implement RemoveAskFor to indicate that we're not interested in an it…

…em anymore (#2384)

* Implement RemoveAskFor to indicate that we're not interested in an item anymore

When an INV item is received from the first node, the item is requested
immediately. If the same item is received from another node, an entry is
added to mapAskFor which marks the item for re-requesting in case the first
node did not respond. When the item is received from the first node,
the item was previously never removed from mapAskFor. Only the later getdata
loop in SendMessages would then gradually remove items from the map. This
is quite delayed however as the entries in mapAskFor have a timeout value.

RemoveAskFor allows to remove all entries from mapAskFor and setAskFor
when we are not interested in the item anymore (e.g. because we received
it already).

* Call RemoveAskFor whenever we receive a message

* Only pass hash instead of CInv object to RemoveAskFor

src/governance.cpp

@@ -132,7 +132,10 @@ void CGovernanceManager::ProcessMessage(CNode* pfrom, const std::string& strComm
 
         uint256 nHash = govobj.GetHash();
 
-        pfrom->setAskFor.erase(nHash);
+        {
+            LOCK(cs_main);
+            connman.RemoveAskFor(nHash);
+        }
 
         if (pfrom->nVersion < MIN_GOVERNANCE_PEER_PROTO_VERSION) {
             LogPrint("gobject", "MNGOVERNANCEOBJECT -- peer=%d using obsolete version %i\n", pfrom->id, pfrom->nVersion);
@@ -220,7 +223,10 @@ void CGovernanceManager::ProcessMessage(CNode* pfrom, const std::string& strComm
 
         uint256 nHash = vote.GetHash();
 
-        pfrom->setAskFor.erase(nHash);
+        {
+            LOCK(cs_main);
+            connman.RemoveAskFor(nHash);
+        }
 
         if (pfrom->nVersion < MIN_GOVERNANCE_PEER_PROTO_VERSION) {
             LogPrint("gobject", "MNGOVERNANCEOBJECTVOTE -- peer=%d using obsolete version %i\n", pfrom->id, pfrom->nVersion);
@@ -1113,10 +1119,13 @@ int CGovernanceManager::RequestGovernanceObjectVotes(const std::vector<CNode*>&
             // only use up to date peers
             if (pnode->nVersion < MIN_GOVERNANCE_PEER_PROTO_VERSION) continue;
             // stop early to prevent setAskFor overflow
-            size_t nProjectedSize = pnode->setAskFor.size() + nProjectedVotes;
-            if (nProjectedSize > SETASKFOR_MAX_SZ / 2) continue;
-            // to early to ask the same node
-            if (mapAskedRecently[nHashGovobj].count(pnode->addr)) continue;
+            {
+                LOCK(cs_main);
+                size_t nProjectedSize = pnode->setAskFor.size() + nProjectedVotes;
+                if (nProjectedSize > SETASKFOR_MAX_SZ / 2) continue;
+                // to early to ask the same node
+                if (mapAskedRecently[nHashGovobj].count(pnode->addr)) continue;
+            }
 
             RequestGovernanceObject(pnode, nHashGovobj, connman, true);
             mapAskedRecently[nHashGovobj][pnode->addr] = nNow + nTimeout;

src/instantx.cpp

@@ -73,7 +73,10 @@ void CInstantSend::ProcessMessage(CNode* pfrom, const std::string& strCommand, C
 
         uint256 nVoteHash = vote.GetHash();
 
-        pfrom->setAskFor.erase(nVoteHash);
+        {
+            LOCK(cs_main);
+            connman.RemoveAskFor(nVoteHash);
+        }
 
         // Ignore any InstantSend messages until masternode list is synced
         if (!masternodeSync.IsMasternodeListSynced()) return;

src/masternode-payments.cpp

@@ -446,7 +446,10 @@ void CMasternodePayments::ProcessMessage(CNode* pfrom, const std::string& strCom
 
         uint256 nHash = vote.GetHash();
 
-        pfrom->setAskFor.erase(nHash);
+        {
+            LOCK(cs_main);
+            connman.RemoveAskFor(nHash);
+        }
 
         // TODO: clear setAskFor for MSG_MASTERNODE_PAYMENT_BLOCK too
 

src/masternodeman.cpp

@@ -977,7 +977,10 @@ void CMasternodeMan::ProcessMessage(CNode* pfrom, const std::string& strCommand,
         CMasternodeBroadcast mnb;
         vRecv >> mnb;
 
-        pfrom->setAskFor.erase(mnb.GetHash());
+        {
+            LOCK(cs_main);
+            connman.RemoveAskFor(mnb.GetHash());
+        }
 
         if(!masternodeSync.IsBlockchainSynced()) return;
 
@@ -1003,7 +1006,10 @@ void CMasternodeMan::ProcessMessage(CNode* pfrom, const std::string& strCommand,
 
         uint256 nHash = mnp.GetHash();
 
-        pfrom->setAskFor.erase(nHash);
+        {
+            LOCK(cs_main);
+            connman.RemoveAskFor(nHash);
+        }
 
         if(!masternodeSync.IsBlockchainSynced()) return;
 
@@ -1066,7 +1072,10 @@ void CMasternodeMan::ProcessMessage(CNode* pfrom, const std::string& strCommand,
         CMasternodeVerification mnv;
         vRecv >> mnv;
 
-        pfrom->setAskFor.erase(mnv.GetHash());
+        {
+            LOCK(cs_main);
+            connman.RemoveAskFor(mnv.GetHash());
+        }
 
         if(!masternodeSync.IsMasternodeListSynced()) return;
 

src/net.cpp

@@ -2655,6 +2655,16 @@ void CConnman::RelayInvFiltered(CInv &inv, const CTransaction& relatedTx, const
     }
 }
 
+void CConnman::RemoveAskFor(const uint256& hash)
+{
+    mapAlreadyAskedFor.erase(hash);
+
+    LOCK(cs_vNodes);
+    for (const auto& pnode : vNodes) {
+        pnode->RemoveAskFor(hash);
+    }
+}
+
 void CConnman::RecordBytesRecv(uint64_t bytes)
 {
     LOCK(cs_totalBytesRecv);
@@ -2906,6 +2916,18 @@ void CNode::AskFor(const CInv& inv)
     mapAskFor.insert(std::make_pair(nRequestTime, inv));
 }
 
+void CNode::RemoveAskFor(const uint256& hash)
+{
+    setAskFor.erase(hash);
+    for (auto it = mapAskFor.begin(); it != mapAskFor.end();) {
+        if (it->second.hash == hash) {
+            it = mapAskFor.erase(it);
+        } else {
+            ++it;
+        }
+    }
+}
+
 bool CConnman::NodeFullyConnected(const CNode* pnode)
 {
     return pnode && pnode->fSuccessfullyConnected && !pnode->fDisconnect;

src/net.h

@@ -310,6 +310,7 @@ class CConnman
     void RelayTransaction(const CTransaction& tx, const CDataStream& ss);
     void RelayInv(CInv &inv, const int minProtoVersion = MIN_PEER_PROTO_VERSION);
     void RelayInvFiltered(CInv &inv, const CTransaction &relatedTx, const int minProtoVersion = MIN_PEER_PROTO_VERSION);
+    void RemoveAskFor(const uint256& hash);
 
     // Addrman functions
     size_t GetAddressCount() const;
@@ -929,6 +930,7 @@ class CNode
     }
 
     void AskFor(const CInv& inv);
+    void RemoveAskFor(const uint256& hash);
 
     void CloseSocketDisconnect();
 

src/net_processing.cpp

@@ -1962,7 +1962,6 @@ bool static ProcessMessage(CNode* pfrom, const std::string& strCommand, CDataStr
 
         CInv inv(nInvType, tx.GetHash());
         pfrom->AddInventoryKnown(inv);
-        pfrom->setAskFor.erase(inv.hash);
 
         // Process custom logic, no matter if tx will be accepted to mempool later or not
         if (strCommand == NetMsgType::TXLOCKREQUEST || fCanAutoLock) {
@@ -2009,11 +2008,11 @@ bool static ProcessMessage(CNode* pfrom, const std::string& strCommand, CDataStr
 
         LOCK(cs_main);
 
+        connman.RemoveAskFor(inv.hash);
+
         bool fMissingInputs = false;
         CValidationState state;
 
-        mapAlreadyAskedFor.erase(inv.hash);
-
         if (!AlreadyHave(inv) && AcceptToMemoryPool(mempool, state, ptx, true, &fMissingInputs)) {
             // Process custom txes, this changes AlreadyHave to "true"
             if (strCommand == NetMsgType::DSTX) {

src/spork.cpp

@@ -123,7 +123,7 @@ void CSporkManager::ProcessSpork(CNode* pfrom, const std::string& strCommand, CD
         std::string strLogMsg;
         {
             LOCK(cs_main);
-            pfrom->setAskFor.erase(hash);
+            connman.RemoveAskFor(hash);
             if(!chainActive.Tip()) return;
             strLogMsg = strprintf("SPORK -- hash: %s id: %d value: %10d bestHeight: %d peer=%d", hash.ToString(), spork.nSporkID, spork.nValue, chainActive.Height(), pfrom->id);
         }