Hyperledger-Fabric源碼分析（Gossip-StateInfo）

StateInfo是用來傳播peer的狀態(tài)信息給其他成員。

struct

type StateInfo struct {
   Timestamp *PeerTime 
   PkiId     []byte   
   // channel_MAC is an authentication code that proves
   // that the peer that sent this message knows
   // the name of the channel.
   Channel_MAC          []byte      
   Properties           *Properties 
}

type Properties struct {
    LedgerHeight         uint64       
    LeftChannel          bool         
    Chaincodes           []*Chaincode 
}

初始化

stateInfMsg := &proto.StateInfo{
        Channel_MAC: GenerateMAC(gc.pkiID, gc.chainID),
        PkiId:       gc.pkiID,
        Timestamp: &proto.PeerTime{
            IncNum: gc.incTime,
            SeqNum: uint64(time.Now().UnixNano()),
        },
        Properties: &proto.Properties{
            LeftChannel:  leftChannel,
            LedgerHeight: ledgerHeight,
            Chaincodes:   chaincodes,
        },
    }
    m := &proto.GossipMessage{
        Nonce: 0,
        Tag:   proto.GossipMessage_CHAN_OR_ORG,
        Content: &proto.GossipMessage_StateInfo{
            StateInfo: stateInfMsg,
        },
    }
  • 可以看到stateinfo的組成
    • pkiid，你可以理解為peer的標(biāo)識id，內(nèi)部其實(shí)就是mspid+cert算出來的一個(gè)摘要。
    • Channel_MAC，pkiid+chainid生成MAC，背后就是sha256計(jì)算hash
    • 時(shí)間戳
    • 屬性代表著三個(gè)觸發(fā)stateinfo消息的地方
      • 該節(jié)點(diǎn)退出通道
      • 有新的block寫入，更新peer的賬本height
      • chaincode更新
        • 這個(gè)看起來不太好理解。我跟進(jìn)了下，chaincode部署成功會觸發(fā)這里。換句話說如果cc部署成功，是通過這個(gè)消息讓成員知道的。

觸發(fā)點(diǎn)

commitblock

func (s *GossipStateProviderImpl) commitBlock(block *common.Block, pvtData util.PvtDataCollections) error {

    // Commit block with available private transactions
    if err := s.ledger.StoreBlock(block, pvtData); err != nil {
        logger.Errorf("Got error while committing(%+v)", errors.WithStack(err))
        return err
    }

    // Update ledger height
    s.mediator.UpdateLedgerHeight(block.Header.Number+1, common2.ChainID(s.chainID))
    logger.Debugf("[%s] Committed block [%d] with %d transaction(s)",
        s.chainID, block.Header.Number, len(block.Data.Data))

    return nil
}
  • 前面就是提交block到賬本了
  • 后來開始UpdateLedgerHeight，開始處理賬本新的height

UpdateLedgerHeight

func (gc *gossipChannel) UpdateLedgerHeight(height uint64) {
    gc.Lock()
    defer gc.Unlock()

    var chaincodes []*proto.Chaincode
    var leftChannel bool
    if prevMsg := gc.stateInfoMsg; prevMsg != nil {
        leftChannel = prevMsg.GetStateInfo().Properties.LeftChannel
        chaincodes = prevMsg.GetStateInfo().Properties.Chaincodes
    }
    gc.updateProperties(height, chaincodes, leftChannel)
}
  • 因?yàn)橹皇歉耯eight，所以其他的狀態(tài)沿用之前的stateInfoMsg
  • 下面開始要廣播前的最后準(zhǔn)備工作了

updateStateInfo

func (gc *gossipChannel) updateStateInfo(msg *proto.SignedGossipMessage) {
   gc.stateInfoMsgStore.Add(msg)
   gc.ledgerHeight = msg.GetStateInfo().Properties.LedgerHeight
   gc.stateInfoMsg = msg
   atomic.StoreInt32(&gc.shouldGossipStateInfo, int32(1))
}
  • stateInfoMsgStore用來保存收到的成員發(fā)來的所有的stateinfo消息，包括自己的。
  • 更新自己的height
  • 每給其他成員分享一次stateinfo的時(shí)候，都會自己保留一份，以備不時(shí)之需。這種情況正好用上。
  • 啟動shouldGossipStateInfo開關(guān)

發(fā)送點(diǎn)

func (gc *gossipChannel) publishStateInfo() {
   if atomic.LoadInt32(&gc.shouldGossipStateInfo) == int32(0) {
      return
   }
   gc.RLock()
   stateInfoMsg := gc.stateInfoMsg
   gc.RUnlock()
   gc.Gossip(stateInfoMsg)
   if len(gc.GetMembership()) > 0 {
      atomic.StoreInt32(&gc.shouldGossipStateInfo, int32(0))
   }
}
  • 可以看到最后會在這里將消息Gossip出去，里面是用emitter模塊去處理。emitter你暫時(shí)不用關(guān)心，里面根據(jù)不同的消息類型來決定點(diǎn)對點(diǎn)發(fā)送還是群發(fā)，不過在這里你只用知道發(fā)出去就好了。有時(shí)間我會專門講這個(gè)模塊。
  • 那么發(fā)送是怎么觸發(fā)的呢？

時(shí)機(jī)

func NewGossipChannel(pkiID common.PKIidType, org api.OrgIdentityType, mcs api.MessageCryptoService,
   chainID common.ChainID, adapter Adapter, joinMsg api.JoinChannelMessage) GossipChannel {
   gc := &gossipChannel{
      incTime:                   uint64(time.Now().UnixNano()),
      selfOrg:                   org,
      pkiID:                     pkiID,
      mcs:                       mcs,
      Adapter:                   adapter,
      logger:                    util.GetLogger(util.ChannelLogger, adapter.GetConf().ID),
      stopChan:                  make(chan struct{}, 1),
      shouldGossipStateInfo:     int32(0),
      stateInfoPublishScheduler: time.NewTicker(adapter.GetConf().PublishStateInfoInterval),
      stateInfoRequestScheduler: time.NewTicker(adapter.GetConf().RequestStateInfoInterval),
      orgs:                      []api.OrgIdentityType{},
      chainID:                   chainID,
   }

   ...

   // Periodically publish state info
   go gc.periodicalInvocation(gc.publishStateInfo, gc.stateInfoPublishScheduler.C)
   ...
}
  • 在初始化GossipChannel的時(shí)候，會定時(shí)來監(jiān)聽是否有新的stateinfo消息需要發(fā)布。
  • GossipChannel看名字你也大概能猜到干嘛的，這是專門給同channel的成員進(jìn)行g(shù)ossip服務(wù)的。

接受點(diǎn)

中間省略了很多地方，這個(gè)消息的專題都是這種風(fēng)格，盡量不要被其他的細(xì)節(jié)給干擾?？傊?，消息已經(jīng)被peer收到，下面我們看下收到后，怎么處理。

GossipService

GossipService是統(tǒng)管gossip服務(wù)的，所有的動作都由這里發(fā)起，消息處理也不例外

消息驗(yàn)證

func (g *gossipServiceImpl) validateStateInfoMsg(msg *proto.SignedGossipMessage) error {
    verifier := func(identity []byte, signature, message []byte) error {
        pkiID := g.idMapper.GetPKIidOfCert(api.PeerIdentityType(identity))
        if pkiID == nil {
            return errors.New("PKI-ID not found in identity mapper")
        }
        return g.idMapper.Verify(pkiID, signature, message)
    }
    identity, err := g.idMapper.Get(msg.GetStateInfo().PkiId)
    if err != nil {
        return errors.WithStack(err)
    }
    return msg.Verify(identity, verifier)
}
  • 這里主要做兩個(gè)事情
  • 一，判斷當(dāng)前消息的Pkiid是否認(rèn)識，這是消息接收的基礎(chǔ)。因?yàn)镚ossip有機(jī)制能同步成員列表，如果有不認(rèn)識節(jié)點(diǎn)，那么就問題大了。
  • 二，對方消息是私鑰進(jìn)行數(shù)字簽名的，這里用本地的公鑰進(jìn)行簽名校驗(yàn)。這也是安全的基礎(chǔ)。

消息處理

func (g *gossipServiceImpl) handleMessage(m proto.ReceivedMessage) {
    ...

    if msg.IsChannelRestricted() {
        if gc := g.chanState.lookupChannelForMsg(m); gc == nil {
            // If we're not in the channel, we should still forward to peers of our org
            // in case it's a StateInfo message
            if g.isInMyorg(discovery.NetworkMember{PKIid: m.GetConnectionInfo().ID}) && msg.IsStateInfoMsg() {
                if g.stateInfoMsgStore.Add(msg) {
                    g.emitter.Add(&emittedGossipMessage{
                        SignedGossipMessage: msg,
                        filter:              m.GetConnectionInfo().ID.IsNotSameFilter,
                    })
                }
            }
            if !g.toDie() {
                g.logger.Debug("No such channel", msg.Channel, "discarding message", msg)
            }
        } else {
            ...
            gc.HandleMessage(m)
        }
        return
    }

    ...
}
  • 前面校驗(yàn)部分過了后，基本上消息沒有大的問題
  • 下面開始正式處理了，這里需要一些背景知識。首先一個(gè)Peer加入一個(gè)channel, 就會有一個(gè)GossipChannel相伴。所以如果查不到這個(gè)gc，那么說明Peer不在這個(gè)channel里面。
  • 這里是描述了兩個(gè)場景
    • 首先該P(yáng)eer不在這個(gè)channel里面，但屬于同一個(gè)Org，也就是組織，那么處于優(yōu)化的目的，可以盡快將消息擴(kuò)散出去，以盡快讓同channel的節(jié)點(diǎn)處理。再次遇到emitter，忽略它。
    • 如果同屬一個(gè)channel，那么開始交給所屬的GossipChannel來處理。

GossipChannel

消息校驗(yàn)

func (gc *gossipChannel) verifyMsg(msg proto.ReceivedMessage) bool {
    ...
    if m.IsStateInfoMsg() {
        si := m.GetStateInfo()
        expectedMAC := GenerateMAC(si.PkiId, gc.chainID)
        if !bytes.Equal(expectedMAC, si.Channel_MAC) {
            gc.logger.Warning("Message contains wrong channel MAC(", si.Channel_MAC, "), expected", expectedMAC)
            return false
        }
        return true
    }
    ...
}

  • 通用校驗(yàn)，我們這里就不費(fèi)功夫了，主要看stateinfo的。

  • 可以看到這里主要做MAC校驗(yàn)，這個(gè)校驗(yàn)感覺沒什么用，并不能保證其完整性。

消息處理

if m.IsDataMsg() || m.IsStateInfoMsg() {
   added := false

   if m.IsDataMsg() {
      ...
   } else { // StateInfoMsg verification should be handled in a layer above
      //  since we don't have access to the id mapper here
      added = gc.stateInfoMsgStore.Add(msg.GetGossipMessage())
   }

   if added {
      // Forward the message
      gc.Forward(msg)
      // DeMultiplex to local subscribers
      gc.DeMultiplex(m)

      if m.IsDataMsg() {
         gc.blocksPuller.Add(msg.GetGossipMessage())
      }
   }

  • 當(dāng)然了這里會直接加到gc的stateInfoMsgStore里面存起來。當(dāng)然了Add也不是那么簡單。簡單看看。

    func (cache *stateInfoCache) Add(msg *proto.SignedGossipMessage) bool {
   if !cache.MessageStore.CheckValid(msg) {
      return false
   }
   if !cache.verify(msg) {
      return false
   }
   added := cache.MessageStore.Add(msg)
   if added {
      pkiID := msg.GetStateInfo().PkiId
      cache.MembershipStore.Put(pkiID, msg)
   }
   return added
}
    • CheckValid會將msg跟本地保存做比較，如果是全新的或比已有的新，會判定有效
    • verify主要是校驗(yàn)消息發(fā)起人是否同屬一個(gè)Channel，還有就是這個(gè)節(jié)點(diǎn)是否有讀取成員狀態(tài)的權(quán)力。
    • 按pkiID的維度冗余一遍

  • 如果Add成功，第一件事就是讓其他人知道。通過emitter轉(zhuǎn)發(fā)出去

  • DeMultiplex是本地的一個(gè)多路分發(fā)的模塊，里面可以增加訂閱器，來訂閱感興趣的消息類型，進(jìn)而處理。不過幸運(yùn)的是，stateinfo沒有人訂閱，所以這里不擴(kuò)散了。

使用

前面講了這么多StateInfo消息，那么問題來了，這消息收下來到底干嘛用呢？舉兩個(gè)例子就清楚了。當(dāng)然里面的Properties還有別的用處，這里先不展開。

func (gc *gossipChannel) EligibleForChannel(member discovery.NetworkMember) bool {
   peerIdentity := gc.GetIdentityByPKIID(member.PKIid)
   if len(peerIdentity) == 0 {
      gc.logger.Warning("Identity for peer", member.PKIid, "doesn't exist")
      return false
   }
   msg := gc.stateInfoMsgStore.MsgByID(member.PKIid)
   if msg == nil {
      return false
   }
   return true
}
  • 這里如果能從stateInfoMsgStore里面找到，說明這個(gè)member同屬一個(gè)通道。
// If we don't have a StateInfo message from the peer,
// no way of validating its eligibility in the channel.
if gc.stateInfoMsgStore.MsgByID(msg.GetConnectionInfo().ID) == nil {
   gc.logger.Debug("Don't have StateInfo message of peer", msg.GetConnectionInfo())
   return
}
if !gc.eligibleForChannelAndSameOrg(discovery.NetworkMember{PKIid: msg.GetConnectionInfo().ID}) {
   gc.logger.Warning(msg.GetConnectionInfo(), "isn't eligible for pulling blocks of", string(gc.chainID))
   return
}
  • 這里也是一樣，上面判定同屬一個(gè)通道，下面判定是否同屬一個(gè)通道下的同一個(gè)組織。

總結(jié)

  • StateInfo主要給別人分享是否有新的cc部署完成，是否退出通道，是否有新的block寫入。
  • 不管如何，第一時(shí)間轉(zhuǎn)發(fā)消息給成員，利于消息擴(kuò)散。
  • 然后自循環(huán)，對外發(fā)布狀態(tài)更新。
最后編輯于 ：
?著作權(quán)歸作者所有,轉(zhuǎn)載或內(nèi)容合作請聯(lián)系作者
【社區(qū)內(nèi)容提示】社區(qū)部分內(nèi)容疑似由AI輔助生成，瀏覽時(shí)請結(jié)合常識與多方信息審慎甄別。
平臺聲明：文章內(nèi)容（如有圖片或視頻亦包括在內(nèi)）由作者上傳并發(fā)布，文章內(nèi)容僅代表作者本人觀點(diǎn)，簡書系信息發(fā)布平臺，僅提供信息存儲服務(wù)。

相關(guān)閱讀更多精彩內(nèi)容

友情鏈接更多精彩內(nèi)容