netsync/chainmgr/block_process.go

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