從零開始學習比特幣：P2P 網路的建立之訊息處理（上篇）

P2P 區塊鏈開發 · 發表 2018-10-25 10:46:48

摘要： P2P 網路的建立是在系統啟動的第 12 步，最後時刻呼叫CConnman::Start 方法開始的。本部分內容在net.cpp 、net_processing.cpp 等檔案中。前面幾個章節學完了ThreadSocketHandler...

P2P 網路的建立是在系統啟動的第 12 步，最後時刻呼叫CConnman::Start 方法開始的。

本部分內容在net.cpp 、net_processing.cpp 等檔案中。

前面幾個章節學完了ofollow,noindex" target="_blank">ThreadSocketHandler執行緒來處理套接字的讀取和傳送 ThreadDNSAddressSeed執行緒查詢DNS節點 ThreadOpenAddedConnections執行緒生成地址對並連線到指定地址

終於，來到了我們非常非常關心的比特幣訊息處理執行緒。通過比特幣訊息處理，我們會理解比特幣的協義，理解比特幣是如何同步區塊，如何傳送交易，從而建立起理解比特幣的至關重要一步。

本部分內容是如此的重要，也是相當的長，所以我們分上下兩部分來介紹具體的訊息處理。

上篇主要以訊息處理執行緒的分析為主，下篇以具體的比特幣訊息即比特幣協義分析為主。

下面我們來看訊息處理執行緒相關的程式碼。

ThreadMessageHandler

處理所有訊息的執行緒。主體是一個while 迴圈，退出條件為flagInterruptMsgProc

為假，迴圈體如下：

生成一個對等節點容器物件vNodesCopy ，型別為CNode* 。然後，設定其值為vNodes 。後者，代表當前節點儲存的所有對等節點資訊。
遍歷所有的vNodesCopy 節點，呼叫當前節點的AddRef 方法，把對等節點的nRefCount 屬性加1。

遍歷所有的vNodesCopy 節點，進行如下處理：

PeerLogicValidation::ProcessMessages
PeerLogicValidation::SendMessages

遍歷所有的vNodesCopy 節點，呼叫當前節點的Release 方法，把對等節點的nRefCount 屬性減1。

1、ProcessMessages

本方法主要處理對等節點接收到的相關訊息，具體程式碼在net_processing.cpp

檔案中。

如果對等節點的已接收請求資料集合不為空，也就是儲存別的對等節點請求資料的集合不為空，則呼叫ProcessGetData 方法，處理別的對等獲取資料的請求。
```
if (!pfrom->vRecvGetData.empty())
ProcessGetData(pfrom, chainparams, connman, interruptMsgProc);
```
vRecvGetData 屬性是一個 inv 訊息（CInv ）的雙端佇列。
下面，我們來看ProcessGetData 方法是怎樣處理收到的訊息。
- 從對等節點的vRecvGetData 集合中，取得其迭代器。std::deque<CInv>::iterator it = pfrom->vRecvGetData.begin();std::vector<CInv> vNotFound;
- 生成一個CNetMsgMaker 型別的物件msgMaker 。其nVersionIn 屬性是對等節點已經發送的版本訊息。
- 遍歷請求資料集合，如果請求資料的型別是交易或者見證隔離交易，進行如下的處理：如果已經終止處理訊息訊號為真，則直接返回。如果當前節點處於暫停狀態（緩衝區太慢而不能響應），則推出迴圈。
  取得當前的inv 訊息。從mapRelay 集合中取得當前inv 訊息。這個mapRelay 是一個 Map 集合，Key 是交易的雜湊，Value 是指向交易的智慧指標。
  
  如果查詢的交易存在於集合中，則呼叫位於net.cpp 檔案中的CConnman::PushMessage 方法，傳送交易；否則，如果節點最後一次MEMPOOL 請求存在，則從記憶體池中取得對應的交易信息，然後呼叫CConnman::PushMessage 方法，傳送交易。
```
while (it != pfrom->vRecvGetData.end() && (it->type == MSG_TX || it->type == MSG_WITNESS_TX)) {
if (interruptMsgProc)
return;
// Don't bother if send buffer is too full to respond anyway
if (pfrom->fPauseSend)
break;
const CInv &inv = *it;
it++;
bool push = false;
auto mi = mapRelay.find(inv.hash);
int nSendFlags = (inv.type == MSG_TX ? SERIALIZE_TRANSACTION_NO_WITNESS : 0);
if (mi != mapRelay.end()) {
connman->PushMessage(pfrom, msgMaker.Make(nSendFlags, NetMsgType::TX, *mi->second));
push = true;
} else if (pfrom->timeLastMempoolReq) {
auto txinfo = mempool.info(inv.hash);
if (txinfo.tx && txinfo.nTime <= pfrom->timeLastMempoolReq) {
connman->PushMessage(pfrom, msgMaker.Make(nSendFlags, NetMsgType::TX, *txinfo.tx));
push = true;
}
}
if (!push) {
vNotFound.push_back(inv);
}
}
```
- 如果沒有達到請求資料集合尾部，且節點物件不是暫停狀態，並且當前請求的inv 訊息型別是區塊、過濾型區塊、緊湊型區塊、隔離見證型區塊之一，則呼叫ProcessGetBlockData 方法處理區塊資料。遍歷收到的所有資料，如果資料型別是交易或者見證隔離交易，進行如下的處理：ProcessGetBlockData 方法，因為處理過程比較長，所以其處理過程放在下面進行詳細說明。
- 清空已收到資料集合從起始部分到當前位置之間的inv 資料。
如果對等節點已斷開，則直接返回。
如果對等節點要處理的資料不為空，則直接返回。這樣維護了響應的順序。
如果對等節點已暫停，則直接返回。緩衝區太滿而不能進行繼續處理。
如果節點物件待處理的訊息列表vProcessMsg 為空，則返回假。否則，取出第一個訊息物件，放入訊息列表msgs 中，並待處理的訊息列表中刪除。將節點物件處理佇列大小減去已刪除的訊息物件收到的資料長度與訊息頭部大小之和。根據節點物件處理佇列大小與允許的接收上限比較，如果大於接收上限，則設定節點暫停接收訊息。
```
if (pfrom->vProcessMsg.empty())
return false;
 msgs.splice(msgs.begin(), pfrom->vProcessMsg, pfrom->vProcessMsg.begin());
 pfrom->nProcessQueueSize -= msgs.front().vRecv.size() + CMessageHeader::HEADER_SIZE;
 pfrom->fPauseRecv = pfrom->nProcessQueueSize > connman->GetReceiveFloodSize();
 fMoreWork = !pfrom->vProcessMsg.empty();
```
生成一個訊息物件，設定其版本為對等節點已接收到的版本。
```
CNetMessage& msg(msgs.front());
 msg.SetVersion(pfrom->GetRecvVersion());
```

驗證訊息的MESSAGESTART 是否有效。如果無效，則設定對等節點為斷開，然後返回。

if (memcmp(msg.hdr.pchMessageStart, chainparams.MessageStart(), CMessageHeader::MESSAGE_START_SIZE) != 0) {
LogPrint(BCLog::NET, "PROCESSMESSAGE: INVALID MESSAGESTART %s peer=%d\n", SanitizeString(msg.hdr.GetCommand()), pfrom->GetId());
pfrom->fDisconnect = true;
return false;
 }

從訊息物件中取得訊息頭部，並進行驗證。如果無效，則返回。

CMessageHeader& hdr = msg.hdr;
 if (!hdr.IsValid(chainparams.MessageStart()))
 {
LogPrint(BCLog::NET, "PROCESSMESSAGE: ERRORS IN HEADER %s peer=%d\n", SanitizeString(hdr.GetCommand()), pfrom->GetId());
return fMoreWork;
 }

從訊息物件中取得具體的命令和訊息大小。

std::string strCommand = hdr.GetCommand();
 unsigned int nMessageSize = hdr.nMessageSize;

呼叫訊息物件自身的雜湊，並與訊息頭部的校驗和進行驗證。如果驗證失敗，則返回。

const uint256& hash = msg.GetMessageHash();
 if (memcmp(hash.begin(), hdr.pchChecksum, CMessageHeader::CHECKSUM_SIZE) != 0)
 {
LogPrint(BCLog::NET, "%s(%s, %u bytes): CHECKSUM ERROR expected %s was %s\n", __func__,
SanitizeString(strCommand), nMessageSize,
HexStr(hash.begin(), hash.begin()+CMessageHeader::CHECKSUM_SIZE),
HexStr(hdr.pchChecksum, hdr.pchChecksum+CMessageHeader::CHECKSUM_SIZE));
return fMoreWork;
 }

呼叫ProcessMessage
方法，進行訊息處理。對於比特幣網路來說，最最重要的方法來了。這個方法處理比特幣的所有具體，比如：版本訊息、獲取區塊訊息等。

因為這個方法是如此的重要，所以我們把留在下一篇文章中進行說明。
呼叫SendRejectsAndCheckIfBanned 方法，進行可能的reject 處理。具體如下：
- 取得節點的狀態物件。
- 如果啟用 BIP61，則遍歷狀態物件中儲存的reject ，呼叫PushMessage 方法，傳送reject 訊息。
- 清空狀態物件中儲存的reject .
- 如果狀態物件的禁止屬性fShouldBan 為真，則：
```
Ban
```

ProcessGetBlockData

生成一些內部變數，並設定為已快取的變數值。
呼叫LookupBlockIndex 方法，查詢訊息對應的區塊索引。如果區塊索引存在，並且索引對應的區塊在鏈上的交易也存在，但區塊還沒有驗證過，那麼設定變數need_activate_chain 為真。const CBlockIndex* pindex = LookupBlockIndex(inv.hash);if (pindex) { if (pindex->nChainTx && !pindex->IsValid(BLOCK_VALID_SCRIPTS) && pindex->IsValid(BLOCK_VALID_TREE)) { need_activate_chain = true; }}
如果需要啟用區塊鏈，那麼呼叫ActivateBestChain 方法來啟用。
如果區塊索引存在，呼叫BlockRequestAllowed 方法檢查是否允許傳送資料。

如果允許傳送，且達到歷史區塊服務限額的情況下，斷開節點連線，並設定傳送標誌為假。

if (send && connman->OutboundTargetReached(true) && ( ((pindexBestHeader != nullptr) && (pindexBestHeader->GetBlockTime() - pindex->GetBlockTime() > HISTORICAL_BLOCK_AGE)) || inv.type == MSG_FILTERED_BLOCK) && !pfrom->fWhitelisted)
 {
LogPrint(BCLog::NET, "historical block serving limit reached, disconnect peer=%d\n", pfrom->GetId());
pfrom->fDisconnect = true;
send = false;
 }

如果允許傳送，且節點不在白名單中，且節點支援的服務是NODE_NETWORK_LIMITED （只支援 288個區塊，即2天內生成的區塊），且不支援NODE_NETWORK 服務，且區塊鏈棧頂的高度與當前區塊索引的高度之差大於網路限制允許的最小區塊數（NODE_NETWORK_LIMITED_MIN_BLOCKS ，288個區塊）加上額外的兩個區塊（為了防止競爭，指的是分叉？所以增加兩個區塊緩衝），則斷開節點連線，並設定傳送標誌為假。
```
if (send && !pfrom->fWhitelisted && (
(((pfrom->GetLocalServices() & NODE_NETWORK_LIMITED) == NODE_NETWORK_LIMITED) &&
((pfrom->GetLocalServices() & NODE_NETWORK) != NODE_NETWORK) &&
(chainActive.Tip()->nHeight - pindex->nHeight > (int)NODE_NETWORK_LIMITED_MIN_BLOCKS + 2) )
)) {
pfrom->fDisconnect = true;
send = false;
 }
```

如果允許傳送，並且這個區塊索引的狀態等於BLOCK_HAVE_DATA （全部資料都blk*.dat 檔案中可用），則進行下面的處理：

生成一個指向區塊物件的智慧指標物件pblock 。
如果最近區塊物件存在，且其雜湊與區塊索引物件的雜湊一樣，那麼設定pblock 為最近區塊物件；
否則，如果訊息物件型別是隔離見證區塊，那麼：呼叫ReadRawBlockFromDisk 方法，從磁碟中讀取原始的區塊資料。如果可以讀到，則呼叫PushMessage 方法，傳送區塊資料。這種情況下，直接傳送了區塊資料，所以不設定pblock 變數。
否則，呼叫ReadBlockFromDisk 方法，從磁碟中讀取原始的區塊資料。如果可以讀到，則設定pblock 為讀取到的資料。
如果pblock 為真，則傳送訊息。如果訊息物件型別是區塊，則呼叫PushMessage 方法，傳送標誌為SERIALIZE_TRANSACTION_NO_WITNESS 的區塊訊息；
如果訊息型別是隔離見證區塊，則呼叫PushMessage 方法，傳送區塊訊息；

如果訊息型別是過濾區塊，則：如果對節物件的 Bloom 過濾器存在，那麼生成默克爾區塊，並設定傳送過濾區塊標誌為真；如果傳送過濾區塊標誌為真，則呼叫PushMessage 方法，傳送默克爾區塊，然後呼叫PushMessage 方法，傳送默克爾區塊的每個交易資料，標誌為SERIALIZE_TRANSACTION_NO_WITNESS ；

如果訊息型別是緊湊區塊，同樣呼叫PushMessage 方法，傳送區塊訊息。
如果訊息的雜湊等於節點的繼續傳送屬性（hashContinue 屬性，代表繼續傳送的雜湊），則：生成一個CInv 向量；然後，構造一個CInv 物件，型別為區塊，雜湊為區塊鏈棧頂元素的雜湊；然後，呼叫PushMessage 方法，傳送inv 訊息；最後，設定節點的繼續傳送屬性為空。

程式碼如下：

if (send && (pindex->nStatus & BLOCK_HAVE_DATA)){
std::shared_ptr<const CBlock> pblock;
if (a_recent_block && a_recent_block->GetHash() == pindex->GetBlockHash()) {
pblock = a_recent_block;
} else if (inv.type == MSG_WITNESS_BLOCK) {
std::vector<uint8_t> block_data;
if (!ReadRawBlockFromDisk(block_data, pindex, chainparams.MessageStart())) {
assert(!"cannot load block from disk");
}
connman->PushMessage(pfrom, msgMaker.Make(NetMsgType::BLOCK, MakeSpan(block_data)));
} else {
std::shared_ptr<CBlock> pblockRead = std::make_shared<CBlock>();
if (!ReadBlockFromDisk(*pblockRead, pindex, consensusParams))
assert(!"cannot load block from disk");
pblock = pblockRead;
}
if (pblock) {
if (inv.type == MSG_BLOCK)
connman->PushMessage(pfrom, msgMaker.Make(SERIALIZE_TRANSACTION_NO_WITNESS, NetMsgType::BLOCK, *pblock));
else if (inv.type == MSG_WITNESS_BLOCK)
connman->PushMessage(pfrom, msgMaker.Make(NetMsgType::BLOCK, *pblock));
else if (inv.type == MSG_FILTERED_BLOCK)
{
bool sendMerkleBlock = false;
CMerkleBlock merkleBlock;
{
LOCK(pfrom->cs_filter);
if (pfrom->pfilter) {
sendMerkleBlock = true;
merkleBlock = CMerkleBlock(*pblock, *pfrom->pfilter);
}
}
if (sendMerkleBlock) {
connman->PushMessage(pfrom, msgMaker.Make(NetMsgType::MERKLEBLOCK, merkleBlock));
typedef std::pair<unsigned int, uint256> PairType;
for (PairType& pair : merkleBlock.vMatchedTxn)
connman->PushMessage(pfrom, msgMaker.Make(SERIALIZE_TRANSACTION_NO_WITNESS, NetMsgType::TX, *pblock->vtx[pair.first]));
}
}
else if (inv.type == MSG_CMPCT_BLOCK)
{
bool fPeerWantsWitness = State(pfrom->GetId())->fWantsCmpctWitness;
int nSendFlags = fPeerWantsWitness ? 0 : SERIALIZE_TRANSACTION_NO_WITNESS;
if (CanDirectFetch(consensusParams) && pindex->nHeight >= chainActive.Height() - MAX_CMPCTBLOCK_DEPTH) {
if ((fPeerWantsWitness || !fWitnessesPresentInARecentCompactBlock) && a_recent_compact_block && a_recent_compact_block->header.GetHash() == pindex->GetBlockHash()) {
connman->PushMessage(pfrom, msgMaker.Make(nSendFlags, NetMsgType::CMPCTBLOCK, *a_recent_compact_block));
} else {
CBlockHeaderAndShortTxIDs cmpctblock(*pblock, fPeerWantsWitness);
connman->PushMessage(pfrom, msgMaker.Make(nSendFlags, NetMsgType::CMPCTBLOCK, cmpctblock));
}
} else {
connman->PushMessage(pfrom, msgMaker.Make(nSendFlags, NetMsgType::BLOCK, *pblock));
}
}
}
if (inv.hash == pfrom->hashContinue)
{
std::vector<CInv> vInv;
vInv.push_back(CInv(MSG_BLOCK, chainActive.Tip()->GetBlockHash()));
connman->PushMessage(pfrom, msgMaker.Make(NetMsgType::INV, vInv));
pfrom->hashContinue.SetNull();
}
}

2、SendMessages

本方法主要處理對等節點發送的相關邏輯，具體程式碼在net_processing.cpp

檔案中。

如果對等節點間還沒有完成握手，或者已經斷開連線，則返回。
```
if (!pto->fSuccessfullyConnected || pto->fDisconnect)
return true;
```
如果對等節點的 Ping 已經被請求，則設定pingSend 變數為真。
如果對等節點沒有期望 Pong 回覆（即對等節點的nPingNonceSent 等於0），且使用者開始 ping 的時間（nPingUsecStart ）加上規定的節點 ping 間隔小於當前時間，則設定pingSend 變數為真。
如果pingSend 為真，處理如下：
- 設定對等節點的 Ping 已經被請求假。
- 設定開始 ping 的時間為當前時間。
- 如果對等節點的版本大於 BIP 0031 規定的版本（60000），則設定對等節點的nPingNonceSent 為隨機變數nonce ，呼叫PushMessage 方法，傳送ping 訊息，訊息中包括nonce ；否則，即對等節點不支援帶隨機數的 Ping 命令，則設定設定對等節點的nPingNonceSent 為0，呼叫PushMessage 方法，傳送ping 訊息，訊息中不包括nonce 。
```
if (pingSend) {
uint64_t nonce = 0;
while (nonce == 0) {
GetRandBytes((unsigned char*)&nonce, sizeof(nonce));
}
pto->fPingQueued = false;
pto->nPingUsecStart = GetTimeMicros();
if (pto->nVersion > BIP0031_VERSION) {
pto->nPingNonceSent = nonce;
connman->PushMessage(pto, msgMaker.Make(NetMsgType::PING, nonce));
} else {
pto->nPingNonceSent = 0;
connman->PushMessage(pto, msgMaker.Make(NetMsgType::PING));
}
}
```
呼叫SendRejectsAndCheckIfBanned 方法，進行可能的reject 處理。如果該函式返回為真，則返回。
獲取節點的狀態物件。
如果當前沒有在 IBD 下載中（IsInitialBlockDownload 函式為假），且節點下次傳送本地地址的時間（nNextLocalAddrSend ）小於當前時間，那麼進行如下處理：
- 呼叫AdvertiseLocal 方法，傳送我們自己本身的地址給對等節點。方法的主要邏輯是呼叫節點物件的PushAddress 方法，把要傳送的地址儲存在vAddrToSend 集合中。
- 呼叫PoissonNextSend 方法，計算下次傳送地址的時間，並設定節點的下次傳送本地地址的時間為該值。
如果節點的下次傳送地址時間小於當前時間，則：
- 呼叫PoissonNextSend 方法，計算下次傳送地址的時間，並儲存為節點的下次傳送地址時間nNextAddrSend 屬性。
- 生成一個地址向量集合vAddr 。
- 遍歷節點待發送的地址向量vAddrToSend ，如果當前地址不在節點的概率“跟蹤最近插入的”集合addrKnown 中，則：
```
vAddr
PushMessage
```
- 清空節點的傳送地址集合vAddrToSend 。
- 如果地址向量集合不空，即地址向量集合的數量不超過 1000個，則呼叫PushMessage 方法，傳送地址訊息。
- 如果節點的待發送的地址向量集合的預分配的記憶體空間（capacity() ）大於40，則呼叫其shrink_to_fit 方法來縮減空間，即只允許傳送一次大的地址包。
```
if (pto->nNextAddrSend < nNow) {
pto->nNextAddrSend = PoissonNextSend(nNow, AVG_ADDRESS_BROADCAST_INTERVAL);
std::vector<CAddress> vAddr;
vAddr.reserve(pto->vAddrToSend.size());
for (const CAddress& addr : pto->vAddrToSend)
{
if (!pto->addrKnown.contains(addr.GetKey()))
{
pto->addrKnown.insert(addr.GetKey());
vAddr.push_back(addr);
// receiver rejects addr messages larger than 1000
if (vAddr.size() >= 1000)
{
connman->PushMessage(pto, msgMaker.Make(NetMsgType::ADDR, vAddr));
vAddr.clear();
}
}
}
pto->vAddrToSend.clear();
if (!vAddr.empty())
connman->PushMessage(pto, msgMaker.Make(NetMsgType::ADDR, vAddr));
if (pto->vAddrToSend.capacity() > 40)
pto->vAddrToSend.shrink_to_fit();
}
```

接下來，開始區塊同步。

如果指向最佳區塊鏈頭部的指標（pindexBestHeader ）為空指標，則設定其為當前活躍區塊鏈的棧頂元素指標。
```
if (pindexBestHeader == nullptr)
pindexBestHeader = chainActive.Tip();
```

如果還沒有從這個節點同步區塊頭部，並且節點的fClient 、fImporting 、fReindex 等屬性為假，進一步，如果已經同步的節點數量為 0 且需要獲取區塊資料，或者最佳區塊頭部的區塊時間距離現在已超過 24 小時，那麼呼叫PushMessage 方法，發出請求GETHEADERS 命令，開始同步區塊頭部。

bool fFetch = state.fPreferredDownload || (nPreferredDownload == 0 && !pto->fClient && !pto->fOneShot); // Download if this is a nice peer, or we have no nice peers and this one might do.
if (!state.fSyncStarted && !pto->fClient && !fImporting && !fReindex) {
if ((nSyncStarted == 0 && fFetch) || pindexBestHeader->GetBlockTime() > GetAdjustedTime() - 24 * 60 * 60) {
state.fSyncStarted = true;
state.nHeadersSyncTimeout = GetTimeMicros() + HEADERS_DOWNLOAD_TIMEOUT_BASE + HEADERS_DOWNLOAD_TIMEOUT_PER_HEADER * (GetAdjustedTime() - pindexBestHeader->GetBlockTime())/(consensusParams.nPowTargetSpacing);
nSyncStarted++;
const CBlockIndex *pindexStart = pindexBestHeader;
if (pindexStart->pprev)
pindexStart = pindexStart->pprev;
LogPrint(BCLog::NET, "initial getheaders (%d) to peer=%d (startheight:%d)\n", pindexStart->nHeight, pto->GetId(), pto->nStartingHeight);
connman->PushMessage(pto, msgMaker.Make(NetMsgType::GETHEADERS, chainActive.GetLocator(pindexStart), uint256()));
}
}

如果當前不是重建索引、重新匯入和 IBD 下載期間，則重新發送尚未進入區塊的錢包交易。if (!fReindex && !fImporting && !IsInitialBlockDownload()){
```
GetMainSignals().Broadcast(nTimeBestReceived, connman);
```
}

如果不需要轉化為Inv 訊息，那麼進行如下處理：遍歷區塊頭部進行區塊公告的集合（vBlockHashesToAnnounce ），按如下處理：

呼叫LookupBlockIndex 方法，查詢當前對應的區塊索引。

如果當前活躍區塊鏈上沒有這個索引，那麼設定需要轉化為Inv 訊息為真，然後退出當前迴圈。

const CBlockIndex* pindex = LookupBlockIndex(hash);
if (chainActive[pindex->nHeight] != pindex) {
fRevertToInv = true;
break;
}

如果最佳索引不是空指標，並且當前區塊索引不等於最佳指標，那麼設定需要轉化為Inv 訊息為真，然後退出當前迴圈。
```
if (pBestIndex != nullptr && pindex->pprev != pBestIndex) {
fRevertToInv = true;
break;
}
```

接下來，設定當前區塊索引為最佳索引，處理哪些區塊索引可以放入頭部集合。

pBestIndex = pindex;
if (fFoundStartingHeader) {
// add this to the headers message
vHeaders.push_back(pindex->GetBlockHeader());
} else if (PeerHasHeader(&state, pindex)) {
continue; // keep looking for the first new block
} else if (pindex->pprev == nullptr || PeerHasHeader(&state, pindex->pprev)) {
// Peer doesn't have this header but they do have the prior one.
// Start sending headers.
fFoundStartingHeader = true;
vHeaders.push_back(pindex->GetBlockHeader());
} else {
// Peer doesn't have this header or the prior one -- nothing will
// connect, so bail out.
fRevertToInv = true;
break;
}

如果不需要轉化成Inv 訊息，並且區塊頭部集合（vHeaders ）不空，進行下面的處理：

如果區塊頭部長度為1，並且需要下載頭部和ID，那麼：生成傳送標誌，如果對等節點想要緊湊的隔離見證型別，則設定傳送標誌為 0，否則設定為SERIALIZE_TRANSACTION_NO_WITNESS 。
如果最近的區塊雜湊與最佳區塊索引的雜湊相等，則呼叫PushMessage 方法，傳送訊息，型別為CMPCTBLOCK ，然後設定區塊是從快取區中載入的標誌為真。

如果區塊不是從快取區中載入的，那麼就需要呼叫ReadBlockFromDisk 方法，從硬碟中載入區塊，然後再呼叫PushMessage 方法，傳送訊息，型別為CMPCTBLOCK 。
否則，如果不是優先下載頭部（即區塊狀態物件fPreferHeaders ）為假，那麼就呼叫PushMessage 方法，傳送訊息，型別為HEADERS ，然後設定區塊狀態物件的pindexBestHeaderSent 屬性為當前的最佳索引區塊。

具體程式碼如下：

if (!fRevertToInv && !vHeaders.empty()) {
if (vHeaders.size() == 1 && state.fPreferHeaderAndIDs) {
int nSendFlags = state.fWantsCmpctWitness ? 0 : SERIALIZE_TRANSACTION_NO_WITNESS;
bool fGotBlockFromCache = false;
{
LOCK(cs_most_recent_block);
if (most_recent_block_hash == pBestIndex->GetBlockHash()) {
if (state.fWantsCmpctWitness || !fWitnessesPresentInMostRecentCompactBlock)
connman->PushMessage(pto, msgMaker.Make(nSendFlags, NetMsgType::CMPCTBLOCK, *most_recent_compact_block));
else {
CBlockHeaderAndShortTxIDs cmpctblock(*most_recent_block, state.fWantsCmpctWitness);
connman->PushMessage(pto, msgMaker.Make(nSendFlags, NetMsgType::CMPCTBLOCK, cmpctblock));
}
fGotBlockFromCache = true;
}
}
if (!fGotBlockFromCache) {
CBlock block;
bool ret = ReadBlockFromDisk(block, pBestIndex, consensusParams);
assert(ret);
CBlockHeaderAndShortTxIDs cmpctblock(block, state.fWantsCmpctWitness);
connman->PushMessage(pto, msgMaker.Make(nSendFlags, NetMsgType::CMPCTBLOCK, cmpctblock));
}
state.pindexBestHeaderSent = pBestIndex;
} else if (state.fPreferHeaders) {
connman->PushMessage(pto, msgMaker.Make(NetMsgType::HEADERS, vHeaders));
state.pindexBestHeaderSent = pBestIndex;
} else
fRevertToInv = true;
}

如果需要轉化成Inv 訊息，進一步，如果使用區塊頭部進行區塊公告的集合（vBlockHashesToAnnounce ），則進行如下處理：
- 返回集合最後一個元素，呼叫LookupBlockIndex 方法，查詢這個元素對應的區塊索引。
- 如果活躍區塊鏈在區塊索引指定的高度上對應的索引不是我們找到的索引，即要公告的區塊不在主鏈上，則列印一個警告。
- 如果這個區塊不在節點的區塊鏈上，那麼就把這個區塊放在區塊庫存清單中。生成一個Inv 訊息，型別是區塊，然後呼叫節點物件的PushInventory 方法，放入節點物件的庫存清單集合中。
清空節點物件的區塊頭部進行區塊公告的集合。

生成vInv 集合，並設定其長度。

std::vector<CInv> vInv;
 vInv.reserve(std::max<size_t>(pto->vInventoryBlockToSend.size(), INVENTORY_BROADCAST_MAX));

遍歷已經公告的區塊 ID 列表，如果沒有達到vInv 集合的最大長度，則加入集合尾部，如果已經達到則呼叫PushMessage 方法，傳送INV 訊息。然後，清空已經公告的區塊 ID 列表。

for (const uint256& hash : pto->vInventoryBlockToSend) {
vInv.push_back(CInv(MSG_BLOCK, hash));
if (vInv.size() == MAX_INV_SZ) {
connman->PushMessage(pto, msgMaker.Make(NetMsgType::INV, vInv));
vInv.clear();
}
 }
 pto->vInventoryBlockToSend.clear();

如果節點物件下次傳送Inv 訊息的時間已經小於當前時間，那麼設定fSendTrickle 變數為真，根據是否為入門節點，設定不同的節點物件下次傳送Inv 訊息。

bool fSendTrickle = pto->fWhitelisted;
 if (pto->nNextInvSend < nNow) {
fSendTrickle = true;
if (pto->fInbound) {
pto->nNextInvSend = connman->PoissonNextSendInbound(nNow, INVENTORY_BROADCAST_INTERVAL);
} else {
pto->nNextInvSend = PoissonNextSend(nNow, INVENTORY_BROADCAST_INTERVAL >> 1);
}
 }

如果傳送時間已到，但是節點請求我們不要傳送中繼交易，那麼清空節點物件的傳送Inv 訊息的集合集合setInventoryTxToSend 。
```
if (fSendTrickle) {
LOCK(pto->cs_filter);
if (!pto->fRelayTxes) pto->setInventoryTxToSend.clear();
 }
```
如果傳送時間已到，並且節點請求過 BIP35 規定的mempool ，那麼：
- 呼叫記憶體池物件的infoAll 方法，返回記憶體池交易資訊集合。
- 設定區塊物件的mempool 為假。
- 獲取節點設定的最小交易費用過濾器。預設為 0。
```
CAmount filterrate = 0;
{
LOCK(pto->cs_feeFilter);
filterrate = pto->minFeeFilter;
}
```
- 遍歷記憶體池交易資訊集合並進行處理。用當前交易資訊生成一個 inv 物件，然後從區塊物件的setInventoryTxToSend 集合中刪除對應的交易資訊。如果設定了最小交易費用，並且當前交易的費用小於設定的最小費用，那麼處理下一個。
```
const uint256& hash = txinfo.tx->GetHash();
CInv inv(MSG_TX, hash);
pto->setInventoryTxToSend.erase(hash);
if (filterrate) {
if (txinfo.feeRate.GetFeePerK() < filterrate)
continue;
}
```
  如果區塊物件設定了布隆過濾器，並且當前交易不符合要求，那麼處理下一個。
```
if (pto->pfilter) {
if (!pto->pfilter->IsRelevantAndUpdate(*txinfo.tx)) continue;
}
```
  把當前交易的雜湊加入區塊物件的filterInventoryKnown 集合；把 inv 物件加入vInv 集合。如果集合已經達到規定的最大數量，那麼呼叫PushMessage 方法，傳送INV 訊息給遠端對等節點，然後清空集合。
```
pto->filterInventoryKnown.insert(hash);
vInv.push_back(inv);
if (vInv.size() == MAX_INV_SZ) {
connman->PushMessage(pto, msgMaker.Make(NetMsgType::INV, vInv));
vInv.clear();
}
```
- 設定區塊物件的timeLastMempoolReq 屬性。

如果需要傳送，即傳送時間已到，那麼：

生成交易的向量集合，並設定其大小。然後從區塊物件的setInventoryTxToSend 集合中取得其迭代器放進新生成的向量集合。

std::vector<std::set<uint256>::iterator> vInvTx;
vInvTx.reserve(pto->setInventoryTxToSend.size());
for (std::set<uint256>::iterator it = pto->setInventoryTxToSend.begin(); it != pto->setInventoryTxToSend.end(); it++) {
vInvTx.push_back(it);
}

設定區塊物件的最小交易費用，預設為0。

CAmount filterrate = 0;
{
LOCK(pto->cs_feeFilter);
filterrate = pto->minFeeFilter;
}

如果vInvTx 集合不空，並且需要中繼的交易數量小於規定的最大 INV 廣播數量，那就進行while 迴圈。下面是迴圈體：

while (!vInvTx.empty() && nRelayedTransactions < INVENTORY_BROADCAST_MAX) {
std::pop_heap(vInvTx.begin(), vInvTx.end(), compareInvMempoolOrder);
std::set<uint256>::iterator it = vInvTx.back();
vInvTx.pop_back();
uint256 hash = *it;
pto->setInventoryTxToSend.erase(it);
if (pto->filterInventoryKnown.contains(hash)) {
continue;
}
auto txinfo = mempool.info(hash);
if (!txinfo.tx) {
continue;
}
if (filterrate && txinfo.feeRate.GetFeePerK() < filterrate) {
continue;
}
if (pto->pfilter && !pto->pfilter->IsRelevantAndUpdate(*txinfo.tx)) continue;
vInv.push_back(CInv(MSG_TX, hash));
nRelayedTransactions++;
{
while (!vRelayExpiration.empty() && vRelayExpiration.front().first < nNow)
{
mapRelay.erase(vRelayExpiration.front().second);
vRelayExpiration.pop_front();
}
auto ret = mapRelay.insert(std::make_pair(hash, std::move(txinfo.tx)));
if (ret.second) {
vRelayExpiration.push_back(std::make_pair(nNow + 15 * 60 * 1000000, ret.first));
}
}
if (vInv.size() == MAX_INV_SZ) {
connman->PushMessage(pto, msgMaker.Make(NetMsgType::INV, vInv));
vInv.clear();
}
pto->filterInventoryKnown.insert(hash);
}

如果vInv 集合不空，那麼就呼叫PushMessage 方法，傳送INV 訊息。
如果區塊狀態物件的停止下載區塊的時間不等於0，並且小於當前時間減去規定的時間，那麼設定區塊物件為斷開，然後返回真。
```
nNow = GetTimeMicros();
 if (state.nStallingSince && state.nStallingSince < nNow - 1000000 * BLOCK_STALLING_TIMEOUT) {
pto->fDisconnect = true;
return true;
 }
```

如果區塊下載超時，那麼設定區塊物件為斷開，然後返回真。

if (state.vBlocksInFlight.size() > 0) {
QueuedBlock &queuedBlock = state.vBlocksInFlight.front();
int nOtherPeersWithValidatedDownloads = nPeersWithValidatedDownloads - (state.nBlocksInFlightValidHeaders > 0);
if (nNow > state.nDownloadingSince + consensusParams.nPowTargetSpacing * (BLOCK_DOWNLOAD_TIMEOUT_BASE + BLOCK_DOWNLOAD_TIMEOUT_PER_PEER * nOtherPeersWithValidatedDownloads)) {
pto->fDisconnect = true;
return true;
}
 }

接下來，檢查區塊頭部下載是否超時。

if (state.fSyncStarted && state.nHeadersSyncTimeout < std::numeric_limits<int64_t>::max()) {
if (pindexBestHeader->GetBlockTime() <= GetAdjustedTime() - 24*60*60) {
if (nNow > state.nHeadersSyncTimeout && nSyncStarted == 1 && (nPreferredDownload - state.fPreferredDownload >= 1)) {
if (!pto->fWhitelisted) {
LogPrintf("Timeout downloading headers from peer=%d, disconnecting\n", pto->GetId());
pto->fDisconnect = true;
return true;
} else {
state.fSyncStarted = false;
nSyncStarted--;
state.nHeadersSyncTimeout = 0;
}
}
} else {
state.nHeadersSyncTimeout = std::numeric_limits<int64_t>::max();
}
 }

獲取所有節點請求的資料，包括區塊與非區塊，並放在vGetData 集合中，然後呼叫PushMessage 方法，傳送給遠端節點。

std::vector<CInv> vGetData;
 if (!pto->fClient && ((fFetch && !pto->m_limited_node) || !IsInitialBlockDownload()) && state.nBlocksInFlight < MAX_BLOCKS_IN_TRANSIT_PER_PEER) {
std::vector<const CBlockIndex*> vToDownload;
NodeId staller = -1;
FindNextBlocksToDownload(pto->GetId(), MAX_BLOCKS_IN_TRANSIT_PER_PEER - state.nBlocksInFlight, vToDownload, staller, consensusParams);
for (const CBlockIndex *pindex : vToDownload) {
uint32_t nFetchFlags = GetFetchFlags(pto);
vGetData.push_back(CInv(MSG_BLOCK | nFetchFlags, pindex->GetBlockHash()));
MarkBlockAsInFlight(pto->GetId(), pindex->GetBlockHash(), pindex);
}
if (state.nBlocksInFlight == 0 && staller != -1) {
if (State(staller)->nStallingSince == 0) {
State(staller)->nStallingSince = nNow;
}
}
 }
 while (!pto->mapAskFor.empty() && (*pto->mapAskFor.begin()).first <= nNow)
 {
const CInv& inv = (*pto->mapAskFor.begin()).second;
if (!AlreadyHave(inv))
{
LogPrint(BCLog::NET, "Requesting %s peer=%d\n", inv.ToString(), pto->GetId());
vGetData.push_back(inv);
if (vGetData.size() >= 1000)
{
connman->PushMessage(pto, msgMaker.Make(NetMsgType::GETDATA, vGetData));
vGetData.clear();
}
} else {
pto->setAskFor.erase(inv.hash);
}
pto->mapAskFor.erase(pto->mapAskFor.begin());
 }
 if (!vGetData.empty())
connman->PushMessage(pto, msgMaker.Make(NetMsgType::GETDATA, vGetData));

我是區小白，Ulord全球社群聯盟（優得社群）核心區塊鏈技術開發者，深入研究比特幣,以太坊,EOS Dash,Rsk,Java, Nodejs,PHP,Python,C++ 我希望能聚集更多區塊鏈開發者，一起學習共同進步。為了更高效的交流探討區塊鏈開發過程中遇到的問題，歡迎將以上問題的答案在帖子下面留言。

從零開始學習比特幣：P2P 網路的建立之訊息處理（上篇）

ThreadMessageHandler

1、ProcessMessages

ProcessGetBlockData

2、SendMessages

您可能也會喜歡…