说明：
基于Hyperledger Fabric 1.2的源码；
自己阅读源码也是在学习和摸索中，有错误的话欢迎指正，也有不少还不懂的地方，欢迎指导和讨论。
主要流程
【接收交易proposal的过程】
- peer在调用chaincode执行交易后，会依据collection配置在有权限的peer间散播私有数据，并存储私有数据到自己的transient store。
- 其他接收到散播私有数据的peer，会将私有数据存储到自己的transient store。
【接收Block的过程】
- peer最终接收到block后，会检查自己是否有权限获取私有数据，有权限的话从transient store中拿数据，如果没有，则去其他peer上获取，最终私有数据会存储到pvtdataStore中。
源码分析
A. 接收交易proposal
在peer节点处理proposal（core/endorser/endorser.go的ProcessProposal函数）的过程中，执行模拟操作（SimulateProposal函数）环节中，调用chaincode执行交易以后，会判断读写集结果中是否包含私有数据，如果包含，则执行以下操作（代码如下）：
if simResult.PvtSimulationResults != nil {
if [cid.Name](http://cid.name/)== "lscc" {
// TODO: remove once we can store collection configuration outside of LSCC
txsim.Done()
return nil, nil, nil, nil, errors.New("Private data is forbidden to be used in instantiate")
}
// 汇总私有数据collection配置和私有数据信息
pvtDataWithConfig, err := e.AssemblePvtRWSet(simResult.PvtSimulationResults, txsim)
// To read collection config need to read collection updates before
// releasing the lock, hence txsim.Done() moved down here
txsim.Done()
if err != nil {
return nil, nil, nil, nil, errors.WithMessage(err, "failed to obtain collections config")
}
endorsedAt, err := e.s.GetLedgerHeight(chainID)
if err != nil {
return nil, nil, nil, nil, errors.WithMessage(err, fmt.Sprint("failed to obtain ledger height for channel", chainID))
}
// Add ledger height at which transaction was endorsed,
// `endorsedAt` is obtained from the block storage and at times this could be 'endorsement Height + 1'.
// However, since we use this height only to select the configuration (3rd parameter in distributePrivateData) and
// manage transient store purge for orphaned private writesets (4th parameter in distributePrivateData), this works for now.
// Ideally, ledger should add support in the simulator as a first class function `GetHeight()`.
pvtDataWithConfig.EndorsedAt = endorsedAt
// 在chainID为名的channel中，依据collection配置，散播私有数据
if err := e.distributePrivateData(chainID, txid, pvtDataWithConfig, endorsedAt); err != nil {
return nil, nil, nil, nil, err
}
}
以上代码内容包括：
- 汇总私有数据collection配置和私有数据信息（core/endorser/pvtrwset_assembler.go的AssemblePvtRWSet函数）：
根据读写集中私有数据的合约名（namespace），从系统合约lscc处获取其对应的collection配置（即实例化合约时提供的collection_config.yaml），整理一番（代码如下）；
// AssemblePvtRWSet prepares TxPvtReadWriteSet for distribution
// augmenting it into TxPvtReadWriteSetWithConfigInfo adding
// information about collections config available related
// to private read-write set
func (as *rwSetAssembler) AssemblePvtRWSet(privData *rwset.TxPvtReadWriteSet, txsim CollectionConfigRetriever) (*transientstore.TxPvtReadWriteSetWithConfigInfo, error) {
txPvtRwSetWithConfig := &transientstore.TxPvtReadWriteSetWithConfigInfo{
PvtRwset: privData,
CollectionConfigs: make(map[string]*common.CollectionConfigPackage),
}
for _, pvtRwset := range privData.NsPvtRwset {
namespace := pvtRwset.Namespace
if _, found := txPvtRwSetWithConfig.CollectionConfigs[namespace]; !found {
cb, err := txsim.GetState("lscc", privdata.BuildCollectionKVSKey(namespace))
if err != nil {
return nil, errors.WithMessage(err, fmt.Sprintf("error while retrieving collection config for chaincode %#v", namespace))
}
if cb == nil {
return nil, errors.New(fmt.Sprintf("no collection config for chaincode %#v", namespace))
}
colCP := &common.CollectionConfigPackage{}
err = proto.Unmarshal(cb, colCP)
if err != nil {
return nil, errors.Wrapf(err, "invalid configuration for collection criteria %#v", namespace)
}
txPvtRwSetWithConfig.CollectionConfigs[namespace] = colCP
}
}
as.trimCollectionConfigs(txPvtRwSetWithConfig)
return txPvtRwSetWithConfig, nil
}
- 在peer之间分发私有数据：
调用distributePrivateData函数，该函数在生成EndorserServer时由外部传值建立（peer/node/start.go的serve函数），即privDataDist函数（代码如下）。
privDataDist := func(channel string, txID string, privateData *transientstore.TxPvtReadWriteSetWithConfigInfo, blkHt uint64) error {
return service.GetGossipService().DistributePrivateData(channel, txID, privateData, blkHt)
}
...
serverEndorser := endorser.NewEndorserServer(privDataDist, endorserSupport)
其中，service.GetGossipService()返回gossipService的唯一实例（gossip/service/gossip_service.go的gossipServiceInstance），调用函数（gossip/service/gossip_service.go的DistributePrivateData函数）完成在channel内部的私有数据传输（代码如下）：
// DistributePrivateData distribute private read write set inside the channel based on the collections policies
func (g *gossipServiceImpl) DistributePrivateData(chainID string, txID string, privData *transientstore.TxPvtReadWriteSetWithConfigInfo, blkHt uint64) error {
g.lock.RLock()
// 根据channel名称获取私有数据的handler
handler, exists := g.privateHandlers[chainID]
g.lock.RUnlock()
if !exists {
return errors.Errorf("No private data handler for %s", chainID)
}
// 依据collection配置散播私有数据
if err := handler.distributor.Distribute(txID, privData, blkHt); err != nil {
logger.Error("Failed to distributed private collection, txID", txID, "channel", chainID, "due to", err)
return err
}
if err := handler.coordinator.StorePvtData(txID, privData, blkHt); err != nil {
logger.Error("Failed to store private data into transient store, txID",
txID, "channel", chainID, "due to", err)
return err
}
return nil
}
以上函数包括如下步骤：
1）根据channel名称chainID获取私有数据的handler；
2）依据collection配置散播私有数据（gossip/privdata/distributor.go的Distribute函数，代码如下）：
// Distribute broadcast reliably private data read write set based on policies
func (d *distributorImpl) Distribute(txID string, privData *transientstore.TxPvtReadWriteSetWithConfigInfo, blkHt uint64) error {
disseminationPlan, err := d.computeDisseminationPlan(txID, privData, blkHt)
if err != nil {
return errors.WithStack(err)
}
return d.disseminate(disseminationPlan)
}
Distribute函数包括以下步骤：
a）制定散播方案（gossip/privdata/distributor.go的computeDisseminationPlan函数）：根据collection配置获取策略，设置过滤器，生成私有数据消息，生成散播方案（dissemination的列表，dissemination包含msg和criteria）；
b）根据散播方案，开启协程启动散播（散播最终调用的函数需要收到接收端的ACK）；
c）存储私有数据到Transient Store（core/transientstore/store.go的PersistWithConfig函数）：创建comisiteKey为键值（包括txid, uuid和blockHeight），存储私有数据，并创建两个用于后期清理的索引（compositeKeyPurgeIndexByHeight和compositeKeyPurgeIndexByTxid）,数据最后存储在leveldb中，位于/var/hyperledger/production/transientStore路径下。
B. 接收Block和私有数据
在peer启动时会启动Gossip服务，其中会启动接收消息的协程
（调用过程：peer/node/start.go的serve函数—>core/peer/peer.go的Initialize函数—>core/peer/peer.go的createChain函数—>gossip/service/gossip_service.go的InitializeChannel函数—>gossip/state/state.go的NewGossipStateProvider函数）
主要关注以下代码：
// Listen for incoming communication
go s.listen()
// Deliver in order messages into the incoming channel
go s.deliverPayloads()
1. listen函数处理
func (s *GossipStateProviderImpl) listen() {
defer s.done.Done()
for {
select {
case msg := <-s.gossipChan:
logger.Debug("Received new message via gossip channel")
go s.queueNewMessage(msg)
case msg := <-s.commChan:
logger.Debug("Dispatching a message", msg)
go s.dispatch(msg)
case <-s.stopCh:
s.stopCh <- struct{}{}
logger.Debug("Stop listening for new messages")
return
}
}
}
1）对于gossipChan的消息
（DataMessage类型，即区块）
协程调用queueNewMessage函数（gossip/state/state.go的queueNewMessage函数），将同一channel的信息放到payload中（gossip/state/state.go的addPayload函数）（代码如下）。
gossipChan, _ := services.Accept(func(message interface{}) bool {
// Get only data messages
return message.(*proto.GossipMessage).IsDataMsg() &&
bytes.Equal(message.(*proto.GossipMessage).Channel, []byte(chainID))
}, false)
...
// New message notification/handler
func (s *GossipStateProviderImpl) queueNewMessage(msg *proto.GossipMessage) {
if !bytes.Equal(msg.Channel, []byte(s.chainID)) {
logger.Warning("Received enqueue for channel",
string(msg.Channel), "while expecting channel", s.chainID, "ignoring enqueue")
return
}
dataMsg := msg.GetDataMsg()
if dataMsg != nil {
if err := s.addPayload(dataMsg.GetPayload(), nonBlocking); err != nil {
logger.Warning("Failed adding payload:", err)
return
}
logger.Debugf("Received new payload with sequence number = [%d]", dataMsg.Payload.SeqNum)
} else {
logger.Debug("Gossip message received is not of data message type, usually this should not happen.")
}
}
2）对于commChan的消息
（RemoteStateRequest类型或RemoteStateResponse类型，即远程peer的请求或响应信息）
该通道的消息是经过remoteStateMsgFilter筛选的（主要检查消息内容，channel权限），协程调用dispatch函数（gossip/state/state.go的dispatch函数），分别处理state message和私有数据信息。这里主要关注私有数据，如果获得私有数据，则进行处理（gossip/state/state.go的privateDataMessage函数）：
a）检查channel，需要一致；
b）获取私有数据信息和collection配置信息；
c）写入Transient Store（最终还是调用core/transientstore/store.go的PersistWithConfig函数）；
d）返回ACK。
remoteStateMsgFilter := func(message interface{}) bool {
receivedMsg := message.(proto.ReceivedMessage)
msg := receivedMsg.GetGossipMessage()
if !(msg.IsRemoteStateMessage() || msg.GetPrivateData() != nil) {
return false
}
// Ensure we deal only with messages that belong to this channel
if !bytes.Equal(msg.Channel, []byte(chainID)) {
return false
}
connInfo := receivedMsg.GetConnectionInfo()
authErr := services.VerifyByChannel(msg.Channel, connInfo.Identity, connInfo.Auth.Signature, connInfo.Auth.SignedData)
if authErr != nil {
logger.Warning("Got unauthorized request from", string(connInfo.Identity))
return false
}
return true
}
// Filter message which are only relevant for nodeMetastate transfer
_, commChan := services.Accept(remoteStateMsgFilter, true)
...
func (s *GossipStateProviderImpl) dispatch(msg proto.ReceivedMessage) {
// Check type of the message
if msg.GetGossipMessage().IsRemoteStateMessage() {
logger.Debug("Handling direct state transfer message")
// Got state transfer request response
s.directMessage(msg)
} else if msg.GetGossipMessage().GetPrivateData() != nil {
logger.Debug("Handling private data collection message")
// Handling private data replication message
s.privateDataMessage(msg)
}
}
2. deliverPayloads函数处理
func (s *GossipStateProviderImpl) deliverPayloads() {
defer s.done.Done()
for {
select {
// Wait for notification that next seq has arrived
case <-s.payloads.Ready():
logger.Debugf("Ready to transfer payloads to the ledger, next sequence number is = [%d]", s.payloads.Next())
// Collect all subsequent payloads
for payload := s.payloads.Pop(); payload != nil; payload = s.payloads.Pop() {
rawBlock := &common.Block{}
if err := pb.Unmarshal(payload.Data, rawBlock); err != nil {
logger.Errorf("Error getting block with seqNum = %d due to (%+v)...dropping block", payload.SeqNum, errors.WithStack(err))
continue
}
if rawBlock.Data == nil || rawBlock.Header == nil {
logger.Errorf("Block with claimed sequence %d has no header (%v) or data (%v)",
payload.SeqNum, rawBlock.Header, rawBlock.Data)
continue
}
logger.Debug("New block with claimed sequence number ", payload.SeqNum, " transactions num ", len(rawBlock.Data.Data))
// Read all private data into slice
var p util.PvtDataCollections
if payload.PrivateData != nil {
err := p.Unmarshal(payload.PrivateData)
if err != nil {
logger.Errorf("Wasn't able to unmarshal private data for block seqNum = %d due to (%+v)...dropping block", payload.SeqNum, errors.WithStack(err))
continue
}
}
if err := s.commitBlock(rawBlock, p); err != nil {
if executionErr, isExecutionErr := err.(*vsccErrors.VSCCExecutionFailureError); isExecutionErr {
logger.Errorf("Failed executing VSCC due to %v. Aborting chain processing", executionErr)
return
}
logger.Panicf("Cannot commit block to the ledger due to %+v", errors.WithStack(err))
}
}
case <-s.stopCh:
s.stopCh <- struct{}{}
logger.Debug("State provider has been stopped, finishing to push new blocks.")
return
}
}
}
从payload中获取区块，读取区块和私有数据，提交区块（gossip/state/state.go的commitBlock函数）：
其中提交区块函数commitBlock包括：存储区块（gossip/privdata/coordinator.go的StoreBlock函数），更新ledger高度。
其中，存储区块函数StoreBlock包括：
1）检查区块数据和header不为空；
2）验证区块（core/committer/txvalidator/validator.go的Validate函数）：协程验证各个交易（core/committer/txvalidator/validator.go的validateTx函数），最后统一整理结果，验证交易还是有些复杂的，大致包括：
a）检查交易信息格式是否正确（core/common/validation/msgvalidation.go的ValidateTransaction）；
b）检查channel是否存在；
c）检查交易是否重复；
d）如果是交易，使用vscc检查（core/committer/txvalidator/vscc_validator.go的VSCCValidateTx函数）；如果是config，则apply（core/committer/txvalidator/validator.go的Apply接口）。
最后会设置metadata的BlockMetadataIndex_TRANSACTIONS_FILTER的值。
可以阅读Validate函数的注释信息：
// Validate performs the validation of a block. The validation
// of each transaction in the block is performed in parallel.
// The approach is as follows: the committer thread starts the
// tx validation function in a goroutine (using a semaphore to cap
// the number of concurrent validating goroutines). The committer
// thread then reads results of validation (in orderer of completion
// of the goroutines) from the results channel. The goroutines
// perform the validation of the txs in the block and enqueue the
// validation result in the results channel. A few note-worthy facts:
// 1) to keep the approach simple, the committer thread enqueues
// all transactions in the block and then moves on to reading the
// results.
// 2) for parallel validation to work, it is important that the
// validation function does not change the state of the system.
// Otherwise the order in which validation is perform matters
// and we have to resort to sequential validation (or some locking).
// This is currently true, because the only function that affects
// state is when a config transaction is received, but they are
// guaranteed to be alone in the block. If/when this assumption
// is violated, this code must be changed.
func (v *TxValidator) Validate(block *common.Block) error {
3）计算区块中自己拥有的私有数据（这里自己拥有指payload中含有）的hash（gossip/privdata/coordinator.go的computeOwnedRWsets函数）；
4）检查缺少的私有数据，从自己的transient store中获取（gossip/privdata/coordinator.go的listMissingPrivateData函数）：
a）检查metadata的BlockMetadataIndex_TRANSACTIONS_FILTER和实际交易数目，获取可以拥有私密数据的交易（gossip/privdata/coordinator.go的forEachTxn函数和inspectTransaction函数）；
b）从Transient Store中获取缺少的私有数据（gossip/privdata/coordinator.go的fetchMissingFromTransientStore函数—>fetchFromTransientStore函数）；
c）整理数据；
5）在指定的时间（peer.gossip.pvtData.pullRetryThreshold）内，从其他peer获取缺少的私有数据（gossip/privdata/coordinator.go的fetchFromPeers函数），存入Transient Store中；
6）提交区块和私有数据（core/committer/committer_impl.go的CommitWithPvtData函数），详情见下分析；
7）依据高度，清理私密数据。
CommitWithPvtData函数处理
// CommitWithPvtData commits blocks atomically with private data
func (lc *LedgerCommitter) CommitWithPvtData(blockAndPvtData *ledger.BlockAndPvtData) error {
// Do validation and whatever needed before
// committing new block
if err := lc.preCommit(blockAndPvtData.Block); err != nil {
return err
}
// Committing new block
if err := lc.PeerLedgerSupport.CommitWithPvtData(blockAndPvtData); err != nil {
return err
}
// post commit actions, such as event publishing
lc.postCommit(blockAndPvtData.Block)
return nil
}
以上代码内容包括：
a）如果是config block，调用系统合约cscc升级配置（core/committer/committer_impl.go的preCommit函数）；
b）提交区块和私密数据（core/ledger/kvledger/kv_ledger.go的CommitWithPvtData函数）。
// CommitWithPvtData commits the block and the corresponding pvt data in an atomic operation
func (l *kvLedger) CommitWithPvtData(pvtdataAndBlock *ledger.BlockAndPvtData) error {
var err error
block := pvtdataAndBlock.Block
blockNo := pvtdataAndBlock.Block.Header.Number
logger.Debugf("Channel [%s]: Validating state for block [%d]", l.ledgerID, blockNo)
// 检查和准备工作
err = l.txtmgmt.ValidateAndPrepare(pvtdataAndBlock, true)
if err != nil {
return err
}
logger.Debugf("Channel [%s]: Committing block [%d] to storage", l.ledgerID, blockNo)
l.blockAPIsRWLock.Lock()
defer l.blockAPIsRWLock.Unlock()
// 提交区块和私密数据
if err = l.blockStore.CommitWithPvtData(pvtdataAndBlock); err != nil {
return err
}
logger.Infof("Channel [%s]: Committed block [%d] with %d transaction(s)", l.ledgerID, block.Header.Number, len(block.Data.Data))
if utils.IsConfigBlock(block) {
if err := l.WriteConfigBlockToSpecFile(block); err != nil {
logger.Errorf("Failed to write config block to file for %s", err)
}
}
logger.Debugf("Channel [%s]: Committing block [%d] transactions to state database", l.ledgerID, blockNo)
// 提交stateDB
if err = l.txtmgmt.Commit(); err != nil {
panic(fmt.Errorf(`Error during commit to txmgr:%s`, err))
}
// History database could be written in parallel with state and/or async as a future optimization
if ledgerconfig.IsHistoryDBEnabled() {
logger.Debugf("Channel [%s]: Committing block [%d] transactions to history database", l.ledgerID, blockNo)
// 提交historyDB
if err := l.historyDB.Commit(block); err != nil {
panic(fmt.Errorf(`Error during commit to history db:%s`, err))
}
}
return nil
}
...
// validateAndPreparePvtBatch pulls out the private write-set for the transactions that are marked as valid
// by the internal public data validator. Finally, it validates (if not already self-endorsed) the pvt rwset against the
// corresponding hash present in the public rwset
func validateAndPreparePvtBatch(block *valinternal.Block, pvtdata map[uint64]*ledger.TxPvtData) (*privacyenabledstate.PvtUpdateBatch, error) {
CommitWithPvtData函数主要关注以下内容：
i）检查和准备工作（core/ledger/kvledger/txmgmt/txmgr/lockbasedtxmgr/lockbased_txmgr.go的ValidateAndPrepare函数）：等待pvtdata清理完毕（core/ledger/kvledger/txmgmt/pvtstatepurgemgmt/purge_mgr.go的WaitForPrepareToFinish函数），检查和准备batch（core/ledger/kvledger/txmgmt/validator/valimpl/default_impl.go的ValidateAndPrepareBatch函数），主要关注私密数据的检查，排除没有权限的数据，验证hash正确性；开启state change监听；
ii）写入区块（core/ledger/ledgerstorage/store.go的CommitWithPvtData函数），私密数据存入leveldb，位于/var/hyperledger/production/pvtdataStore路径下，进行清理操作；
iii）提交交易数据（core/ledger/kvledger/txmgmt/txmgr/lockbasedtxmgr/lockbased_txmgr.go的Commit函数）：开启私密数据清理准备，对过期的数据添加清理标记，提交更新数据updates（包括私密数据和hash数据），清理操作；
iv）提交历史数据。
c）发送事件消息（core/committer/committer_impl.go的postCommit函数）。
网友评论