diff options
Diffstat (limited to 'src/net_processing.cpp')
-rw-r--r-- | src/net_processing.cpp | 213 |
1 files changed, 182 insertions, 31 deletions
diff --git a/src/net_processing.cpp b/src/net_processing.cpp index 0e222bdfa4..5927a14a6e 100644 --- a/src/net_processing.cpp +++ b/src/net_processing.cpp @@ -6,6 +6,7 @@ #include <net_processing.h> #include <addrman.h> +#include <banman.h> #include <arith_uint256.h> #include <blockencodings.h> #include <chainparams.h> @@ -63,6 +64,21 @@ static constexpr int STALE_RELAY_AGE_LIMIT = 30 * 24 * 60 * 60; /// Age after which a block is considered historical for purposes of rate /// limiting block relay. Set to one week, denominated in seconds. static constexpr int HISTORICAL_BLOCK_AGE = 7 * 24 * 60 * 60; +/** Maximum number of in-flight transactions from a peer */ +static constexpr int32_t MAX_PEER_TX_IN_FLIGHT = 100; +/** Maximum number of announced transactions from a peer */ +static constexpr int32_t MAX_PEER_TX_ANNOUNCEMENTS = 2 * MAX_INV_SZ; +/** How many microseconds to delay requesting transactions from inbound peers */ +static constexpr int64_t INBOUND_PEER_TX_DELAY = 2 * 1000000; +/** How long to wait (in microseconds) before downloading a transaction from an additional peer */ +static constexpr int64_t GETDATA_TX_INTERVAL = 60 * 1000000; +/** Maximum delay (in microseconds) for transaction requests to avoid biasing some peers over others. */ +static constexpr int64_t MAX_GETDATA_RANDOM_DELAY = 2 * 1000000; +static_assert(INBOUND_PEER_TX_DELAY >= MAX_GETDATA_RANDOM_DELAY, +"To preserve security, MAX_GETDATA_RANDOM_DELAY should not exceed INBOUND_PEER_DELAY"); +/** Limit to avoid sending big packets. Not used in processing incoming GETDATA for compatibility */ +static const unsigned int MAX_GETDATA_SZ = 1000; + struct COrphanTx { // When modifying, adapt the copy of this definition in tests/DoS_tests. @@ -273,6 +289,66 @@ struct CNodeState { //! Time of last new block announcement int64_t m_last_block_announcement; + /* + * State associated with transaction download. + * + * Tx download algorithm: + * + * When inv comes in, queue up (process_time, txid) inside the peer's + * CNodeState (m_tx_process_time) as long as m_tx_announced for the peer + * isn't too big (MAX_PEER_TX_ANNOUNCEMENTS). + * + * The process_time for a transaction is set to nNow for outbound peers, + * nNow + 2 seconds for inbound peers. This is the time at which we'll + * consider trying to request the transaction from the peer in + * SendMessages(). The delay for inbound peers is to allow outbound peers + * a chance to announce before we request from inbound peers, to prevent + * an adversary from using inbound connections to blind us to a + * transaction (InvBlock). + * + * When we call SendMessages() for a given peer, + * we will loop over the transactions in m_tx_process_time, looking + * at the transactions whose process_time <= nNow. We'll request each + * such transaction that we don't have already and that hasn't been + * requested from another peer recently, up until we hit the + * MAX_PEER_TX_IN_FLIGHT limit for the peer. Then we'll update + * g_already_asked_for for each requested txid, storing the time of the + * GETDATA request. We use g_already_asked_for to coordinate transaction + * requests amongst our peers. + * + * For transactions that we still need but we have already recently + * requested from some other peer, we'll reinsert (process_time, txid) + * back into the peer's m_tx_process_time at the point in the future at + * which the most recent GETDATA request would time out (ie + * GETDATA_TX_INTERVAL + the request time stored in g_already_asked_for). + * We add an additional delay for inbound peers, again to prefer + * attempting download from outbound peers first. + * We also add an extra small random delay up to 2 seconds + * to avoid biasing some peers over others. (e.g., due to fixed ordering + * of peer processing in ThreadMessageHandler). + * + * When we receive a transaction from a peer, we remove the txid from the + * peer's m_tx_in_flight set and from their recently announced set + * (m_tx_announced). We also clear g_already_asked_for for that entry, so + * that if somehow the transaction is not accepted but also not added to + * the reject filter, then we will eventually redownload from other + * peers. + */ + struct TxDownloadState { + /* Track when to attempt download of announced transactions (process + * time in micros -> txid) + */ + std::multimap<int64_t, uint256> m_tx_process_time; + + //! Store all the transactions a peer has recently announced + std::set<uint256> m_tx_announced; + + //! Store transactions which were requested by us + std::set<uint256> m_tx_in_flight; + }; + + TxDownloadState m_tx_download; + CNodeState(CAddress addrIn, std::string addrNameIn) : address(addrIn), name(addrNameIn) { fCurrentlyConnected = false; nMisbehavior = 0; @@ -300,6 +376,9 @@ struct CNodeState { } }; +// Keeps track of the time (in microseconds) when transactions were requested last time +limitedmap<uint256, int64_t> g_already_asked_for GUARDED_BY(cs_main)(MAX_INV_SZ); + /** Map maintaining per-node state. */ static std::map<NodeId, CNodeState> mapNodeState GUARDED_BY(cs_main); @@ -590,6 +669,58 @@ static void FindNextBlocksToDownload(NodeId nodeid, unsigned int count, std::vec } } +void EraseTxRequest(const uint256& txid) EXCLUSIVE_LOCKS_REQUIRED(cs_main) +{ + g_already_asked_for.erase(txid); +} + +int64_t GetTxRequestTime(const uint256& txid) EXCLUSIVE_LOCKS_REQUIRED(cs_main) +{ + auto it = g_already_asked_for.find(txid); + if (it != g_already_asked_for.end()) { + return it->second; + } + return 0; +} + +void UpdateTxRequestTime(const uint256& txid, int64_t request_time) EXCLUSIVE_LOCKS_REQUIRED(cs_main) +{ + auto it = g_already_asked_for.find(txid); + if (it == g_already_asked_for.end()) { + g_already_asked_for.insert(std::make_pair(txid, request_time)); + } else { + g_already_asked_for.update(it, request_time); + } +} + + +void RequestTx(CNodeState* state, const uint256& txid, int64_t nNow) EXCLUSIVE_LOCKS_REQUIRED(cs_main) +{ + CNodeState::TxDownloadState& peer_download_state = state->m_tx_download; + if (peer_download_state.m_tx_announced.size() >= MAX_PEER_TX_ANNOUNCEMENTS || peer_download_state.m_tx_announced.count(txid)) { + // Too many queued announcements from this peer, or we already have + // this announcement + return; + } + peer_download_state.m_tx_announced.insert(txid); + + int64_t process_time; + int64_t last_request_time = GetTxRequestTime(txid); + // First time requesting this tx + if (last_request_time == 0) { + process_time = nNow; + } else { + // Randomize the delay to avoid biasing some peers over others (such as due to + // fixed ordering of peer processing in ThreadMessageHandler) + process_time = last_request_time + GETDATA_TX_INTERVAL + GetRand(MAX_GETDATA_RANDOM_DELAY); + } + + // We delay processing announcements from non-preferred (eg inbound) peers + if (!state->fPreferredDownload) process_time += INBOUND_PEER_TX_DELAY; + + peer_download_state.m_tx_process_time.emplace(process_time, txid); +} + } // namespace // This function is used for testing the stale tip eviction logic, see @@ -841,9 +972,8 @@ static bool BlockRequestAllowed(const CBlockIndex* pindex, const Consensus::Para (GetBlockProofEquivalentTime(*pindexBestHeader, *pindex, *pindexBestHeader, consensusParams) < STALE_RELAY_AGE_LIMIT); } -PeerLogicValidation::PeerLogicValidation(CConnman* connmanIn, CScheduler &scheduler, bool enable_bip61) - : connman(connmanIn), m_stale_tip_check_time(0), m_enable_bip61(enable_bip61) { - +PeerLogicValidation::PeerLogicValidation(CConnman* connmanIn, BanMan* banman, CScheduler &scheduler, bool enable_bip61) + : connman(connmanIn), m_banman(banman), m_stale_tip_check_time(0), m_enable_bip61(enable_bip61) { // Initialize global variables that cannot be constructed at startup. recentRejects.reset(new CRollingBloomFilter(120000, 0.000001)); @@ -1945,6 +2075,7 @@ bool static ProcessMessage(CNode* pfrom, const std::string& strCommand, CDataStr LOCK(cs_main); uint32_t nFetchFlags = GetFetchFlags(pfrom); + int64_t nNow = GetTimeMicros(); for (CInv &inv : vInv) { @@ -1976,7 +2107,7 @@ bool static ProcessMessage(CNode* pfrom, const std::string& strCommand, CDataStr if (fBlocksOnly) { LogPrint(BCLog::NET, "transaction (%s) inv sent in violation of protocol peer=%d\n", inv.hash.ToString(), pfrom->GetId()); } else if (!fAlreadyHave && !fImporting && !fReindex && !IsInitialBlockDownload()) { - pfrom->AskFor(inv); + RequestTx(State(pfrom->GetId()), inv.hash, nNow); } } } @@ -2211,8 +2342,10 @@ bool static ProcessMessage(CNode* pfrom, const std::string& strCommand, CDataStr bool fMissingInputs = false; CValidationState state; - pfrom->setAskFor.erase(inv.hash); - mapAlreadyAskedFor.erase(inv.hash); + CNodeState* nodestate = State(pfrom->GetId()); + nodestate->m_tx_download.m_tx_announced.erase(inv.hash); + nodestate->m_tx_download.m_tx_in_flight.erase(inv.hash); + EraseTxRequest(inv.hash); std::list<CTransactionRef> lRemovedTxn; @@ -2303,10 +2436,12 @@ bool static ProcessMessage(CNode* pfrom, const std::string& strCommand, CDataStr } if (!fRejectedParents) { uint32_t nFetchFlags = GetFetchFlags(pfrom); + int64_t nNow = GetTimeMicros(); + for (const CTxIn& txin : tx.vin) { CInv _inv(MSG_TX | nFetchFlags, txin.prevout.hash); pfrom->AddInventoryKnown(_inv); - if (!AlreadyHave(_inv)) pfrom->AskFor(_inv); + if (!AlreadyHave(_inv)) RequestTx(State(pfrom->GetId()), _inv.hash, nNow); } AddOrphanTx(ptx, pfrom->GetId()); @@ -2943,7 +3078,7 @@ bool static ProcessMessage(CNode* pfrom, const std::string& strCommand, CDataStr return true; } -static bool SendRejectsAndCheckIfBanned(CNode* pnode, CConnman* connman, bool enable_bip61) EXCLUSIVE_LOCKS_REQUIRED(cs_main) +bool PeerLogicValidation::SendRejectsAndCheckIfBanned(CNode* pnode, bool enable_bip61) EXCLUSIVE_LOCKS_REQUIRED(cs_main) { AssertLockHeld(cs_main); CNodeState &state = *State(pnode->GetId()); @@ -2961,14 +3096,16 @@ static bool SendRejectsAndCheckIfBanned(CNode* pnode, CConnman* connman, bool en LogPrintf("Warning: not punishing whitelisted peer %s!\n", pnode->addr.ToString()); else if (pnode->m_manual_connection) LogPrintf("Warning: not punishing manually-connected peer %s!\n", pnode->addr.ToString()); - else { + else if (pnode->addr.IsLocal()) { + // Disconnect but don't ban _this_ local node + LogPrintf("Warning: disconnecting but not banning local peer %s!\n", pnode->addr.ToString()); pnode->fDisconnect = true; - if (pnode->addr.IsLocal()) - LogPrintf("Warning: not banning local peer %s!\n", pnode->addr.ToString()); - else - { - connman->Ban(pnode->addr, BanReasonNodeMisbehaving); + } else { + // Disconnect and ban all nodes sharing the address + if (m_banman) { + m_banman->Ban(pnode->addr, BanReasonNodeMisbehaving); } + connman->DisconnectNode(pnode->addr); } return true; } @@ -3092,7 +3229,7 @@ bool PeerLogicValidation::ProcessMessages(CNode* pfrom, std::atomic<bool>& inter } LOCK(cs_main); - SendRejectsAndCheckIfBanned(pfrom, connman, m_enable_bip61); + SendRejectsAndCheckIfBanned(pfrom, m_enable_bip61); return fMoreWork; } @@ -3293,8 +3430,7 @@ bool PeerLogicValidation::SendMessages(CNode* pto) if (!lockMain) return true; - if (SendRejectsAndCheckIfBanned(pto, connman, m_enable_bip61)) - return true; + if (SendRejectsAndCheckIfBanned(pto, m_enable_bip61)) return true; CNodeState &state = *State(pto->GetId()); // Address refresh broadcast @@ -3730,24 +3866,39 @@ bool PeerLogicValidation::SendMessages(CNode* pto) // // Message: getdata (non-blocks) // - while (!pto->mapAskFor.empty() && (*pto->mapAskFor.begin()).first <= nNow) - { - const CInv& inv = (*pto->mapAskFor.begin()).second; - if (!AlreadyHave(inv)) - { - LogPrint(BCLog::NET, "Requesting %s peer=%d\n", inv.ToString(), pto->GetId()); - vGetData.push_back(inv); - if (vGetData.size() >= 1000) - { - connman->PushMessage(pto, msgMaker.Make(NetMsgType::GETDATA, vGetData)); - vGetData.clear(); + auto& tx_process_time = state.m_tx_download.m_tx_process_time; + while (!tx_process_time.empty() && tx_process_time.begin()->first <= nNow && state.m_tx_download.m_tx_in_flight.size() < MAX_PEER_TX_IN_FLIGHT) { + const uint256& txid = tx_process_time.begin()->second; + CInv inv(MSG_TX | GetFetchFlags(pto), txid); + if (!AlreadyHave(inv)) { + // If this transaction was last requested more than 1 minute ago, + // then request. + int64_t last_request_time = GetTxRequestTime(inv.hash); + if (last_request_time <= nNow - GETDATA_TX_INTERVAL) { + LogPrint(BCLog::NET, "Requesting %s peer=%d\n", inv.ToString(), pto->GetId()); + vGetData.push_back(inv); + if (vGetData.size() >= MAX_GETDATA_SZ) { + connman->PushMessage(pto, msgMaker.Make(NetMsgType::GETDATA, vGetData)); + vGetData.clear(); + } + UpdateTxRequestTime(inv.hash, nNow); + state.m_tx_download.m_tx_in_flight.insert(inv.hash); + } else { + // This transaction is in flight from someone else; queue + // up processing to happen after the download times out + // (with a slight delay for inbound peers, to prefer + // requests to outbound peers). + RequestTx(&state, txid, nNow); } } else { - //If we're not going to ask, don't expect a response. - pto->setAskFor.erase(inv.hash); + // We have already seen this transaction, no need to download. + state.m_tx_download.m_tx_announced.erase(inv.hash); + state.m_tx_download.m_tx_in_flight.erase(inv.hash); } - pto->mapAskFor.erase(pto->mapAskFor.begin()); + tx_process_time.erase(tx_process_time.begin()); } + + if (!vGetData.empty()) connman->PushMessage(pto, msgMaker.Make(NetMsgType::GETDATA, vGetData)); |