从今天开始分析orderer的consensus机制，1.0的时候有solo和kafka两种，都比较简单，1.4里面有etcd，之前看过是raft的实现，正好温习下。这个系列从solo开始吧。
启动
当然了，Orderer启动的时候根据配置来决定用什么共识实现。solo最终会在这里会调起。
func (ch *chain) Start() { go ch.main() }
接受事件
func (ch *chain) Order(env *cb.Envelope, configSeq uint64) error {
select {
case ch.sendChan <- &message{
configSeq: configSeq,
normalMsg: env,
}:
...
}
这里接受来自四面八方的消息，都在这里排队，等待出包(也就是打包成block)。最终消息都是通知到chain的sendChan通道。
消息处理
if msg.configMsg == nil {
// NormalMsg
if msg.configSeq < seq {
_, err = ch.support.ProcessNormalMsg(msg.normalMsg)
if err != nil {
logger.Warningf("Discarding bad normal message: %s", err)
continue
}
}
batches, pending := ch.support.BlockCutter().Ordered(msg.normalMsg)
for _, batch := range batches {
block := ch.support.CreateNextBlock(batch)
ch.support.WriteBlock(block, nil)
}
...
}
这里消息分两种，一种是NormalMsg，一种是ConfigMsg。
msg.configSeq < seq这个很关键，是比对最新的configblocknum，一致就说明从设置configSeq开始到这里为止，起码没有config的变动，因为前面设置configSeq的时候已经做过有效性校验了，所以这里就不用再做了。但是如果msg.configSeq < seq真的成立，说明中间有configblock写入了账本，导致整个环境变化，前面的过滤不能保证有效，所以这里再做一次。
func CreateStandardChannelFilters(filterSupport channelconfig.Resources) *RuleSet {
ordererConfig, ok := filterSupport.OrdererConfig()
if !ok {
logger.Panicf("Missing orderer config")
}
return NewRuleSet([]Rule{
EmptyRejectRule,
NewExpirationRejectRule(filterSupport),
NewSizeFilter(ordererConfig),
NewSigFilter(policies.ChannelWriters, filterSupport),
})
}
这是里面过滤器组的定义，有兴趣的可以去看看，这里略过。
出包
batches, pending := ch.support.BlockCutter().Ordered(msg.normalMsg)
batches可以看成是blocks，而pending用来表示里面还有没有等待处理的消息。
这里是消息出包的关键，下面我们进去看看
func (r *receiver) Ordered(msg *cb.Envelope) (messageBatches [][]*cb.Envelope, pending bool) {
ordererConfig, ok := r.sharedConfigFetcher.OrdererConfig()
batchSize := ordererConfig.BatchSize()
messageSizeBytes := messageSizeBytes(msg)
if messageSizeBytes > batchSize.PreferredMaxBytes {
// cut pending batch, if it has any messages
if len(r.pendingBatch) > 0 {
messageBatch := r.Cut()
messageBatches = append(messageBatches, messageBatch)
}
// create new batch with single message
messageBatches = append(messageBatches, []*cb.Envelope{msg})
// Record that this batch took no time to fill
r.Metrics.BlockFillDuration.With("channel", r.ChannelID).Observe(0)
return
}
messageWillOverflowBatchSizeBytes := r.pendingBatchSizeBytes+messageSizeBytes > batchSize.PreferredMaxBytes
if messageWillOverflowBatchSizeBytes {
messageBatch := r.Cut()
r.PendingBatchStartTime = time.Now()
messageBatches = append(messageBatches, messageBatch)
}
r.pendingBatch = append(r.pendingBatch, msg)
r.pendingBatchSizeBytes += messageSizeBytes
pending = true
if uint32(len(r.pendingBatch)) >= batchSize.MaxMessageCount {
logger.Debugf("Batch size met, cutting batch")
messageBatch := r.Cut()
messageBatches = append(messageBatches, messageBatch)
pending = false
}
return
}
Orderer: &OrdererDefaults
# Orderer Type: The orderer implementation to start
# Available types are "solo" and "kafka"
OrdererType: solo
Addresses:
- orderer.example.com:7050
# Batch Timeout: The amount of time to wait before creating a batch
BatchTimeout: 2s
# Batch Size: Controls the number of messages batched into a block
BatchSize:
# Max Message Count: The maximum number of messages to permit in a batch
MaxMessageCount: 10
# Absolute Max Bytes: The absolute maximum number of bytes allowed for
# the serialized messages in a batch.
AbsoluteMaxBytes: 99 MB
# Preferred Max Bytes: The preferred maximum number of bytes allowed for
# the serialized messages in a batch. A message larger than the preferred
# max bytes will result in a batch larger than preferred max bytes.
PreferredMaxBytes: 512 KB
- 首先拿到Orderer的batchSize配置
- 计算进来的msg的大小足够大，且超过了PreferredMaxBytes
- 将已有的pending的msg都拿出来出包，然后加上新进的msg单独成包，一起返回
- 如果进来的消息大小合适，但跟现有pending的消息加总超过了PreferredMaxBytes
- 将已有的pending的msg都拿出来出包, 准备返回
- 将新进的msg加到pending中
- 如果pending的消息数超过了MaxMessageCount
- 将已有的pending的msg都拿出来，返回
for _, batch := range batches {
block := ch.support.CreateNextBlock(batch)
ch.support.WriteBlock(block, nil)
}
前面已经说了，batch就是block。代码很简单，就是遍历batch，组装block写到本地账本中。这里专指orderer账本，至于orderer账本里面的block怎么扩散给peer，那会用到orderer的deliver服务，到时我会单独再讲，这里就不扩散了。
switch {
case timer != nil && !pending:
// Timer is already running but there are no messages pending, stop the timer
timer = nil
case timer == nil && pending:
// Timer is not already running and there are messages pending, so start it
timer = time.After(ch.support.SharedConfig().BatchTimeout())
logger.Debugf("Just began %s batch timer", ch.support.SharedConfig().BatchTimeout().String())
default:
// Do nothing when:
// 1. Timer is already running and there are messages pending
// 2. Timer is not set and there are no messages pending
}
最后Cut触发的时机当然少不了timer，不然pending消息有可能永远都不能出包。这里很简单，就是如果有消息pending，就开始计时，过了batchtimeout后，触发下面的逻辑
case <-timer:
//clear the timer
timer = nil
batch := ch.support.BlockCutter().Cut()
if len(batch) == 0 {
logger.Warningf("Batch timer expired with no pending requests, this might indicate a bug")
continue
}
logger.Debugf("Batch timer expired, creating block")
block := ch.support.CreateNextBlock(batch)
ch.support.WriteBlock(block, nil)
很熟悉对不对。整个SOLO到此为止。
