diff options
Diffstat (limited to 'src/net.cpp')
| -rw-r--r-- | src/net.cpp | 2024 |
1 files changed, 1103 insertions, 921 deletions
diff --git a/src/net.cpp b/src/net.cpp index e5659efc0..de5fc2969 100644 --- a/src/net.cpp +++ b/src/net.cpp @@ -1,5 +1,5 @@ // Copyright (c) 2009-2010 Satoshi Nakamoto -// Copyright (c) 2009-2014 The Bitcoin Core developers +// Copyright (c) 2009-2016 The Bitcoin Core developers // Distributed under the MIT software license, see the accompanying // file COPYING or http://www.opensource.org/licenses/mit-license.php. @@ -14,8 +14,10 @@ #include "clientversion.h" #include "consensus/consensus.h" #include "crypto/common.h" +#include "crypto/sha256.h" #include "hash.h" #include "primitives/transaction.h" +#include "netbase.h" #include "scheduler.h" #include "ui_interface.h" #include "utilstrencodings.h" @@ -33,12 +35,15 @@ #include <miniupnpc/upnperrors.h> #endif -#include <boost/filesystem.hpp> -#include <boost/thread.hpp> -// Dump addresses to peers.dat every 15 minutes (900s) +#include <math.h> + +// Dump addresses to peers.dat and banlist.dat every 15 minutes (900s) #define DUMP_ADDRESSES_INTERVAL 900 +// We add a random period time (0 to 1 seconds) to feeler connections to prevent synchronization. +#define FEELER_SLEEP_WINDOW 1 + #if !defined(HAVE_MSG_NOSIGNAL) && !defined(MSG_NOSIGNAL) #define MSG_NOSIGNAL 0 #endif @@ -54,64 +59,28 @@ #endif #endif -using namespace std; - -namespace { - const int MAX_OUTBOUND_CONNECTIONS = 8; - - struct ListenSocket { - SOCKET socket; - bool whitelisted; - - ListenSocket(SOCKET socket, bool whitelisted) : socket(socket), whitelisted(whitelisted) {} - }; -} +const static std::string NET_MESSAGE_COMMAND_OTHER = "*other*"; +static const uint64_t RANDOMIZER_ID_NETGROUP = 0x6c0edd8036ef4036ULL; // SHA256("netgroup")[0:8] +static const uint64_t RANDOMIZER_ID_LOCALHOSTNONCE = 0xd93e69e2bbfa5735ULL; // SHA256("localhostnonce")[0:8] // // Global state variables // bool fDiscover = true; bool fListen = true; -uint64_t nLocalServices = NODE_NETWORK; +bool fRelayTxes = true; CCriticalSection cs_mapLocalHost; -map<CNetAddr, LocalServiceInfo> mapLocalHost; -static bool vfReachable[NET_MAX] = {}; +std::map<CNetAddr, LocalServiceInfo> mapLocalHost; static bool vfLimited[NET_MAX] = {}; -static CNode* pnodeLocalHost = NULL; -uint64_t nLocalHostNonce = 0; -static std::vector<ListenSocket> vhListenSocket; -CAddrMan addrman; -int nMaxConnections = DEFAULT_MAX_PEER_CONNECTIONS; -bool fAddressesInitialized = false; std::string strSubVersion; -vector<CNode*> vNodes; -CCriticalSection cs_vNodes; -map<CInv, CDataStream> mapRelay; -deque<pair<int64_t, CInv> > vRelayExpiration; -CCriticalSection cs_mapRelay; -limitedmap<CInv, int64_t> mapAlreadyAskedFor(MAX_INV_SZ); - -static deque<string> vOneShots; -CCriticalSection cs_vOneShots; - -set<CNetAddr> setservAddNodeAddresses; -CCriticalSection cs_setservAddNodeAddresses; - -vector<std::string> vAddedNodes; -CCriticalSection cs_vAddedNodes; - -NodeId nLastNodeId = 0; -CCriticalSection cs_nLastNodeId; - -static CSemaphore *semOutbound = NULL; -boost::condition_variable messageHandlerCondition; +limitedmap<uint256, int64_t> mapAlreadyAskedFor(MAX_INV_SZ); // Signals for message handling static CNodeSignals g_signals; CNodeSignals& GetNodeSignals() { return g_signals; } -void AddOneShot(const std::string& strDest) +void CConnman::AddOneShot(const std::string& strDest) { LOCK(cs_vOneShots); vOneShots.push_back(strDest); @@ -132,7 +101,7 @@ bool GetLocal(CService& addr, const CNetAddr *paddrPeer) int nBestReachability = -1; { LOCK(cs_mapLocalHost); - for (map<CNetAddr, LocalServiceInfo>::iterator it = mapLocalHost.begin(); it != mapLocalHost.end(); it++) + for (std::map<CNetAddr, LocalServiceInfo>::iterator it = mapLocalHost.begin(); it != mapLocalHost.end(); it++) { int nScore = (*it).second.nScore; int nReachability = (*it).first.GetReachabilityFrom(paddrPeer); @@ -161,7 +130,7 @@ static std::vector<CAddress> convertSeed6(const std::vector<SeedSpec6> &vSeedsIn { struct in6_addr ip; memcpy(&ip, i->addr, sizeof(ip)); - CAddress addr(CService(ip, i->port)); + CAddress addr(CService(ip, i->port), NODE_NETWORK); addr.nTime = GetTime() - GetRand(nOneWeek) - nOneWeek; vSeedsOut.push_back(addr); } @@ -172,15 +141,14 @@ static std::vector<CAddress> convertSeed6(const std::vector<SeedSpec6> &vSeedsIn // Otherwise, return the unroutable 0.0.0.0 but filled in with // the normal parameters, since the IP may be changed to a useful // one by discovery. -CAddress GetLocalAddress(const CNetAddr *paddrPeer) +CAddress GetLocalAddress(const CNetAddr *paddrPeer, ServiceFlags nLocalServices) { - CAddress ret(CService("0.0.0.0",GetListenPort()),0); + CAddress ret(CService(CNetAddr(),GetListenPort()), NODE_NONE); CService addr; if (GetLocal(addr, paddrPeer)) { - ret = CAddress(addr); + ret = CAddress(addr, nLocalServices); } - ret.nServices = nLocalServices; ret.nTime = GetAdjustedTime(); return ret; } @@ -196,40 +164,34 @@ int GetnScore(const CService& addr) // Is our peer's addrLocal potentially useful as an external IP source? bool IsPeerAddrLocalGood(CNode *pnode) { - return fDiscover && pnode->addr.IsRoutable() && pnode->addrLocal.IsRoutable() && - !IsLimited(pnode->addrLocal.GetNetwork()); + CService addrLocal = pnode->GetAddrLocal(); + return fDiscover && pnode->addr.IsRoutable() && addrLocal.IsRoutable() && + !IsLimited(addrLocal.GetNetwork()); } // pushes our own address to a peer -void AdvertizeLocal(CNode *pnode) +void AdvertiseLocal(CNode *pnode) { if (fListen && pnode->fSuccessfullyConnected) { - CAddress addrLocal = GetLocalAddress(&pnode->addr); + CAddress addrLocal = GetLocalAddress(&pnode->addr, pnode->GetLocalServices()); // If discovery is enabled, sometimes give our peer the address it // tells us that it sees us as in case it has a better idea of our // address than we do. if (IsPeerAddrLocalGood(pnode) && (!addrLocal.IsRoutable() || GetRand((GetnScore(addrLocal) > LOCAL_MANUAL) ? 8:2) == 0)) { - addrLocal.SetIP(pnode->addrLocal); + addrLocal.SetIP(pnode->GetAddrLocal()); } if (addrLocal.IsRoutable()) { - LogPrintf("AdvertizeLocal: advertizing address %s\n", addrLocal.ToString()); - pnode->PushAddress(addrLocal); + LogPrint("net", "AdvertiseLocal: advertising address %s\n", addrLocal.ToString()); + FastRandomContext insecure_rand; + pnode->PushAddress(addrLocal, insecure_rand); } } } -void SetReachable(enum Network net, bool fFlag) -{ - LOCK(cs_mapLocalHost); - vfReachable[net] = fFlag; - if (net == NET_IPV6 && fFlag) - vfReachable[NET_IPV4] = true; -} - // learn a new local address bool AddLocal(const CService& addr, int nScore) { @@ -252,7 +214,6 @@ bool AddLocal(const CService& addr, int nScore) info.nScore = nScore + (fAlready ? 1 : 0); info.nPort = addr.GetPort(); } - SetReachable(addr.GetNetwork()); } return true; @@ -315,7 +276,7 @@ bool IsLocal(const CService& addr) bool IsReachable(enum Network net) { LOCK(cs_mapLocalHost); - return vfReachable[net] && !vfLimited[net]; + return !vfLimited[net]; } /** check whether a given address is in a network we can probably connect to */ @@ -325,23 +286,8 @@ bool IsReachable(const CNetAddr& addr) return IsReachable(net); } -void AddressCurrentlyConnected(const CService& addr) -{ - addrman.Connected(addr); -} - -uint64_t CNode::nTotalBytesRecv = 0; -uint64_t CNode::nTotalBytesSent = 0; -CCriticalSection CNode::cs_totalBytesRecv; -CCriticalSection CNode::cs_totalBytesSent; - -uint64_t CNode::nMaxOutboundLimit = 0; -uint64_t CNode::nMaxOutboundTotalBytesSentInCycle = 0; -uint64_t CNode::nMaxOutboundTimeframe = 60*60*24; //1 day -uint64_t CNode::nMaxOutboundCycleStartTime = 0; - -CNode* FindNode(const CNetAddr& ip) +CNode* CConnman::FindNode(const CNetAddr& ip) { LOCK(cs_vNodes); BOOST_FOREACH(CNode* pnode, vNodes) @@ -350,7 +296,7 @@ CNode* FindNode(const CNetAddr& ip) return NULL; } -CNode* FindNode(const CSubNet& subNet) +CNode* CConnman::FindNode(const CSubNet& subNet) { LOCK(cs_vNodes); BOOST_FOREACH(CNode* pnode, vNodes) @@ -359,16 +305,18 @@ CNode* FindNode(const CSubNet& subNet) return NULL; } -CNode* FindNode(const std::string& addrName) +CNode* CConnman::FindNode(const std::string& addrName) { LOCK(cs_vNodes); - BOOST_FOREACH(CNode* pnode, vNodes) - if (pnode->addrName == addrName) + BOOST_FOREACH(CNode* pnode, vNodes) { + if (pnode->GetAddrName() == addrName) { return (pnode); + } + } return NULL; } -CNode* FindNode(const CService& addr) +CNode* CConnman::FindNode(const CService& addr) { LOCK(cs_vNodes); BOOST_FOREACH(CNode* pnode, vNodes) @@ -377,7 +325,17 @@ CNode* FindNode(const CService& addr) return NULL; } -CNode* ConnectNode(CAddress addrConnect, const char *pszDest) +bool CConnman::CheckIncomingNonce(uint64_t nonce) +{ + LOCK(cs_vNodes); + BOOST_FOREACH(CNode* pnode, vNodes) { + if (!pnode->fSuccessfullyConnected && !pnode->fInbound && pnode->GetLocalNonce() == nonce) + return false; + } + return true; +} + +CNode* CConnman::ConnectNode(CAddress addrConnect, const char *pszDest, bool fCountFailure) { if (pszDest == NULL) { if (IsLocal(addrConnect)) @@ -387,8 +345,8 @@ CNode* ConnectNode(CAddress addrConnect, const char *pszDest) CNode* pnode = FindNode((CService)addrConnect); if (pnode) { - pnode->AddRef(); - return pnode; + LogPrintf("Failed to open new connection, already connected\n"); + return NULL; } } @@ -409,76 +367,85 @@ CNode* ConnectNode(CAddress addrConnect, const char *pszDest) return NULL; } - addrman.Attempt(addrConnect); - - // Add node - CNode* pnode = new CNode(hSocket, addrConnect, pszDest ? pszDest : "", false); - pnode->AddRef(); - - { + if (pszDest && addrConnect.IsValid()) { + // It is possible that we already have a connection to the IP/port pszDest resolved to. + // In that case, drop the connection that was just created, and return the existing CNode instead. + // Also store the name we used to connect in that CNode, so that future FindNode() calls to that + // name catch this early. LOCK(cs_vNodes); - vNodes.push_back(pnode); + CNode* pnode = FindNode((CService)addrConnect); + if (pnode) + { + pnode->MaybeSetAddrName(std::string(pszDest)); + CloseSocket(hSocket); + LogPrintf("Failed to open new connection, already connected\n"); + return NULL; + } } - pnode->nTimeConnected = GetTime(); + addrman.Attempt(addrConnect, fCountFailure); + + // Add node + NodeId id = GetNewNodeId(); + uint64_t nonce = GetDeterministicRandomizer(RANDOMIZER_ID_LOCALHOSTNONCE).Write(id).Finalize(); + CNode* pnode = new CNode(id, nLocalServices, GetBestHeight(), hSocket, addrConnect, CalculateKeyedNetGroup(addrConnect), nonce, pszDest ? pszDest : "", false); + pnode->nServicesExpected = ServiceFlags(addrConnect.nServices & nRelevantServices); + pnode->AddRef(); return pnode; } else if (!proxyConnectionFailed) { // If connecting to the node failed, and failure is not caused by a problem connecting to // the proxy, mark this as an attempt. - addrman.Attempt(addrConnect); + addrman.Attempt(addrConnect, fCountFailure); } return NULL; } +void CConnman::DumpBanlist() +{ + SweepBanned(); // clean unused entries (if bantime has expired) + + if (!BannedSetIsDirty()) + return; + + int64_t nStart = GetTimeMillis(); + + CBanDB bandb; + banmap_t banmap; + SetBannedSetDirty(false); + GetBanned(banmap); + if (!bandb.Write(banmap)) + SetBannedSetDirty(true); + + LogPrint("net", "Flushed %d banned node ips/subnets to banlist.dat %dms\n", + banmap.size(), GetTimeMillis() - nStart); +} + void CNode::CloseSocketDisconnect() { fDisconnect = true; + LOCK(cs_hSocket); if (hSocket != INVALID_SOCKET) { LogPrint("net", "disconnecting peer=%d\n", id); CloseSocket(hSocket); } - - // in case this fails, we'll empty the recv buffer when the CNode is deleted - TRY_LOCK(cs_vRecvMsg, lockRecv); - if (lockRecv) - vRecvMsg.clear(); } -void CNode::PushVersion() +void CConnman::ClearBanned() { - int nBestHeight = g_signals.GetHeight().get_value_or(0); - - int64_t nTime = (fInbound ? GetAdjustedTime() : GetTime()); - CAddress addrYou = (addr.IsRoutable() && !IsProxy(addr) ? addr : CAddress(CService("0.0.0.0",0))); - CAddress addrMe = GetLocalAddress(&addr); - GetRandBytes((unsigned char*)&nLocalHostNonce, sizeof(nLocalHostNonce)); - if (fLogIPs) - LogPrint("net", "send version message: version %d, blocks=%d, us=%s, them=%s, peer=%d\n", PROTOCOL_VERSION, nBestHeight, addrMe.ToString(), addrYou.ToString(), id); - else - LogPrint("net", "send version message: version %d, blocks=%d, us=%s, peer=%d\n", PROTOCOL_VERSION, nBestHeight, addrMe.ToString(), id); - PushMessage("version", PROTOCOL_VERSION, nLocalServices, nTime, addrYou, addrMe, - nLocalHostNonce, strSubVersion, nBestHeight, !GetBoolArg("-blocksonly", DEFAULT_BLOCKSONLY)); -} - - - - - -banmap_t CNode::setBanned; -CCriticalSection CNode::cs_setBanned; -bool CNode::setBannedIsDirty; - -void CNode::ClearBanned() -{ - LOCK(cs_setBanned); - setBanned.clear(); - setBannedIsDirty = true; + { + LOCK(cs_setBanned); + setBanned.clear(); + setBannedIsDirty = true; + } + DumpBanlist(); //store banlist to disk + if(clientInterface) + clientInterface->BannedListChanged(); } -bool CNode::IsBanned(CNetAddr ip) +bool CConnman::IsBanned(CNetAddr ip) { bool fResult = false; { @@ -495,7 +462,7 @@ bool CNode::IsBanned(CNetAddr ip) return fResult; } -bool CNode::IsBanned(CSubNet subnet) +bool CConnman::IsBanned(CSubNet subnet) { bool fResult = false; { @@ -511,12 +478,12 @@ bool CNode::IsBanned(CSubNet subnet) return fResult; } -void CNode::Ban(const CNetAddr& addr, const BanReason &banReason, int64_t bantimeoffset, bool sinceUnixEpoch) { +void CConnman::Ban(const CNetAddr& addr, const BanReason &banReason, int64_t bantimeoffset, bool sinceUnixEpoch) { CSubNet subNet(addr); Ban(subNet, banReason, bantimeoffset, sinceUnixEpoch); } -void CNode::Ban(const CSubNet& subNet, const BanReason &banReason, int64_t bantimeoffset, bool sinceUnixEpoch) { +void CConnman::Ban(const CSubNet& subNet, const BanReason &banReason, int64_t bantimeoffset, bool sinceUnixEpoch) { CBanEntry banEntry(GetTime()); banEntry.banReason = banReason; if (bantimeoffset <= 0) @@ -526,42 +493,60 @@ void CNode::Ban(const CSubNet& subNet, const BanReason &banReason, int64_t banti } banEntry.nBanUntil = (sinceUnixEpoch ? 0 : GetTime() )+bantimeoffset; - LOCK(cs_setBanned); - if (setBanned[subNet].nBanUntil < banEntry.nBanUntil) - setBanned[subNet] = banEntry; - - setBannedIsDirty = true; + { + LOCK(cs_setBanned); + if (setBanned[subNet].nBanUntil < banEntry.nBanUntil) { + setBanned[subNet] = banEntry; + setBannedIsDirty = true; + } + else + return; + } + if(clientInterface) + clientInterface->BannedListChanged(); + { + LOCK(cs_vNodes); + BOOST_FOREACH(CNode* pnode, vNodes) { + if (subNet.Match((CNetAddr)pnode->addr)) + pnode->fDisconnect = true; + } + } + if(banReason == BanReasonManuallyAdded) + DumpBanlist(); //store banlist to disk immediately if user requested ban } -bool CNode::Unban(const CNetAddr &addr) { +bool CConnman::Unban(const CNetAddr &addr) { CSubNet subNet(addr); return Unban(subNet); } -bool CNode::Unban(const CSubNet &subNet) { - LOCK(cs_setBanned); - if (setBanned.erase(subNet)) +bool CConnman::Unban(const CSubNet &subNet) { { + LOCK(cs_setBanned); + if (!setBanned.erase(subNet)) + return false; setBannedIsDirty = true; - return true; } - return false; + if(clientInterface) + clientInterface->BannedListChanged(); + DumpBanlist(); //store banlist to disk immediately + return true; } -void CNode::GetBanned(banmap_t &banMap) +void CConnman::GetBanned(banmap_t &banMap) { LOCK(cs_setBanned); banMap = setBanned; //create a thread safe copy } -void CNode::SetBanned(const banmap_t &banMap) +void CConnman::SetBanned(const banmap_t &banMap) { LOCK(cs_setBanned); setBanned = banMap; setBannedIsDirty = true; } -void CNode::SweepBanned() +void CConnman::SweepBanned() { int64_t now = GetTime(); @@ -569,34 +554,33 @@ void CNode::SweepBanned() banmap_t::iterator it = setBanned.begin(); while(it != setBanned.end()) { + CSubNet subNet = (*it).first; CBanEntry banEntry = (*it).second; if(now > banEntry.nBanUntil) { setBanned.erase(it++); setBannedIsDirty = true; + LogPrint("net", "%s: Removed banned node ip/subnet from banlist.dat: %s\n", __func__, subNet.ToString()); } else ++it; } } -bool CNode::BannedSetIsDirty() +bool CConnman::BannedSetIsDirty() { LOCK(cs_setBanned); return setBannedIsDirty; } -void CNode::SetBannedSetDirty(bool dirty) +void CConnman::SetBannedSetDirty(bool dirty) { LOCK(cs_setBanned); //reuse setBanned lock for the isDirty flag setBannedIsDirty = dirty; } -std::vector<CSubNet> CNode::vWhitelistedRange; -CCriticalSection CNode::cs_vWhitelistedRange; - -bool CNode::IsWhitelistedRange(const CNetAddr &addr) { +bool CConnman::IsWhitelistedRange(const CNetAddr &addr) { LOCK(cs_vWhitelistedRange); BOOST_FOREACH(const CSubNet& subnet, vWhitelistedRange) { if (subnet.Match(addr)) @@ -605,29 +589,72 @@ bool CNode::IsWhitelistedRange(const CNetAddr &addr) { return false; } -void CNode::AddWhitelistedRange(const CSubNet &subnet) { +void CConnman::AddWhitelistedRange(const CSubNet &subnet) { LOCK(cs_vWhitelistedRange); vWhitelistedRange.push_back(subnet); } + +std::string CNode::GetAddrName() const { + LOCK(cs_addrName); + return addrName; +} + +void CNode::MaybeSetAddrName(const std::string& addrNameIn) { + LOCK(cs_addrName); + if (addrName.empty()) { + addrName = addrNameIn; + } +} + +CService CNode::GetAddrLocal() const { + LOCK(cs_addrLocal); + return addrLocal; +} + +void CNode::SetAddrLocal(const CService& addrLocalIn) { + LOCK(cs_addrLocal); + if (addrLocal.IsValid()) { + error("Addr local already set for node: %i. Refusing to change from %s to %s", id, addrLocal.ToString(), addrLocalIn.ToString()); + } else { + addrLocal = addrLocalIn; + } +} + #undef X #define X(name) stats.name = name void CNode::copyStats(CNodeStats &stats) { stats.nodeid = this->GetId(); X(nServices); - X(fRelayTxes); + X(addr); + { + LOCK(cs_filter); + X(fRelayTxes); + } X(nLastSend); X(nLastRecv); X(nTimeConnected); X(nTimeOffset); - X(addrName); + stats.addrName = GetAddrName(); X(nVersion); - X(cleanSubVer); + { + LOCK(cs_SubVer); + X(cleanSubVer); + } X(fInbound); + X(fAddnode); X(nStartingHeight); - X(nSendBytes); - X(nRecvBytes); + { + LOCK(cs_vSend); + X(mapSendBytesPerMsgCmd); + X(nSendBytes); + } + { + LOCK(cs_vRecv); + X(mapRecvBytesPerMsgCmd); + X(nRecvBytes); + } X(fWhitelisted); // It is common for nodes with good ping times to suddenly become lagged, @@ -643,23 +670,28 @@ void CNode::copyStats(CNodeStats &stats) // Raw ping time is in microseconds, but show it to user as whole seconds (Bitcoin users should be well used to small numbers with many decimal places by now :) stats.dPingTime = (((double)nPingUsecTime) / 1e6); - stats.dPingMin = (((double)nMinPingUsecTime) / 1e6); + stats.dMinPing = (((double)nMinPingUsecTime) / 1e6); stats.dPingWait = (((double)nPingUsecWait) / 1e6); // Leave string empty if addrLocal invalid (not filled in yet) - stats.addrLocal = addrLocal.IsValid() ? addrLocal.ToString() : ""; + CService addrLocalUnlocked = GetAddrLocal(); + stats.addrLocal = addrLocalUnlocked.IsValid() ? addrLocalUnlocked.ToString() : ""; } #undef X -// requires LOCK(cs_vRecvMsg) -bool CNode::ReceiveMsgBytes(const char *pch, unsigned int nBytes) +bool CNode::ReceiveMsgBytes(const char *pch, unsigned int nBytes, bool& complete) { + complete = false; + int64_t nTimeMicros = GetTimeMicros(); + LOCK(cs_vRecv); + nLastRecv = nTimeMicros / 1000000; + nRecvBytes += nBytes; while (nBytes > 0) { // get current incomplete message, or create a new one if (vRecvMsg.empty() || vRecvMsg.back().complete()) - vRecvMsg.push_back(CNetMessage(Params().MessageStart(), SER_NETWORK, nRecvVersion)); + vRecvMsg.push_back(CNetMessage(Params().MessageStart(), SER_NETWORK, INIT_PROTO_VERSION)); CNetMessage& msg = vRecvMsg.back(); @@ -682,14 +714,50 @@ bool CNode::ReceiveMsgBytes(const char *pch, unsigned int nBytes) nBytes -= handled; if (msg.complete()) { - msg.nTime = GetTimeMicros(); - messageHandlerCondition.notify_one(); + + //store received bytes per message command + //to prevent a memory DOS, only allow valid commands + mapMsgCmdSize::iterator i = mapRecvBytesPerMsgCmd.find(msg.hdr.pchCommand); + if (i == mapRecvBytesPerMsgCmd.end()) + i = mapRecvBytesPerMsgCmd.find(NET_MESSAGE_COMMAND_OTHER); + assert(i != mapRecvBytesPerMsgCmd.end()); + i->second += msg.hdr.nMessageSize + CMessageHeader::HEADER_SIZE; + + msg.nTime = nTimeMicros; + complete = true; } } return true; } +void CNode::SetSendVersion(int nVersionIn) +{ + // Send version may only be changed in the version message, and + // only one version message is allowed per session. We can therefore + // treat this value as const and even atomic as long as it's only used + // once a version message has been successfully processed. Any attempt to + // set this twice is an error. + if (nSendVersion != 0) { + error("Send version already set for node: %i. Refusing to change from %i to %i", id, nSendVersion, nVersionIn); + } else { + nSendVersion = nVersionIn; + } +} + +int CNode::GetSendVersion() const +{ + // The send version should always be explicitly set to + // INIT_PROTO_VERSION rather than using this value until SetSendVersion + // has been called. + if (nSendVersion == 0) { + error("Requesting unset send version for node: %i. Using %i", id, INIT_PROTO_VERSION); + return INIT_PROTO_VERSION; + } + return nSendVersion; +} + + int CNetMessage::readHeader(const char *pch, unsigned int nBytes) { // copy data to temporary parsing buffer @@ -731,12 +799,21 @@ int CNetMessage::readData(const char *pch, unsigned int nBytes) vRecv.resize(std::min(hdr.nMessageSize, nDataPos + nCopy + 256 * 1024)); } + hasher.Write((const unsigned char*)pch, nCopy); memcpy(&vRecv[nDataPos], pch, nCopy); nDataPos += nCopy; return nCopy; } +const uint256& CNetMessage::GetMessageHash() const +{ + assert(complete()); + if (data_hash.IsNull()) + hasher.Finalize(data_hash.begin()); + return data_hash; +} + @@ -746,22 +823,30 @@ int CNetMessage::readData(const char *pch, unsigned int nBytes) // requires LOCK(cs_vSend) -void SocketSendData(CNode *pnode) +size_t CConnman::SocketSendData(CNode *pnode) const { - std::deque<CSerializeData>::iterator it = pnode->vSendMsg.begin(); + auto it = pnode->vSendMsg.begin(); + size_t nSentSize = 0; while (it != pnode->vSendMsg.end()) { - const CSerializeData &data = *it; + const auto &data = *it; assert(data.size() > pnode->nSendOffset); - int nBytes = send(pnode->hSocket, &data[pnode->nSendOffset], data.size() - pnode->nSendOffset, MSG_NOSIGNAL | MSG_DONTWAIT); + int nBytes = 0; + { + LOCK(pnode->cs_hSocket); + if (pnode->hSocket == INVALID_SOCKET) + break; + nBytes = send(pnode->hSocket, reinterpret_cast<const char*>(data.data()) + pnode->nSendOffset, data.size() - pnode->nSendOffset, MSG_NOSIGNAL | MSG_DONTWAIT); + } if (nBytes > 0) { - pnode->nLastSend = GetTime(); + pnode->nLastSend = GetSystemTimeInSeconds(); pnode->nSendBytes += nBytes; pnode->nSendOffset += nBytes; - pnode->RecordBytesSent(nBytes); + nSentSize += nBytes; if (pnode->nSendOffset == data.size()) { pnode->nSendOffset = 0; pnode->nSendSize -= data.size(); + pnode->fPauseSend = pnode->nSendSize > nSendBufferMaxSize; it++; } else { // could not send full message; stop sending more @@ -787,91 +872,65 @@ void SocketSendData(CNode *pnode) assert(pnode->nSendSize == 0); } pnode->vSendMsg.erase(pnode->vSendMsg.begin(), it); + return nSentSize; } -static list<CNode*> vNodesDisconnected; - -class CNodeRef { -public: - CNodeRef(CNode *pnode) : _pnode(pnode) { - LOCK(cs_vNodes); - _pnode->AddRef(); - } - - ~CNodeRef() { - LOCK(cs_vNodes); - _pnode->Release(); - } - - CNode& operator *() const {return *_pnode;}; - CNode* operator ->() const {return _pnode;}; - - CNodeRef& operator =(const CNodeRef& other) - { - if (this != &other) { - LOCK(cs_vNodes); - - _pnode->Release(); - _pnode = other._pnode; - _pnode->AddRef(); - } - return *this; - } - - CNodeRef(const CNodeRef& other): - _pnode(other._pnode) - { - LOCK(cs_vNodes); - _pnode->AddRef(); - } -private: - CNode *_pnode; +struct NodeEvictionCandidate +{ + NodeId id; + int64_t nTimeConnected; + int64_t nMinPingUsecTime; + int64_t nLastBlockTime; + int64_t nLastTXTime; + bool fRelevantServices; + bool fRelayTxes; + bool fBloomFilter; + CAddress addr; + uint64_t nKeyedNetGroup; }; -static bool ReverseCompareNodeMinPingTime(const CNodeRef &a, const CNodeRef &b) +static bool ReverseCompareNodeMinPingTime(const NodeEvictionCandidate &a, const NodeEvictionCandidate &b) { - return a->nMinPingUsecTime > b->nMinPingUsecTime; + return a.nMinPingUsecTime > b.nMinPingUsecTime; } -static bool ReverseCompareNodeTimeConnected(const CNodeRef &a, const CNodeRef &b) +static bool ReverseCompareNodeTimeConnected(const NodeEvictionCandidate &a, const NodeEvictionCandidate &b) { - return a->nTimeConnected > b->nTimeConnected; + return a.nTimeConnected > b.nTimeConnected; } -class CompareNetGroupKeyed -{ - std::vector<unsigned char> vchSecretKey; -public: - CompareNetGroupKeyed() - { - vchSecretKey.resize(32, 0); - GetRandBytes(vchSecretKey.data(), vchSecretKey.size()); - } - - bool operator()(const CNodeRef &a, const CNodeRef &b) - { - std::vector<unsigned char> vchGroupA, vchGroupB; - CSHA256 hashA, hashB; - std::vector<unsigned char> vchA(32), vchB(32); - - vchGroupA = a->addr.GetGroup(); - vchGroupB = b->addr.GetGroup(); - - hashA.Write(begin_ptr(vchGroupA), vchGroupA.size()); - hashB.Write(begin_ptr(vchGroupB), vchGroupB.size()); - - hashA.Write(begin_ptr(vchSecretKey), vchSecretKey.size()); - hashB.Write(begin_ptr(vchSecretKey), vchSecretKey.size()); +static bool CompareNetGroupKeyed(const NodeEvictionCandidate &a, const NodeEvictionCandidate &b) { + return a.nKeyedNetGroup < b.nKeyedNetGroup; +} - hashA.Finalize(begin_ptr(vchA)); - hashB.Finalize(begin_ptr(vchB)); +static bool CompareNodeBlockTime(const NodeEvictionCandidate &a, const NodeEvictionCandidate &b) +{ + // There is a fall-through here because it is common for a node to have many peers which have not yet relayed a block. + if (a.nLastBlockTime != b.nLastBlockTime) return a.nLastBlockTime < b.nLastBlockTime; + if (a.fRelevantServices != b.fRelevantServices) return b.fRelevantServices; + return a.nTimeConnected > b.nTimeConnected; +} - return vchA < vchB; - } -}; +static bool CompareNodeTXTime(const NodeEvictionCandidate &a, const NodeEvictionCandidate &b) +{ + // There is a fall-through here because it is common for a node to have more than a few peers that have not yet relayed txn. + if (a.nLastTXTime != b.nLastTXTime) return a.nLastTXTime < b.nLastTXTime; + if (a.fRelayTxes != b.fRelayTxes) return b.fRelayTxes; + if (a.fBloomFilter != b.fBloomFilter) return a.fBloomFilter; + return a.nTimeConnected > b.nTimeConnected; +} -static bool AttemptToEvictConnection(bool fPreferNewConnection) { - std::vector<CNodeRef> vEvictionCandidates; +/** Try to find a connection to evict when the node is full. + * Extreme care must be taken to avoid opening the node to attacker + * triggered network partitioning. + * The strategy used here is to protect a small number of peers + * for each of several distinct characteristics which are difficult + * to forge. In order to partition a node the attacker must be + * simultaneously better at all of them than honest peers. + */ +bool CConnman::AttemptToEvictConnection() +{ + std::vector<NodeEvictionCandidate> vEvictionCandidates; { LOCK(cs_vNodes); @@ -882,9 +941,11 @@ static bool AttemptToEvictConnection(bool fPreferNewConnection) { continue; if (node->fDisconnect) continue; - if (node->addr.IsLocal()) - continue; - vEvictionCandidates.push_back(CNodeRef(node)); + NodeEvictionCandidate candidate = {node->id, node->nTimeConnected, node->nMinPingUsecTime, + node->nLastBlockTime, node->nLastTXTime, + (node->nServices & nRelevantServices) == nRelevantServices, + node->fRelayTxes, node->pfilter != NULL, node->addr, node->nKeyedNetGroup}; + vEvictionCandidates.push_back(candidate); } } @@ -893,69 +954,86 @@ static bool AttemptToEvictConnection(bool fPreferNewConnection) { // Protect connections with certain characteristics // Deterministically select 4 peers to protect by netgroup. - // An attacker cannot predict which netgroups will be protected. - static CompareNetGroupKeyed comparerNetGroupKeyed; - std::sort(vEvictionCandidates.begin(), vEvictionCandidates.end(), comparerNetGroupKeyed); + // An attacker cannot predict which netgroups will be protected + std::sort(vEvictionCandidates.begin(), vEvictionCandidates.end(), CompareNetGroupKeyed); vEvictionCandidates.erase(vEvictionCandidates.end() - std::min(4, static_cast<int>(vEvictionCandidates.size())), vEvictionCandidates.end()); if (vEvictionCandidates.empty()) return false; - // Protect the 8 nodes with the best ping times. + // Protect the 8 nodes with the lowest minimum ping time. // An attacker cannot manipulate this metric without physically moving nodes closer to the target. std::sort(vEvictionCandidates.begin(), vEvictionCandidates.end(), ReverseCompareNodeMinPingTime); vEvictionCandidates.erase(vEvictionCandidates.end() - std::min(8, static_cast<int>(vEvictionCandidates.size())), vEvictionCandidates.end()); if (vEvictionCandidates.empty()) return false; + // Protect 4 nodes that most recently sent us transactions. + // An attacker cannot manipulate this metric without performing useful work. + std::sort(vEvictionCandidates.begin(), vEvictionCandidates.end(), CompareNodeTXTime); + vEvictionCandidates.erase(vEvictionCandidates.end() - std::min(4, static_cast<int>(vEvictionCandidates.size())), vEvictionCandidates.end()); + + if (vEvictionCandidates.empty()) return false; + + // Protect 4 nodes that most recently sent us blocks. + // An attacker cannot manipulate this metric without performing useful work. + std::sort(vEvictionCandidates.begin(), vEvictionCandidates.end(), CompareNodeBlockTime); + vEvictionCandidates.erase(vEvictionCandidates.end() - std::min(4, static_cast<int>(vEvictionCandidates.size())), vEvictionCandidates.end()); + + if (vEvictionCandidates.empty()) return false; + // Protect the half of the remaining nodes which have been connected the longest. - // This replicates the existing implicit behavior. + // This replicates the non-eviction implicit behavior, and precludes attacks that start later. std::sort(vEvictionCandidates.begin(), vEvictionCandidates.end(), ReverseCompareNodeTimeConnected); vEvictionCandidates.erase(vEvictionCandidates.end() - static_cast<int>(vEvictionCandidates.size() / 2), vEvictionCandidates.end()); if (vEvictionCandidates.empty()) return false; - // Identify the network group with the most connections - std::vector<unsigned char> naMostConnections; + // Identify the network group with the most connections and youngest member. + // (vEvictionCandidates is already sorted by reverse connect time) + uint64_t naMostConnections; unsigned int nMostConnections = 0; - std::map<std::vector<unsigned char>, std::vector<CNodeRef> > mapAddrCounts; - BOOST_FOREACH(const CNodeRef &node, vEvictionCandidates) { - mapAddrCounts[node->addr.GetGroup()].push_back(node); - - if (mapAddrCounts[node->addr.GetGroup()].size() > nMostConnections) { - nMostConnections = mapAddrCounts[node->addr.GetGroup()].size(); - naMostConnections = node->addr.GetGroup(); + int64_t nMostConnectionsTime = 0; + std::map<uint64_t, std::vector<NodeEvictionCandidate> > mapNetGroupNodes; + BOOST_FOREACH(const NodeEvictionCandidate &node, vEvictionCandidates) { + mapNetGroupNodes[node.nKeyedNetGroup].push_back(node); + int64_t grouptime = mapNetGroupNodes[node.nKeyedNetGroup][0].nTimeConnected; + size_t groupsize = mapNetGroupNodes[node.nKeyedNetGroup].size(); + + if (groupsize > nMostConnections || (groupsize == nMostConnections && grouptime > nMostConnectionsTime)) { + nMostConnections = groupsize; + nMostConnectionsTime = grouptime; + naMostConnections = node.nKeyedNetGroup; } } // Reduce to the network group with the most connections - vEvictionCandidates = mapAddrCounts[naMostConnections]; + vEvictionCandidates = std::move(mapNetGroupNodes[naMostConnections]); - // Do not disconnect peers if there is only 1 connection from their network group - if (vEvictionCandidates.size() <= 1) - // unless we prefer the new connection (for whitelisted peers) - if (!fPreferNewConnection) - return false; - - // Disconnect the most recent connection from the network group with the most connections - std::sort(vEvictionCandidates.begin(), vEvictionCandidates.end(), ReverseCompareNodeTimeConnected); - vEvictionCandidates[0]->fDisconnect = true; - - return true; + // Disconnect from the network group with the most connections + NodeId evicted = vEvictionCandidates.front().id; + LOCK(cs_vNodes); + for(std::vector<CNode*>::const_iterator it(vNodes.begin()); it != vNodes.end(); ++it) { + if ((*it)->GetId() == evicted) { + (*it)->fDisconnect = true; + return true; + } + } + return false; } -static void AcceptConnection(const ListenSocket& hListenSocket) { +void CConnman::AcceptConnection(const ListenSocket& hListenSocket) { struct sockaddr_storage sockaddr; socklen_t len = sizeof(sockaddr); SOCKET hSocket = accept(hListenSocket.socket, (struct sockaddr*)&sockaddr, &len); CAddress addr; int nInbound = 0; - int nMaxInbound = nMaxConnections - MAX_OUTBOUND_CONNECTIONS; + int nMaxInbound = nMaxConnections - (nMaxOutbound + nMaxFeeler); if (hSocket != INVALID_SOCKET) if (!addr.SetSockAddr((const struct sockaddr*)&sockaddr)) LogPrintf("Warning: Unknown socket family\n"); - bool whitelisted = hListenSocket.whitelisted || CNode::IsWhitelistedRange(addr); + bool whitelisted = hListenSocket.whitelisted || IsWhitelistedRange(addr); { LOCK(cs_vNodes); BOOST_FOREACH(CNode* pnode, vNodes) @@ -971,6 +1049,12 @@ static void AcceptConnection(const ListenSocket& hListenSocket) { return; } + if (!fNetworkActive) { + LogPrintf("connection from %s dropped: not accepting new connections\n", addr.ToString()); + CloseSocket(hSocket); + return; + } + if (!IsSelectableSocket(hSocket)) { LogPrintf("connection from %s dropped: non-selectable socket\n", addr.ToString()); @@ -987,7 +1071,7 @@ static void AcceptConnection(const ListenSocket& hListenSocket) { setsockopt(hSocket, IPPROTO_TCP, TCP_NODELAY, (void*)&set, sizeof(int)); #endif - if (CNode::IsBanned(addr) && !whitelisted) + if (IsBanned(addr) && !whitelisted) { LogPrintf("connection from %s dropped (banned)\n", addr.ToString()); CloseSocket(hSocket); @@ -996,7 +1080,7 @@ static void AcceptConnection(const ListenSocket& hListenSocket) { if (nInbound >= nMaxInbound) { - if (!AttemptToEvictConnection(whitelisted)) { + if (!AttemptToEvictConnection()) { // No connection to evict, disconnect the new connection LogPrint("net", "failed to find an eviction candidate - connection dropped (full)\n"); CloseSocket(hSocket); @@ -1004,9 +1088,13 @@ static void AcceptConnection(const ListenSocket& hListenSocket) { } } - CNode* pnode = new CNode(hSocket, addr, "", true); + NodeId id = GetNewNodeId(); + uint64_t nonce = GetDeterministicRandomizer(RANDOMIZER_ID_LOCALHOSTNONCE).Write(id).Finalize(); + + CNode* pnode = new CNode(id, nLocalServices, GetBestHeight(), hSocket, addr, CalculateKeyedNetGroup(addr), nonce, "", true); pnode->AddRef(); pnode->fWhitelisted = whitelisted; + GetNodeSignals().InitializeNode(pnode, *this); LogPrint("net", "connection from %s accepted\n", addr.ToString()); @@ -1016,10 +1104,10 @@ static void AcceptConnection(const ListenSocket& hListenSocket) { } } -void ThreadSocketHandler() +void CConnman::ThreadSocketHandler() { unsigned int nPrevNodeCount = 0; - while (true) + while (!interruptNet) { // // Disconnect nodes @@ -1027,11 +1115,10 @@ void ThreadSocketHandler() { LOCK(cs_vNodes); // Disconnect unused nodes - vector<CNode*> vNodesCopy = vNodes; + std::vector<CNode*> vNodesCopy = vNodes; BOOST_FOREACH(CNode* pnode, vNodesCopy) { - if (pnode->fDisconnect || - (pnode->GetRefCount() <= 0 && pnode->vRecvMsg.empty() && pnode->nSendSize == 0 && pnode->ssSend.empty())) + if (pnode->fDisconnect) { // remove from vNodes vNodes.erase(remove(vNodes.begin(), vNodes.end(), pnode), vNodes.end()); @@ -1043,45 +1130,44 @@ void ThreadSocketHandler() pnode->CloseSocketDisconnect(); // hold in disconnected pool until all refs are released - if (pnode->fNetworkNode || pnode->fInbound) - pnode->Release(); + pnode->Release(); vNodesDisconnected.push_back(pnode); } } } { // Delete disconnected nodes - list<CNode*> vNodesDisconnectedCopy = vNodesDisconnected; + std::list<CNode*> vNodesDisconnectedCopy = vNodesDisconnected; BOOST_FOREACH(CNode* pnode, vNodesDisconnectedCopy) { // wait until threads are done using it - if (pnode->GetRefCount() <= 0) - { + if (pnode->GetRefCount() <= 0) { bool fDelete = false; { - TRY_LOCK(pnode->cs_vSend, lockSend); - if (lockSend) - { - TRY_LOCK(pnode->cs_vRecvMsg, lockRecv); - if (lockRecv) - { - TRY_LOCK(pnode->cs_inventory, lockInv); - if (lockInv) - fDelete = true; + TRY_LOCK(pnode->cs_inventory, lockInv); + if (lockInv) { + TRY_LOCK(pnode->cs_vSend, lockSend); + if (lockSend) { + fDelete = true; } } } - if (fDelete) - { + if (fDelete) { vNodesDisconnected.remove(pnode); - delete pnode; + DeleteNode(pnode); } } } } - if(vNodes.size() != nPrevNodeCount) { - nPrevNodeCount = vNodes.size(); - uiInterface.NotifyNumConnectionsChanged(nPrevNodeCount); + size_t vNodesSize; + { + LOCK(cs_vNodes); + vNodesSize = vNodes.size(); + } + if(vNodesSize != nPrevNodeCount) { + nPrevNodeCount = vNodesSize; + if(clientInterface) + clientInterface->NotifyNumConnectionsChanged(nPrevNodeCount); } // @@ -1102,7 +1188,7 @@ void ThreadSocketHandler() BOOST_FOREACH(const ListenSocket& hListenSocket, vhListenSocket) { FD_SET(hListenSocket.socket, &fdsetRecv); - hSocketMax = max(hSocketMax, hListenSocket.socket); + hSocketMax = std::max(hSocketMax, hListenSocket.socket); have_fds = true; } @@ -1110,47 +1196,46 @@ void ThreadSocketHandler() LOCK(cs_vNodes); BOOST_FOREACH(CNode* pnode, vNodes) { - if (pnode->hSocket == INVALID_SOCKET) - continue; - FD_SET(pnode->hSocket, &fdsetError); - hSocketMax = max(hSocketMax, pnode->hSocket); - have_fds = true; - // Implement the following logic: // * If there is data to send, select() for sending data. As this only // happens when optimistic write failed, we choose to first drain the // write buffer in this case before receiving more. This avoids // needlessly queueing received data, if the remote peer is not themselves // receiving data. This means properly utilizing TCP flow control signalling. - // * Otherwise, if there is no (complete) message in the receive buffer, - // or there is space left in the buffer, select() for receiving data. - // * (if neither of the above applies, there is certainly one message - // in the receiver buffer ready to be processed). - // Together, that means that at least one of the following is always possible, - // so we don't deadlock: - // * We send some data. - // * We wait for data to be received (and disconnect after timeout). - // * We process a message in the buffer (message handler thread). + // * Otherwise, if there is space left in the receive buffer, select() for + // receiving data. + // * Hand off all complete messages to the processor, to be handled without + // blocking here. + + bool select_recv = !pnode->fPauseRecv; + bool select_send; { - TRY_LOCK(pnode->cs_vSend, lockSend); - if (lockSend && !pnode->vSendMsg.empty()) { - FD_SET(pnode->hSocket, &fdsetSend); - continue; - } + LOCK(pnode->cs_vSend); + select_send = !pnode->vSendMsg.empty(); } - { - TRY_LOCK(pnode->cs_vRecvMsg, lockRecv); - if (lockRecv && ( - pnode->vRecvMsg.empty() || !pnode->vRecvMsg.front().complete() || - pnode->GetTotalRecvSize() <= ReceiveFloodSize())) - FD_SET(pnode->hSocket, &fdsetRecv); + + LOCK(pnode->cs_hSocket); + if (pnode->hSocket == INVALID_SOCKET) + continue; + + FD_SET(pnode->hSocket, &fdsetError); + hSocketMax = std::max(hSocketMax, pnode->hSocket); + have_fds = true; + + if (select_send) { + FD_SET(pnode->hSocket, &fdsetSend); + continue; + } + if (select_recv) { + FD_SET(pnode->hSocket, &fdsetRecv); } } } int nSelect = select(have_fds ? hSocketMax + 1 : 0, &fdsetRecv, &fdsetSend, &fdsetError, &timeout); - boost::this_thread::interruption_point(); + if (interruptNet) + return; if (nSelect == SOCKET_ERROR) { @@ -1163,7 +1248,8 @@ void ThreadSocketHandler() } FD_ZERO(&fdsetSend); FD_ZERO(&fdsetError); - MilliSleep(timeout.tv_usec/1000); + if (!interruptNet.sleep_for(std::chrono::milliseconds(timeout.tv_usec/1000))) + return; } // @@ -1180,7 +1266,7 @@ void ThreadSocketHandler() // // Service each socket // - vector<CNode*> vNodesCopy; + std::vector<CNode*> vNodesCopy; { LOCK(cs_vNodes); vNodesCopy = vNodes; @@ -1189,29 +1275,58 @@ void ThreadSocketHandler() } BOOST_FOREACH(CNode* pnode, vNodesCopy) { - boost::this_thread::interruption_point(); + if (interruptNet) + return; // // Receive // - if (pnode->hSocket == INVALID_SOCKET) - continue; - if (FD_ISSET(pnode->hSocket, &fdsetRecv) || FD_ISSET(pnode->hSocket, &fdsetError)) + bool recvSet = false; + bool sendSet = false; + bool errorSet = false; + { + LOCK(pnode->cs_hSocket); + if (pnode->hSocket == INVALID_SOCKET) + continue; + recvSet = FD_ISSET(pnode->hSocket, &fdsetRecv); + sendSet = FD_ISSET(pnode->hSocket, &fdsetSend); + errorSet = FD_ISSET(pnode->hSocket, &fdsetError); + } + if (recvSet || errorSet) { - TRY_LOCK(pnode->cs_vRecvMsg, lockRecv); - if (lockRecv) { { // typical socket buffer is 8K-64K char pchBuf[0x10000]; - int nBytes = recv(pnode->hSocket, pchBuf, sizeof(pchBuf), MSG_DONTWAIT); + int nBytes = 0; + { + LOCK(pnode->cs_hSocket); + if (pnode->hSocket == INVALID_SOCKET) + continue; + nBytes = recv(pnode->hSocket, pchBuf, sizeof(pchBuf), MSG_DONTWAIT); + } if (nBytes > 0) { - if (!pnode->ReceiveMsgBytes(pchBuf, nBytes)) + bool notify = false; + if (!pnode->ReceiveMsgBytes(pchBuf, nBytes, notify)) pnode->CloseSocketDisconnect(); - pnode->nLastRecv = GetTime(); - pnode->nRecvBytes += nBytes; - pnode->RecordBytesRecv(nBytes); + RecordBytesRecv(nBytes); + if (notify) { + size_t nSizeAdded = 0; + auto it(pnode->vRecvMsg.begin()); + for (; it != pnode->vRecvMsg.end(); ++it) { + if (!it->complete()) + break; + nSizeAdded += it->vRecv.size() + CMessageHeader::HEADER_SIZE; + } + { + LOCK(pnode->cs_vProcessMsg); + pnode->vProcessMsg.splice(pnode->vProcessMsg.end(), pnode->vRecvMsg, pnode->vRecvMsg.begin(), it); + pnode->nProcessQueueSize += nSizeAdded; + pnode->fPauseRecv = pnode->nProcessQueueSize > nReceiveFloodSize; + } + WakeMessageHandler(); + } } else if (nBytes == 0) { @@ -1238,19 +1353,19 @@ void ThreadSocketHandler() // // Send // - if (pnode->hSocket == INVALID_SOCKET) - continue; - if (FD_ISSET(pnode->hSocket, &fdsetSend)) + if (sendSet) { - TRY_LOCK(pnode->cs_vSend, lockSend); - if (lockSend) - SocketSendData(pnode); + LOCK(pnode->cs_vSend); + size_t nBytes = SocketSendData(pnode); + if (nBytes) { + RecordBytesSent(nBytes); + } } // // Inactivity checking // - int64_t nTime = GetTime(); + int64_t nTime = GetSystemTimeInSeconds(); if (nTime - pnode->nTimeConnected > 60) { if (pnode->nLastRecv == 0 || pnode->nLastSend == 0) @@ -1273,6 +1388,11 @@ void ThreadSocketHandler() LogPrintf("ping timeout: %fs\n", 0.000001 * (GetTimeMicros() - pnode->nPingUsecStart)); pnode->fDisconnect = true; } + else if (!pnode->fSuccessfullyConnected) + { + LogPrintf("version handshake timeout from %d\n", pnode->id); + pnode->fDisconnect = true; + } } } { @@ -1283,8 +1403,14 @@ void ThreadSocketHandler() } } - - +void CConnman::WakeMessageHandler() +{ + { + std::lock_guard<std::mutex> lock(mutexMsgProc); + fMsgProcWake = true; + } + condMsgProc.notify_one(); +} @@ -1329,15 +1455,18 @@ void ThreadMapPort() { if(externalIPAddress[0]) { - LogPrintf("UPnP: ExternalIPAddress = %s\n", externalIPAddress); - AddLocal(CNetAddr(externalIPAddress), LOCAL_UPNP); + CNetAddr resolved; + if(LookupHost(externalIPAddress, resolved, false)) { + LogPrintf("UPnP: ExternalIPAddress = %s\n", resolved.ToString().c_str()); + AddLocal(resolved, LOCAL_UPNP); + } } else LogPrintf("UPnP: GetExternalIPAddress failed.\n"); } } - string strDesc = "Bitcoin " + FormatFullVersion(); + std::string strDesc = "Bitcoin " + FormatFullVersion(); try { while (true) { @@ -1355,7 +1484,7 @@ void ThreadMapPort() LogPrintf("AddPortMapping(%s, %s, %s) failed with code %d (%s)\n", port, port, lanaddr, r, strupnperror(r)); else - LogPrintf("UPnP Port Mapping successful.\n");; + LogPrintf("UPnP Port Mapping successful.\n"); MilliSleep(20*60*1000); // Refresh every 20 minutes } @@ -1409,21 +1538,42 @@ void MapPort(bool) -void ThreadDNSAddressSeed() +static std::string GetDNSHost(const CDNSSeedData& data, ServiceFlags* requiredServiceBits) +{ + //use default host for non-filter-capable seeds or if we use the default service bits (NODE_NETWORK) + if (!data.supportsServiceBitsFiltering || *requiredServiceBits == NODE_NETWORK) { + *requiredServiceBits = NODE_NETWORK; + return data.host; + } + + // See chainparams.cpp, most dnsseeds only support one or two possible servicebits hostnames + return strprintf("x%x.%s", *requiredServiceBits, data.host); +} + + +void CConnman::ThreadDNSAddressSeed() { // goal: only query DNS seeds if address need is acute + // Avoiding DNS seeds when we don't need them improves user privacy by + // creating fewer identifying DNS requests, reduces trust by giving seeds + // less influence on the network topology, and reduces traffic to the seeds. if ((addrman.size() > 0) && (!GetBoolArg("-forcednsseed", DEFAULT_FORCEDNSSEED))) { - MilliSleep(11 * 1000); + if (!interruptNet.sleep_for(std::chrono::seconds(11))) + return; LOCK(cs_vNodes); - if (vNodes.size() >= 2) { + int nRelevant = 0; + for (auto pnode : vNodes) { + nRelevant += pnode->fSuccessfullyConnected && ((pnode->nServices & nRelevantServices) == nRelevantServices); + } + if (nRelevant >= 2) { LogPrintf("P2P peers available. Skipped DNS seeding.\n"); return; } } - const vector<CDNSSeedData> &vSeeds = Params().DNSSeeds(); + const std::vector<CDNSSeedData> &vSeeds = Params().DNSSeeds(); int found = 0; LogPrintf("Loading addresses from DNS seeds (could take a while)\n"); @@ -1432,20 +1582,29 @@ void ThreadDNSAddressSeed() if (HaveNameProxy()) { AddOneShot(seed.host); } else { - vector<CNetAddr> vIPs; - vector<CAddress> vAdd; - if (LookupHost(seed.host.c_str(), vIPs)) + std::vector<CNetAddr> vIPs; + std::vector<CAddress> vAdd; + ServiceFlags requiredServiceBits = nRelevantServices; + if (LookupHost(GetDNSHost(seed, &requiredServiceBits).c_str(), vIPs, 0, true)) { BOOST_FOREACH(const CNetAddr& ip, vIPs) { int nOneDay = 24*3600; - CAddress addr = CAddress(CService(ip, Params().GetDefaultPort())); + CAddress addr = CAddress(CService(ip, Params().GetDefaultPort()), requiredServiceBits); addr.nTime = GetTime() - 3*nOneDay - GetRand(4*nOneDay); // use a random age between 3 and 7 days old vAdd.push_back(addr); found++; } } - addrman.Add(vAdd, CNetAddr(seed.name, true)); + // TODO: The seed name resolve may fail, yielding an IP of [::], which results in + // addrman assigning the same source to results from different seeds. + // This should switch to a hard-coded stable dummy IP for each seed name, so that the + // resolve is not required at all. + if (!vIPs.empty()) { + CService seedSource; + Lookup(seed.name.c_str(), seedSource, 0, true); + addrman.Add(vAdd, seedSource); + } } } @@ -1463,7 +1622,7 @@ void ThreadDNSAddressSeed() -void DumpAddresses() +void CConnman::DumpAddresses() { int64_t nStart = GetTimeMillis(); @@ -1474,20 +1633,15 @@ void DumpAddresses() addrman.size(), GetTimeMillis() - nStart); } -void DumpData() +void CConnman::DumpData() { DumpAddresses(); - - if (CNode::BannedSetIsDirty()) - { - DumpBanlist(); - CNode::SetBannedSetDirty(false); - } + DumpBanlist(); } -void static ProcessOneShot() +void CConnman::ProcessOneShot() { - string strDest; + std::string strDest; { LOCK(cs_vOneShots); if (vOneShots.empty()) @@ -1498,49 +1652,58 @@ void static ProcessOneShot() CAddress addr; CSemaphoreGrant grant(*semOutbound, true); if (grant) { - if (!OpenNetworkConnection(addr, &grant, strDest.c_str(), true)) + if (!OpenNetworkConnection(addr, false, &grant, strDest.c_str(), true)) AddOneShot(strDest); } } -void ThreadOpenConnections() +void CConnman::ThreadOpenConnections() { // Connect to specific addresses - if (mapArgs.count("-connect") && mapMultiArgs["-connect"].size() > 0) + if (mapMultiArgs.count("-connect") && mapMultiArgs.at("-connect").size() > 0) { for (int64_t nLoop = 0;; nLoop++) { ProcessOneShot(); - BOOST_FOREACH(const std::string& strAddr, mapMultiArgs["-connect"]) + BOOST_FOREACH(const std::string& strAddr, mapMultiArgs.at("-connect")) { - CAddress addr; - OpenNetworkConnection(addr, NULL, strAddr.c_str()); + CAddress addr(CService(), NODE_NONE); + OpenNetworkConnection(addr, false, NULL, strAddr.c_str()); for (int i = 0; i < 10 && i < nLoop; i++) { - MilliSleep(500); + if (!interruptNet.sleep_for(std::chrono::milliseconds(500))) + return; } } - MilliSleep(500); + if (!interruptNet.sleep_for(std::chrono::milliseconds(500))) + return; } } // Initiate network connections int64_t nStart = GetTime(); - while (true) + + // Minimum time before next feeler connection (in microseconds). + int64_t nNextFeeler = PoissonNextSend(nStart*1000*1000, FEELER_INTERVAL); + while (!interruptNet) { ProcessOneShot(); - MilliSleep(500); + if (!interruptNet.sleep_for(std::chrono::milliseconds(500))) + return; CSemaphoreGrant grant(*semOutbound); - boost::this_thread::interruption_point(); + if (interruptNet) + return; // Add seed nodes if DNS seeds are all down (an infrastructure attack?). if (addrman.size() == 0 && (GetTime() - nStart > 60)) { static bool done = false; if (!done) { LogPrintf("Adding fixed seed nodes as DNS doesn't seem to be available.\n"); - addrman.Add(convertSeed6(Params().FixedSeeds()), CNetAddr("127.0.0.1")); + CNetAddr local; + LookupHost("127.0.0.1", local, false); + addrman.Add(convertSeed6(Params().FixedSeeds()), local); done = true; } } @@ -1553,23 +1716,50 @@ void ThreadOpenConnections() // Only connect out to one peer per network group (/16 for IPv4). // Do this here so we don't have to critsect vNodes inside mapAddresses critsect. int nOutbound = 0; - set<vector<unsigned char> > setConnected; + std::set<std::vector<unsigned char> > setConnected; { LOCK(cs_vNodes); BOOST_FOREACH(CNode* pnode, vNodes) { - if (!pnode->fInbound) { + if (!pnode->fInbound && !pnode->fAddnode) { + // Netgroups for inbound and addnode peers are not excluded because our goal here + // is to not use multiple of our limited outbound slots on a single netgroup + // but inbound and addnode peers do not use our outbound slots. Inbound peers + // also have the added issue that they're attacker controlled and could be used + // to prevent us from connecting to particular hosts if we used them here. setConnected.insert(pnode->addr.GetGroup()); nOutbound++; } } } - int64_t nANow = GetAdjustedTime(); + // Feeler Connections + // + // Design goals: + // * Increase the number of connectable addresses in the tried table. + // + // Method: + // * Choose a random address from new and attempt to connect to it if we can connect + // successfully it is added to tried. + // * Start attempting feeler connections only after node finishes making outbound + // connections. + // * Only make a feeler connection once every few minutes. + // + bool fFeeler = false; + if (nOutbound >= nMaxOutbound) { + int64_t nTime = GetTimeMicros(); // The current time right now (in microseconds). + if (nTime > nNextFeeler) { + nNextFeeler = PoissonNextSend(nTime, FEELER_INTERVAL); + fFeeler = true; + } else { + continue; + } + } + int64_t nANow = GetAdjustedTime(); int nTries = 0; - while (true) + while (!interruptNet) { - CAddrInfo addr = addrman.Select(); + CAddrInfo addr = addrman.Select(fFeeler); // if we selected an invalid address, restart if (!addr.IsValid() || setConnected.count(addr.GetGroup()) || IsLocal(addr)) @@ -1585,10 +1775,18 @@ void ThreadOpenConnections() if (IsLimited(addr)) continue; + // only connect to full nodes + if ((addr.nServices & REQUIRED_SERVICES) != REQUIRED_SERVICES) + continue; + // only consider very recently tried nodes after 30 failed attempts if (nANow - addr.nLastTry < 600 && nTries < 30) continue; + // only consider nodes missing relevant services after 40 failed attempts and only if less than half the outbound are up. + if ((addr.nServices & nRelevantServices) != nRelevantServices && (nTries < 40 || nOutbound >= (nMaxOutbound >> 1))) + continue; + // do not allow non-default ports, unless after 50 invalid addresses selected already if (addr.GetPort() != Params().GetDefaultPort() && nTries < 50) continue; @@ -1597,121 +1795,156 @@ void ThreadOpenConnections() break; } - if (addrConnect.IsValid()) - OpenNetworkConnection(addrConnect, &grant); + if (addrConnect.IsValid()) { + + if (fFeeler) { + // Add small amount of random noise before connection to avoid synchronization. + int randsleep = GetRandInt(FEELER_SLEEP_WINDOW * 1000); + if (!interruptNet.sleep_for(std::chrono::milliseconds(randsleep))) + return; + LogPrint("net", "Making feeler connection to %s\n", addrConnect.ToString()); + } + + OpenNetworkConnection(addrConnect, (int)setConnected.size() >= std::min(nMaxConnections - 1, 2), &grant, NULL, false, fFeeler); + } } } -void ThreadOpenAddedConnections() +std::vector<AddedNodeInfo> CConnman::GetAddedNodeInfo() { + std::vector<AddedNodeInfo> ret; + + std::list<std::string> lAddresses(0); { LOCK(cs_vAddedNodes); - vAddedNodes = mapMultiArgs["-addnode"]; + ret.reserve(vAddedNodes.size()); + BOOST_FOREACH(const std::string& strAddNode, vAddedNodes) + lAddresses.push_back(strAddNode); } - if (HaveNameProxy()) { - while(true) { - list<string> lAddresses(0); - { - LOCK(cs_vAddedNodes); - BOOST_FOREACH(const std::string& strAddNode, vAddedNodes) - lAddresses.push_back(strAddNode); + + // Build a map of all already connected addresses (by IP:port and by name) to inbound/outbound and resolved CService + std::map<CService, bool> mapConnected; + std::map<std::string, std::pair<bool, CService>> mapConnectedByName; + { + LOCK(cs_vNodes); + for (const CNode* pnode : vNodes) { + if (pnode->addr.IsValid()) { + mapConnected[pnode->addr] = pnode->fInbound; } - BOOST_FOREACH(const std::string& strAddNode, lAddresses) { - CAddress addr; - CSemaphoreGrant grant(*semOutbound); - OpenNetworkConnection(addr, &grant, strAddNode.c_str()); - MilliSleep(500); + std::string addrName = pnode->GetAddrName(); + if (!addrName.empty()) { + mapConnectedByName[std::move(addrName)] = std::make_pair(pnode->fInbound, static_cast<const CService&>(pnode->addr)); } - MilliSleep(120000); // Retry every 2 minutes } } - for (unsigned int i = 0; true; i++) - { - list<string> lAddresses(0); - { - LOCK(cs_vAddedNodes); - BOOST_FOREACH(const std::string& strAddNode, vAddedNodes) - lAddresses.push_back(strAddNode); + BOOST_FOREACH(const std::string& strAddNode, lAddresses) { + CService service(LookupNumeric(strAddNode.c_str(), Params().GetDefaultPort())); + if (service.IsValid()) { + // strAddNode is an IP:port + auto it = mapConnected.find(service); + if (it != mapConnected.end()) { + ret.push_back(AddedNodeInfo{strAddNode, service, true, it->second}); + } else { + ret.push_back(AddedNodeInfo{strAddNode, CService(), false, false}); + } + } else { + // strAddNode is a name + auto it = mapConnectedByName.find(strAddNode); + if (it != mapConnectedByName.end()) { + ret.push_back(AddedNodeInfo{strAddNode, it->second.second, true, it->second.first}); + } else { + ret.push_back(AddedNodeInfo{strAddNode, CService(), false, false}); + } } + } - list<vector<CService> > lservAddressesToAdd(0); - BOOST_FOREACH(const std::string& strAddNode, lAddresses) { - vector<CService> vservNode(0); - if(Lookup(strAddNode.c_str(), vservNode, Params().GetDefaultPort(), fNameLookup, 0)) - { - lservAddressesToAdd.push_back(vservNode); - { - LOCK(cs_setservAddNodeAddresses); - BOOST_FOREACH(const CService& serv, vservNode) - setservAddNodeAddresses.insert(serv); + return ret; +} + +void CConnman::ThreadOpenAddedConnections() +{ + { + LOCK(cs_vAddedNodes); + if (mapMultiArgs.count("-addnode")) + vAddedNodes = mapMultiArgs.at("-addnode"); + } + + while (true) + { + CSemaphoreGrant grant(*semAddnode); + std::vector<AddedNodeInfo> vInfo = GetAddedNodeInfo(); + bool tried = false; + for (const AddedNodeInfo& info : vInfo) { + if (!info.fConnected) { + if (!grant.TryAcquire()) { + // If we've used up our semaphore and need a new one, lets not wait here since while we are waiting + // the addednodeinfo state might change. + break; } + // If strAddedNode is an IP/port, decode it immediately, so + // OpenNetworkConnection can detect existing connections to that IP/port. + tried = true; + CService service(LookupNumeric(info.strAddedNode.c_str(), Params().GetDefaultPort())); + OpenNetworkConnection(CAddress(service, NODE_NONE), false, &grant, info.strAddedNode.c_str(), false, false, true); + if (!interruptNet.sleep_for(std::chrono::milliseconds(500))) + return; } } - // Attempt to connect to each IP for each addnode entry until at least one is successful per addnode entry - // (keeping in mind that addnode entries can have many IPs if fNameLookup) - { - LOCK(cs_vNodes); - BOOST_FOREACH(CNode* pnode, vNodes) - for (list<vector<CService> >::iterator it = lservAddressesToAdd.begin(); it != lservAddressesToAdd.end(); it++) - BOOST_FOREACH(const CService& addrNode, *(it)) - if (pnode->addr == addrNode) - { - it = lservAddressesToAdd.erase(it); - it--; - break; - } - } - BOOST_FOREACH(vector<CService>& vserv, lservAddressesToAdd) - { - CSemaphoreGrant grant(*semOutbound); - OpenNetworkConnection(CAddress(vserv[i % vserv.size()]), &grant); - MilliSleep(500); - } - MilliSleep(120000); // Retry every 2 minutes + // Retry every 60 seconds if a connection was attempted, otherwise two seconds + if (!interruptNet.sleep_for(std::chrono::seconds(tried ? 60 : 2))) + return; } } // if successful, this moves the passed grant to the constructed node -bool OpenNetworkConnection(const CAddress& addrConnect, CSemaphoreGrant *grantOutbound, const char *pszDest, bool fOneShot) +bool CConnman::OpenNetworkConnection(const CAddress& addrConnect, bool fCountFailure, CSemaphoreGrant *grantOutbound, const char *pszDest, bool fOneShot, bool fFeeler, bool fAddnode) { // // Initiate outbound network connection // - boost::this_thread::interruption_point(); + if (interruptNet) { + return false; + } + if (!fNetworkActive) { + return false; + } if (!pszDest) { if (IsLocal(addrConnect) || - FindNode((CNetAddr)addrConnect) || CNode::IsBanned(addrConnect) || + FindNode((CNetAddr)addrConnect) || IsBanned(addrConnect) || FindNode(addrConnect.ToStringIPPort())) return false; } else if (FindNode(std::string(pszDest))) return false; - CNode* pnode = ConnectNode(addrConnect, pszDest); - boost::this_thread::interruption_point(); + CNode* pnode = ConnectNode(addrConnect, pszDest, fCountFailure); if (!pnode) return false; if (grantOutbound) grantOutbound->MoveTo(pnode->grantOutbound); - pnode->fNetworkNode = true; if (fOneShot) pnode->fOneShot = true; + if (fFeeler) + pnode->fFeeler = true; + if (fAddnode) + pnode->fAddnode = true; + + GetNodeSignals().InitializeNode(pnode, *this); + { + LOCK(cs_vNodes); + vNodes.push_back(pnode); + } return true; } - -void ThreadMessageHandler() +void CConnman::ThreadMessageHandler() { - boost::mutex condition_mutex; - boost::unique_lock<boost::mutex> lock(condition_mutex); - - SetThreadPriority(THREAD_PRIORITY_BELOW_NORMAL); - while (true) + while (!flagInterruptMsgProc) { - vector<CNode*> vNodesCopy; + std::vector<CNode*> vNodesCopy; { LOCK(cs_vNodes); vNodesCopy = vNodes; @@ -1720,12 +1953,7 @@ void ThreadMessageHandler() } } - // Poll the connected nodes for messages - CNode* pnodeTrickle = NULL; - if (!vNodesCopy.empty()) - pnodeTrickle = vNodesCopy[GetRand(vNodesCopy.size())]; - - bool fSleep = true; + bool fMoreWork = false; BOOST_FOREACH(CNode* pnode, vNodesCopy) { @@ -1733,31 +1961,18 @@ void ThreadMessageHandler() continue; // Receive messages - { - TRY_LOCK(pnode->cs_vRecvMsg, lockRecv); - if (lockRecv) - { - if (!g_signals.ProcessMessages(pnode)) - pnode->CloseSocketDisconnect(); - - if (pnode->nSendSize < SendBufferSize()) - { - if (!pnode->vRecvGetData.empty() || (!pnode->vRecvMsg.empty() && pnode->vRecvMsg[0].complete())) - { - fSleep = false; - } - } - } - } - boost::this_thread::interruption_point(); + bool fMoreNodeWork = GetNodeSignals().ProcessMessages(pnode, *this, flagInterruptMsgProc); + fMoreWork |= (fMoreNodeWork && !pnode->fPauseSend); + if (flagInterruptMsgProc) + return; // Send messages { - TRY_LOCK(pnode->cs_vSend, lockSend); - if (lockSend) - g_signals.SendMessages(pnode, pnode == pnodeTrickle || pnode->fWhitelisted); + LOCK(pnode->cs_sendProcessing); + GetNodeSignals().SendMessages(pnode, *this, flagInterruptMsgProc); } - boost::this_thread::interruption_point(); + if (flagInterruptMsgProc) + return; } { @@ -1766,8 +1981,11 @@ void ThreadMessageHandler() pnode->Release(); } - if (fSleep) - messageHandlerCondition.timed_wait(lock, boost::posix_time::microsec_clock::universal_time() + boost::posix_time::milliseconds(100)); + std::unique_lock<std::mutex> lock(mutexMsgProc); + if (!fMoreWork) { + condMsgProc.wait_until(lock, std::chrono::steady_clock::now() + std::chrono::milliseconds(100), [this] { return fMsgProcWake; }); + } + fMsgProcWake = false; } } @@ -1776,7 +1994,7 @@ void ThreadMessageHandler() -bool BindListenPort(const CService &addrBind, string& strError, bool fWhitelisted) +bool CConnman::BindListenPort(const CService &addrBind, std::string& strError, bool fWhitelisted) { strError = ""; int nOne = 1; @@ -1848,7 +2066,7 @@ bool BindListenPort(const CService &addrBind, string& strError, bool fWhiteliste { int nErr = WSAGetLastError(); if (nErr == WSAEADDRINUSE) - strError = strprintf(_("Unable to bind to %s on this computer. Bitcoin Core is probably already running."), addrBind.ToString()); + strError = strprintf(_("Unable to bind to %s on this computer. %s is probably already running."), addrBind.ToString(), _(PACKAGE_NAME)); else strError = strprintf(_("Unable to bind to %s on this computer (bind returned error %s)"), addrBind.ToString(), NetworkErrorString(nErr)); LogPrintf("%s\n", strError); @@ -1874,7 +2092,7 @@ bool BindListenPort(const CService &addrBind, string& strError, bool fWhiteliste return true; } -void static Discover(boost::thread_group& threadGroup) +void Discover(boost::thread_group& threadGroup) { if (!fDiscover) return; @@ -1884,8 +2102,8 @@ void static Discover(boost::thread_group& threadGroup) char pszHostName[256] = ""; if (gethostname(pszHostName, sizeof(pszHostName)) != SOCKET_ERROR) { - vector<CNetAddr> vaddr; - if (LookupHost(pszHostName, vaddr)) + std::vector<CNetAddr> vaddr; + if (LookupHost(pszHostName, vaddr, 0, true)) { BOOST_FOREACH (const CNetAddr &addr, vaddr) { @@ -1925,83 +2143,151 @@ void static Discover(boost::thread_group& threadGroup) #endif } -void StartNode(boost::thread_group& threadGroup, CScheduler& scheduler) +void CConnman::SetNetworkActive(bool active) +{ + if (fDebug) { + LogPrint("net", "SetNetworkActive: %s\n", active); + } + + if (!active) { + fNetworkActive = false; + + LOCK(cs_vNodes); + // Close sockets to all nodes + BOOST_FOREACH(CNode* pnode, vNodes) { + pnode->CloseSocketDisconnect(); + } + } else { + fNetworkActive = true; + } + + uiInterface.NotifyNetworkActiveChanged(fNetworkActive); +} + +CConnman::CConnman(uint64_t nSeed0In, uint64_t nSeed1In) : nSeed0(nSeed0In), nSeed1(nSeed1In) +{ + fNetworkActive = true; + setBannedIsDirty = false; + fAddressesInitialized = false; + nLastNodeId = 0; + nSendBufferMaxSize = 0; + nReceiveFloodSize = 0; + semOutbound = NULL; + semAddnode = NULL; + nMaxConnections = 0; + nMaxOutbound = 0; + nMaxAddnode = 0; + nBestHeight = 0; + clientInterface = NULL; + flagInterruptMsgProc = false; +} + +NodeId CConnman::GetNewNodeId() { - uiInterface.InitMessage(_("Loading addresses...")); - // Load addresses for peers.dat + return nLastNodeId.fetch_add(1, std::memory_order_relaxed); +} + +bool CConnman::Start(CScheduler& scheduler, std::string& strNodeError, Options connOptions) +{ + nTotalBytesRecv = 0; + nTotalBytesSent = 0; + nMaxOutboundTotalBytesSentInCycle = 0; + nMaxOutboundCycleStartTime = 0; + + nRelevantServices = connOptions.nRelevantServices; + nLocalServices = connOptions.nLocalServices; + nMaxConnections = connOptions.nMaxConnections; + nMaxOutbound = std::min((connOptions.nMaxOutbound), nMaxConnections); + nMaxAddnode = connOptions.nMaxAddnode; + nMaxFeeler = connOptions.nMaxFeeler; + + nSendBufferMaxSize = connOptions.nSendBufferMaxSize; + nReceiveFloodSize = connOptions.nReceiveFloodSize; + + nMaxOutboundLimit = connOptions.nMaxOutboundLimit; + nMaxOutboundTimeframe = connOptions.nMaxOutboundTimeframe; + + SetBestHeight(connOptions.nBestHeight); + + clientInterface = connOptions.uiInterface; + if (clientInterface) + clientInterface->InitMessage(_("Loading addresses...")); + // Load addresses from peers.dat int64_t nStart = GetTimeMillis(); { CAddrDB adb; - if (!adb.Read(addrman)) + if (adb.Read(addrman)) + LogPrintf("Loaded %i addresses from peers.dat %dms\n", addrman.size(), GetTimeMillis() - nStart); + else { + addrman.Clear(); // Addrman can be in an inconsistent state after failure, reset it LogPrintf("Invalid or missing peers.dat; recreating\n"); + DumpAddresses(); + } } - - //try to read stored banlist + if (clientInterface) + clientInterface->InitMessage(_("Loading banlist...")); + // Load addresses from banlist.dat + nStart = GetTimeMillis(); CBanDB bandb; banmap_t banmap; - if (!bandb.Read(banmap)) + if (bandb.Read(banmap)) { + SetBanned(banmap); // thread save setter + SetBannedSetDirty(false); // no need to write down, just read data + SweepBanned(); // sweep out unused entries + + LogPrint("net", "Loaded %d banned node ips/subnets from banlist.dat %dms\n", + banmap.size(), GetTimeMillis() - nStart); + } else { LogPrintf("Invalid or missing banlist.dat; recreating\n"); + SetBannedSetDirty(true); // force write + DumpBanlist(); + } - CNode::SetBanned(banmap); //thread save setter - CNode::SetBannedSetDirty(false); //no need to write down just read or nonexistent data - CNode::SweepBanned(); //sweap out unused entries + uiInterface.InitMessage(_("Starting network threads...")); - LogPrintf("Loaded %i addresses from peers.dat %dms\n", - addrman.size(), GetTimeMillis() - nStart); fAddressesInitialized = true; if (semOutbound == NULL) { // initialize semaphore - int nMaxOutbound = min(MAX_OUTBOUND_CONNECTIONS, nMaxConnections); - semOutbound = new CSemaphore(nMaxOutbound); + semOutbound = new CSemaphore(std::min((nMaxOutbound + nMaxFeeler), nMaxConnections)); + } + if (semAddnode == NULL) { + // initialize semaphore + semAddnode = new CSemaphore(nMaxAddnode); } - - if (pnodeLocalHost == NULL) - pnodeLocalHost = new CNode(INVALID_SOCKET, CAddress(CService("127.0.0.1", 0), nLocalServices)); - - Discover(threadGroup); // // Start threads // + InterruptSocks5(false); + interruptNet.reset(); + flagInterruptMsgProc = false; + + { + std::unique_lock<std::mutex> lock(mutexMsgProc); + fMsgProcWake = false; + } + + // Send and receive from sockets, accept connections + threadSocketHandler = std::thread(&TraceThread<std::function<void()> >, "net", std::function<void()>(std::bind(&CConnman::ThreadSocketHandler, this))); if (!GetBoolArg("-dnsseed", true)) LogPrintf("DNS seeding disabled\n"); else - threadGroup.create_thread(boost::bind(&TraceThread<void (*)()>, "dnsseed", &ThreadDNSAddressSeed)); - - // Map ports with UPnP - MapPort(GetBoolArg("-upnp", DEFAULT_UPNP)); - - // Send and receive from sockets, accept connections - threadGroup.create_thread(boost::bind(&TraceThread<void (*)()>, "net", &ThreadSocketHandler)); + threadDNSAddressSeed = std::thread(&TraceThread<std::function<void()> >, "dnsseed", std::function<void()>(std::bind(&CConnman::ThreadDNSAddressSeed, this))); // Initiate outbound connections from -addnode - threadGroup.create_thread(boost::bind(&TraceThread<void (*)()>, "addcon", &ThreadOpenAddedConnections)); + threadOpenAddedConnections = std::thread(&TraceThread<std::function<void()> >, "addcon", std::function<void()>(std::bind(&CConnman::ThreadOpenAddedConnections, this))); - // Initiate outbound connections - threadGroup.create_thread(boost::bind(&TraceThread<void (*)()>, "opencon", &ThreadOpenConnections)); + // Initiate outbound connections unless connect=0 + if (!mapMultiArgs.count("-connect") || mapMultiArgs.at("-connect").size() != 1 || mapMultiArgs.at("-connect")[0] != "0") + threadOpenConnections = std::thread(&TraceThread<std::function<void()> >, "opencon", std::function<void()>(std::bind(&CConnman::ThreadOpenConnections, this))); // Process messages - threadGroup.create_thread(boost::bind(&TraceThread<void (*)()>, "msghand", &ThreadMessageHandler)); + threadMessageHandler = std::thread(&TraceThread<std::function<void()> >, "msghand", std::function<void()>(std::bind(&CConnman::ThreadMessageHandler, this))); // Dump network addresses - scheduler.scheduleEvery(&DumpData, DUMP_ADDRESSES_INTERVAL); -} - -bool StopNode() -{ - LogPrintf("StopNode()\n"); - MapPort(false); - if (semOutbound) - for (int i=0; i<MAX_OUTBOUND_CONNECTIONS; i++) - semOutbound->post(); - - if (fAddressesInitialized) - { - DumpData(); - fAddressesInitialized = false; - } + scheduler.scheduleEvery(boost::bind(&CConnman::DumpData, this), DUMP_ADDRESSES_INTERVAL); return true; } @@ -2013,28 +2299,6 @@ public: ~CNetCleanup() { - // Close sockets - BOOST_FOREACH(CNode* pnode, vNodes) - if (pnode->hSocket != INVALID_SOCKET) - CloseSocket(pnode->hSocket); - BOOST_FOREACH(ListenSocket& hListenSocket, vhListenSocket) - if (hListenSocket.socket != INVALID_SOCKET) - if (!CloseSocket(hListenSocket.socket)) - LogPrintf("CloseSocket(hListenSocket) failed with error %s\n", NetworkErrorString(WSAGetLastError())); - - // clean up some globals (to help leak detection) - BOOST_FOREACH(CNode *pnode, vNodes) - delete pnode; - BOOST_FOREACH(CNode *pnode, vNodesDisconnected) - delete pnode; - vNodes.clear(); - vNodesDisconnected.clear(); - vhListenSocket.clear(); - delete semOutbound; - semOutbound = NULL; - delete pnodeLocalHost; - pnodeLocalHost = NULL; - #ifdef WIN32 // Shutdown Windows Sockets WSACleanup(); @@ -2043,58 +2307,193 @@ public: } instance_of_cnetcleanup; +void CConnman::Interrupt() +{ + { + std::lock_guard<std::mutex> lock(mutexMsgProc); + flagInterruptMsgProc = true; + } + condMsgProc.notify_all(); + + interruptNet(); + InterruptSocks5(true); + + if (semOutbound) + for (int i=0; i<(nMaxOutbound + nMaxFeeler); i++) + semOutbound->post(); +} + +void CConnman::Stop() +{ + if (threadMessageHandler.joinable()) + threadMessageHandler.join(); + if (threadOpenConnections.joinable()) + threadOpenConnections.join(); + if (threadOpenAddedConnections.joinable()) + threadOpenAddedConnections.join(); + if (threadDNSAddressSeed.joinable()) + threadDNSAddressSeed.join(); + if (threadSocketHandler.joinable()) + threadSocketHandler.join(); + + if (semAddnode) + for (int i=0; i<nMaxAddnode; i++) + semOutbound->post(); + if (fAddressesInitialized) + { + DumpData(); + fAddressesInitialized = false; + } + // Close sockets + BOOST_FOREACH(CNode* pnode, vNodes) + pnode->CloseSocketDisconnect(); + BOOST_FOREACH(ListenSocket& hListenSocket, vhListenSocket) + if (hListenSocket.socket != INVALID_SOCKET) + if (!CloseSocket(hListenSocket.socket)) + LogPrintf("CloseSocket(hListenSocket) failed with error %s\n", NetworkErrorString(WSAGetLastError())); + // clean up some globals (to help leak detection) + BOOST_FOREACH(CNode *pnode, vNodes) { + DeleteNode(pnode); + } + BOOST_FOREACH(CNode *pnode, vNodesDisconnected) { + DeleteNode(pnode); + } + vNodes.clear(); + vNodesDisconnected.clear(); + vhListenSocket.clear(); + delete semOutbound; + semOutbound = NULL; + delete semAddnode; + semAddnode = NULL; +} +void CConnman::DeleteNode(CNode* pnode) +{ + assert(pnode); + bool fUpdateConnectionTime = false; + GetNodeSignals().FinalizeNode(pnode->GetId(), fUpdateConnectionTime); + if(fUpdateConnectionTime) + addrman.Connected(pnode->addr); + delete pnode; +} +CConnman::~CConnman() +{ + Interrupt(); + Stop(); +} -void RelayTransaction(const CTransaction& tx) +size_t CConnman::GetAddressCount() const { - CDataStream ss(SER_NETWORK, PROTOCOL_VERSION); - ss.reserve(10000); - ss << tx; - RelayTransaction(tx, ss); + return addrman.size(); } -void RelayTransaction(const CTransaction& tx, const CDataStream& ss) +void CConnman::SetServices(const CService &addr, ServiceFlags nServices) { - CInv inv(MSG_TX, tx.GetHash()); - { - LOCK(cs_mapRelay); - // Expire old relay messages - while (!vRelayExpiration.empty() && vRelayExpiration.front().first < GetTime()) - { - mapRelay.erase(vRelayExpiration.front().second); - vRelayExpiration.pop_front(); + addrman.SetServices(addr, nServices); +} + +void CConnman::MarkAddressGood(const CAddress& addr) +{ + addrman.Good(addr); +} + +void CConnman::AddNewAddress(const CAddress& addr, const CAddress& addrFrom, int64_t nTimePenalty) +{ + addrman.Add(addr, addrFrom, nTimePenalty); +} + +void CConnman::AddNewAddresses(const std::vector<CAddress>& vAddr, const CAddress& addrFrom, int64_t nTimePenalty) +{ + addrman.Add(vAddr, addrFrom, nTimePenalty); +} + +std::vector<CAddress> CConnman::GetAddresses() +{ + return addrman.GetAddr(); +} + +bool CConnman::AddNode(const std::string& strNode) +{ + LOCK(cs_vAddedNodes); + for(std::vector<std::string>::const_iterator it = vAddedNodes.begin(); it != vAddedNodes.end(); ++it) { + if (strNode == *it) + return false; + } + + vAddedNodes.push_back(strNode); + return true; +} + +bool CConnman::RemoveAddedNode(const std::string& strNode) +{ + LOCK(cs_vAddedNodes); + for(std::vector<std::string>::iterator it = vAddedNodes.begin(); it != vAddedNodes.end(); ++it) { + if (strNode == *it) { + vAddedNodes.erase(it); + return true; } + } + return false; +} + +size_t CConnman::GetNodeCount(NumConnections flags) +{ + LOCK(cs_vNodes); + if (flags == CConnman::CONNECTIONS_ALL) // Shortcut if we want total + return vNodes.size(); - // Save original serialized message so newer versions are preserved - mapRelay.insert(std::make_pair(inv, ss)); - vRelayExpiration.push_back(std::make_pair(GetTime() + 15 * 60, inv)); + int nNum = 0; + for(std::vector<CNode*>::const_iterator it = vNodes.begin(); it != vNodes.end(); ++it) + if (flags & ((*it)->fInbound ? CONNECTIONS_IN : CONNECTIONS_OUT)) + nNum++; + + return nNum; +} + +void CConnman::GetNodeStats(std::vector<CNodeStats>& vstats) +{ + vstats.clear(); + LOCK(cs_vNodes); + vstats.reserve(vNodes.size()); + for(std::vector<CNode*>::iterator it = vNodes.begin(); it != vNodes.end(); ++it) { + CNode* pnode = *it; + vstats.emplace_back(); + pnode->copyStats(vstats.back()); } +} + +bool CConnman::DisconnectNode(const std::string& strNode) +{ LOCK(cs_vNodes); - BOOST_FOREACH(CNode* pnode, vNodes) - { - if(!pnode->fRelayTxes) - continue; - LOCK(pnode->cs_filter); - if (pnode->pfilter) - { - if (pnode->pfilter->IsRelevantAndUpdate(tx)) - pnode->PushInventory(inv); - } else - pnode->PushInventory(inv); + if (CNode* pnode = FindNode(strNode)) { + pnode->fDisconnect = true; + return true; } + return false; +} +bool CConnman::DisconnectNode(NodeId id) +{ + LOCK(cs_vNodes); + for(CNode* pnode : vNodes) { + if (id == pnode->id) { + pnode->fDisconnect = true; + return true; + } + } + return false; } -void CNode::RecordBytesRecv(uint64_t bytes) +void CConnman::RecordBytesRecv(uint64_t bytes) { LOCK(cs_totalBytesRecv); nTotalBytesRecv += bytes; } -void CNode::RecordBytesSent(uint64_t bytes) +void CConnman::RecordBytesSent(uint64_t bytes) { LOCK(cs_totalBytesSent); nTotalBytesSent += bytes; @@ -2111,29 +2510,25 @@ void CNode::RecordBytesSent(uint64_t bytes) nMaxOutboundTotalBytesSentInCycle += bytes; } -void CNode::SetMaxOutboundTarget(uint64_t limit) +void CConnman::SetMaxOutboundTarget(uint64_t limit) { LOCK(cs_totalBytesSent); - uint64_t recommendedMinimum = (nMaxOutboundTimeframe / 600) * MAX_BLOCK_SIZE; nMaxOutboundLimit = limit; - - if (limit > 0 && limit < recommendedMinimum) - LogPrintf("Max outbound target is very small (%s bytes) and will be overshot. Recommended minimum is %s bytes.\n", nMaxOutboundLimit, recommendedMinimum); } -uint64_t CNode::GetMaxOutboundTarget() +uint64_t CConnman::GetMaxOutboundTarget() { LOCK(cs_totalBytesSent); return nMaxOutboundLimit; } -uint64_t CNode::GetMaxOutboundTimeframe() +uint64_t CConnman::GetMaxOutboundTimeframe() { LOCK(cs_totalBytesSent); return nMaxOutboundTimeframe; } -uint64_t CNode::GetMaxOutboundTimeLeftInCycle() +uint64_t CConnman::GetMaxOutboundTimeLeftInCycle() { LOCK(cs_totalBytesSent); if (nMaxOutboundLimit == 0) @@ -2147,7 +2542,7 @@ uint64_t CNode::GetMaxOutboundTimeLeftInCycle() return (cycleEndTime < now) ? 0 : cycleEndTime - GetTime(); } -void CNode::SetMaxOutboundTimeframe(uint64_t timeframe) +void CConnman::SetMaxOutboundTimeframe(uint64_t timeframe) { LOCK(cs_totalBytesSent); if (nMaxOutboundTimeframe != timeframe) @@ -2159,7 +2554,7 @@ void CNode::SetMaxOutboundTimeframe(uint64_t timeframe) nMaxOutboundTimeframe = timeframe; } -bool CNode::OutboundTargetReached(bool historicalBlockServingLimit) +bool CConnman::OutboundTargetReached(bool historicalBlockServingLimit) { LOCK(cs_totalBytesSent); if (nMaxOutboundLimit == 0) @@ -2167,9 +2562,9 @@ bool CNode::OutboundTargetReached(bool historicalBlockServingLimit) if (historicalBlockServingLimit) { - // keep a large enought buffer to at least relay each block once + // keep a large enough buffer to at least relay each block once uint64_t timeLeftInCycle = GetMaxOutboundTimeLeftInCycle(); - uint64_t buffer = timeLeftInCycle / 600 * MAX_BLOCK_SIZE; + uint64_t buffer = timeLeftInCycle / 600 * MAX_BLOCK_SERIALIZED_SIZE; if (buffer >= nMaxOutboundLimit || nMaxOutboundTotalBytesSentInCycle >= nMaxOutboundLimit - buffer) return true; } @@ -2179,7 +2574,7 @@ bool CNode::OutboundTargetReached(bool historicalBlockServingLimit) return false; } -uint64_t CNode::GetOutboundTargetBytesLeft() +uint64_t CConnman::GetOutboundTargetBytesLeft() { LOCK(cs_totalBytesSent); if (nMaxOutboundLimit == 0) @@ -2188,180 +2583,66 @@ uint64_t CNode::GetOutboundTargetBytesLeft() return (nMaxOutboundTotalBytesSentInCycle >= nMaxOutboundLimit) ? 0 : nMaxOutboundLimit - nMaxOutboundTotalBytesSentInCycle; } -uint64_t CNode::GetTotalBytesRecv() +uint64_t CConnman::GetTotalBytesRecv() { LOCK(cs_totalBytesRecv); return nTotalBytesRecv; } -uint64_t CNode::GetTotalBytesSent() +uint64_t CConnman::GetTotalBytesSent() { LOCK(cs_totalBytesSent); return nTotalBytesSent; } -void CNode::Fuzz(int nChance) -{ - if (!fSuccessfullyConnected) return; // Don't fuzz initial handshake - if (GetRand(nChance) != 0) return; // Fuzz 1 of every nChance messages - - switch (GetRand(3)) - { - case 0: - // xor a random byte with a random value: - if (!ssSend.empty()) { - CDataStream::size_type pos = GetRand(ssSend.size()); - ssSend[pos] ^= (unsigned char)(GetRand(256)); - } - break; - case 1: - // delete a random byte: - if (!ssSend.empty()) { - CDataStream::size_type pos = GetRand(ssSend.size()); - ssSend.erase(ssSend.begin()+pos); - } - break; - case 2: - // insert a random byte at a random position - { - CDataStream::size_type pos = GetRand(ssSend.size()); - char ch = (char)GetRand(256); - ssSend.insert(ssSend.begin()+pos, ch); - } - break; - } - // Chance of more than one change half the time: - // (more changes exponentially less likely): - Fuzz(2); -} - -// -// CAddrDB -// - -CAddrDB::CAddrDB() +ServiceFlags CConnman::GetLocalServices() const { - pathAddr = GetDataDir() / "peers.dat"; + return nLocalServices; } -bool CAddrDB::Write(const CAddrMan& addr) +void CConnman::SetBestHeight(int height) { - // Generate random temporary filename - unsigned short randv = 0; - GetRandBytes((unsigned char*)&randv, sizeof(randv)); - std::string tmpfn = strprintf("peers.dat.%04x", randv); - - // serialize addresses, checksum data up to that point, then append csum - CDataStream ssPeers(SER_DISK, CLIENT_VERSION); - ssPeers << FLATDATA(Params().MessageStart()); - ssPeers << addr; - uint256 hash = Hash(ssPeers.begin(), ssPeers.end()); - ssPeers << hash; - - // open temp output file, and associate with CAutoFile - boost::filesystem::path pathTmp = GetDataDir() / tmpfn; - FILE *file = fopen(pathTmp.string().c_str(), "wb"); - CAutoFile fileout(file, SER_DISK, CLIENT_VERSION); - if (fileout.IsNull()) - return error("%s: Failed to open file %s", __func__, pathTmp.string()); - - // Write and commit header, data - try { - fileout << ssPeers; - } - catch (const std::exception& e) { - return error("%s: Serialize or I/O error - %s", __func__, e.what()); - } - FileCommit(fileout.Get()); - fileout.fclose(); - - // replace existing peers.dat, if any, with new peers.dat.XXXX - if (!RenameOver(pathTmp, pathAddr)) - return error("%s: Rename-into-place failed", __func__); - - return true; + nBestHeight.store(height, std::memory_order_release); } -bool CAddrDB::Read(CAddrMan& addr) +int CConnman::GetBestHeight() const { - // open input file, and associate with CAutoFile - FILE *file = fopen(pathAddr.string().c_str(), "rb"); - CAutoFile filein(file, SER_DISK, CLIENT_VERSION); - if (filein.IsNull()) - return error("%s: Failed to open file %s", __func__, pathAddr.string()); - - // use file size to size memory buffer - uint64_t fileSize = boost::filesystem::file_size(pathAddr); - uint64_t dataSize = 0; - // Don't try to resize to a negative number if file is small - if (fileSize >= sizeof(uint256)) - dataSize = fileSize - sizeof(uint256); - vector<unsigned char> vchData; - vchData.resize(dataSize); - uint256 hashIn; - - // read data and checksum from file - try { - filein.read((char *)&vchData[0], dataSize); - filein >> hashIn; - } - catch (const std::exception& e) { - return error("%s: Deserialize or I/O error - %s", __func__, e.what()); - } - filein.fclose(); - - CDataStream ssPeers(vchData, SER_DISK, CLIENT_VERSION); - - // verify stored checksum matches input data - uint256 hashTmp = Hash(ssPeers.begin(), ssPeers.end()); - if (hashIn != hashTmp) - return error("%s: Checksum mismatch, data corrupted", __func__); - - unsigned char pchMsgTmp[4]; - try { - // de-serialize file header (network specific magic number) and .. - ssPeers >> FLATDATA(pchMsgTmp); - - // ... verify the network matches ours - if (memcmp(pchMsgTmp, Params().MessageStart(), sizeof(pchMsgTmp))) - return error("%s: Invalid network magic number", __func__); - - // de-serialize address data into one CAddrMan object - ssPeers >> addr; - } - catch (const std::exception& e) { - return error("%s: Deserialize or I/O error - %s", __func__, e.what()); - } - - return true; + return nBestHeight.load(std::memory_order_acquire); } -unsigned int ReceiveFloodSize() { return 1000*GetArg("-maxreceivebuffer", DEFAULT_MAXRECEIVEBUFFER); } -unsigned int SendBufferSize() { return 1000*GetArg("-maxsendbuffer", DEFAULT_MAXSENDBUFFER); } +unsigned int CConnman::GetReceiveFloodSize() const { return nReceiveFloodSize; } +unsigned int CConnman::GetSendBufferSize() const{ return nSendBufferMaxSize; } -CNode::CNode(SOCKET hSocketIn, const CAddress& addrIn, const std::string& addrNameIn, bool fInboundIn) : - ssSend(SER_NETWORK, INIT_PROTO_VERSION), +CNode::CNode(NodeId idIn, ServiceFlags nLocalServicesIn, int nMyStartingHeightIn, SOCKET hSocketIn, const CAddress& addrIn, uint64_t nKeyedNetGroupIn, uint64_t nLocalHostNonceIn, const std::string& addrNameIn, bool fInboundIn) : + nTimeConnected(GetSystemTimeInSeconds()), + addr(addrIn), + fInbound(fInboundIn), + id(idIn), + nKeyedNetGroup(nKeyedNetGroupIn), addrKnown(5000, 0.001), - setInventoryKnown(SendBufferSize() / 1000) -{ - nServices = 0; + filterInventoryKnown(50000, 0.000001), + nLocalHostNonce(nLocalHostNonceIn), + nLocalServices(nLocalServicesIn), + nMyStartingHeight(nMyStartingHeightIn), + nSendVersion(0) +{ + nServices = NODE_NONE; + nServicesExpected = NODE_NONE; hSocket = hSocketIn; nRecvVersion = INIT_PROTO_VERSION; nLastSend = 0; nLastRecv = 0; nSendBytes = 0; nRecvBytes = 0; - nTimeConnected = GetTime(); nTimeOffset = 0; - addr = addrIn; addrName = addrNameIn == "" ? addr.ToStringIPPort() : addrNameIn; nVersion = 0; strSubVer = ""; fWhitelisted = false; fOneShot = false; + fAddnode = false; fClient = false; // set by version message - fInbound = fInboundIn; - fNetworkNode = false; + fFeeler = false; fSuccessfullyConnected = false; fDisconnect = false; nRefCount = 0; @@ -2369,30 +2650,38 @@ CNode::CNode(SOCKET hSocketIn, const CAddress& addrIn, const std::string& addrNa nSendOffset = 0; hashContinue = uint256(); nStartingHeight = -1; + filterInventoryKnown.reset(); + fSendMempool = false; fGetAddr = false; + nNextLocalAddrSend = 0; + nNextAddrSend = 0; + nNextInvSend = 0; fRelayTxes = false; + fSentAddr = false; pfilter = new CBloomFilter(); + timeLastMempoolReq = 0; + nLastBlockTime = 0; + nLastTXTime = 0; nPingNonceSent = 0; nPingUsecStart = 0; nPingUsecTime = 0; fPingQueued = false; nMinPingUsecTime = std::numeric_limits<int64_t>::max(); + minFeeFilter = 0; + lastSentFeeFilter = 0; + nextSendTimeFeeFilter = 0; + fPauseRecv = false; + fPauseSend = false; + nProcessQueueSize = 0; - { - LOCK(cs_nLastNodeId); - id = nLastNodeId++; - } + BOOST_FOREACH(const std::string &msg, getAllNetMessageTypes()) + mapRecvBytesPerMsgCmd[msg] = 0; + mapRecvBytesPerMsgCmd[NET_MESSAGE_COMMAND_OTHER] = 0; if (fLogIPs) LogPrint("net", "Added connection to %s peer=%d\n", addrName, id); else LogPrint("net", "Added connection peer=%d\n", id); - - // Be shy and don't send version until we hear - if (hSocket != INVALID_SOCKET && !fInbound) - PushVersion(); - - GetNodeSignals().InitializeNode(GetId(), this); } CNode::~CNode() @@ -2401,8 +2690,6 @@ CNode::~CNode() if (pfilter) delete pfilter; - - GetNodeSignals().FinalizeNode(GetId()); } void CNode::AskFor(const CInv& inv) @@ -2416,7 +2703,7 @@ void CNode::AskFor(const CInv& inv) // We're using mapAskFor as a priority queue, // the key is the earliest time the request can be sent int64_t nRequestTime; - limitedmap<CInv, int64_t>::const_iterator it = mapAlreadyAskedFor.find(inv); + limitedmap<uint256, int64_t>::const_iterator it = mapAlreadyAskedFor.find(inv.hash); if (it != mapAlreadyAskedFor.end()) nRequestTime = it->second; else @@ -2435,182 +2722,77 @@ void CNode::AskFor(const CInv& inv) if (it != mapAlreadyAskedFor.end()) mapAlreadyAskedFor.update(it, nRequestTime); else - mapAlreadyAskedFor.insert(std::make_pair(inv, nRequestTime)); + mapAlreadyAskedFor.insert(std::make_pair(inv.hash, nRequestTime)); mapAskFor.insert(std::make_pair(nRequestTime, inv)); } -void CNode::BeginMessage(const char* pszCommand) EXCLUSIVE_LOCK_FUNCTION(cs_vSend) +bool CConnman::NodeFullyConnected(const CNode* pnode) { - ENTER_CRITICAL_SECTION(cs_vSend); - assert(ssSend.size() == 0); - ssSend << CMessageHeader(Params().MessageStart(), pszCommand, 0); - LogPrint("net", "sending: %s ", SanitizeString(pszCommand)); + return pnode && pnode->fSuccessfullyConnected && !pnode->fDisconnect; } -void CNode::AbortMessage() UNLOCK_FUNCTION(cs_vSend) +void CConnman::PushMessage(CNode* pnode, CSerializedNetMsg&& msg) { - ssSend.clear(); + size_t nMessageSize = msg.data.size(); + size_t nTotalSize = nMessageSize + CMessageHeader::HEADER_SIZE; + LogPrint("net", "sending %s (%d bytes) peer=%d\n", SanitizeString(msg.command.c_str()), nMessageSize, pnode->id); - LEAVE_CRITICAL_SECTION(cs_vSend); + std::vector<unsigned char> serializedHeader; + serializedHeader.reserve(CMessageHeader::HEADER_SIZE); + uint256 hash = Hash(msg.data.data(), msg.data.data() + nMessageSize); + CMessageHeader hdr(Params().MessageStart(), msg.command.c_str(), nMessageSize); + memcpy(hdr.pchChecksum, hash.begin(), CMessageHeader::CHECKSUM_SIZE); - LogPrint("net", "(aborted)\n"); -} + CVectorWriter{SER_NETWORK, INIT_PROTO_VERSION, serializedHeader, 0, hdr}; -void CNode::EndMessage() UNLOCK_FUNCTION(cs_vSend) -{ - // The -*messagestest options are intentionally not documented in the help message, - // since they are only used during development to debug the networking code and are - // not intended for end-users. - if (mapArgs.count("-dropmessagestest") && GetRand(GetArg("-dropmessagestest", 2)) == 0) + size_t nBytesSent = 0; { - LogPrint("net", "dropmessages DROPPING SEND MESSAGE\n"); - AbortMessage(); - return; - } - if (mapArgs.count("-fuzzmessagestest")) - Fuzz(GetArg("-fuzzmessagestest", 10)); + LOCK(pnode->cs_vSend); + bool optimisticSend(pnode->vSendMsg.empty()); - if (ssSend.size() == 0) - { - LEAVE_CRITICAL_SECTION(cs_vSend); - return; - } - // Set the size - unsigned int nSize = ssSend.size() - CMessageHeader::HEADER_SIZE; - WriteLE32((uint8_t*)&ssSend[CMessageHeader::MESSAGE_SIZE_OFFSET], nSize); - - // Set the checksum - uint256 hash = Hash(ssSend.begin() + CMessageHeader::HEADER_SIZE, ssSend.end()); - unsigned int nChecksum = 0; - memcpy(&nChecksum, &hash, sizeof(nChecksum)); - assert(ssSend.size () >= CMessageHeader::CHECKSUM_OFFSET + sizeof(nChecksum)); - memcpy((char*)&ssSend[CMessageHeader::CHECKSUM_OFFSET], &nChecksum, sizeof(nChecksum)); - - LogPrint("net", "(%d bytes) peer=%d\n", nSize, id); + //log total amount of bytes per command + pnode->mapSendBytesPerMsgCmd[msg.command] += nTotalSize; + pnode->nSendSize += nTotalSize; - std::deque<CSerializeData>::iterator it = vSendMsg.insert(vSendMsg.end(), CSerializeData()); - ssSend.GetAndClear(*it); - nSendSize += (*it).size(); + if (pnode->nSendSize > nSendBufferMaxSize) + pnode->fPauseSend = true; + pnode->vSendMsg.push_back(std::move(serializedHeader)); + if (nMessageSize) + pnode->vSendMsg.push_back(std::move(msg.data)); - // If write queue empty, attempt "optimistic write" - if (it == vSendMsg.begin()) - SocketSendData(this); - - LEAVE_CRITICAL_SECTION(cs_vSend); -} - -// -// CBanDB -// - -CBanDB::CBanDB() -{ - pathBanlist = GetDataDir() / "banlist.dat"; + // If write queue empty, attempt "optimistic write" + if (optimisticSend == true) + nBytesSent = SocketSendData(pnode); + } + if (nBytesSent) + RecordBytesSent(nBytesSent); } -bool CBanDB::Write(const banmap_t& banSet) +bool CConnman::ForNode(NodeId id, std::function<bool(CNode* pnode)> func) { - // Generate random temporary filename - unsigned short randv = 0; - GetRandBytes((unsigned char*)&randv, sizeof(randv)); - std::string tmpfn = strprintf("banlist.dat.%04x", randv); - - // serialize banlist, checksum data up to that point, then append csum - CDataStream ssBanlist(SER_DISK, CLIENT_VERSION); - ssBanlist << FLATDATA(Params().MessageStart()); - ssBanlist << banSet; - uint256 hash = Hash(ssBanlist.begin(), ssBanlist.end()); - ssBanlist << hash; - - // open temp output file, and associate with CAutoFile - boost::filesystem::path pathTmp = GetDataDir() / tmpfn; - FILE *file = fopen(pathTmp.string().c_str(), "wb"); - CAutoFile fileout(file, SER_DISK, CLIENT_VERSION); - if (fileout.IsNull()) - return error("%s: Failed to open file %s", __func__, pathTmp.string()); - - // Write and commit header, data - try { - fileout << ssBanlist; - } - catch (const std::exception& e) { - return error("%s: Serialize or I/O error - %s", __func__, e.what()); + CNode* found = nullptr; + LOCK(cs_vNodes); + for (auto&& pnode : vNodes) { + if(pnode->id == id) { + found = pnode; + break; + } } - FileCommit(fileout.Get()); - fileout.fclose(); - - // replace existing banlist.dat, if any, with new banlist.dat.XXXX - if (!RenameOver(pathTmp, pathBanlist)) - return error("%s: Rename-into-place failed", __func__); + return found != nullptr && NodeFullyConnected(found) && func(found); +} - return true; +int64_t PoissonNextSend(int64_t nNow, int average_interval_seconds) { + return nNow + (int64_t)(log1p(GetRand(1ULL << 48) * -0.0000000000000035527136788 /* -1/2^48 */) * average_interval_seconds * -1000000.0 + 0.5); } -bool CBanDB::Read(banmap_t& banSet) +CSipHasher CConnman::GetDeterministicRandomizer(uint64_t id) const { - // open input file, and associate with CAutoFile - FILE *file = fopen(pathBanlist.string().c_str(), "rb"); - CAutoFile filein(file, SER_DISK, CLIENT_VERSION); - if (filein.IsNull()) - return error("%s: Failed to open file %s", __func__, pathBanlist.string()); - - // use file size to size memory buffer - uint64_t fileSize = boost::filesystem::file_size(pathBanlist); - uint64_t dataSize = 0; - // Don't try to resize to a negative number if file is small - if (fileSize >= sizeof(uint256)) - dataSize = fileSize - sizeof(uint256); - vector<unsigned char> vchData; - vchData.resize(dataSize); - uint256 hashIn; - - // read data and checksum from file - try { - filein.read((char *)&vchData[0], dataSize); - filein >> hashIn; - } - catch (const std::exception& e) { - return error("%s: Deserialize or I/O error - %s", __func__, e.what()); - } - filein.fclose(); - - CDataStream ssBanlist(vchData, SER_DISK, CLIENT_VERSION); - - // verify stored checksum matches input data - uint256 hashTmp = Hash(ssBanlist.begin(), ssBanlist.end()); - if (hashIn != hashTmp) - return error("%s: Checksum mismatch, data corrupted", __func__); - - unsigned char pchMsgTmp[4]; - try { - // de-serialize file header (network specific magic number) and .. - ssBanlist >> FLATDATA(pchMsgTmp); - - // ... verify the network matches ours - if (memcmp(pchMsgTmp, Params().MessageStart(), sizeof(pchMsgTmp))) - return error("%s: Invalid network magic number", __func__); - - // de-serialize address data into one CAddrMan object - ssBanlist >> banSet; - } - catch (const std::exception& e) { - return error("%s: Deserialize or I/O error - %s", __func__, e.what()); - } - - return true; + return CSipHasher(nSeed0, nSeed1).Write(id); } -void DumpBanlist() +uint64_t CConnman::CalculateKeyedNetGroup(const CAddress& ad) const { - int64_t nStart = GetTimeMillis(); - - CNode::SweepBanned(); //clean unused entries (if bantime has expired) + std::vector<unsigned char> vchNetGroup(ad.GetGroup()); - CBanDB bandb; - banmap_t banmap; - CNode::GetBanned(banmap); - bandb.Write(banmap); - - LogPrint("net", "Flushed %d banned node ips/subnets to banlist.dat %dms\n", - banmap.size(), GetTimeMillis() - nStart); + return GetDeterministicRandomizer(RANDOMIZER_ID_NETGROUP).Write(&vchNetGroup[0], vchNetGroup.size()).Finalize(); } |