比特币源码情景分析之致密区块(Compact block)
致密区块是一种降低区块数据同步带宽的解决方案。
我们在SPV钱包章节https://blog.csdn.net/ITleaks/article/details/79922902就提到了SPV钱包因为资源的限制(硬件性能，移动网络带宽贵慢)需要减少数据处理和传输，其使用bloom filter 机制过滤SPV节点不需要的数据，大幅提升了性能。全节点因为需要维护所有地址(钱包)的交易数据，因而没法通过bloom filter这种方式来优化性能。但是还是有人想到了致密区块，其核心工作原理如下：
每个节点都会保存pending transaction到本地mempool。当miner挖出新块打包交易后，其他节点会同步下载这个新区块的交易数据内容。但是其实新块里的交易数据大部分应该都已经保存在本地mempool，是没必要下载。致密区块就是利用这一特性实现的，新块通知会将块包含的交易缩短交易哈希id(stxids)发送到其他节点，其他节点对比本地mempool中的交易hash和传输过来的缩短交易哈希id(stxids), 发现本地已经存在就过滤掉，只会向通知者下载本地不存在的交易数据，极大的减少了数据的传输。
CMPCTBLOCK流程
- 传统块中继(同步)
A节点通过INV告知B节点有新块，B然后通过getdata从A获取完整的区块交易数据
-
高带宽CMPCT块中继节点B使用sendcmpt(1)来告诉节点A它想使用Compact方式接受区块数据, 其实就是向A表明自己支持Compact模式。当一个新的区块抵达时，节点A验证区块保存后便会将区块头、缩短交易哈希id（txids)等信息发送给节点B。节点B尝试比对交易数据，并请求（getblocktxn）本地不存在的交易，它们是由节点A发送（blocktxn）的。在此背景下，在将区块添加到各自的区块链副本之前，两个节点都完成了完整的区块验证，并和以前一样保持了相同的全节点安全性。
-
低带宽CMPCT块中继低带宽CMPCT块中继和高带宽CMPCT块中继基本相同，只是多了一个INV(MSG_BLOCK)操作。在发送CMPCTBLOCK前先通过INV消息告知节点B新块头信息，节点B判断区块是否已经存在后再主动要求A发送CMPTBLOCK。相比CMPTBLOCK，INV(MSG_BLOCK)只包含块头少量信息，该方案能进一步减少数据的传输，降低带宽
wiki原文
1. The sendcmpct message is defined as a message containing a 1-byte integer followed by a 8-byte integer where pchCommand == "sendcmpct".
2. The first integer SHALL be interpreted as a boolean (and MUST have a value of either 1 or 0)
3. The second integer SHALL be interpreted as a little-endian version number. Nodes sending a sendcmpct message MUST currently set this value to 1.
4. Upon receipt of a "sendcmpct" message with the first and second integers set to 1, the node SHOULD announce new blocks by sending a cmpctblock message.
5. Upon receipt of a "sendcmpct" message with the first integer set to 0, the node SHOULD NOT announce new blocks by sending a cmpctblock message, but SHOULD announce new blocks by sending invs or headers, as defined by BIP130.
6. Upon receipt of a "sendcmpct" message with the second integer set to something other than 1, nodes MUST treat the peer as if they had not received the message (as it indicates the peer will provide an unexpected encoding in cmpctblock, and/or other, messages). This allows future versions to send duplicate sendcmpct messages with different versions as a part of a version handshake for future versions. See Protocol Versioning section, below, for more info on the specifics of the version-negotiation mechanics.
7. Nodes SHOULD check for a protocol version of >= 70014 before sending sendcmpct messages.
8. Nodes MUST NOT send a request for a MSG_CMPCT_BLOCK object to a peer before having received a sendcmpct message from that peer.
9. Nodes MUST NOT request a MSG_CMPCT_BLOCK object before having sent all sendcmpct messages to that peer which they intend to send, as the peer cannot know what version protocol to use in the response.
CMPCTBLOCK源码分析
CMPCT模式支持消息SENDCMPCT
节点B向节点A发送SENDCMPCT消息，节点A会在pfrom记住这一状态
else if (strCommand == NetMsgType::SENDCMPCT)
{
bool fAnnounceUsingCMPCTBLOCK = false;
uint64_t nCMPCTBLOCKVersion = 0;
vRecv >> fAnnounceUsingCMPCTBLOCK >> nCMPCTBLOCKVersion;
if (nCMPCTBLOCKVersion == 1 || ((pfrom->GetLocalServices() & NODE_WITNESS) && nCMPCTBLOCKVersion == 2)) {
LOCK(cs_main);
// fProvidesHeaderAndIDs is used to "lock in" version of compact blocks we send (fWantsCmpctWitness)
if (!State(pfrom->GetId())->fProvidesHeaderAndIDs) {
State(pfrom->GetId())->fProvidesHeaderAndIDs = true;
State(pfrom->GetId())->fWantsCmpctWitness = nCMPCTBLOCKVersion == 2;
}
if (State(pfrom->GetId())->fWantsCmpctWitness == (nCMPCTBLOCKVersion == 2)) // ignore later version announces
State(pfrom->GetId())->fPreferHeaderAndIDs = fAnnounceUsingCMPCTBLOCK;
if (!State(pfrom->GetId())->fSupportsDesiredCmpctVersion) {
if (pfrom->GetLocalServices() & NODE_WITNESS)
State(pfrom->GetId())->fSupportsDesiredCmpctVersion = (nCMPCTBLOCKVersion == 2);
else
State(pfrom->GetId())->fSupportsDesiredCmpctVersion = (nCMPCTBLOCKVersion == 1);
}
}
}
高带宽CMPCT块中继
A节点发送CMPCTBLOCK消息
节点A收到新块，将通知节点B CMPCTBLOCK消息
void PeerLogicValidation::NewPoWValidBlock(const CBlockIndex *pindex, const std::shared_ptr<const CBlock>& pblock) {
std::shared_ptr<const CBlockHeaderAndShortTxIDs> pcmpctblock = std::make_shared<const CBlockHeaderAndShortTxIDs> (*pblock, true)
bool fWitnessEnabled = IsWitnessEnabled(pindex->pprev, Params().GetConsensus());
uint256 hashBlock(pblock->GetHash());
//给所有其他节点包括B节点发送CMPCTLBOCK消息
connman->ForEachNode([this, &pcmpctblock, pindex, &msgMaker, fWitnessEnabled, &hashBlock](CNode* pnode) {
// TODO: Avoid the repeated-serialization here
if (pnode->nVersion < INVALID_CB_NO_BAN_VERSION || pnode->fDisconnect)
return;
ProcessBlockAvailability(pnode->GetId());
CNodeState &state = *State(pnode->GetId());
// If the peer has, or we announced to them the previous block already,
// but we don't think they have this one, go ahead and announce it
if (state.fPreferHeaderAndIDs && (!fWitnessEnabled || state.fWantsCmpctWitness) &&
!PeerHasHeader(&state, pindex) && PeerHasHeader(&state, pindex->pprev)) {
LogPrint(BCLog::NET, "%s sending header-and-ids %s to peer=%d\n", "PeerLogicValidation::NewPoWValidBlock",
hashBlock.ToString(), pnode->GetId());
connman->PushMessage(pnode, msgMaker.Make(NetMsgType::CMPCTBLOCK, *pcmpctblock));
state.pindexBestHeaderSent = pindex;
}
});
}
CBlockHeaderAndShortTxIDs::CBlockHeaderAndShortTxIDs(const CBlock& block, bool fUseWTXID) :
nonce(GetRand(std::numeric_limits<uint64_t>::max())),
shorttxids(block.vtx.size() - 1), prefilledtxn(1), header(block) {
FillShortTxIDSelector();
//TODO: Use our mempool prior to block acceptance to predictively fill more than just the coinbase
prefilledtxn[0] = {0, block.vtx[0]};
for (size_t i = 1; i < block.vtx.size(); i++) {
const CTransaction& tx = *block.vtx[i];
shorttxids[i - 1] = GetShortID(fUseWTXID ? tx.GetWitnessHash() : tx.GetHash());
}
}
//取256位的交易hash的低64位作为shorttxid
uint64_t CBlockHeaderAndShortTxIDs::GetShortID(const uint256& txhash) const {
static_assert(SHORTTXIDS_LENGTH == 6, "shorttxids calculation assumes 6-byte shorttxids");
return SipHashUint256(shorttxidk0, shorttxidk1, txhash) & 0xffffffffffffL;
}
B节点收到CMPCTBLOCK消息
else if (strCommand == NetMsgType::CMPCTBLOCK && !fImporting && !fReindex) // Ignore blocks received while importing
{
CBlockHeaderAndShortTxIDs cmpctblock;
vRecv >> cmpctblock;
bool received_new_header = false;
{
LOCK(cs_main);
//查找cmpctblock的prev是否存在,必须存在，否则不连续
if (!LookupBlockIndex(cmpctblock.header.hashPrevBlock)) {
// Doesn't connect (or is genesis), instead of DoSing in AcceptBlockHeader, request deeper headers
if (!IsInitialBlockDownload())
connman->PushMessage(pfrom, msgMaker.Make(NetMsgType::GETHEADERS, chainActive.GetLocator(pindexBestHeader), uint256()));
return true;
}
//块是否在本地存在
if (!LookupBlockIndex(cmpctblock.header.GetHash())) {
received_new_header = true;
}
}
const CBlockIndex *pindex = nullptr;
CValidationState state;
//处理头部
if (!ProcessNewBlockHeaders({cmpctblock.header}, state, chainparams, &pindex)) {
int nDoS;
if (state.IsInvalid(nDoS)) {
if (nDoS > 0) {
LOCK(cs_main);
Misbehaving(pfrom->GetId(), nDoS, strprintf("Peer %d sent us invalid header via cmpctblock\n", pfrom->GetId()));
} else {
LogPrint(BCLog::NET, "Peer %d sent us invalid header via cmpctblock\n", pfrom->GetId());
}
return true;
}
}
…..
//接下来是重头戏，比较short hashid和本地交易hashid，找出本地不存在的交易
// We want to be a bit conservative just to be extra careful about DoS
// possibilities in compact block processing...
if (pindex->nHeight <= chainActive.Height() + 2) {
if ((!fAlreadyInFlight && nodestate->nBlocksInFlight < MAX_BLOCKS_IN_TRANSIT_PER_PEER) ||
(fAlreadyInFlight && blockInFlightIt->second.first == pfrom->GetId())) {
//这个就会筛选出所有本地没有的交易
PartiallyDownloadedBlock& partialBlock = *(*queuedBlockIt)->partialBlock;
ReadStatus status = partialBlock.InitData(cmpctblock, vExtraTxnForCompact);
BlockTransactionsRequest req;
for (size_t i = 0; i < cmpctblock.BlockTxCount(); i++) {
if (!partialBlock.IsTxAvailable(i))
req.indexes.push_back(i);
}
if (req.indexes.empty()) {
// Dirty hack to jump to BLOCKTXN code (TODO: move message handling into their own functions)
BlockTransactions txn;
txn.blockhash = cmpctblock.header.GetHash();
blockTxnMsg << txn;
fProcessBLOCKTXN = true;
} else {
req.blockhash = pindex->GetBlockHash();
//请求本地不存在的交易信息
connman->PushMessage(pfrom, msgMaker.Make(NetMsgType::GETBLOCKTXN, req));
}
}
} // cs_main
}
ReadStatus PartiallyDownloadedBlock::InitData(const CBlockHeaderAndShortTxIDs& cmpctblock, const std::vector<std::pair<uint256, CTransactionRef>>& extra_txn) {
//建立<shorttxid,index>map,为啥需要构建这样一个对象呢，因为我们后续的getblocktxn的参数是txindex,而不是txhash，因为txindex占用空间少
// Calculate map of txids -> positions and check mempool to see what we have (or don't)
// Because well-formed cmpctblock messages will have a (relatively) uniform distribution
// of short IDs, any highly-uneven distribution of elements can be safely treated as a
// READ_STATUS_FAILED.
std::unordered_map<uint64_t, uint16_t> shorttxids(cmpctblock.shorttxids.size());
uint16_t index_offset = 0;
for (size_t i = 0; i < cmpctblock.shorttxids.size(); i++) {
while (txn_available[i + index_offset])
index_offset++;
shorttxids[cmpctblock.shorttxids[i]] = i + index_offset;
// To determine the chance that the number of entries in a bucket exceeds N,
// we use the fact that the number of elements in a single bucket is
// binomially distributed (with n = the number of shorttxids S, and p =
// 1 / the number of buckets), that in the worst case the number of buckets is
// equal to S (due to std::unordered_map having a default load factor of 1.0),
// and that the chance for any bucket to exceed N elements is at most
// buckets * (the chance that any given bucket is above N elements).
// Thus: P(max_elements_per_bucket > N) <= S * (1 - cdf(binomial(n=S,p=1/S), N)).
// If we assume blocks of up to 16000, allowing 12 elements per bucket should
// only fail once per ~1 million block transfers (per peer and connection).
if (shorttxids.bucket_size(shorttxids.bucket(cmpctblock.shorttxids[i])) > 12)
return READ_STATUS_FAILED;
}
// TODO: in the shortid-collision case, we should instead request both transactions
// which collided. Falling back to full-block-request here is overkill.
if (shorttxids.size() != cmpctblock.shorttxids.size())
return READ_STATUS_FAILED; // Short ID collision
std::vector<bool> have_txn(txn_available.size());
{
LOCK(pool->cs);
//用mempool过滤本地存在的transaction
const std::vector<std::pair<uint256, CTxMemPool::txiter> >& vTxHashes = pool->vTxHashes;
for (size_t i = 0; i < vTxHashes.size(); i++) {
uint64_t shortid = cmpctblock.GetShortID(vTxHashes[i].first);
std::unordered_map<uint64_t, uint16_t>::iterator idit = shorttxids.find(shortid);
if (idit != shorttxids.end()) {
if (!have_txn[idit->second]) {
txn_available[idit->second] = vTxHashes[i].second->GetSharedTx();
have_txn[idit->second] = true;
mempool_count++;
} else {
// If we find two mempool txn that match the short id, just request it.
// This should be rare enough that the extra bandwidth doesn't matter,
// but eating a round-trip due to FillBlock failure would be annoying
if (txn_available[idit->second]) {
txn_available[idit->second].reset();
mempool_count--;
}
}
}
// Though ideally we'd continue scanning for the two-txn-match-shortid case,
// the performance win of an early exit here is too good to pass up and worth
// the extra risk.
if (mempool_count == shorttxids.size())
break;
}
return READ_STATUS_OK;
}
A节点处理GETBLOCKTXN消息
else if (strCommand == NetMsgType::GETBLOCKTXN)
{
BlockTransactionsRequest req;
vRecv >> req;
std::shared_ptr<const CBlock> recent_block;
{
LOCK(cs_most_recent_block);
if (most_recent_block_hash == req.blockhash)
recent_block = most_recent_block;
// Unlock cs_most_recent_block to avoid cs_main lock inversion
}
//在recent_block,直接发recent_block
if (recent_block) {
SendBlockTransactions(*recent_block, req, pfrom, connman);
return true;
}
LOCK(cs_main);
const CBlockIndex* pindex = LookupBlockIndex(req.blockhash);
if (!pindex || !(pindex->nStatus & BLOCK_HAVE_DATA)) {
LogPrint(BCLog::NET, "Peer %d sent us a getblocktxn for a block we don't have", pfrom->GetId());
return true;
}
if (pindex->nHeight < chainActive.Height() - MAX_BLOCKTXN_DEPTH) {
//如果B节点请求的是老区块，为了避免B节点恶意大量请求getblocktxn(大量高耗时的磁盘读取操作)，直接转值MSG_BLOCK消息逻辑.那为啥转至MSG_BLOCK就能避免恶意getblocktxn，因为节点收到MSG_BLOCK消息也是需要处理每条交易, 不存在getblocktxn只需处理少量交易数据的优势，且还需要多前面几个消息对话，恶意节点还不如直接一开始就使用getdata(MSG_BLOCK)消息，这样最后的结果就是，所有节点只对tip区块使用getblocktxn消息
//简单说来，getblocktxn消息A节点工作量远大于B节点, 而getdata(MSG_BLOCK)消息A节点和B节点工作量差不多，所以作恶的可能性就少了
// If an older block is requested (should never happen in practice,
// but can happen in tests) send a block response instead of a
// blocktxn response. Sending a full block response instead of a
// small blocktxn response is preferable in the case where a peer
// might maliciously send lots of getblocktxn requests to trigger
// expensive disk reads, because it will require the peer to
// actually receive all the data read from disk over the network.
LogPrint(BCLog::NET, "Peer %d sent us a getblocktxn for a block > %i deep", pfrom->GetId(), MAX_BLOCKTXN_DEPTH);
CInv inv;
inv.type = State(pfrom->GetId())->fWantsCmpctWitness ? MSG_WITNESS_BLOCK : MSG_BLOCK;
inv.hash = req.blockhash;
pfrom->vRecvGetData.push_back(inv);
// The message processing loop will go around again (without pausing) and we'll respond then (without cs_main)
return true;
}
CBlock block;
bool ret = ReadBlockFromDisk(block, pindex, chainparams.GetConsensus());
assert(ret);
SendBlockTransactions(block, req, pfrom, connman);
}
inline void static SendBlockTransactions(const CBlock& block, const BlockTransactionsRequest& req, CNode* pfrom, CConnman* connman) {
BlockTransactions resp(req);
for (size_t i = 0; i < req.indexes.size(); i++) {
if (req.indexes[i] >= block.vtx.size()) {
LOCK(cs_main);
Misbehaving(pfrom->GetId(), 100, strprintf("Peer %d sent us a getblocktxn with out-of-bounds tx indices", pfrom->GetId()));
return;
}
//对应index的交易数据写入
resp.txn[i] = block.vtx[req.indexes[i]];
}
LOCK(cs_main);
const CNetMsgMaker msgMaker(pfrom->GetSendVersion());
int nSendFlags = State(pfrom->GetId())->fWantsCmpctWitness ? 0 : SERIALIZE_TRANSACTION_NO_WITNESS;
//返回交易数据
connman->PushMessage(pfrom, msgMaker.Make(nSendFlags, NetMsgType::BLOCKTXN, resp));
}
B节点处理BLOCKTXN消息
else if (strCommand == NetMsgType::BLOCKTXN && !fImporting && !fReindex) // Ignore blocks received while importing
{
BlockTransactions resp;
vRecv >> resp;
std::shared_ptr<CBlock> pblock = std::make_shared<CBlock>();
bool fBlockRead = false;
{
PartiallyDownloadedBlock& partialBlock = *it->second.second->partialBlock;
//将收到的交易构造pblock
ReadStatus status = partialBlock.FillBlock(*pblock, resp.txn);
}
if (fBlockRead) {
bool fNewBlock = false;
// Since we requested this block (it was in mapBlocksInFlight), force it to be processed,
// even if it would not be a candidate for new tip (missing previous block, chain not long enough, etc)
// This bypasses some anti-DoS logic in AcceptBlock (eg to prevent
// disk-space attacks), but this should be safe due to the
// protections in the compact block handler -- see related comment
// in compact block optimistic reconstruction handling.
//添加到本地
ProcessNewBlock(chainparams, pblock, /*fForceProcessing=*/true, &fNewBlock)
}
}
ReadStatus PartiallyDownloadedBlock::FillBlock(CBlock& block, const std::vector<CTransactionRef>& vtx_missing) {
assert(!header.IsNull());
uint256 hash = header.GetHash();
block = header;
block.vtx.resize(txn_available.size());
size_t tx_missing_offset = 0;
for (size_t i = 0; i < txn_available.size(); i++) {
if (!txn_available[i]) {
//其他节点发送过来的交易
if (vtx_missing.size() <= tx_missing_offset)
return READ_STATUS_INVALID;
block.vtx[i] = vtx_missing[tx_missing_offset++];
} else
//本地有的交易
block.vtx[i] = std::move(txn_available[i]);
}
return READ_STATUS_OK;
}
低带宽CMPCT块中继
A节点通知B节点新块后，经过INV(MSG_BLOCK), GETHEADERS消息传递后，B节点经过会接受到区块头信息，然后就会发送getdata(cmpct)
bool static ProcessHeadersMessage(CNode *pfrom, CConnman *connman, const std::vector<CBlockHeader>& headers, const CChainParams& chainparams, bool punish_duplicate_invalid)
if (!ProcessNewBlockHeaders(headers, state, chainparams, &pindexLast, &first_invalid_header)) {
bool fCanDirectFetch = CanDirectFetch(chainparams.GetConsensus());
// If this set of headers is valid and ends in a block with at least as
// much work as our tip, download as much as possible.
if (fCanDirectFetch && pindexLast->IsValid(BLOCK_VALID_TREE) && chainActive.Tip()->nChainWork <= pindexLast->nChainWork) {
std::vector<const CBlockIndex*> vToFetch;
const CBlockIndex *pindexWalk = pindexLast;
// Calculate all the blocks we'd need to switch to pindexLast, up to a limit.
while (pindexWalk && !chainActive.Contains(pindexWalk) && vToFetch.size() <= MAX_BLOCKS_IN_TRANSIT_PER_PEER) {
if (!(pindexWalk->nStatus & BLOCK_HAVE_DATA) &&
!mapBlocksInFlight.count(pindexWalk->GetBlockHash()) &&
(!IsWitnessEnabled(pindexWalk->pprev, chainparams.GetConsensus()) || State(pfrom->GetId())->fHaveWitness)) {
// We don't have this block, and it's not yet in flight.
vToFetch.push_back(pindexWalk);
}
pindexWalk = pindexWalk->pprev;
}
// If pindexWalk still isn't on our main chain, we're looking at a
// very large reorg at a time we think we're close to caught up to
// the main chain -- this shouldn't really happen. Bail out on the
// direct fetch and rely on parallel download instead.
if (!chainActive.Contains(pindexWalk)) {
} else {
std::vector<CInv> vGetData;
// Download as much as possible, from earliest to latest.
for (const CBlockIndex *pindex : reverse_iterate(vToFetch)) {
if (nodestate->nBlocksInFlight >= MAX_BLOCKS_IN_TRANSIT_PER_PEER) {
// Can't download any more from this peer
break;
}
uint32_t nFetchFlags = GetFetchFlags(pfrom);
vGetData.push_back(CInv(MSG_BLOCK | nFetchFlags, pindex->GetBlockHash()));
MarkBlockAsInFlight(pfrom->GetId(), pindex->GetBlockHash(), pindex);
LogPrint(BCLog::NET, "Requesting block %s from peer=%d\n",
pindex->GetBlockHash().ToString(), pfrom->GetId());
}
if (vGetData.size() > 1) {
LogPrint(BCLog::NET, "Downloading blocks toward %s (%d) via headers direct fetch\n",
pindexLast->GetBlockHash().ToString(), pindexLast->nHeight);
}
if (vGetData.size() > 0) {
//低带宽CMPCT块中继模式
if (nodestate->fSupportsDesiredCmpctVersion && vGetData.size() == 1 && mapBlocksInFlight.size() == 1 && pindexLast->pprev->IsValid(BLOCK_VALID_CHAIN)) {
// In any case, we want to download using a compact block, not a regular one
vGetData[0] = CInv(MSG_CMPCT_BLOCK, vGetData[0].hash);
}
connman->PushMessage(pfrom, msgMaker.Make(NetMsgType::GETDATA, vGetData));
}
}
}
}
节点A收到getdata(CMPCT)后会将CMPCTBLOCK发送给A节点
else if (inv.type == MSG_CMPCT_BLOCK)
{
// If a peer is asking for old blocks, we're almost guaranteed
// they won't have a useful mempool to match against a compact block,
// and we don't feel like constructing the object for them, so
// instead we respond with the full, non-compact 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));
}
}
剩下的流程就和高带宽CMPCT块中继一样