說明：
基于Hyperledger Fabric 1.2的源碼；
自己閱讀源碼也是在學(xué)習(xí)和摸索中，有錯(cuò)誤的話歡迎指正，也有不少還不懂的地方，歡迎指導(dǎo)和討論。
主要流程
【接收交易proposal的過程】
- peer在調(diào)用chaincode執(zhí)行交易后，會(huì)依據(jù)collection配置在有權(quán)限的peer間散播私有數(shù)據(jù)，并存儲(chǔ)私有數(shù)據(jù)到自己的transient store。
- 其他接收到散播私有數(shù)據(jù)的peer，會(huì)將私有數(shù)據(jù)存儲(chǔ)到自己的transient store。
【接收Block的過程】
- peer最終接收到block后，會(huì)檢查自己是否有權(quán)限獲取私有數(shù)據(jù)，有權(quán)限的話從transient store中拿數(shù)據(jù)，如果沒有，則去其他peer上獲取，最終私有數(shù)據(jù)會(huì)存儲(chǔ)到pvtdataStore中。
源碼分析
A. 接收交易proposal
在peer節(jié)點(diǎn)處理proposal（core/endorser/endorser.go的ProcessProposal函數(shù)）的過程中，執(zhí)行模擬操作（SimulateProposal函數(shù)）環(huán)節(jié)中，調(diào)用chaincode執(zhí)行交易以后，會(huì)判斷讀寫集結(jié)果中是否包含私有數(shù)據(jù)，如果包含，則執(zhí)行以下操作（代碼如下）：
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")
}
// 匯總私有數(shù)據(jù)collection配置和私有數(shù)據(jù)信息
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中，依據(jù)collection配置，散播私有數(shù)據(jù)
if err := e.distributePrivateData(chainID, txid, pvtDataWithConfig, endorsedAt); err != nil {
return nil, nil, nil, nil, err
}
}
以上代碼內(nèi)容包括：
- 匯總私有數(shù)據(jù)collection配置和私有數(shù)據(jù)信息（core/endorser/pvtrwset_assembler.go的AssemblePvtRWSet函數(shù)）：
根據(jù)讀寫集中私有數(shù)據(jù)的合約名（namespace），從系統(tǒng)合約lscc處獲取其對應(yīng)的collection配置（即實(shí)例化合約時(shí)提供的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之間分發(fā)私有數(shù)據(jù)：
調(diào)用distributePrivateData函數(shù)，該函數(shù)在生成EndorserServer時(shí)由外部傳值建立（peer/node/start.go的serve函數(shù)），即privDataDist函數(shù)（代碼如下）。
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的唯一實(shí)例（gossip/service/gossip_service.go的gossipServiceInstance），調(diào)用函數(shù)（gossip/service/gossip_service.go的DistributePrivateData函數(shù)）完成在channel內(nèi)部的私有數(shù)據(jù)傳輸（代碼如下）：
// 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()
// 根據(jù)channel名稱獲取私有數(shù)據(jù)的handler
handler, exists := g.privateHandlers[chainID]
g.lock.RUnlock()
if !exists {
return errors.Errorf("No private data handler for %s", chainID)
}
// 依據(jù)collection配置散播私有數(shù)據(jù)
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
}
以上函數(shù)包括如下步驟：
1）根據(jù)channel名稱chainID獲取私有數(shù)據(jù)的handler；
2）依據(jù)collection配置散播私有數(shù)據(jù)（gossip/privdata/distributor.go的Distribute函數(shù)，代碼如下）：
// 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函數(shù)包括以下步驟：
a）制定散播方案（gossip/privdata/distributor.go的computeDisseminationPlan函數(shù)）：根據(jù)collection配置獲取策略，設(shè)置過濾器，生成私有數(shù)據(jù)消息，生成散播方案（dissemination的列表，dissemination包含msg和criteria）；
b）根據(jù)散播方案，開啟協(xié)程啟動(dòng)散播（散播最終調(diào)用的函數(shù)需要收到接收端的ACK）；
c）存儲(chǔ)私有數(shù)據(jù)到Transient Store（core/transientstore/store.go的PersistWithConfig函數(shù)）：創(chuàng)建comisiteKey為鍵值（包括txid, uuid和blockHeight），存儲(chǔ)私有數(shù)據(jù)，并創(chuàng)建兩個(gè)用于后期清理的索引（compositeKeyPurgeIndexByHeight和compositeKeyPurgeIndexByTxid）,數(shù)據(jù)最后存儲(chǔ)在leveldb中，位于/var/hyperledger/production/transientStore路徑下。
B. 接收Block和私有數(shù)據(jù)
在peer啟動(dòng)時(shí)會(huì)啟動(dòng)Gossip服務(wù)，其中會(huì)啟動(dòng)接收消息的協(xié)程
（調(diào)用過程：peer/node/start.go的serve函數(shù)—>core/peer/peer.go的Initialize函數(shù)—>core/peer/peer.go的createChain函數(shù)—>gossip/service/gossip_service.go的InitializeChannel函數(shù)—>gossip/state/state.go的NewGossipStateProvider函數(shù)）
主要關(guān)注以下代碼：
// Listen for incoming communication
go s.listen()
// Deliver in order messages into the incoming channel
go s.deliverPayloads()
1. listen函數(shù)處理
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類型，即區(qū)塊）
協(xié)程調(diào)用queueNewMessage函數(shù)（gossip/state/state.go的queueNewMessage函數(shù)），將同一channel的信息放到payload中（gossip/state/state.go的addPayload函數(shù)）（代碼如下）。
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類型，即遠(yuǎn)程peer的請求或響應(yīng)信息）
該通道的消息是經(jīng)過remoteStateMsgFilter篩選的（主要檢查消息內(nèi)容，channel權(quán)限），協(xié)程調(diào)用dispatch函數(shù)（gossip/state/state.go的dispatch函數(shù)），分別處理state message和私有數(shù)據(jù)信息。這里主要關(guān)注私有數(shù)據(jù)，如果獲得私有數(shù)據(jù)，則進(jìn)行處理（gossip/state/state.go的privateDataMessage函數(shù)）：
a）檢查channel，需要一致；
b）獲取私有數(shù)據(jù)信息和collection配置信息；
c）寫入Transient Store（最終還是調(diào)用core/transientstore/store.go的PersistWithConfig函數(shù)）；
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函數(shù)處理
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中獲取區(qū)塊，讀取區(qū)塊和私有數(shù)據(jù)，提交區(qū)塊（gossip/state/state.go的commitBlock函數(shù)）：
其中提交區(qū)塊函數(shù)commitBlock包括：存儲(chǔ)區(qū)塊（gossip/privdata/coordinator.go的StoreBlock函數(shù)），更新ledger高度。
其中，存儲(chǔ)區(qū)塊函數(shù)StoreBlock包括：
1）檢查區(qū)塊數(shù)據(jù)和header不為空；
2）驗(yàn)證區(qū)塊（core/committer/txvalidator/validator.go的Validate函數(shù)）：協(xié)程驗(yàn)證各個(gè)交易（core/committer/txvalidator/validator.go的validateTx函數(shù)），最后統(tǒng)一整理結(jié)果，驗(yàn)證交易還是有些復(fù)雜的，大致包括：
??a）檢查交易信息格式是否正確（core/common/validation/msgvalidation.go的ValidateTransaction）；
??b）檢查channel是否存在；
??c）檢查交易是否重復(fù)；
??d）如果是交易，使用vscc檢查（core/committer/txvalidator/vscc_validator.go的VSCCValidateTx函數(shù)）；如果是config，則apply（core/committer/txvalidator/validator.go的Apply接口）。
最后會(huì)設(shè)置metadata的BlockMetadataIndex_TRANSACTIONS_FILTER的值。
可以閱讀Validate函數(shù)的注釋信息：
// 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）計(jì)算區(qū)塊中自己擁有的私有數(shù)據(jù)（這里自己擁有指payload中含有）的hash（gossip/privdata/coordinator.go的computeOwnedRWsets函數(shù)）；
4）檢查缺少的私有數(shù)據(jù)，從自己的transient store中獲?。╣ossip/privdata/coordinator.go的listMissingPrivateData函數(shù)）：
??a）檢查metadata的BlockMetadataIndex_TRANSACTIONS_FILTER和實(shí)際交易數(shù)目，獲取可以擁有私密數(shù)據(jù)的交易（gossip/privdata/coordinator.go的forEachTxn函數(shù)和inspectTransaction函數(shù)）；
??b）從Transient Store中獲取缺少的私有數(shù)據(jù)（gossip/privdata/coordinator.go的fetchMissingFromTransientStore函數(shù)—>fetchFromTransientStore函數(shù)）；
??c）整理數(shù)據(jù)；
5）在指定的時(shí)間（peer.gossip.pvtData.pullRetryThreshold）內(nèi)，從其他peer獲取缺少的私有數(shù)據(jù)（gossip/privdata/coordinator.go的fetchFromPeers函數(shù)），存入Transient Store中；
6）提交區(qū)塊和私有數(shù)據(jù)（core/committer/committer_impl.go的CommitWithPvtData函數(shù)），詳情見下分析；
7）依據(jù)高度，清理私密數(shù)據(jù)。
CommitWithPvtData函數(shù)處理
// 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
}
以上代碼內(nèi)容包括：
??a）如果是config block，調(diào)用系統(tǒng)合約cscc升級(jí)配置（core/committer/committer_impl.go的preCommit函數(shù)）；
??b）提交區(qū)塊和私密數(shù)據(jù)（core/ledger/kvledger/kv_ledger.go的CommitWithPvtData函數(shù)）。
// 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)
// 檢查和準(zhǔn)備工作
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()
// 提交區(qū)塊和私密數(shù)據(jù)
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函數(shù)主要關(guān)注以下內(nèi)容：
????i）檢查和準(zhǔn)備工作（core/ledger/kvledger/txmgmt/txmgr/lockbasedtxmgr/lockbased_txmgr.go的ValidateAndPrepare函數(shù)）：等待pvtdata清理完畢（core/ledger/kvledger/txmgmt/pvtstatepurgemgmt/purge_mgr.go的WaitForPrepareToFinish函數(shù)），檢查和準(zhǔn)備batch（core/ledger/kvledger/txmgmt/validator/valimpl/default_impl.go的ValidateAndPrepareBatch函數(shù)），主要關(guān)注私密數(shù)據(jù)的檢查，排除沒有權(quán)限的數(shù)據(jù)，驗(yàn)證hash正確性；開啟state change監(jiān)聽；
????ii）寫入?yún)^(qū)塊（core/ledger/ledgerstorage/store.go的CommitWithPvtData函數(shù)），私密數(shù)據(jù)存入leveldb，位于/var/hyperledger/production/pvtdataStore路徑下，進(jìn)行清理操作；
????iii）提交交易數(shù)據(jù)（core/ledger/kvledger/txmgmt/txmgr/lockbasedtxmgr/lockbased_txmgr.go的Commit函數(shù)）：開啟私密數(shù)據(jù)清理準(zhǔn)備，對過期的數(shù)據(jù)添加清理標(biāo)記，提交更新數(shù)據(jù)updates（包括私密數(shù)據(jù)和hash數(shù)據(jù)），清理操作；
????iv）提交歷史數(shù)據(jù)。
??c）發(fā)送事件消息（core/committer/committer_impl.go的postCommit函數(shù)）。
