core, les: add block processing feedback

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() {

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)

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
 }