diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/interfaces/chain.cpp | 8 | ||||
| -rw-r--r-- | src/interfaces/chain.h | 13 | ||||
| -rw-r--r-- | src/net.h | 2 | ||||
| -rw-r--r-- | src/net_processing.cpp | 140 | ||||
| -rw-r--r-- | src/wallet/rpcdump.cpp | 18 | ||||
| -rw-r--r-- | src/wallet/wallet.cpp | 16 | ||||
| -rw-r--r-- | src/wallet/wallet.h | 2 |
7 files changed, 127 insertions, 72 deletions
diff --git a/src/interfaces/chain.cpp b/src/interfaces/chain.cpp index da810bc5e6..d3d9e86e8d 100644 --- a/src/interfaces/chain.cpp +++ b/src/interfaces/chain.cpp @@ -8,6 +8,7 @@ #include <chainparams.h> #include <primitives/block.h> #include <sync.h> +#include <txmempool.h> #include <uint256.h> #include <util/system.h> #include <validation.h> @@ -177,6 +178,13 @@ public: LOCK(cs_main); return GuessVerificationProgress(Params().TxData(), LookupBlockIndex(block_hash)); } + void requestMempoolTransactions(std::function<void(const CTransactionRef&)> fn) override + { + LOCK2(::cs_main, ::mempool.cs); + for (const CTxMemPoolEntry& entry : ::mempool.mapTx) { + fn(entry.GetSharedTx()); + } + } }; } // namespace diff --git a/src/interfaces/chain.h b/src/interfaces/chain.h index 3a54b9164e..57e9703758 100644 --- a/src/interfaces/chain.h +++ b/src/interfaces/chain.h @@ -16,6 +16,9 @@ class CBlock; class CScheduler; class uint256; struct CBlockLocator; +class CTransaction; + +using CTransactionRef = std::shared_ptr<const CTransaction>; namespace interfaces { @@ -127,6 +130,16 @@ public: //! Estimate fraction of total transactions verified if blocks up to //! the specified block hash are verified. virtual double guessVerificationProgress(const uint256& block_hash) = 0; + + //! Synchronously send TransactionAddedToMempool notifications about all + //! current mempool transactions to the specified handler and return after + //! the last one is sent. These notifications aren't coordinated with async + //! notifications sent by handleNotifications, so out of date async + //! notifications from handleNotifications can arrive during and after + //! synchronous notifications from requestMempoolTransactions. Clients need + //! to be prepared to handle this by ignoring notifications about unknown + //! removed transactions and already added new transactions. + virtual void requestMempoolTransactions(std::function<void(const CTransactionRef&)> fn) = 0; }; //! Interface to let node manage chain clients (wallets, or maybe tools for @@ -739,6 +739,8 @@ public: 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; diff --git a/src/net_processing.cpp b/src/net_processing.cpp index a37451e39a..53ff6a52ac 100644 --- a/src/net_processing.cpp +++ b/src/net_processing.cpp @@ -1713,6 +1713,67 @@ bool static ProcessHeadersMessage(CNode *pfrom, CConnman *connman, const std::ve return true; } +void static ProcessOrphanTx(CConnman* connman, std::set<uint256>& orphan_work_set, std::list<CTransactionRef>& removed_txn) EXCLUSIVE_LOCKS_REQUIRED(cs_main, g_cs_orphans) +{ + AssertLockHeld(cs_main); + AssertLockHeld(g_cs_orphans); + std::set<NodeId> setMisbehaving; + bool done = false; + while (!done && !orphan_work_set.empty()) { + const uint256 orphanHash = *orphan_work_set.begin(); + orphan_work_set.erase(orphan_work_set.begin()); + + auto orphan_it = mapOrphanTransactions.find(orphanHash); + if (orphan_it == mapOrphanTransactions.end()) continue; + + const CTransactionRef porphanTx = orphan_it->second.tx; + const CTransaction& orphanTx = *porphanTx; + NodeId fromPeer = orphan_it->second.fromPeer; + bool fMissingInputs2 = false; + // Use a dummy CValidationState so someone can't setup nodes to counter-DoS based on orphan + // resolution (that is, feeding people an invalid transaction based on LegitTxX in order to get + // anyone relaying LegitTxX banned) + CValidationState stateDummy; + + if (setMisbehaving.count(fromPeer)) continue; + if (AcceptToMemoryPool(mempool, stateDummy, porphanTx, &fMissingInputs2, &removed_txn, false /* bypass_limits */, 0 /* nAbsurdFee */)) { + LogPrint(BCLog::MEMPOOL, " accepted orphan tx %s\n", orphanHash.ToString()); + RelayTransaction(orphanTx, connman); + for (unsigned int i = 0; i < orphanTx.vout.size(); i++) { + auto it_by_prev = mapOrphanTransactionsByPrev.find(COutPoint(orphanHash, i)); + if (it_by_prev != mapOrphanTransactionsByPrev.end()) { + for (const auto& elem : it_by_prev->second) { + orphan_work_set.insert(elem->first); + } + } + } + EraseOrphanTx(orphanHash); + done = true; + } else if (!fMissingInputs2) { + int nDos = 0; + if (stateDummy.IsInvalid(nDos) && nDos > 0) { + // Punish peer that gave us an invalid orphan tx + Misbehaving(fromPeer, nDos); + setMisbehaving.insert(fromPeer); + LogPrint(BCLog::MEMPOOL, " invalid orphan tx %s\n", orphanHash.ToString()); + } + // Has inputs but not accepted to mempool + // Probably non-standard or insufficient fee + LogPrint(BCLog::MEMPOOL, " removed orphan tx %s\n", orphanHash.ToString()); + if (!orphanTx.HasWitness() && !stateDummy.CorruptionPossible()) { + // Do not use rejection cache for witness transactions or + // witness-stripped transactions, as they can have been malleated. + // See https://github.com/bitcoin/bitcoin/issues/8279 for details. + assert(recentRejects); + recentRejects->insert(orphanHash); + } + EraseOrphanTx(orphanHash); + done = true; + } + mempool.check(pcoinsTip.get()); + } +} + bool static ProcessMessage(CNode* pfrom, const std::string& strCommand, CDataStream& vRecv, int64_t nTimeReceived, const CChainParams& chainparams, CConnman* connman, const std::atomic<bool>& interruptMsgProc, bool enable_bip61) { LogPrint(BCLog::NET, "received: %s (%u bytes) peer=%d\n", SanitizeString(strCommand), vRecv.size(), pfrom->GetId()); @@ -2342,8 +2403,6 @@ bool static ProcessMessage(CNode* pfrom, const std::string& strCommand, CDataStr return true; } - std::deque<COutPoint> vWorkQueue; - std::vector<uint256> vEraseQueue; CTransactionRef ptx; vRecv >> ptx; const CTransaction& tx = *ptx; @@ -2368,7 +2427,12 @@ bool static ProcessMessage(CNode* pfrom, const std::string& strCommand, CDataStr mempool.check(pcoinsTip.get()); RelayTransaction(tx, connman); for (unsigned int i = 0; i < tx.vout.size(); i++) { - vWorkQueue.emplace_back(inv.hash, i); + auto it_by_prev = mapOrphanTransactionsByPrev.find(COutPoint(inv.hash, i)); + if (it_by_prev != mapOrphanTransactionsByPrev.end()) { + for (const auto& elem : it_by_prev->second) { + pfrom->orphan_work_set.insert(elem->first); + } + } } pfrom->nLastTXTime = GetTime(); @@ -2379,65 +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 - std::set<NodeId> setMisbehaving; - while (!vWorkQueue.empty()) { - auto itByPrev = mapOrphanTransactionsByPrev.find(vWorkQueue.front()); - vWorkQueue.pop_front(); - if (itByPrev == mapOrphanTransactionsByPrev.end()) - continue; - for (auto mi = itByPrev->second.begin(); - mi != itByPrev->second.end(); - ++mi) - { - const CTransactionRef& porphanTx = (*mi)->second.tx; - const CTransaction& orphanTx = *porphanTx; - const uint256& orphanHash = orphanTx.GetHash(); - NodeId fromPeer = (*mi)->second.fromPeer; - bool fMissingInputs2 = false; - // Use a dummy CValidationState so someone can't setup nodes to counter-DoS based on orphan - // resolution (that is, feeding people an invalid transaction based on LegitTxX in order to get - // anyone relaying LegitTxX banned) - CValidationState stateDummy; - - - if (setMisbehaving.count(fromPeer)) - continue; - if (AcceptToMemoryPool(mempool, stateDummy, porphanTx, &fMissingInputs2, &lRemovedTxn, false /* bypass_limits */, 0 /* nAbsurdFee */)) { - LogPrint(BCLog::MEMPOOL, " accepted orphan tx %s\n", orphanHash.ToString()); - RelayTransaction(orphanTx, connman); - for (unsigned int i = 0; i < orphanTx.vout.size(); i++) { - vWorkQueue.emplace_back(orphanHash, i); - } - vEraseQueue.push_back(orphanHash); - } - else if (!fMissingInputs2) - { - int nDos = 0; - if (stateDummy.IsInvalid(nDos) && nDos > 0) - { - // Punish peer that gave us an invalid orphan tx - Misbehaving(fromPeer, nDos); - setMisbehaving.insert(fromPeer); - LogPrint(BCLog::MEMPOOL, " invalid orphan tx %s\n", orphanHash.ToString()); - } - // Has inputs but not accepted to mempool - // Probably non-standard or insufficient fee - LogPrint(BCLog::MEMPOOL, " removed orphan tx %s\n", orphanHash.ToString()); - vEraseQueue.push_back(orphanHash); - if (!orphanTx.HasWitness() && !stateDummy.CorruptionPossible()) { - // Do not use rejection cache for witness transactions or - // witness-stripped transactions, as they can have been malleated. - // See https://github.com/bitcoin/bitcoin/issues/8279 for details. - assert(recentRejects); - recentRejects->insert(orphanHash); - } - } - mempool.check(pcoinsTip.get()); - } - } - - for (const uint256& hash : vEraseQueue) - EraseOrphanTx(hash); + ProcessOrphanTx(connman, pfrom->orphan_work_set, lRemovedTxn); } else if (fMissingInputs) { @@ -3145,11 +3151,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) diff --git a/src/wallet/rpcdump.cpp b/src/wallet/rpcdump.cpp index 5b0592e06d..bdf901dee4 100644 --- a/src/wallet/rpcdump.cpp +++ b/src/wallet/rpcdump.cpp @@ -346,7 +346,11 @@ UniValue importaddress(const JSONRPCRequest& request) if (fRescan) { RescanWallet(*pwallet, reserver); - pwallet->ReacceptWalletTransactions(); + { + auto locked_chain = pwallet->chain().lock(); + LOCK(pwallet->cs_wallet); + pwallet->ReacceptWalletTransactions(*locked_chain); + } } return NullUniValue; @@ -529,7 +533,11 @@ UniValue importpubkey(const JSONRPCRequest& request) if (fRescan) { RescanWallet(*pwallet, reserver); - pwallet->ReacceptWalletTransactions(); + { + auto locked_chain = pwallet->chain().lock(); + LOCK(pwallet->cs_wallet); + pwallet->ReacceptWalletTransactions(*locked_chain); + } } return NullUniValue; @@ -1464,7 +1472,11 @@ UniValue importmulti(const JSONRPCRequest& mainRequest) } if (fRescan && fRunScan && requests.size()) { int64_t scannedTime = pwallet->RescanFromTime(nLowestTimestamp, reserver, true /* update */); - pwallet->ReacceptWalletTransactions(); + { + auto locked_chain = pwallet->chain().lock(); + LOCK(pwallet->cs_wallet); + pwallet->ReacceptWalletTransactions(*locked_chain); + } if (pwallet->IsAbortingRescan()) { throw JSONRPCError(RPC_MISC_ERROR, "Rescan aborted by user."); diff --git a/src/wallet/wallet.cpp b/src/wallet/wallet.cpp index 388422bec8..b1b2a63166 100644 --- a/src/wallet/wallet.cpp +++ b/src/wallet/wallet.cpp @@ -1859,13 +1859,11 @@ CWallet::ScanResult CWallet::ScanForWalletTransactions(const uint256& start_bloc return result; } -void CWallet::ReacceptWalletTransactions() +void CWallet::ReacceptWalletTransactions(interfaces::Chain::Lock& locked_chain) { // If transactions aren't being broadcasted, don't let them into local mempool either if (!fBroadcastTransactions) return; - auto locked_chain = chain().lock(); - LOCK(cs_wallet); std::map<int64_t, CWalletTx*> mapSorted; // Sort pending wallet transactions based on their initial wallet insertion order @@ -1875,7 +1873,7 @@ void CWallet::ReacceptWalletTransactions() CWalletTx& wtx = item.second; assert(wtx.GetHash() == wtxid); - int nDepth = wtx.GetDepthInMainChain(*locked_chain); + int nDepth = wtx.GetDepthInMainChain(locked_chain); if (!wtx.IsCoinBase() && (nDepth == 0 && !wtx.isAbandoned())) { mapSorted.insert(std::make_pair(wtx.nOrderPos, &wtx)); @@ -1886,7 +1884,7 @@ void CWallet::ReacceptWalletTransactions() for (const std::pair<const int64_t, CWalletTx*>& item : mapSorted) { CWalletTx& wtx = *(item.second); CValidationState state; - wtx.AcceptToMemoryPool(*locked_chain, maxTxFee, state); + wtx.AcceptToMemoryPool(locked_chain, maxTxFee, state); } } @@ -4425,9 +4423,15 @@ std::shared_ptr<CWallet> CWallet::CreateWalletFromFile(interfaces::Chain& chain, void CWallet::postInitProcess() { + auto locked_chain = chain().lock(); + LOCK(cs_wallet); + // Add wallet transactions that aren't already in a block to mempool // Do this here as mempool requires genesis block to be loaded - ReacceptWalletTransactions(); + ReacceptWalletTransactions(*locked_chain); + + // Update wallet transactions with current mempool transactions. + chain().requestMempoolTransactions([this](const CTransactionRef& tx) { TransactionAddedToMempool(tx); }); } bool CWallet::BackupWallet(const std::string& strDest) diff --git a/src/wallet/wallet.h b/src/wallet/wallet.h index 2a5d6caaf8..9a4d252618 100644 --- a/src/wallet/wallet.h +++ b/src/wallet/wallet.h @@ -943,7 +943,7 @@ public: }; ScanResult ScanForWalletTransactions(const uint256& first_block, const uint256& last_block, const WalletRescanReserver& reserver, bool fUpdate); void TransactionRemovedFromMempool(const CTransactionRef &ptx) override; - void ReacceptWalletTransactions(); + void ReacceptWalletTransactions(interfaces::Chain::Lock& locked_chain) EXCLUSIVE_LOCKS_REQUIRED(cs_wallet); void ResendWalletTransactions(int64_t nBestBlockTime, CConnman* connman) override EXCLUSIVE_LOCKS_REQUIRED(cs_main); // ResendWalletTransactionsBefore may only be called if fBroadcastTransactions! std::vector<uint256> ResendWalletTransactionsBefore(interfaces::Chain::Lock& locked_chain, int64_t nTime, CConnman* connman); |