netsync/chainmgr/block_process.go

   1 package chainmgr
   2
   3 import (
   4         "sync"
   5
   6         log "github.com/sirupsen/logrus"
   7
   8         "github.com/vapor/errors"
   9         "github.com/vapor/netsync/peers"
  10         "github.com/vapor/p2p/security"
  11 )
  12
  13 var errOrphanBlock = errors.New("fast sync inserting orphan block")
  14
  15 type BlockProcessor interface {
  16         process(chan struct{}, chan struct{}, *sync.WaitGroup)
  17 }
  18
  19 type blockProcessor struct {
  20         chain   Chain
  21         storage Storage
  22         peers   *peers.PeerSet
  23 }
  24
  25 func newBlockProcessor(chain Chain, storage Storage, peers *peers.PeerSet) *blockProcessor {
  26         return &blockProcessor{
  27                 chain:   chain,
  28                 peers:   peers,
  29                 storage: storage,
  30         }
  31 }
  32
  33 func (bp *blockProcessor) insert(blockStorage *blockStorage) error {
  34         isOrphan, err := bp.chain.ProcessBlock(blockStorage.block)
  35         if isOrphan {
  36                 bp.peers.ProcessIllegal(blockStorage.peerID, security.LevelMsgIllegal, errOrphanBlock.Error())
  37                 return errOrphanBlock
  38         }
  39
  40         if err != nil {
  41                 bp.peers.ProcessIllegal(blockStorage.peerID, security.LevelMsgIllegal, err.Error())
  42         }
  43         return err
  44 }
  45
  46 func (bp *blockProcessor) process(downloadNotifyCh chan struct{}, ProcessStop chan struct{}, wg *sync.WaitGroup) {
  47         defer func() {
  48                 close(ProcessStop)
  49                 wg.Done()
  50         }()
  51
  52         for {
  53                 for {
  54                         nextHeight := bp.chain.BestBlockHeight() + 1
  55                         block, err := bp.storage.readBlock(nextHeight)
  56                         if err != nil {
  57                                 break
  58                         }
  59
  60                         if err := bp.insert(block); err != nil {
  61                                 log.WithFields(log.Fields{"module": logModule, "err": err}).Error("failed on process block")
  62                                 return
  63                         }
  64
  65                         bp.storage.deleteBlock(nextHeight)
  66                 }
  67
  68                 if _, ok := <-downloadNotifyCh; !ok {
  69                         return
  70                 }
  71         }
  72 }