Method blocksHandler

protocols/cpc/syncer/syncer.go:343–401

(errCh chan error, successCh chan struct{}, mode SyncMode)

}

func (s *Synchronizer) blocksHandler(errCh chan error, successCh chan struct{}, mode SyncMode) {
	for {
		if atomic.LoadInt32(&s.finishFetch) == int32(1) && s.blocksQueue.empty() {
			successCh <- struct{}{}
			return
		}
		task, err := s.blocksQueue.get()
		if err != nil {
			if err == ErrTimeout {
				continue
			}
			log.Error("block queue get error", "err", err)
			errCh <- err
			return
		}
		s.progressLock.Lock()
		if mode == FastSync {
			s.progress.CurrentBlock = s.blockchain.CurrentFastBlock().NumberU64()
		} else {
			s.progress.CurrentBlock = s.blockchain.CurrentBlock().NumberU64()
		}
		s.progressLock.Unlock()

		if mode == FullSync {
			if _, err := s.blockchain.InsertChain(task.data.(types.Blocks)); err != nil {
				log.Debug("insert chain", "err", err)
				errCh <- err
				return
			}
		} else {
			var start = time.Now()
			data := task.data.(blocksWithReceipts)
			_, err := s.blockchain.InsertHeaderChain(data.headers, 16)
			if err != nil {
				errCh <- err
				return
			}
			log.Debug("validate and insert headers", "elapsed", common.PrettyDuration(time.Since(start)))
			start = time.Now()
			blocksNew := make([]*types.Block, len(data.headers))
			for i, header := range data.headers {
				blocksNew[i] = types.NewBlockWithHeader(header).WithBody(data.bodies[i])
			}
			if _, err := s.blockchain.InsertReceiptChain(blocksNew, data.receipts); err != nil {
				log.Error("insert receipts error", "err", err)
				errCh <- err
				return
			}
			// send metrics msg to monitor(prometheus)
			if chainmetrics.NeedPushMetrics() {
				go func() {
					chainmetrics.UpdateBlockNumberGauge(float64(s.blockchain.CurrentFastBlock().NumberU64()))
					chainmetrics.ReportBlockNumberGauge("blocknumber")
				}()
			}
			log.Debug("insert blocks with receipts", "elapsed", common.PrettyDuration(time.Since(start)))
		}
	}