[mempool] Track "unbroadcast" transactions

amitiuttarwar · amitiuttarwar · commit 89eeb4a3335f · 2020-04-23T14:42:25.000-07:00
- Mempool tracks locally submitted transactions (wallet or rpc)
- Transactions are removed from set when the node receives a GETDATA request
  from a peer, or if the transaction is removed from the mempool.
diff --git a/src/net_processing.cpp b/src/net_processing.cpp
@@ -1556,7 +1556,7 @@ void static ProcessGetBlockData(CNode* pfrom, const CChainParams& chainparams, c
     }
 }
 
-void static ProcessGetData(CNode* pfrom, const CChainParams& chainparams, CConnman* connman, const CTxMemPool& mempool, const std::atomic<bool>& interruptMsgProc) LOCKS_EXCLUDED(cs_main)
+void static ProcessGetData(CNode* pfrom, const CChainParams& chainparams, CConnman* connman, CTxMemPool& mempool, const std::atomic<bool>& interruptMsgProc) LOCKS_EXCLUDED(cs_main)
 {
     AssertLockNotHeld(cs_main);
 
@@ -1605,7 +1605,13 @@ void static ProcessGetData(CNode* pfrom, const CChainParams& chainparams, CConnm
                     push = true;
                 }
             }
-            if (!push) {
+
+            if (push) {
+                // We interpret fulfilling a GETDATA for a transaction as a
+                // successful initial broadcast and remove it from our
+                // unbroadcast set.
+                mempool.RemoveUnbroadcastTx(inv.hash);
+            } else {
                 vNotFound.push_back(inv);
             }
         }
diff --git a/src/node/transaction.cpp b/src/node/transaction.cpp
@@ -78,6 +78,10 @@ TransactionError BroadcastTransaction(NodeContext& node, const CTransactionRef t
     }
 
     if (relay) {
+        // the mempool tracks locally submitted transactions to make a
+        // best-effort of initial broadcast
+        node.mempool->AddUnbroadcastTx(hashTx);
+
         RelayTransaction(hashTx, *node.connman);
     }
 
diff --git a/src/txmempool.cpp b/src/txmempool.cpp
@@ -417,6 +417,8 @@ void CTxMemPool::removeUnchecked(txiter it, MemPoolRemovalReason reason)
     for (const CTxIn& txin : it->GetTx().vin)
         mapNextTx.erase(txin.prevout);
 
+    RemoveUnbroadcastTx(hash, true /* add logging because unchecked */ );
+
     if (vTxHashes.size() > 1) {
         vTxHashes[it->vTxHashesIdx] = std::move(vTxHashes.back());
         vTxHashes[it->vTxHashesIdx].second->vTxHashesIdx = it->vTxHashesIdx;
@@ -919,6 +921,15 @@ size_t CTxMemPool::DynamicMemoryUsage() const {
     return memusage::MallocUsage(sizeof(CTxMemPoolEntry) + 12 * sizeof(void*)) * mapTx.size() + memusage::DynamicUsage(mapNextTx) + memusage::DynamicUsage(mapDeltas) + memusage::DynamicUsage(mapLinks) + memusage::DynamicUsage(vTxHashes) + cachedInnerUsage;
 }
 
+void CTxMemPool::RemoveUnbroadcastTx(const uint256& txid, const bool unchecked) {
+    LOCK(cs);
+
+    if (m_unbroadcast_txids.erase(txid))
+    {
+        LogPrint(BCLog::MEMPOOL, "Removed %i from set of unbroadcast txns%s\n", txid.GetHex(), (unchecked ? " before confirmation that txn was sent out" : ""));
+    }
+}
+
 void CTxMemPool::RemoveStaged(setEntries &stage, bool updateDescendants, MemPoolRemovalReason reason) {
     AssertLockHeld(cs);
     UpdateForRemoveFromMempool(stage, updateDescendants);
diff --git a/src/txmempool.h b/src/txmempool.h
@@ -549,6 +549,9 @@ class CTxMemPool
 
     std::vector<indexed_transaction_set::const_iterator> GetSortedDepthAndScore() const EXCLUSIVE_LOCKS_REQUIRED(cs);
 
+    /** track locally submitted transactions to periodically retry initial broadcast */
+    std::set<uint256> m_unbroadcast_txids GUARDED_BY(cs);
+
 public:
     indirectmap<COutPoint, const CTransaction*> mapNextTx GUARDED_BY(cs);
     std::map<uint256, CAmount> mapDeltas;
@@ -698,6 +701,21 @@ class CTxMemPool
 
     size_t DynamicMemoryUsage() const;
 
+    /** Adds a transaction to the unbroadcast set */
+    void AddUnbroadcastTx(const uint256& txid) {
+        LOCK(cs);
+        m_unbroadcast_txids.insert(txid);
+    }
+
+    /** Removes a transaction from the unbroadcast set */
+    void RemoveUnbroadcastTx(const uint256& txid, const bool unchecked = false);
+
+    /** Returns transactions in unbroadcast set */
+    const std::set<uint256> GetUnbroadcastTxs() const {
+        LOCK(cs);
+        return m_unbroadcast_txids;
+    }
+
 private:
     /** UpdateForDescendants is used by UpdateTransactionsFromBlock to update
      *  the descendants for a single transaction that has been added to the

Original file line number	Diff line number	Diff line change
`@@ -1556,7 +1556,7 @@ void static ProcessGetBlockData(CNode* pfrom, const CChainParams& chainparams, c`
`1556`	`1556`	`}`
`1557`	`1557`	`}`
`1558`	`1558`
`1559`		`-void static ProcessGetData(CNode* pfrom, const CChainParams& chainparams, CConnman* connman, const CTxMemPool& mempool, const std::atomic<bool>& interruptMsgProc) LOCKS_EXCLUDED(cs_main)`
	`1559`	`+void static ProcessGetData(CNode* pfrom, const CChainParams& chainparams, CConnman* connman, CTxMemPool& mempool, const std::atomic<bool>& interruptMsgProc) LOCKS_EXCLUDED(cs_main)`
`1560`	`1560`	`{`
`1561`	`1561`	`AssertLockNotHeld(cs_main);`
`1562`	`1562`
`@@ -1605,7 +1605,13 @@ void static ProcessGetData(CNode* pfrom, const CChainParams& chainparams, CConnm`
`1605`	`1605`	`push = true;`
`1606`	`1606`	`}`
`1607`	`1607`	`}`
`1608`		`- if (!push) {`
	`1608`	`+`
	`1609`	`+ if (push) {`
	`1610`	`+ // We interpret fulfilling a GETDATA for a transaction as a`
	`1611`	`+ // successful initial broadcast and remove it from our`
	`1612`	`+ // unbroadcast set.`
	`1613`	`+ mempool.RemoveUnbroadcastTx(inv.hash);`
	`1614`	`+ } else {`
`1609`	`1615`	`vNotFound.push_back(inv);`
`1610`	`1616`	`}`
`1611`	`1617`	`}`
Original file line number	Diff line number	Diff line change
`@@ -78,6 +78,10 @@ TransactionError BroadcastTransaction(NodeContext& node, const CTransactionRef t`
`78`	`78`	`}`
`79`	`79`
`80`	`80`	`if (relay) {`
	`81`	`+ // the mempool tracks locally submitted transactions to make a`
	`82`	`+ // best-effort of initial broadcast`
	`83`	`+ node.mempool->AddUnbroadcastTx(hashTx);`
	`84`	`+`
`81`	`85`	`RelayTransaction(hashTx, *node.connman);`
`82`	`86`	`}`
`83`	`87`