接下来我们来看看Emitter模块，这里初略的看就是gossip的消息发送的模块。下面我们来具体分析下
启动时机
func NewGossipService(conf *Config, s *grpc.Server, sa api.SecurityAdvisor,
mcs api.MessageCryptoService, selfIdentity api.PeerIdentityType,
secureDialOpts api.PeerSecureDialOpts) Gossip {
...
g.emitter = newBatchingEmitter(conf.PropagateIterations,
conf.MaxPropagationBurstSize, conf.MaxPropagationBurstLatency,
g.sendGossipBatch)
...
}
- 首先要知道peer在启动的同时会初始化gossip服务，当然也会初始化emitter模块。
- 这里有几个配置要先讲讲。
- PropagateIterations：一个gossip消息最多可以发几次
- MaxPropagationBurstSize：emitter的buff最多可以存多少个
- MaxPropagationBurstLatency：连续消息push的间隔时间
- sendGossipBatch是回调函数，后面再说
初始化
func newBatchingEmitter(iterations, burstSize int, latency time.Duration, cb emitBatchCallback) batchingEmitter {
if iterations < 0 {
panic(errors.Errorf("Got a negative iterations number"))
}
p := &batchingEmitterImpl{
cb: cb,
delay: latency,
iterations: iterations,
burstSize: burstSize,
lock: &sync.Mutex{},
buff: make([]*batchedMessage, 0),
stopFlag: int32(0),
}
if iterations != 0 {
go p.periodicEmit()
}
return p
}
- batchingEmitterImpl是实现类，初看下，这是一个带buff的，说明它会做一定程度的缓冲，等到某个时机满足再一次push出去，这样对于不是那么急迫的消息，是可以提高效率的，也是常见的一种优化。
periodicEmit
func (p *batchingEmitterImpl) periodicEmit() {
for !p.toDie() {
time.Sleep(p.delay)
p.lock.Lock()
p.emit()
p.lock.Unlock()
}
}
func (p *batchingEmitterImpl) emit() {
if p.toDie() {
return
}
if len(p.buff) == 0 {
return
}
msgs2beEmitted := make([]interface{}, len(p.buff))
for i, v := range p.buff {
msgs2beEmitted[i] = v.data
}
p.cb(msgs2beEmitted)
p.decrementCounters()
}
- 这里很简单，就是每隔一段时间进行一次emit操作，也就是将消息push出去，看来buff的时机是时间而已。
- 具体怎么push，会用cb来做回调。
- decrementCounters这里也很玄妙，这个后面再讲
gossipBatch
func (g *gossipServiceImpl) gossipBatch(msgs []*emittedGossipMessage) {
if g.disc == nil {
g.logger.Error("Discovery has not been initialized yet, aborting!")
return
}
var blocks []*emittedGossipMessage
var stateInfoMsgs []*emittedGossipMessage
var orgMsgs []*emittedGossipMessage
var leadershipMsgs []*emittedGossipMessage
isABlock := func(o interface{}) bool {
return o.(*emittedGossipMessage).IsDataMsg()
}
isAStateInfoMsg := func(o interface{}) bool {
return o.(*emittedGossipMessage).IsStateInfoMsg()
}
aliveMsgsWithNoEndpointAndInOurOrg := func(o interface{}) bool {
msg := o.(*emittedGossipMessage)
if !msg.IsAliveMsg() {
return false
}
member := msg.GetAliveMsg().Membership
return member.Endpoint == "" && g.isInMyorg(discovery.NetworkMember{PKIid: member.PkiId})
}
isOrgRestricted := func(o interface{}) bool {
return aliveMsgsWithNoEndpointAndInOurOrg(o) || o.(*emittedGossipMessage).IsOrgRestricted()
}
isLeadershipMsg := func(o interface{}) bool {
return o.(*emittedGossipMessage).IsLeadershipMsg()
}
// Gossip blocks
blocks, msgs = partitionMessages(isABlock, msgs)
g.gossipInChan(blocks, func(gc channel.GossipChannel) filter.RoutingFilter {
return filter.CombineRoutingFilters(gc.EligibleForChannel, gc.IsMemberInChan, g.isInMyorg)
})
// Gossip Leadership messages
leadershipMsgs, msgs = partitionMessages(isLeadershipMsg, msgs)
g.gossipInChan(leadershipMsgs, func(gc channel.GossipChannel) filter.RoutingFilter {
return filter.CombineRoutingFilters(gc.EligibleForChannel, gc.IsMemberInChan, g.isInMyorg)
})
// Gossip StateInfo messages
stateInfoMsgs, msgs = partitionMessages(isAStateInfoMsg, msgs)
for _, stateInfMsg := range stateInfoMsgs {
peerSelector := g.isInMyorg
gc := g.chanState.lookupChannelForGossipMsg(stateInfMsg.GossipMessage)
if gc != nil && g.hasExternalEndpoint(stateInfMsg.GossipMessage.GetStateInfo().PkiId) {
peerSelector = gc.IsMemberInChan
}
peerSelector = filter.CombineRoutingFilters(peerSelector, func(member discovery.NetworkMember) bool {
return stateInfMsg.filter(member.PKIid)
})
peers2Send := filter.SelectPeers(g.conf.PropagatePeerNum, g.disc.GetMembership(), peerSelector)
g.comm.Send(stateInfMsg.SignedGossipMessage, peers2Send...)
}
// Gossip messages restricted to our org
orgMsgs, msgs = partitionMessages(isOrgRestricted, msgs)
peers2Send := filter.SelectPeers(g.conf.PropagatePeerNum, g.disc.GetMembership(), g.isInMyorg)
for _, msg := range orgMsgs {
g.comm.Send(msg.SignedGossipMessage, g.removeSelfLoop(msg, peers2Send)...)
}
// Finally, gossip the remaining messages
for _, msg := range msgs {
if !msg.IsAliveMsg() {
g.logger.Error("Unknown message type", msg)
continue
}
selectByOriginOrg := g.peersByOriginOrgPolicy(discovery.NetworkMember{PKIid: msg.GetAliveMsg().Membership.PkiId})
selector := filter.CombineRoutingFilters(selectByOriginOrg, func(member discovery.NetworkMember) bool {
return msg.filter(member.PKIid)
})
peers2Send := filter.SelectPeers(g.conf.PropagatePeerNum, g.disc.GetMembership(), selector)
g.sendAndFilterSecrets(msg.SignedGossipMessage, peers2Send...)
}
}
- 这里就是cb回调方法的真身了。看起来很啰嗦，其实就是针对这次批量发送的消息，设计发送的策略。下面具体来看看。
- 将其中的block消息分离出来，gossipInChan出去，这个后面会讲，就是将消息分发出去。
- 挑个StateInfo消息来分析吧，其他都类似，这里就不浪费笔墨了
- 当然了，第一步是过滤出这一批所有的同类消息
- 默认gossip是通知到同组织成员isInMyorg，那如果当前节点配置了CORE_PEER_GOSSIP_EXTERNALENDPOINT，意味你想让同channel不同Org的其他成员知晓。g.hasExternalEndpoint(stateInfMsg.GossipMessage.GetStateInfo().PkiId)就是做这个事情，会选取IsMemberInChan
- 然后就是随机选取N个节点，然后调用comm模块进行发送。
func (g *gossipServiceImpl) gossipInChan(messages []*emittedGossipMessage, chanRoutingFactory channelRoutingFilterFactory) {
if len(messages) == 0 {
return
}
totalChannels := extractChannels(messages)
var channel common.ChainID
var messagesOfChannel []*emittedGossipMessage
for len(totalChannels) > 0 {
// Take first channel
channel, totalChannels = totalChannels[0], totalChannels[1:]
// Extract all messages of that channel
grabMsgs := func(o interface{}) bool {
return bytes.Equal(o.(*emittedGossipMessage).Channel, channel)
}
messagesOfChannel, messages = partitionMessages(grabMsgs, messages)
if len(messagesOfChannel) == 0 {
continue
}
// Grab channel object for that channel
gc := g.chanState.getGossipChannelByChainID(channel)
if gc == nil {
g.logger.Warning("Channel", channel, "wasn't found")
continue
}
// Select the peers to send the messages to
// For leadership messages we will select all peers that pass routing factory - e.g. all peers in channel and org
membership := g.disc.GetMembership()
var peers2Send []*comm.RemotePeer
if messagesOfChannel[0].IsLeadershipMsg() {
peers2Send = filter.SelectPeers(len(membership), membership, chanRoutingFactory(gc))
} else {
peers2Send = filter.SelectPeers(g.conf.PropagatePeerNum, membership, chanRoutingFactory(gc))
}
// Send the messages to the remote peers
for _, msg := range messagesOfChannel {
filteredPeers := g.removeSelfLoop(msg, peers2Send)
g.comm.Send(msg.SignedGossipMessage, filteredPeers...)
}
}
}
- 这里就是将同类消息按channel在分组，如果是leadership消息的话，要特殊处理，一次发给全部成员。
- 如果是其他类型的消息的话，就没有必要，随机选取n个节点发送就好，让他们慢慢发酵。
- 底层就是调用comm模块去send
总结
其实很简单，就是个带buff的Gossip消息的发送模块，封装得很好，基本上只需要往里面add就好，剩下的这里来搞定。
需要注意的是，不同的消息类型逻辑上有些许差别。不过大致都类似。
