netsync/block_fetcher.go

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