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