les, core: use event.Feed for block processing start/stop events

core/blockchain.go

@@ -111,6 +111,7 @@ type BlockChain struct {
 	chainSideFeed event.Feed
 	chainHeadFeed event.Feed
 	logsFeed      event.Feed
+	blockProcFeed event.Feed
 	scope         event.SubscriptionScope
 	genesisBlock  *types.Block
 
@@ -138,7 +139,6 @@ type BlockChain struct {
 	processor Processor // block processor interface
 	validator Validator // block and state validator interface
 	vmConfig  vm.Config
-	procFeedback chan bool
 
 	badBlocks      *lru.Cache              // Bad block cache
 	shouldPreserve func(*types.Block) bool // Function used to determine whether should preserve the given block.
@@ -371,14 +371,6 @@ func (bc *BlockChain) CurrentFastBlock() *types.Block {
 	return bc.currentFastBlock.Load().(*types.Block)
 }
 
-// SetProcFeedback adds a feedback channel where true is sent each time block
-// processing begins and false is sent when it is finished.
-func (bc *BlockChain) SetProcFeedback(procFeedback chan bool) {
-	bc.procmu.Lock()
-	defer bc.procmu.Unlock()
-	bc.procFeedback = procFeedback
-}
-
 // SetProcessor sets the processor required for making state modifications.
 func (bc *BlockChain) SetProcessor(processor Processor) {
 	bc.procmu.Lock()
@@ -1100,23 +1092,8 @@ func (bc *BlockChain) InsertChain(chain types.Blocks) (int, error) {
 		return 0, nil
 	}
 
-	// send block processing feedback if needed
+	bc.blockProcFeed.Send(true)
-	bc.procmu.RLock()
+	defer bc.blockProcFeed.Send(false)
-	procFeedback := bc.procFeedback
-	bc.procmu.RUnlock()
-
-	if procFeedback != nil {
-		select {
-		case procFeedback <- true:
-		default:
-		}
-		defer func() {
-			select {
-			case procFeedback <- false:
-			default:
-			}
-		}()
-	}
 
 	// Remove already known canon-blocks
 	var (
@@ -1753,3 +1730,9 @@ func (bc *BlockChain) SubscribeChainSideEvent(ch chan<- ChainSideEvent) event.Su
 func (bc *BlockChain) SubscribeLogsEvent(ch chan<- []*types.Log) event.Subscription {
 	return bc.scope.Track(bc.logsFeed.Subscribe(ch))
 }
+
+// SubscribeBlockProcessingEvent registers a subscription of bool where true means
+// block processing has started while false means it has stopped.
+func (bc *BlockChain) SubscribeBlockProcessingEvent(ch chan<- bool) event.Subscription {
+	return bc.scope.Track(bc.blockProcFeed.Subscribe(ch))
+}

les/server.go

@@ -146,8 +146,8 @@ func (s *LesServer) startEventLoop() {
 	s.protocolManager.wg.Add(1)
 
 	var processing bool
-	procFeedback := make(chan bool, 100)
+	blockProcFeed := make(chan bool, 100)
-	s.protocolManager.blockchain.(*core.BlockChain).SetProcFeedback(procFeedback)
+	s.protocolManager.blockchain.(*core.BlockChain).SubscribeBlockProcessingEvent(blockProcFeed)
 	totalRechargeCh := make(chan uint64, 100)
 	totalRecharge := s.costTracker.subscribeTotalRecharge(totalRechargeCh)
 	totalCapacityCh := make(chan uint64, 100)
@@ -163,7 +163,7 @@ func (s *LesServer) startEventLoop() {
 				s.fcManager.SetRechargeCurve(flowcontrol.PieceWiseLinear{{0, 0}, {totalRecharge / 10, totalRecharge}, {totalRecharge, totalRecharge}})
 			}
 			select {
-			case processing = <-procFeedback:
+			case processing = <-blockProcFeed:
 			case totalRecharge = <-totalRechargeCh:
 			case totalCapacity = <-totalCapacityCh:
 				s.priorityClientPool.setLimits(s.maxPeers, totalCapacity)