diff options
| -rwxr-xr-x | qa/rpc-tests/rpcnamedargs.py | 2 | ||||
| -rw-r--r-- | qa/rpc-tests/test_framework/util.py | 23 | ||||
| -rw-r--r-- | src/net.cpp | 161 | ||||
| -rw-r--r-- | src/net.h | 43 | ||||
| -rw-r--r-- | src/net_processing.cpp | 21 |
5 files changed, 175 insertions, 75 deletions
diff --git a/qa/rpc-tests/rpcnamedargs.py b/qa/rpc-tests/rpcnamedargs.py index 0484204668..da2d8f040f 100755 --- a/qa/rpc-tests/rpcnamedargs.py +++ b/qa/rpc-tests/rpcnamedargs.py @@ -37,7 +37,7 @@ class NamedArgumentTest(BitcoinTestFramework): h = node.help(command='getinfo') assert(h.startswith('getinfo\n')) - assert_raises_jsonrpc(-8, node.help, random='getinfo') + assert_raises_jsonrpc(-8, 'Unknown named parameter', node.help, random='getinfo') h = node.getblockhash(height=0) node.getblock(blockhash=h) diff --git a/qa/rpc-tests/test_framework/util.py b/qa/rpc-tests/test_framework/util.py index aca82c8b6f..dc8555c44c 100644 --- a/qa/rpc-tests/test_framework/util.py +++ b/qa/rpc-tests/test_framework/util.py @@ -550,13 +550,30 @@ def assert_raises_message(exc, message, fun, *args, **kwds): else: raise AssertionError("No exception raised") -def assert_raises_jsonrpc(code, fun, *args, **kwds): - '''Check for specific JSONRPC exception code''' +def assert_raises_jsonrpc(code, message, fun, *args, **kwds): + """Run an RPC and verify that a specific JSONRPC exception code and message is raised. + + Calls function `fun` with arguments `args` and `kwds`. Catches a JSONRPCException + and verifies that the error code and message are as expected. Throws AssertionError if + no JSONRPCException was returned or if the error code/message are not as expected. + + Args: + code (int), optional: the error code returned by the RPC call (defined + in src/rpc/protocol.h). Set to None if checking the error code is not required. + message (string), optional: [a substring of] the error string returned by the + RPC call. Set to None if checking the error string is not required + fun (function): the function to call. This should be the name of an RPC. + args*: positional arguments for the function. + kwds**: named arguments for the function. + """ try: fun(*args, **kwds) except JSONRPCException as e: - if e.error["code"] != code: + # JSONRPCException was thrown as expected. Check the code and message values are correct. + if (code is not None) and (code != e.error["code"]): raise AssertionError("Unexpected JSONRPC error code %i" % e.error["code"]) + if (message is not None) and (message not in e.error['message']): + raise AssertionError("Expected substring not found:"+e.error['message']) except Exception as e: raise AssertionError("Unexpected exception raised: "+type(e).__name__) else: diff --git a/src/net.cpp b/src/net.cpp index 2242afb863..505eb971c0 100644 --- a/src/net.cpp +++ b/src/net.cpp @@ -164,8 +164,9 @@ 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 @@ -180,7 +181,7 @@ void AdvertiseLocal(CNode *pnode) if (IsPeerAddrLocalGood(pnode) && (!addrLocal.IsRoutable() || GetRand((GetnScore(addrLocal) > LOCAL_MANUAL) ? 8:2) == 0)) { - addrLocal.SetIP(pnode->addrLocal); + addrLocal.SetIP(pnode->GetAddrLocal()); } if (addrLocal.IsRoutable()) { @@ -307,9 +308,11 @@ CNode* CConnman::FindNode(const CSubNet& subNet) 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; } @@ -373,9 +376,7 @@ CNode* CConnman::ConnectNode(CAddress addrConnect, const char *pszDest, bool fCo CNode* pnode = FindNode((CService)addrConnect); if (pnode) { - if (pnode->addrName.empty()) { - pnode->addrName = std::string(pszDest); - } + pnode->MaybeSetAddrName(std::string(pszDest)); CloseSocket(hSocket); LogPrintf("Failed to open new connection, already connected\n"); return NULL; @@ -389,7 +390,6 @@ CNode* CConnman::ConnectNode(CAddress addrConnect, const char *pszDest, bool fCo 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->nTimeConnected = GetSystemTimeInSeconds(); pnode->AddRef(); return pnode; @@ -425,6 +425,7 @@ void CConnman::DumpBanlist() void CNode::CloseSocketDisconnect() { fDisconnect = true; + LOCK(cs_hSocket); if (hSocket != INVALID_SOCKET) { LogPrint("net", "disconnecting peer=%d\n", id); @@ -593,6 +594,33 @@ void CConnman::AddWhitelistedRange(const CSubNet &subnet) { 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) @@ -600,21 +628,33 @@ void CNode::copyStats(CNodeStats &stats) stats.nodeid = this->GetId(); X(nServices); X(addr); - X(fRelayTxes); + { + 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(mapSendBytesPerMsgCmd); - X(nRecvBytes); - X(mapRecvBytesPerMsgCmd); + { + 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, @@ -634,7 +674,8 @@ void CNode::copyStats(CNodeStats &stats) 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 @@ -642,6 +683,7 @@ 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) { @@ -789,7 +831,13 @@ size_t CConnman::SocketSendData(CNode *pnode) const while (it != pnode->vSendMsg.end()) { const auto &data = *it; assert(data.size() > pnode->nSendOffset); - int nBytes = send(pnode->hSocket, reinterpret_cast<const char*>(data.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 = GetSystemTimeInSeconds(); pnode->nSendBytes += nBytes; @@ -1148,12 +1196,6 @@ void CConnman::ThreadSocketHandler() LOCK(cs_vNodes); BOOST_FOREACH(CNode* pnode, vNodes) { - if (pnode->hSocket == INVALID_SOCKET) - continue; - FD_SET(pnode->hSocket, &fdsetError); - hSocketMax = std::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 @@ -1164,16 +1206,28 @@ void CConnman::ThreadSocketHandler() // receiving data. // * Hand off all complete messages to the processor, to be handled without // blocking here. + + bool select_recv = !pnode->fPauseRecv; + bool select_send; { LOCK(pnode->cs_vSend); - if (!pnode->vSendMsg.empty()) { - FD_SET(pnode->hSocket, &fdsetSend); - continue; - } + select_send = !pnode->vSendMsg.empty(); } - { - if (!pnode->fPauseRecv) - 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); } } } @@ -1227,15 +1281,30 @@ void CConnman::ThreadSocketHandler() // // 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) { { { // 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) { bool notify = false; @@ -1284,9 +1353,7 @@ void CConnman::ThreadSocketHandler() // // Send // - if (pnode->hSocket == INVALID_SOCKET) - continue; - if (FD_ISSET(pnode->hSocket, &fdsetSend)) + if (sendSet) { LOCK(pnode->cs_vSend); size_t nBytes = SocketSendData(pnode); @@ -1760,8 +1827,9 @@ std::vector<AddedNodeInfo> CConnman::GetAddedNodeInfo() if (pnode->addr.IsValid()) { mapConnected[pnode->addr] = pnode->fInbound; } - if (!pnode->addrName.empty()) { - mapConnectedByName[pnode->addrName] = std::make_pair(pnode->fInbound, static_cast<const CService&>(pnode->addr)); + std::string addrName = pnode->GetAddrName(); + if (!addrName.empty()) { + mapConnectedByName[std::move(addrName)] = std::make_pair(pnode->fInbound, static_cast<const CService&>(pnode->addr)); } } } @@ -2275,8 +2343,7 @@ void CConnman::Stop() // Close sockets BOOST_FOREACH(CNode* pnode, vNodes) - if (pnode->hSocket != INVALID_SOCKET) - CloseSocket(pnode->hSocket); + pnode->CloseSocketDisconnect(); BOOST_FOREACH(ListenSocket& hListenSocket, vhListenSocket) if (hListenSocket.socket != INVALID_SOCKET) if (!CloseSocket(hListenSocket.socket)) @@ -2389,9 +2456,8 @@ void CConnman::GetNodeStats(std::vector<CNodeStats>& vstats) vstats.reserve(vNodes.size()); for(std::vector<CNode*>::iterator it = vNodes.begin(); it != vNodes.end(); ++it) { CNode* pnode = *it; - CNodeStats stats; - pnode->copyStats(stats); - vstats.push_back(stats); + vstats.emplace_back(); + pnode->copyStats(vstats.back()); } } @@ -2543,6 +2609,7 @@ unsigned int CConnman::GetReceiveFloodSize() const { return nReceiveFloodSize; } unsigned int CConnman::GetSendBufferSize() const{ return nSendBufferMaxSize; } 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), @@ -2562,7 +2629,6 @@ CNode::CNode(NodeId idIn, ServiceFlags nLocalServicesIn, int nMyStartingHeightIn nLastRecv = 0; nSendBytes = 0; nRecvBytes = 0; - nTimeConnected = GetSystemTimeInSeconds(); nTimeOffset = 0; addrName = addrNameIn == "" ? addr.ToStringIPPort() : addrNameIn; nVersion = 0; @@ -2677,9 +2743,6 @@ void CConnman::PushMessage(CNode* pnode, CSerializedNetMsg&& msg) size_t nBytesSent = 0; { LOCK(pnode->cs_vSend); - if(pnode->hSocket == INVALID_SOCKET) { - return; - } bool optimisticSend(pnode->vSendMsg.empty()); //log total amount of bytes per command @@ -564,7 +564,7 @@ class CNode friend class CConnman; public: // socket - ServiceFlags nServices; + std::atomic<ServiceFlags> nServices; ServiceFlags nServicesExpected; SOCKET hSocket; size_t nSendSize; // total size of all vSendMsg entries @@ -572,6 +572,8 @@ public: uint64_t nSendBytes; std::deque<std::vector<unsigned char>> vSendMsg; CCriticalSection cs_vSend; + CCriticalSection cs_hSocket; + CCriticalSection cs_vRecv; CCriticalSection cs_vProcessMsg; std::list<CNetMessage> vProcessMsg; @@ -583,19 +585,18 @@ public: uint64_t nRecvBytes; std::atomic<int> nRecvVersion; - int64_t nLastSend; - int64_t nLastRecv; - int64_t nTimeConnected; - int64_t nTimeOffset; + std::atomic<int64_t> nLastSend; + std::atomic<int64_t> nLastRecv; + const int64_t nTimeConnected; + std::atomic<int64_t> nTimeOffset; const CAddress addr; - std::string addrName; - CService addrLocal; std::atomic<int> nVersion; // strSubVer is whatever byte array we read from the wire. However, this field is intended // to be printed out, displayed to humans in various forms and so on. So we sanitize it and // store the sanitized version in cleanSubVer. The original should be used when dealing with // the network or wire types and the cleaned string used when displayed or logged. std::string strSubVer, cleanSubVer; + CCriticalSection cs_SubVer; // used for both cleanSubVer and strSubVer bool fWhitelisted; // This peer can bypass DoS banning. bool fFeeler; // If true this node is being used as a short lived feeler. bool fOneShot; @@ -613,7 +614,7 @@ public: CSemaphoreGrant grantOutbound; CCriticalSection cs_filter; CBloomFilter* pfilter; - int nRefCount; + std::atomic<int> nRefCount; const NodeId id; const uint64_t nKeyedNetGroup; @@ -626,7 +627,7 @@ protected: public: uint256 hashContinue; - int nStartingHeight; + std::atomic<int> nStartingHeight; // flood relay std::vector<CAddress> vAddrToSend; @@ -664,15 +665,15 @@ public: // Ping time measurement: // The pong reply we're expecting, or 0 if no pong expected. - uint64_t nPingNonceSent; + std::atomic<uint64_t> nPingNonceSent; // Time (in usec) the last ping was sent, or 0 if no ping was ever sent. - int64_t nPingUsecStart; + std::atomic<int64_t> nPingUsecStart; // Last measured round-trip time. - int64_t nPingUsecTime; + std::atomic<int64_t> nPingUsecTime; // Best measured round-trip time. - int64_t nMinPingUsecTime; + std::atomic<int64_t> nMinPingUsecTime; // Whether a ping is requested. - bool fPingQueued; + std::atomic<bool> fPingQueued; // Minimum fee rate with which to filter inv's to this node CAmount minFeeFilter; CCriticalSection cs_feeFilter; @@ -693,6 +694,12 @@ private: const int nMyStartingHeight; int nSendVersion; std::list<CNetMessage> vRecvMsg; // Used only by SocketHandler thread + + mutable CCriticalSection cs_addrName; + std::string addrName; + + CService addrLocal; + mutable CCriticalSection cs_addrLocal; public: NodeId GetId() const { @@ -726,6 +733,10 @@ public: void SetSendVersion(int nVersionIn); int GetSendVersion() const; + CService GetAddrLocal() const; + //! May not be called more than once + void SetAddrLocal(const CService& addrLocalIn); + CNode* AddRef() { nRefCount++; @@ -795,6 +806,10 @@ public: { return nLocalServices; } + + std::string GetAddrName() const; + //! Sets the addrName only if it was not previously set + void MaybeSetAddrName(const std::string& addrNameIn); }; diff --git a/src/net_processing.cpp b/src/net_processing.cpp index bb14e69d83..7d76aa0b48 100644 --- a/src/net_processing.cpp +++ b/src/net_processing.cpp @@ -36,7 +36,7 @@ # error "Bitcoin cannot be compiled without assertions." #endif -int64_t nTimeBestReceived = 0; // Used only to inform the wallet of when we last received a block +std::atomic<int64_t> nTimeBestReceived(0); // Used only to inform the wallet of when we last received a block struct IteratorComparator { @@ -264,7 +264,7 @@ void PushNodeVersion(CNode *pnode, CConnman& connman, int64_t nTime) void InitializeNode(CNode *pnode, CConnman& connman) { CAddress addr = pnode->addr; - std::string addrName = pnode->addrName; + std::string addrName = pnode->GetAddrName(); NodeId nodeid = pnode->GetId(); { LOCK(cs_main); @@ -1211,6 +1211,7 @@ bool static ProcessMessage(CNode* pfrom, const std::string& strCommand, CDataStr int nVersion; int nSendVersion; std::string strSubVer; + std::string cleanSubVer; int nStartingHeight = -1; bool fRelay = true; @@ -1246,6 +1247,7 @@ bool static ProcessMessage(CNode* pfrom, const std::string& strCommand, CDataStr vRecv >> addrFrom >> nNonce; if (!vRecv.empty()) { vRecv >> LIMITED_STRING(strSubVer, MAX_SUBVERSION_LENGTH); + cleanSubVer = SanitizeString(strSubVer); } if (!vRecv.empty()) { vRecv >> nStartingHeight; @@ -1272,9 +1274,12 @@ bool static ProcessMessage(CNode* pfrom, const std::string& strCommand, CDataStr connman.PushMessage(pfrom, CNetMsgMaker(INIT_PROTO_VERSION).Make(NetMsgType::VERACK)); pfrom->nServices = nServices; - pfrom->addrLocal = addrMe; - pfrom->strSubVer = strSubVer; - pfrom->cleanSubVer = SanitizeString(strSubVer); + pfrom->SetAddrLocal(addrMe); + { + LOCK(pfrom->cs_SubVer); + pfrom->strSubVer = strSubVer; + pfrom->cleanSubVer = cleanSubVer; + } pfrom->nStartingHeight = nStartingHeight; pfrom->fClient = !(nServices & NODE_NETWORK); { @@ -1310,7 +1315,7 @@ bool static ProcessMessage(CNode* pfrom, const std::string& strCommand, CDataStr LogPrint("net", "ProcessMessages: advertising address %s\n", addr.ToString()); pfrom->PushAddress(addr, insecure_rand); } else if (IsPeerAddrLocalGood(pfrom)) { - addr.SetIP(pfrom->addrLocal); + addr.SetIP(addrMe); LogPrint("net", "ProcessMessages: advertising address %s\n", addr.ToString()); pfrom->PushAddress(addr, insecure_rand); } @@ -1330,7 +1335,7 @@ bool static ProcessMessage(CNode* pfrom, const std::string& strCommand, CDataStr remoteAddr = ", peeraddr=" + pfrom->addr.ToString(); LogPrintf("receive version message: %s: version %d, blocks=%d, us=%s, peer=%d%s\n", - pfrom->cleanSubVer, pfrom->nVersion, + cleanSubVer, pfrom->nVersion, pfrom->nStartingHeight, addrMe.ToString(), pfrom->id, remoteAddr); @@ -2450,7 +2455,7 @@ bool static ProcessMessage(CNode* pfrom, const std::string& strCommand, CDataStr if (pingUsecTime > 0) { // Successful ping time measurement, replace previous pfrom->nPingUsecTime = pingUsecTime; - pfrom->nMinPingUsecTime = std::min(pfrom->nMinPingUsecTime, pingUsecTime); + pfrom->nMinPingUsecTime = std::min(pfrom->nMinPingUsecTime.load(), pingUsecTime); } else { // This should never happen sProblem = "Timing mishap"; |