Skip to content

Commit

Permalink
Change in transaction pull scheduling to prevent InvBlock-related att…
Browse files Browse the repository at this point in the history 
…acks

Co-authored-by: Suhas Daftuar <sdaftuar@gmail.com>
  • Loading branch information
@naumenkogs @sdaftuar
naumenkogs and sdaftuar committed Feb 1, 2019
1 parent 6e6b3b9 commit 159111e
Show file tree
Hide file tree
Showing 6 changed files with 164 additions and 84 deletions.
36 changes: 0 additions & 36 deletions src/net.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -96,8 +96,6 @@ std::map<CNetAddr, LocalServiceInfo> mapLocalHost GUARDED_BY(cs_mapLocalHost);
static bool vfLimited[NET_MAX] GUARDED_BY(cs_mapLocalHost) = {};
std::string strSubVersion;

limitedmap<uint256, int64_t> mapAlreadyAskedFor(MAX_INV_SZ);

void CConnman::AddOneShot(const std::string& strDest)
{
LOCK(cs_vOneShots);
Expand Down Expand Up @@ -2641,40 +2639,6 @@ CNode::~CNode()
CloseSocket(hSocket);
}

void CNode::AskFor(const CInv& inv)
{
if (mapAskFor.size() > MAPASKFOR_MAX_SZ || setAskFor.size() > SETASKFOR_MAX_SZ)
return;
// a peer may not have multiple non-responded queue positions for a single inv item
if (!setAskFor.insert(inv.hash).second)
return;

// We're using mapAskFor as a priority queue,
// the key is the earliest time the request can be sent
int64_t nRequestTime;
limitedmap<uint256, int64_t>::const_iterator it = mapAlreadyAskedFor.find(inv.hash);
if (it != mapAlreadyAskedFor.end())
nRequestTime = it->second;
else
nRequestTime = 0;
LogPrint(BCLog::NET, "askfor %s %d (%s) peer=%d\n", inv.ToString(), nRequestTime, FormatISO8601Time(nRequestTime/1000000), id);

// Make sure not to reuse time indexes to keep things in the same order
int64_t nNow = GetTimeMicros() - 1000000;
static int64_t nLastTime;
++nLastTime;
nNow = std::max(nNow, nLastTime);
nLastTime = nNow;

// Each retry is 2 minutes after the last
nRequestTime = std::max(nRequestTime + 2 * 60 * 1000000, nNow);
if (it != mapAlreadyAskedFor.end())
mapAlreadyAskedFor.update(it, nRequestTime);
else
mapAlreadyAskedFor.insert(std::make_pair(inv.hash, nRequestTime));
mapAskFor.insert(std::make_pair(nRequestTime, inv));
}

bool CConnman::NodeFullyConnected(const CNode* pnode)
{
return pnode && pnode->fSuccessfullyConnected && !pnode->fDisconnect;
Expand Down
10 changes: 0 additions & 10 deletions src/net.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -67,10 +67,6 @@ static const bool DEFAULT_UPNP = USE_UPNP;
#else
static const bool DEFAULT_UPNP = false;
#endif
/** The maximum number of entries in mapAskFor */
static const size_t MAPASKFOR_MAX_SZ = MAX_INV_SZ;
/** The maximum number of entries in setAskFor (larger due to getdata latency)*/
static const size_t SETASKFOR_MAX_SZ = 2 * MAX_INV_SZ;
/** The maximum number of peer connections to maintain. */
static const unsigned int DEFAULT_MAX_PEER_CONNECTIONS = 125;
/** The default for -maxuploadtarget. 0 = Unlimited */
Expand Down Expand Up @@ -514,8 +510,6 @@ extern bool fDiscover;
extern bool fListen;
extern bool fRelayTxes;

extern limitedmap<uint256, int64_t> mapAlreadyAskedFor;

/** Subversion as sent to the P2P network in `version` messages */
extern std::string strSubVersion;

Expand Down Expand Up @@ -703,8 +697,6 @@ class CNode
// and in the order requested.
std::vector<uint256> vInventoryBlockToSend GUARDED_BY(cs_inventory);
CCriticalSection cs_inventory;
std::set<uint256> setAskFor;
std::multimap<int64_t, CInv> mapAskFor;
int64_t nNextInvSend{0};
// Used for headers announcements - unfiltered blocks to relay
std::vector<uint256> vBlockHashesToAnnounce GUARDED_BY(cs_inventory);
Expand Down Expand Up @@ -851,8 +843,6 @@ class CNode
vBlockHashesToAnnounce.push_back(hash);
}

void AskFor(const CInv& inv);

void CloseSocketDisconnect();

void copyStats(CNodeStats &stats);
Expand Down
185 changes: 164 additions & 21 deletions src/net_processing.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -64,6 +64,18 @@ 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.
* To preserve security, should not exceed INBOUND_PEER_TX_DELAY */
static constexpr int64_t MAX_GETDATA_RANDOM_DELAY = 2 * 1000000;


struct COrphanTx {
// When modifying, adapt the copy of this definition in tests/DoS_tests.
Expand Down Expand Up @@ -274,6 +286,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_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_in_flight;
};

TxDownloadState m_tx_download;

CNodeState(CAddress addrIn, std::string addrNameIn) : address(addrIn), name(addrNameIn) {
fCurrentlyConnected = false;
nMisbehavior = 0;
Expand Down Expand Up @@ -301,6 +373,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);

Expand Down Expand Up @@ -591,6 +666,57 @@ static void FindNextBlocksToDownload(NodeId nodeid, unsigned int count, std::vec
}
}

void EraseTxRequest(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) 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;
}

int64_t process_time;
int64_t last_request_time = GetTxRequestTime(txid);
// First time requesting this tx
if (last_request_time == 0) {
peer_download_state.m_tx_announced.insert(txid);
process_time = GetTimeMicros();
} 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
Expand Down Expand Up @@ -1570,7 +1696,7 @@ bool static ProcessHeadersMessage(CNode *pfrom, CConnman *connman, const std::ve
return true;
}

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)
bool static ProcessMessage(PeerLogicValidation *peer_logic, 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());
if (gArgs.IsArgSet("-dropmessagestest") && GetRand(gArgs.GetArg("-dropmessagestest", 0)) == 0)
Expand Down Expand Up @@ -1976,7 +2102,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);
}
}
}
Expand Down Expand Up @@ -2211,8 +2337,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_in_flight.erase(inv.hash);
EraseTxRequest(inv.hash);

std::list<CTransactionRef> lRemovedTxn;

Expand Down Expand Up @@ -2306,7 +2434,7 @@ bool static ProcessMessage(CNode* pfrom, const std::string& strCommand, CDataStr
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);
}
AddOrphanTx(ptx, pfrom->GetId());

Expand Down Expand Up @@ -2566,7 +2694,7 @@ bool static ProcessMessage(CNode* pfrom, const std::string& strCommand, CDataStr
} // cs_main

if (fProcessBLOCKTXN)
return ProcessMessage(pfrom, NetMsgType::BLOCKTXN, blockTxnMsg, nTimeReceived, chainparams, connman, interruptMsgProc, enable_bip61);
return ProcessMessage(peer_logic, pfrom, NetMsgType::BLOCKTXN, blockTxnMsg, nTimeReceived, chainparams, connman, interruptMsgProc, enable_bip61);

if (fRevertToHeaderProcessing) {
// Headers received from HB compact block peers are permitted to be
Expand Down Expand Up @@ -3052,7 +3180,7 @@ bool PeerLogicValidation::ProcessMessages(CNode* pfrom, std::atomic<bool>& inter
bool fRet = false;
try
{
fRet = ProcessMessage(pfrom, strCommand, vRecv, msg.nTime, chainparams, connman, interruptMsgProc, m_enable_bip61);
fRet = ProcessMessage(this, pfrom, strCommand, vRecv, msg.nTime, chainparams, connman, interruptMsgProc, m_enable_bip61);
if (interruptMsgProc)
return false;
if (!pfrom->vRecvGetData.empty())
Expand Down Expand Up @@ -3731,24 +3859,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();
std::multimap<int64_t, uint256>& 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_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() >= 1000) {
connman->PushMessage(pto, msgMaker.Make(NetMsgType::GETDATA, vGetData));
vGetData.clear();
}
UpdateTxRequestTime(inv.hash, nNow);
state.m_tx_download.m_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);
}
} 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_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));

Expand Down
5 changes: 0 additions & 5 deletions src/test/util_tests.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -169,11 +169,6 @@ BOOST_AUTO_TEST_CASE(util_FormatISO8601Date)
BOOST_CHECK_EQUAL(FormatISO8601Date(1317425777), "2011-09-30");
}

BOOST_AUTO_TEST_CASE(util_FormatISO8601Time)
{
BOOST_CHECK_EQUAL(FormatISO8601Time(1317425777), "23:36:17Z");
}

struct TestArgsManager : public ArgsManager
{
TestArgsManager() { m_network_only_args.clear(); }
Expand Down
11 changes: 0 additions & 11 deletions src/util/time.cpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -97,14 +97,3 @@ std::string FormatISO8601Date(int64_t nTime) {
#endif
return strprintf("%04i-%02i-%02i", ts.tm_year + 1900, ts.tm_mon + 1, ts.tm_mday);
}

std::string FormatISO8601Time(int64_t nTime) {
struct tm ts;
time_t time_val = nTime;
#ifdef _MSC_VER
gmtime_s(&ts, &time_val);
#else
gmtime_r(&time_val, &ts);
#endif
return strprintf("%02i:%02i:%02iZ", ts.tm_hour, ts.tm_min, ts.tm_sec);
}
1 change: 0 additions & 1 deletion src/util/time.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,5 @@ void MilliSleep(int64_t n);
*/
std::string FormatISO8601DateTime(int64_t nTime);
std::string FormatISO8601Date(int64_t nTime);
std::string FormatISO8601Time(int64_t nTime);

#endif // BITCOIN_UTIL_TIME_H

0 comments on commit 159111e

Please sign in to comment.