Method run

consensus/vbft/node_sync.go:113–242

()

}

func (self *Syncer) run() {
	self.server.quitWg.Add(1)
	defer self.server.quitWg.Done()

	for {
		select {
		case <-self.syncCheckReqC:
		case req := <-self.blockSyncReqC:
			if req.targetBlockNum == 0 {
				// cancel fetcher for peer
				for _, id := range req.targetPeers {
					self.cancelFetcherForPeer(self.peers[id])
				}
				continue
			}

			log.Infof("server %d, got sync req(%d, %d) to %v",
				self.server.Index, req.startBlockNum, req.targetBlockNum, req.targetPeers)
			if req.startBlockNum <= self.server.GetCommittedBlockNo() {
				req.startBlockNum = self.server.GetCommittedBlockNo() + 1
				log.Infof("server %d, sync req start change to %d",
					self.server.Index, req.startBlockNum)
			}
			for ; req.startBlockNum <= req.targetBlockNum; req.startBlockNum++ {
				blk, _ := self.server.blockPool.getSealedBlock(req.startBlockNum)
				if blk == nil {
					log.Infof("server %d, on starting syncing %d, nil block from ledger",
						self.server.Index, req.startBlockNum)
					break
				}
				if err := self.server.fastForwardBlock(blk); err != nil {
					log.Infof("server %d, on starting syncing %d, %s",
						self.server.Index, req.startBlockNum, err)
					break
				}
			}
			if req.startBlockNum > req.targetBlockNum {
				continue
			}
			self.onNewBlockSyncReq(req)

		case syncMsg := <-self.syncMsgC:
			if p, present := self.peers[syncMsg.fromPeer]; present {
				if p.active {
					p.msgC <- syncMsg.msg
				} else {
					// report err
					p.msgC <- nil
				}
			} else {
				// report error
			}

		case blkMsgFromPeer := <-self.blockFromPeerC:
			blkNum := blkMsgFromPeer.block.getBlockNum()
			if blkNum < self.nextReqBlkNum {
				continue
			}