StateInfo是用来传播peer的状态信息给其他成员。
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的标识id，内部其实就是mspid+cert算出来的一个摘要。
- Channel_MAC，pkiid+chainid生成MAC，背后就是sha256计算hash
- 时间戳
- 属性代表着三个触发stateinfo消息的地方
- 该节点退出通道
- 有新的block写入，更新peer的账本height
- chaincode更新
- 这个看起来不太好理解。我跟进了下，chaincode部署成功会触发这里。换句话说如果cc部署成功，是通过这个消息让成员知道的。
触发点
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)
}
- 因为只是更新height，所以其他的状态沿用之前的stateInfoMsg
- 下面开始要广播前的最后准备工作了
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用来保存收到的成员发来的所有的stateinfo消息，包括自己的。
- 更新自己的height
- 每给其他成员分享一次stateinfo的时候，都会自己保留一份，以备不时之需。这种情况正好用上。
- 启动shouldGossipStateInfo开关
发送点
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你暂时不用关心，里面根据不同的消息类型来决定点对点发送还是群发，不过在这里你只用知道发出去就好了。有时间我会专门讲这个模块。
- 那么发送是怎么触发的呢？
时机
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的时候，会定时来监听是否有新的stateinfo消息需要发布。
- GossipChannel看名字你也大概能猜到干嘛的，这是专门给同channel的成员进行gossip服务的。
接受点
中间省略了很多地方，这个消息的专题都是这种风格，尽量不要被其他的细节给干扰。总之，消息已经被peer收到，下面我们看下收到后，怎么处理。
GossipService
GossipService是统管gossip服务的，所有的动作都由这里发起，消息处理也不例外
消息验证
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)
}
- 这里主要做两个事情
- 一，判断当前消息的Pkiid是否认识，这是消息接收的基础。因为Gossip有机制能同步成员列表，如果有不认识节点，那么就问题大了。
- 二，对方消息是私钥进行数字签名的，这里用本地的公钥进行签名校验。这也是安全的基础。
消息处理
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
}
...
}
- 前面校验部分过了后，基本上消息没有大的问题
- 下面开始正式处理了，这里需要一些背景知识。首先一个Peer加入一个channel, 就会有一个GossipChannel相伴。所以如果查不到这个gc，那么说明Peer不在这个channel里面。
- 这里是描述了两个场景
- 首先该Peer不在这个channel里面，但属于同一个Org，也就是组织，那么处于优化的目的，可以尽快将消息扩散出去，以尽快让同channel的节点处理。再次遇到emitter，忽略它。
- 如果同属一个channel，那么开始交给所属的GossipChannel来处理。
GossipChannel
消息校验
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
}
...
}
通用校验，我们这里就不费功夫了，主要看stateinfo的。
可以看到这里主要做MAC校验，这个校验感觉没什么用，并不能保证其完整性。
消息处理
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())
}
}
当然了这里会直接加到gc的stateInfoMsgStore里面存起来。当然了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主要是校验消息发起人是否同属一个Channel，还有就是这个节点是否有读取成员状态的权力。
- 按pkiID的维度冗余一遍
如果Add成功，第一件事就是让其他人知道。通过emitter转发出去
DeMultiplex是本地的一个多路分发的模块，里面可以增加订阅器，来订阅感兴趣的消息类型，进而处理。不过幸运的是，stateinfo没有人订阅，所以这里不扩散了。
使用
前面讲了这么多StateInfo消息，那么问题来了，这消息收下来到底干嘛用呢？举两个例子就清楚了。当然里面的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里面找到，说明这个member同属一个通道。
// 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
}
- 这里也是一样，上面判定同属一个通道，下面判定是否同属一个通道下的同一个组织。
总结
- StateInfo主要给别人分享是否有新的cc部署完成，是否退出通道，是否有新的block写入。
- 不管如何，第一时间转发消息给成员，利于消息扩散。
- 然后自循环，对外发布状态更新。
