网站首页 返回列表像“草根”一样，紧贴着地面，低调的存在，冬去春来，枯荣无恙。

Hyperledger Fabric私有数据（Private Data Collections）源码分析

2020-06-10 11:28:46 admin 598

##### 说明：

基于Hyperledger Fabric 1.2的源码；  
自己阅读源码也是在学习和摸索中，有错误的话欢迎指正，也有不少还不懂的地方，欢迎指导和讨论。

# 主要流程

【接收交易proposal的过程】

1. peer在调用chaincode执行交易后，会依据collection配置在有权限的peer间散播私有数据，并存储私有数据到自己的transient store。
  2. 其他接收到散播私有数据的peer，会将私有数据存储到自己的transient store。

【接收Block的过程】

3. 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
       }
    }

以上代码内容包括：

1. 汇总私有数据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
    }

2. 在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函数）。

转载文章，原文链接： Hyperledger Fabric私有数据（Private Data Collections）源码分析

关键字词源码 Hyperledger

分享到：

上一篇：Hyperledger Fabric架构详解

下一篇：kubelet 中垃圾回收机制的设计与实现

如需留言，请登录，没有账号？请注册

0 条评论 0 人参与