经过前面的铺垫，现在我们可以来看看BlockChainSync::onPeerBlockHeaders()这个函数的实现了，这个函数是EthereumPeer接收到BlockHeadersPacket包时被调用的，用来处理接收到的区块头。
这个函数有点长，还是一段一段来看吧。
if (m_daoChallengedPeers.find(_peer) != m_daoChallengedPeers.end())
{
if (verifyDaoChallengeResponse(_r))
syncPeer(_peer, false);
else
_peer->disable("Peer from another fork.");
m_daoChallengedPeers.erase(_peer);
return;
}
这段是处理DAO分叉，在以太坊C++源码解析（六）以太坊The DAO分叉里已经讲过。
clearPeerDownload(_peer);
if (m_state != SyncState::Blocks && m_state != SyncState::Waiting)
{
LOG(m_logger) << "Ignoring unexpected blocks";
return;
}
if (m_state == SyncState::Waiting)
{
LOG(m_loggerDetail) << "Ignored blocks while waiting";
return;
}
if (itemCount == 0)
{
LOG(m_loggerDetail) << "Peer does not have the blocks requested";
_peer->addRating(-1);
}
这部分中clearPeerDownload()函数用来清除该peer的m_downloadingHeaders，m_downloadingBodies等，以便重新获取，然后是同步状态的检测。
接着是遍历接收的数据头：
for (unsigned i = 0; i < itemCount; i++)
{
BlockHeader info(_r[i].data(), HeaderData);
unsigned blockNumber = static_cast<unsigned>(info.number());
if (blockNumber < m_chainStartBlock)
{
LOG(m_logger) << "Skipping too old header " << blockNumber;
continue;
}
if (haveItem(m_headers, blockNumber))
{
LOG(m_logger) << "Skipping header " << blockNumber << " (already downloaded)";
continue;
}
if (blockNumber <= m_lastImportedBlock && m_haveCommonHeader)
{
LOG(m_logger) << "Skipping header " << blockNumber << " (already imported)";
continue;
}
if (blockNumber > m_highestBlock)
m_highestBlock = blockNumber;
// ...
info是BlockHeader类对象，blockNumber是块号，检查是否该块头已经在下载缓冲区m_headers中等等。
下面这段就比较关键了，可以完整贴出来：
auto status = host().bq().blockStatus(info.hash());
if (status == QueueStatus::Importing || status == QueueStatus::Ready || host().chain().isKnown(info.hash()))
{
m_haveCommonHeader = true;
m_lastImportedBlock = (unsigned)info.number();
m_lastImportedBlockHash = info.hash();
if (!m_headers.empty() && m_headers.begin()->first == m_lastImportedBlock + 1 &&
m_headers.begin()->second[0].parent != m_lastImportedBlockHash)
{
// Start of the header chain in m_headers doesn't match our known chain,
// probably we've downloaded other fork
clog(VerbosityWarning, "sync")
<< "Unknown parent of the downloaded headers, restarting sync";
restartSync();
return;
}
}
在上一篇提到过区块链同步的回退操作，那么回退到什么时候刹住车呢？就是在这里判断的，假如这个块头在BlockQueue中导入过或者正在导入或者直接就在本地区块链中，那么就认为该区块头是可信的，就不需要再继续回退了，并将m_haveCommonHeader设为true。
既然这个区块头是可信的，那么就可以用它来检测m_headers里还在排队的区块里有没有不是一条心的。假如m_headers中块号最小的那个块恰好是这个区块的下一个块(块号+1)，并且那个块的父块又不是这个区块，那就说明m_headers里下载的区块有问题了，这个时候的处理是清空缓冲区，重新开始同步restartSync()。
如果这个区块头是新下载的区块头，并且当前不是在回退操作中，那么也要做检查，这里的检查要稍微复杂一些，既需要检查下载区块的下一个区块，又要检查下载区块的上一个区块：
if (m_haveCommonHeader)
{
Header const* prevBlock = findItem(m_headers, blockNumber - 1);
if ((prevBlock && prevBlock->hash != info.parentHash()) || (blockNumber == m_lastImportedBlock + 1 && info.parentHash() != m_lastImportedBlockHash))
{
// mismatching parent id, delete the previous block and don't add this one
clog(VerbosityWarning, "sync") << "Unknown block header " << blockNumber << " "
<< info.hash() << " (Restart syncing)";
_peer->addRating(-1);
restartSync();
return ;
}
Header const* nextBlock = findItem(m_headers, blockNumber + 1);
if (nextBlock && nextBlock->parent != info.hash())
{
LOG(m_loggerDetail)
<< "Unknown block header " << blockNumber + 1 << " " << nextBlock->hash;
// clear following headers
unsigned n = blockNumber + 1;
auto headers = m_headers.at(n);
for (auto const& h : headers)
{
BlockHeader deletingInfo(h.data, HeaderData);
m_headerIdToNumber.erase(headerId);
m_downloadingBodies.erase(n);
m_downloadingHeaders.erase(n);
++n;
}
removeAllStartingWith(m_headers, blockNumber + 1);
removeAllStartingWith(m_bodies, blockNumber + 1);
}
}
这里还是将m_headers中已下载的区块拿来做比较，如果m_headers中有下载区块的下一个区块，并且下一个区块的父区块不是下载区块或者下载区块是当前需要下载的下一个区块，但是下载区块的父区块不是上一个已下载的区块，那么也需要清空缓冲区，重新开始同步。这段文字比较拗口，还是直接看代码更容易理解。
检查下载区块的下一个区块也是类似的方法，但是有点区别的是这里如果检查未通过并不是重新开始同步，而是将m_headers，m_bodies等缓冲区里下载区块的后续区块清除掉。
检查通过之后，就可以将下载区块头加入m_headers中了：
mergeInto(m_headers, blockNumber, std::move(hdr));
if (headerId.transactionsRoot == EmptyTrie && headerId.uncles == EmptyListSHA3)
{
//empty body, just mark as downloaded
RLPStream r(2);
r.appendRaw(RLPEmptyList);
r.appendRaw(RLPEmptyList);
bytes body;
r.swapOut(body);
mergeInto(m_bodies, blockNumber, std::move(body));
}
else
m_headerIdToNumber[headerId] = blockNumber;
这里有一种特殊的区块，就是区块本身并不包含交易，也就是区块体是空的，那么我们就可以直接将一个空的区块体添加到m_bodies中，而不需要再去下载该区块体了。
所有工作都完成以后是两个函数的调用：
collectBlocks();
continueSync();
其中collectBlocks()是判断是否有合适的区块头和区块体需要合并写入二级缓冲区BlockQueue，continueSync()是继续调用syncPeer()向其他peer请求区块头或者区块体。
网友评论