Hyperledger-Fabric源码分析（orderer-c

从今天开始分析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

1. 首先拿到Orderer的batchSize配置
2. 计算进来的msg的大小足够大，且超过了PreferredMaxBytes

• 将已有的pending的msg都拿出来出包，然后加上新进的msg单独成包，一起返回

3. 如果进来的消息大小合适，但跟现有pending的消息加总超过了PreferredMaxBytes

• 将已有的pending的msg都拿出来出包, 准备返回
• 将新进的msg加到pending中

4. 如果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到此为止。