diff options
Diffstat (limited to 'src/main.cpp')
-rw-r--r-- | src/main.cpp | 290 |
1 files changed, 152 insertions, 138 deletions
diff --git a/src/main.cpp b/src/main.cpp index 22baf0f3eb..b29091b4fe 100644 --- a/src/main.cpp +++ b/src/main.cpp @@ -3029,6 +3029,115 @@ bool static AlreadyHave(const CInv& inv) unsigned char pchMessageStart[4] = { 0xf9, 0xbe, 0xb4, 0xd9 }; +void static ProcessGetData(CNode* pfrom) +{ + std::deque<CInv>::iterator it = pfrom->vRecvGetData.begin(); + + vector<CInv> vNotFound; + + while (it != pfrom->vRecvGetData.end()) { + // Don't bother if send buffer is too full to respond anyway + if (pfrom->nSendSize >= SendBufferSize()) + break; + + const CInv &inv = *it; + { + if (fShutdown) + break; + it++; + + if (inv.type == MSG_BLOCK || inv.type == MSG_FILTERED_BLOCK) + { + // Send block from disk + map<uint256, CBlockIndex*>::iterator mi = mapBlockIndex.find(inv.hash); + if (mi != mapBlockIndex.end()) + { + CBlock block; + block.ReadFromDisk((*mi).second); + if (inv.type == MSG_BLOCK) + pfrom->PushMessage("block", block); + else // MSG_FILTERED_BLOCK) + { + LOCK(pfrom->cs_filter); + if (pfrom->pfilter) + { + CMerkleBlock merkleBlock(block, *pfrom->pfilter); + pfrom->PushMessage("merkleblock", merkleBlock); + // CMerkleBlock just contains hashes, so also push any transactions in the block the client did not see + // This avoids hurting performance by pointlessly requiring a round-trip + // Note that there is currently no way for a node to request any single transactions we didnt send here - + // they must either disconnect and retry or request the full block. + // Thus, the protocol spec specified allows for us to provide duplicate txn here, + // however we MUST always provide at least what the remote peer needs + typedef std::pair<unsigned int, uint256> PairType; + BOOST_FOREACH(PairType& pair, merkleBlock.vMatchedTxn) + if (!pfrom->setInventoryKnown.count(CInv(MSG_TX, pair.second))) + pfrom->PushMessage("tx", block.vtx[pair.first]); + } + // else + // no response + } + + // Trigger them to send a getblocks request for the next batch of inventory + if (inv.hash == pfrom->hashContinue) + { + // Bypass PushInventory, this must send even if redundant, + // and we want it right after the last block so they don't + // wait for other stuff first. + vector<CInv> vInv; + vInv.push_back(CInv(MSG_BLOCK, hashBestChain)); + pfrom->PushMessage("inv", vInv); + pfrom->hashContinue = 0; + } + } + } + else if (inv.IsKnownType()) + { + // Send stream from relay memory + bool pushed = false; + { + LOCK(cs_mapRelay); + map<CInv, CDataStream>::iterator mi = mapRelay.find(inv); + if (mi != mapRelay.end()) { + pfrom->PushMessage(inv.GetCommand(), (*mi).second); + pushed = true; + } + } + if (!pushed && inv.type == MSG_TX) { + LOCK(mempool.cs); + if (mempool.exists(inv.hash)) { + CTransaction tx = mempool.lookup(inv.hash); + CDataStream ss(SER_NETWORK, PROTOCOL_VERSION); + ss.reserve(1000); + ss << tx; + pfrom->PushMessage("tx", ss); + pushed = true; + } + } + if (!pushed) { + vNotFound.push_back(inv); + } + } + + // Track requests for our stuff. + Inventory(inv.hash); + } + } + + pfrom->vRecvGetData.erase(pfrom->vRecvGetData.begin(), it); + + if (!vNotFound.empty()) { + // Let the peer know that we didn't find what it asked for, so it doesn't + // have to wait around forever. Currently only SPV clients actually care + // about this message: it's needed when they are recursively walking the + // dependencies of relevant unconfirmed transactions. SPV clients want to + // do that because they want to know about (and store and rebroadcast and + // risk analyze) the dependencies of transactions relevant to them, without + // having to download the entire memory pool. + pfrom->PushMessage("notfound", vNotFound); + } +} + bool static ProcessMessage(CNode* pfrom, string strCommand, CDataStream& vRecv) { RandAddSeedPerfmon(); @@ -3104,7 +3213,7 @@ bool static ProcessMessage(CNode* pfrom, string strCommand, CDataStream& vRecv) // Change version pfrom->PushMessage("verack"); - pfrom->vSend.SetVersion(min(pfrom->nVersion, PROTOCOL_VERSION)); + pfrom->ssSend.SetVersion(min(pfrom->nVersion, PROTOCOL_VERSION)); if (!pfrom->fInbound) { @@ -3168,7 +3277,7 @@ bool static ProcessMessage(CNode* pfrom, string strCommand, CDataStream& vRecv) else if (strCommand == "verack") { - pfrom->vRecv.SetVersion(min(pfrom->nVersion, PROTOCOL_VERSION)); + pfrom->SetRecvVersion(min(pfrom->nVersion, PROTOCOL_VERSION)); } @@ -3302,101 +3411,11 @@ bool static ProcessMessage(CNode* pfrom, string strCommand, CDataStream& vRecv) if (fDebugNet || (vInv.size() != 1)) printf("received getdata (%"PRIszu" invsz)\n", vInv.size()); - vector<CInv> vNotFound; - BOOST_FOREACH(const CInv& inv, vInv) - { - if (fShutdown) - return true; - if (fDebugNet || (vInv.size() == 1)) - printf("received getdata for: %s\n", inv.ToString().c_str()); + if ((fDebugNet && vInv.size() > 0) || (vInv.size() == 1)) + printf("received getdata for: %s\n", vInv[0].ToString().c_str()); - if (inv.type == MSG_BLOCK || inv.type == MSG_FILTERED_BLOCK) - { - // Send block from disk - map<uint256, CBlockIndex*>::iterator mi = mapBlockIndex.find(inv.hash); - if (mi != mapBlockIndex.end()) - { - CBlock block; - block.ReadFromDisk((*mi).second); - if (inv.type == MSG_BLOCK) - pfrom->PushMessage("block", block); - else // MSG_FILTERED_BLOCK) - { - LOCK(pfrom->cs_filter); - if (pfrom->pfilter) - { - CMerkleBlock merkleBlock(block, *pfrom->pfilter); - pfrom->PushMessage("merkleblock", merkleBlock); - // CMerkleBlock just contains hashes, so also push any transactions in the block the client did not see - // This avoids hurting performance by pointlessly requiring a round-trip - // Note that there is currently no way for a node to request any single transactions we didnt send here - - // they must either disconnect and retry or request the full block. - // Thus, the protocol spec specified allows for us to provide duplicate txn here, - // however we MUST always provide at least what the remote peer needs - typedef std::pair<unsigned int, uint256> PairType; - BOOST_FOREACH(PairType& pair, merkleBlock.vMatchedTxn) - if (!pfrom->setInventoryKnown.count(CInv(MSG_TX, pair.second))) - pfrom->PushMessage("tx", block.vtx[pair.first]); - } - // else - // no response - } - - // Trigger them to send a getblocks request for the next batch of inventory - if (inv.hash == pfrom->hashContinue) - { - // Bypass PushInventory, this must send even if redundant, - // and we want it right after the last block so they don't - // wait for other stuff first. - vector<CInv> vInv; - vInv.push_back(CInv(MSG_BLOCK, hashBestChain)); - pfrom->PushMessage("inv", vInv); - pfrom->hashContinue = 0; - } - } - } - else if (inv.IsKnownType()) - { - // Send stream from relay memory - bool pushed = false; - { - LOCK(cs_mapRelay); - map<CInv, CDataStream>::iterator mi = mapRelay.find(inv); - if (mi != mapRelay.end()) { - pfrom->PushMessage(inv.GetCommand(), (*mi).second); - pushed = true; - } - } - if (!pushed && inv.type == MSG_TX) { - LOCK(mempool.cs); - if (mempool.exists(inv.hash)) { - CTransaction tx = mempool.lookup(inv.hash); - CDataStream ss(SER_NETWORK, PROTOCOL_VERSION); - ss.reserve(1000); - ss << tx; - pfrom->PushMessage("tx", ss); - pushed = true; - } - } - if (!pushed) { - vNotFound.push_back(inv); - } - } - - // Track requests for our stuff. - Inventory(inv.hash); - - if (!vNotFound.empty()) { - // Let the peer know that we didn't find what it asked for, so it doesn't - // have to wait around forever. Currently only SPV clients actually care - // about this message: it's needed when they are recursively walking the - // dependencies of relevant unconfirmed transactions. SPV clients want to - // do that because they want to know about (and store and rebroadcast and - // risk analyze) the dependencies of transactions relevant to them, without - // having to download the entire memory pool. - pfrom->PushMessage("notfound", vNotFound); - } - } + pfrom->vRecvGetData.insert(pfrom->vRecvGetData.end(), vInv.begin(), vInv.end()); + ProcessGetData(pfrom); } @@ -3705,13 +3724,11 @@ bool static ProcessMessage(CNode* pfrom, string strCommand, CDataStream& vRecv) return true; } +// requires LOCK(cs_vRecvMsg) bool ProcessMessages(CNode* pfrom) { - CDataStream& vRecv = pfrom->vRecv; - if (vRecv.empty()) - return true; //if (fDebug) - // printf("ProcessMessages(%u bytes)\n", vRecv.size()); + // printf("ProcessMessages(%zu messages)\n", pfrom->vRecvMsg.size()); // // Message format @@ -3721,33 +3738,41 @@ bool ProcessMessages(CNode* pfrom) // (4) checksum // (x) data // + bool fOk = true; - loop - { + if (!pfrom->vRecvGetData.empty()) + ProcessGetData(pfrom); + + std::deque<CNetMessage>::iterator it = pfrom->vRecvMsg.begin(); + while (!pfrom->fDisconnect && it != pfrom->vRecvMsg.end()) { // Don't bother if send buffer is too full to respond anyway - if (pfrom->vSend.size() >= SendBufferSize()) + if (pfrom->nSendSize >= SendBufferSize()) + break; + + // get next message + CNetMessage& msg = *it; + + //if (fDebug) + // printf("ProcessMessages(message %u msgsz, %zu bytes, complete:%s)\n", + // msg.hdr.nMessageSize, msg.vRecv.size(), + // msg.complete() ? "Y" : "N"); + + // end, if an incomplete message is found + if (!msg.complete()) break; + // at this point, any failure means we can delete the current message + it++; + // Scan for message start - CDataStream::iterator pstart = search(vRecv.begin(), vRecv.end(), BEGIN(pchMessageStart), END(pchMessageStart)); - int nHeaderSize = vRecv.GetSerializeSize(CMessageHeader()); - if (vRecv.end() - pstart < nHeaderSize) - { - if ((int)vRecv.size() > nHeaderSize) - { - printf("\n\nPROCESSMESSAGE MESSAGESTART NOT FOUND\n\n"); - vRecv.erase(vRecv.begin(), vRecv.end() - nHeaderSize); - } + if (memcmp(msg.hdr.pchMessageStart, pchMessageStart, sizeof(pchMessageStart)) != 0) { + printf("\n\nPROCESSMESSAGE: INVALID MESSAGESTART\n\n"); + fOk = false; break; } - if (pstart - vRecv.begin() > 0) - printf("\n\nPROCESSMESSAGE SKIPPED %"PRIpdd" BYTES\n\n", pstart - vRecv.begin()); - vRecv.erase(vRecv.begin(), pstart); // Read header - vector<char> vHeaderSave(vRecv.begin(), vRecv.begin() + nHeaderSize); - CMessageHeader hdr; - vRecv >> hdr; + CMessageHeader& hdr = msg.hdr; if (!hdr.IsValid()) { printf("\n\nPROCESSMESSAGE: ERRORS IN HEADER %s\n\n\n", hdr.GetCommand().c_str()); @@ -3757,19 +3782,9 @@ bool ProcessMessages(CNode* pfrom) // Message size unsigned int nMessageSize = hdr.nMessageSize; - if (nMessageSize > MAX_SIZE) - { - printf("ProcessMessages(%s, %u bytes) : nMessageSize > MAX_SIZE\n", strCommand.c_str(), nMessageSize); - continue; - } - if (nMessageSize > vRecv.size()) - { - // Rewind and wait for rest of message - vRecv.insert(vRecv.begin(), vHeaderSave.begin(), vHeaderSave.end()); - break; - } // Checksum + CDataStream& vRecv = msg.vRecv; uint256 hash = Hash(vRecv.begin(), vRecv.begin() + nMessageSize); unsigned int nChecksum = 0; memcpy(&nChecksum, &hash, sizeof(nChecksum)); @@ -3780,20 +3795,16 @@ bool ProcessMessages(CNode* pfrom) continue; } - // Copy message to its own buffer - CDataStream vMsg(vRecv.begin(), vRecv.begin() + nMessageSize, vRecv.nType, vRecv.nVersion); - vRecv.ignore(nMessageSize); - // Process message bool fRet = false; try { { LOCK(cs_main); - fRet = ProcessMessage(pfrom, strCommand, vMsg); + fRet = ProcessMessage(pfrom, strCommand, vRecv); } if (fShutdown) - return true; + break; } catch (std::ios_base::failure& e) { @@ -3822,8 +3833,11 @@ bool ProcessMessages(CNode* pfrom) printf("ProcessMessage(%s, %u bytes) FAILED\n", strCommand.c_str(), nMessageSize); } - vRecv.Compact(); - return true; + // In case the connection got shut down, its receive buffer was wiped + if (!pfrom->fDisconnect) + pfrom->vRecvMsg.erase(pfrom->vRecvMsg.begin(), it); + + return fOk; } @@ -3837,7 +3851,7 @@ bool SendMessages(CNode* pto, bool fSendTrickle) // Keep-alive ping. We send a nonce of zero because we don't use it anywhere // right now. - if (pto->nLastSend && GetTime() - pto->nLastSend > 30 * 60 && pto->vSend.empty()) { + if (pto->nLastSend && GetTime() - pto->nLastSend > 30 * 60 && pto->vSendMsg.empty()) { uint64 nonce = 0; if (pto->nVersion > BIP0031_VERSION) pto->PushMessage("ping", nonce); |