研究Fabric中Etcd的Raft应用

// Chain defines a way to inject messages for ordering.
type Chain interface {
    // 负责对普通交易消息进行处理排序。当排序服务在 BroadCast 接口收到消息进行校验和过滤之后，就交由对应 Chain 实例进行处理。
	Order(env *cb.Envelope, configSeq uint64) error
	Configure(config *cb.Envelope, configSeq uint64) error
	WaitReady() error
	Errored() <-chan struct{}
   // Start()负责启动此 Chain 服务。
	Start()
	Halt()
}

// Chain implements consensus.Chain interface.
type Chain struct {
   configurator Configurator
   rpc RPC // 节点与外部节点进行通信的对象，RPC 是一个接口，包含两个方法SendConsensus 和 SendSubmit。前面这种用于节点间 raft 信息的通讯，后者用于转发交易请求给 leader 节点。
   raftID    uint64
   channelID string
   lastKnownLeader uint64
   ActiveNodes     atomic.Value
   submitC  chan *submit // 接收 Orderer 客户端提交的共识请求消息的通道
   applyC   chan apply // 接收 raft 节点间应用消息的通道
   observeC chan<- raft.SoftState
   haltC    chan struct{}
   doneC    chan struct{}
   startC   chan struct{}
   snapC    chan *raftpb.Snapshot //接收 raft 节点快照数据的通道
   gcC      chan *gc
   …
   Node *node // 封装了底层 raft 库的节点实例
   …
}

type node struct {
   chainID string
   logger  *flogging.FabricLogger
   metrics *Metrics
   unreachableLock sync.RWMutex
   unreachable     map[uint64]struct{}
   tracker *Tracker
   storage *RaftStorage
   config  *raft.Config
   rpc RPC
   chain *Chain // 前面定义的Fabric自己实现的Chain结构体
   tickInterval time.Duration
   clock        clock.Clock
   metadata *etcdraft.BlockMetadata
   subscriberC chan chan uint64
   raft.Node // etcd底层的Raft中的节点接口
}

// Start instructs the orderer to begin serving the chain and keep it current.
func (c *Chain) Start() {
    
    ...
    
    //启动Node节点
    c.Node.start(c.fresh, isJoin)

	close(c.startC)
	close(c.errorC)

	go c.gc()  //从MemoryStorage中获取快照，并将其持久化到wal和磁盘。
	go c.run()

	es := c.newEvictionSuspector()

	interval := DefaultLeaderlessCheckInterval
	if c.opts.LeaderCheckInterval != 0 {
		interval = c.opts.LeaderCheckInterval
	}

	c.periodicChecker = &PeriodicCheck{
		Logger:        c.logger,
		Report:        es.confirmSuspicion,
		ReportCleared: es.clearSuspicion,
		CheckInterval: interval,
		Condition:     c.suspectEviction,
	}
	c.periodicChecker.Run()

}

func (n *node) start(fresh, join bool) {
   …
   var campaign bool
   if fresh {// 是否是新节点标记位
      if join {// 是否是新加入节点标记位
         raftPeers = nil
         n.logger.Info("Starting raft node to join an existing channel")
      } else {
         n.logger.Info("Starting raft node as part of a new channel")
         sha := sha256.Sum256([]byte(n.chainID))
         number, _ := proto.DecodeVarint(sha[24:])
         if n.config.ID == number%uint64(len(raftPeers))+1 {
            campaign = true
         }
      }
      // 调用raft/node中的启动节点函数，初始化raft
      n.Node = raft.StartNode(n.config, raftPeers)
   } else {
      n.logger.Info("Restarting raft node")
      n.Node = raft.RestartNode(n.config)
   }
   n.subscriberC = make(chan chan uint64)
// run方法中会启动一个循环用来接收raft节点发来的消息，在这里经过进一步处理后，转发给Chain层进行处理，消息的转发机制都是通过通道来完成的。
   go n.run(campaign)
}

// StartNode returns a new Node given configuration and a list of raft peers.
// It appends a ConfChangeAddNode entry for each given peer to the initial log.
func StartNode(c *Config, peers []Peer) Node {
   if len(peers) == 0 {
      panic("no peers given; use RestartNode instead")
   }
   rn, err := NewRawNode(c)
   if err != nil {
      panic(err)
   }
   rn.Bootstrap(peers)

   n := newNode(rn)

   go n.run()
   return &n
}

func (c *Chain) run() {
    ...
    
    becomeLeader := func() (chan<- *common.Block, context.CancelFunc) {
        ch := make(chan *common.Block, c.opts.MaxInflightBlocks)
        go func(ctx context.Context, ch <-chan *common.Block) {
            for {
                select {
                case b := <-ch:   //监听通道，接受Proposal的区块信息
                	data := protoutil.MarshalOrPanic(b)
                	if err := c.Node.Propose(ctx, data); err != nil {
                		c.logger.Errorf("Failed to propose block [%d] to raft and discard %d blocks in queue: %s", b.Header.Number, len(ch), err)
                		return
                	}
                	c.logger.Debugf("Proposed block [%d] to raft consensus", b.Header.Number)
                
                case <-ctx.Done():
                	c.logger.Debugf("Quit proposing blocks, discarded %d blocks in the queue", len(ch))
                	return
                }
            }
        }(ctx, ch)

        }

    ...
}

func (n *node) run(campaign bool) {
    for {
        select {
        case <-raftTicker.C():
            // grab raft Status before ticking it, so `RecentActive` attributes
            // are not reset yet.
            status := n.Status()

            n.Tick()
            n.tracker.Check(&status)

        case rd := <-n.Ready():
            startStoring := n.clock.Now()
            // 第一步，存wal
            if err := n.storage.Store(rd.Entries, rd.HardState, rd.Snapshot); err != nil {
                n.logger.Panicf("Failed to persist etcd/raft data: %s", err)
            }
            duration := n.clock.Since(startStoring).Seconds()
            n.metrics.DataPersistDuration.Observe(float64(duration))
            if duration > halfElectionTimeout {
                n.logger.Warningf("WAL sync took %v seconds and the network is configured to start elections after %v seconds. Your disk is too slow and may cause loss of quorum and trigger leadership election.", duration, electionTimeout)
            }

            if !raft.IsEmptySnap(rd.Snapshot) {
                n.chain.snapC <- &rd.Snapshot
            }

            if notifyLeaderChangeC != nil && rd.SoftState != nil {
                if l := atomic.LoadUint64(&rd.SoftState.Lead); l != raft.None {
                    select {
                    case notifyLeaderChangeC <- l:
                    default:
                    }

                    notifyLeaderChangeC = nil
                }
            }

            // skip empty apply
            if len(rd.CommittedEntries) != 0 || rd.SoftState != nil {
                n.chain.applyC <- apply{rd.CommittedEntries, rd.SoftState}
            }

            if campaign && rd.SoftState != nil {
                leader := atomic.LoadUint64(&rd.SoftState.Lead) // etcdraft requires atomic access to this var
                if leader != raft.None {
                    n.logger.Infof("Leader %d is present, quit campaign", leader)
                    campaign = false
                    close(elected)
                }
            }

            n.Advance() 

            // TODO(jay_guo) leader can write to disk in parallel with replicating
            // to the followers and them writing to their disks. Check 10.2.1 in thesis
            n.send(rd.Messages)

        case notifyLeaderChangeC = <-n.subscriberC:

        case <-n.chain.haltC:
            raftTicker.Stop()
            n.Stop()
            n.storage.Close()
            n.logger.Infof("Raft node stopped")
            close(n.chain.doneC) // close after all the artifacts are closed
            return
        }
    }

}

// Order submits normal type transactions for ordering.
func (c *Chain) Order(env *common.Envelope, configSeq uint64) error {
   c.Metrics.NormalProposalsReceived.Add(1)
   return c.Submit(&orderer.SubmitRequest{LastValidationSeq: configSeq, Payload: env, Channel: c.channelID}, 0)
}
// Configure submits config type transactions for ordering.
func (c *Chain) Configure(env *common.Envelope, configSeq uint64) error {
   c.Metrics.ConfigProposalsReceived.Add(1)
   return c.Submit(&orderer.SubmitRequest{LastValidationSeq: configSeq, Payload: env, Channel: c.channelID}, 0)
}

func (c *Chain) Submit(req *orderer.SubmitRequest, sender uint64) error {
…
   leadC := make(chan uint64, 1)
   select {
   case c.submitC <- &submit{req, leadC}: // 将消息封装并且写入submitC通道
      lead := <-leadC
      if lead == raft.None {
         c.Metrics.ProposalFailures.Add(1)
         return errors.Errorf("no Raft leader")
      }
      if lead != c.raftID { // 当前节点不是Leader，则转发消息给Leader
         if err := c.forwardToLeader(lead, req); err != nil {
            return err
         }
      }
   …
   return nil
}

func (c *Chain) ordered(msg *orderer.SubmitRequest) (batches [][]* common.Envelope, pending bool, err error) {
  if c.isConfig(msg.Payload) {
      // 配置消息
      …
      batch := c.support.BlockCutter().Cut()
      batches = [][]*common.Envelope{}
      if len(batch) != 0 {
         batches = append(batches, batch)
      }
      batches = append(batches, []*common.Envelope{msg.Payload})
      return batches, false, nil
   }
   // 普通交易信息
   if msg.LastValidationSeq < seq {
     …
   }
   batches, pending = c.support.BlockCutter().Ordered(msg.Payload)
   return batches, pending, nil
}

func (c *Chain) propose(ch chan<- *common.Block, bc *blockCreator, batches ...[]*common.Envelope) {
   for _, batch := range batches {
      b := bc.createNextBlock(batch) // 根据当前批次创建一个区块
      c.logger.Infof("Created block [%d], there are %d blocks in flight", b.Header.Number, c.blockInflight)
      select {
      case ch <- b: // 将block传递给c.ch通道，Leader可以通过这个通道收到这个区块
      default:
         c.logger.Panic("Programming error: limit of in-flight blocks does not properly take effect or block is proposed by follower")
      }
      // if it is config block, then we should wait for the commit of the block
      if protoutil.IsConfigBlock(b) {
         c.configInflight = true
      }
      c.blockInflight++
   }
}

func (n *node) Propose(ctx context.Context, data []byte) error {
   return n.stepWait(ctx, pb.Message{Type: pb.MsgProp, Entries: []pb.Entry{{Data: data}}})
}

func (c *Chain) apply(ents []raftpb.Entry) {
   …
   for i := range ents {
      switch ents[i].Type {
      case raftpb.EntryNormal:// 如果是普通entry消息
         …
         block := protoutil.UnmarshalBlockOrPanic(ents[i].Data)
         c.writeBlock(block, ents[i].Index) // 写入区块到本地
 c.Metrics.CommittedBlockNumber.Set(float64(block.Header.Number))
      case raftpb.EntryConfChange:// 如果是配置entry消息
         var cc raftpb.ConfChange
         if err := cc.Unmarshal(ents[i].Data); err != nil {
            c.logger.Warnf("Failed to unmarshal ConfChange data: %s", err)
            continue
         }
         c.confState = *c.Node.ApplyConfChange(cc)
         switch cc.Type {
         case raftpb.ConfChangeAddNode:
            c.logger.Infof("Applied config change to add node %d, current nodes in channel: %+v", cc.NodeID, c.confState.Nodes)
         case raftpb.ConfChangeRemoveNode:
            c.logger.Infof("Applied config change to remove node %d, current nodes in channel: %+v", cc.NodeID, c.confState.Nodes)
         default:
            c.logger.Panic("Programming error, encountered unsupported raft config change")
         }
…
      if ents[i].Index > c.appliedIndex {
         c.appliedIndex = ents[i].Index
      }
   }
}