diff --git a/core/blockchain.go b/core/blockchain.go index 34832252a7..c1595e66d3 100644 --- a/core/blockchain.go +++ b/core/blockchain.go @@ -123,10 +123,11 @@ 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 + engine consensus.Engine + processor Processor // block processor interface + validator Validator // block and state validator interface + vmConfig vm.Config + procFeedback chan bool badBlocks *lru.Cache // Bad block cache } @@ -348,6 +349,14 @@ 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() @@ -1009,6 +1018,25 @@ func (bc *BlockChain) insertChain(chain types.Blocks) (int, []interface{}, []*ty if len(chain) == 0 { return 0, nil, nil, 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: + } + }() + } + // Do a sanity check that the provided chain is actually ordered and linked for i := 1; i < len(chain); i++ { if chain[i].NumberU64() != chain[i-1].NumberU64()+1 || chain[i].ParentHash() != chain[i-1].Hash() { diff --git a/les/flowcontrol/manager.go b/les/flowcontrol/manager.go index 8b9501bb1b..ee7405e2fc 100644 --- a/les/flowcontrol/manager.go +++ b/les/flowcontrol/manager.go @@ -25,9 +25,9 @@ import ( ) const ( - cmDisabled = iota // client manager is disabled, no requests are accepted - cmNormal // normal operation, maximum available bandwidth can be allocated - cmBlockProcessing // requests are accepted but the buffers are only recharged with the guaranteed minimum rate + CmDisabled = iota // client manager is disabled, no requests are accepted + CmNormal // normal operation, maximum available bandwidth can be allocated + CmBlockProcessing // requests are accepted but the buffers are only recharged with the guaranteed minimum rate ) // cmNodeFields are ClientNode fields used by the client manager @@ -111,7 +111,7 @@ func NewClientManager(maxParallelReqs int, targetParallelReqs float64, clock mcl maxParallelReqs: maxParallelReqs, targetParallelReqs: targetParallelReqs, } - cm.SetMode(cmNormal) + cm.SetMode(CmNormal) return cm } @@ -127,8 +127,8 @@ func (cm *ClientManager) SetMode(newMode int) { } cm.updateRecharge(cm.clock.Now()) - enabled := cm.mode != cmDisabled - newEnabled := cm.mode != cmDisabled + enabled := cm.mode != CmDisabled + newEnabled := cm.mode != CmDisabled if !enabled && newEnabled && cm.enabledCh != nil { close(cm.enabledCh) cm.enabledCh = nil @@ -138,15 +138,15 @@ func (cm *ClientManager) SetMode(newMode int) { } switch newMode { - case cmDisabled: + case CmDisabled: cm.totalRecharge = 0 cm.bufCorrEnabled = false cm.forceMinRecharge = false - case cmNormal: + case CmNormal: cm.totalRecharge = cm.targetParallelReqs * 1000000 cm.bufCorrEnabled = true cm.forceMinRecharge = false - case cmBlockProcessing: + case CmBlockProcessing: cm.totalRecharge = 0 cm.bufCorrEnabled = false cm.forceMinRecharge = true @@ -158,7 +158,7 @@ func (cm *ClientManager) SetMode(newMode int) { if cm.parallelReqs == 0 { cm.child.SetMode(newMode) } else { - cm.child.SetMode(cmDisabled) + cm.child.SetMode(CmDisabled) } } } @@ -167,9 +167,9 @@ func (cm *ClientManager) setParallelReqs(p int, time mclock.AbsTime) { if p == cm.parallelReqs { return } - if cm.child != nil && cm.mode != cmDisabled { + if cm.child != nil && cm.mode != CmDisabled { if cm.parallelReqs == 0 { - cm.child.SetMode(cmDisabled) + cm.child.SetMode(CmDisabled) } if p == 0 { cm.child.SetMode(cm.mode) diff --git a/les/server.go b/les/server.go index 9e0dcc96bc..18fead3f2b 100644 --- a/les/server.go +++ b/les/server.go @@ -41,7 +41,7 @@ import ( type LesServer struct { config *eth.Config protocolManager *ProtocolManager - fcManager *flowcontrol.ClientManager // nil if our node is client only + fcManager *flowcontrol.ClientManager fcCostStats *requestCostStats defParams *flowcontrol.ServerParams lesTopics []discv5.Topic @@ -105,10 +105,32 @@ func NewLesServer(eth *eth.Ethereum, config *eth.Config) (*LesServer, error) { mpr = 4 } srv.fcManager = flowcontrol.NewClientManager(mpr, tpr, &mclock.MonotonicClock{}, nil) + pm.blockProcLoop(srv.fcManager) srv.fcCostStats = newCostStats(eth.ChainDb()) return srv, nil } +func (pm *ProtocolManager) blockProcLoop(cm *flowcontrol.ClientManager) { + pm.wg.Add(1) + procFeedback := make(chan bool, 10) + pm.blockchain.(*core.BlockChain).SetProcFeedback(procFeedback) + go func() { + for { + select { + case processing := <-procFeedback: + if processing { + cm.SetMode(flowcontrol.CmBlockProcessing) + } else { + cm.SetMode(flowcontrol.CmNormal) + } + case <-pm.quitSync: + pm.wg.Done() + return + } + } + }() +} + func (s *LesServer) Protocols() []p2p.Protocol { return s.protocolManager.SubProtocols }