netsync/consensusmgr/block_fetcher.go

   1 package consensusmgr
   2
   3 import (
   4         "github.com/sirupsen/logrus"
   5         "gopkg.in/karalabe/cookiejar.v2/collections/prque"
   6
   7         "github.com/vapor/netsync/peers"
   8         "github.com/vapor/p2p/security"
   9         "github.com/vapor/protocol/bc"
  10 )
  11
  12 const (
  13         maxBlockDistance = 64
  14         maxMsgSetSize    = 128
  15         newBlockChSize   = 64
  16 )
  17
  18 // blockFetcher is responsible for accumulating block announcements from various peers
  19 // and scheduling them for retrieval.
  20 type blockFetcher struct {
  21         chain Chain
  22         peers *peers.PeerSet
  23
  24         newBlockCh chan *blockMsg
  25         queue      *prque.Prque
  26         msgSet     map[bc.Hash]*blockMsg
  27 }
  28
  29 //NewBlockFetcher creates a block fetcher to retrieve blocks of the new propose.
  30 func newBlockFetcher(chain Chain, peers *peers.PeerSet) *blockFetcher {
  31         f := &blockFetcher{
  32                 chain:      chain,
  33                 peers:      peers,
  34                 newBlockCh: make(chan *blockMsg, newBlockChSize),
  35                 queue:      prque.New(),
  36                 msgSet:     make(map[bc.Hash]*blockMsg),
  37         }
  38         go f.blockProcessor()
  39         return f
  40 }
  41
  42 func (f *blockFetcher) blockProcessor() {
  43         for {
  44                 for !f.queue.Empty() {
  45                         msg := f.queue.PopItem().(*blockMsg)
  46                         if msg.block.Height > f.chain.BestBlockHeight()+1 {
  47                                 f.queue.Push(msg, -float32(msg.block.Height))
  48                                 break
  49                         }
  50
  51                         f.insert(msg)
  52                         delete(f.msgSet, msg.block.Hash())
  53                 }
  54                 f.add(<-f.newBlockCh)
  55         }
  56 }
  57
  58 func (f *blockFetcher) add(msg *blockMsg) {
  59         bestHeight := f.chain.BestBlockHeight()
  60         if len(f.msgSet) > maxMsgSetSize || bestHeight > msg.block.Height || msg.block.Height-bestHeight > maxBlockDistance {
  61                 return
  62         }
  63
  64         blockHash := msg.block.Hash()
  65         if _, ok := f.msgSet[blockHash]; !ok {
  66                 f.msgSet[blockHash] = msg
  67                 f.queue.Push(msg, -float32(msg.block.Height))
  68                 logrus.WithFields(logrus.Fields{
  69                         "module":       logModule,
  70                         "block height": msg.block.Height,
  71                         "block hash":   blockHash.String(),
  72                 }).Debug("blockFetcher receive propose block")
  73         }
  74 }
  75
  76 func (f *blockFetcher) insert(msg *blockMsg) {
  77         isOrphan, err := f.chain.ProcessBlock(msg.block)
  78         if err != nil {
  79                 peer := f.peers.GetPeer(msg.peerID)
  80                 if peer == nil {
  81                         return
  82                 }
  83
  84                 f.peers.ProcessIllegal(msg.peerID, security.LevelMsgIllegal, err.Error())
  85                 return
  86         }
  87
  88         if isOrphan {
  89                 return
  90         }
  91
  92         proposeMsg, err := NewBlockProposeMsg(msg.block)
  93         if err != nil {
  94                 logrus.WithFields(logrus.Fields{"module": logModule, "err": err}).Error("failed on create BlockProposeMsg")
  95                 return
  96         }
  97
  98         if err := f.peers.BroadcastMsg(NewBroadcastMsg(proposeMsg, consensusChannel)); err != nil {
  99                 logrus.WithFields(logrus.Fields{"module": logModule, "err": err}).Error("failed on broadcast proposed block")
 100                 return
 101         }
 102 }
 103
 104 func (f *blockFetcher) processNewBlock(msg *blockMsg) {
 105         f.newBlockCh <- msg
 106 }