diff --git a/core/blockchain.go b/core/blockchain.go index 1252454e3e..7b4f4b3038 100644 --- a/core/blockchain.go +++ b/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 @@ -134,11 +135,10 @@ type BlockChain struct { procInterrupt int32 // interrupt signaler for block processing wg sync.WaitGroup // chain processing wait group for shutting down - engine consensus.Engine - processor Processor // block processor interface - validator Validator // block and state validator interface - vmConfig vm.Config - procFeedback chan bool + engine consensus.Engine + processor Processor // block processor interface + validator Validator // block and state validator interface + vmConfig vm.Config 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.procmu.RLock() - procFeedback := bc.procFeedback - bc.procmu.RUnlock() - - if procFeedback != nil { - select { - case procFeedback <- true: - default: - } - defer func() { - select { - case procFeedback <- false: - default: - } - }() - } + bc.blockProcFeed.Send(true) + defer bc.blockProcFeed.Send(false) // 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)) +} diff --git a/les/server.go b/les/server.go index 6d606dac36..197d229837 100644 --- a/les/server.go +++ b/les/server.go @@ -146,8 +146,8 @@ func (s *LesServer) startEventLoop() { s.protocolManager.wg.Add(1) var processing bool - procFeedback := make(chan bool, 100) - s.protocolManager.blockchain.(*core.BlockChain).SetProcFeedback(procFeedback) + blockProcFeed := make(chan bool, 100) + 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)