les, les/flowcontrol: improved client manager and load tests

cmd/utils/flags.go

@@ -177,7 +177,7 @@ var (
 	}
 	LightServFlag = cli.IntFlag{
 		Name:  "lightserv",
-		Usage: "Maximum percentage of time allowed for serving LES requests (0-90)",
+		Usage: "Maximum percentage of time allowed for serving LES requests (multi-threaded processing allows values over 100)",
 		Value: 0,
 	}
 	LightPeersFlag = cli.IntFlag{

common/mclock/mclock.go

@@ -30,3 +30,29 @@ type AbsTime time.Duration
 func Now() AbsTime {
 	return AbsTime(monotime.Now())
 }
+
+// Clock interface makes it possible to replace the monotonic system clock with
+// a simulated clock
+type Clock interface {
+	Now() AbsTime
+	Sleep(time.Duration)
+	After(time.Duration) <-chan time.Time
+}
+
+// MonotonicClock implements Clock using the system clock
+type MonotonicClock struct{}
+
+// Now implements Clock
+func (MonotonicClock) Now() AbsTime {
+	return AbsTime(monotime.Now())
+}
+
+// Sleep implements Clock
+func (MonotonicClock) Sleep(d time.Duration) {
+	time.Sleep(d)
+}
+
+// After implements Clock
+func (MonotonicClock) After(d time.Duration) <-chan time.Time {
+	return time.After(d)
+}

common/mclock/simclock.go

@@ -0,0 +1,136 @@
+// Copyright 2016 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+// package mclock is a wrapper for a monotonic clock source
+package mclock
+
+import (
+	"runtime"
+	"sync"
+	"time"
+)
+
+type event struct {
+	do func()
+	at AbsTime
+}
+
+// SimulatedClock implements a virtual Clock for reproducible time-sensitive tests.
+// It simulates a scheduler on a virtual timescale where actual processing takes zero time.
+//
+// Note: since there is no way in Go to know when all goroutines have reached a waiting
+// state (which should theoretically happen in each virtual moment), the algorithm runs
+// GoSched a fixed number of times after each step and limits time steps in order to
+// minimize precision loss (see maxStep and goSchedCount).
+type SimulatedClock struct {
+	now       AbsTime
+	scheduled []event
+	stop      bool
+	lock      sync.RWMutex
+}
+
+const (
+	maxStep      = time.Microsecond * 10
+	goSchedCount = 10
+)
+
+// NewSimulatedClock creates a new simulated clock
+func NewSimulatedClock() *SimulatedClock {
+	s := &SimulatedClock{scheduled: make([]event, 0, 100)}
+
+	go func() {
+		lastScheduled := 0
+		for {
+			for i := 0; i < goSchedCount; i++ {
+				runtime.Gosched()
+			}
+			//time.Sleep(time.Microsecond * 10)
+			s.lock.Lock()
+			if s.stop {
+				s.lock.Unlock()
+				return
+			}
+			scheduled := len(s.scheduled)
+			if scheduled > 0 && scheduled == lastScheduled {
+				ev := s.scheduled[0]
+				if ev.at <= s.now+AbsTime(maxStep) {
+					s.scheduled = s.scheduled[1:]
+					s.now = ev.at
+					ev.do()
+				} else {
+					s.now += AbsTime(maxStep)
+				}
+			}
+			lastScheduled = scheduled
+			s.lock.Unlock()
+		}
+	}()
+
+	return s
+}
+
+// Stop stops the clock (Sleeps and Afters will never return after this)
+func (s *SimulatedClock) Stop() {
+	s.lock.Lock()
+	s.stop = true
+	s.lock.Unlock()
+}
+
+// Now implements Clock
+func (s *SimulatedClock) Now() AbsTime {
+	s.lock.RLock()
+	defer s.lock.RUnlock()
+
+	return s.now
+}
+
+// Sleep implements Clock
+func (s *SimulatedClock) Sleep(d time.Duration) {
+	done := make(chan struct{})
+	s.insert(d, func() {
+		close(done)
+	})
+	<-done
+}
+
+// After implements Clock
+func (s *SimulatedClock) After(d time.Duration) <-chan time.Time {
+	after := make(chan time.Time, 1)
+	s.insert(d, func() {
+		after <- time.Unix(0, int64(s.now))
+	})
+	return after
+}
+
+func (s *SimulatedClock) insert(d time.Duration, do func()) {
+	s.lock.Lock()
+	defer s.lock.Unlock()
+
+	at := s.now + AbsTime(d)
+	l, h := 0, len(s.scheduled)
+	ll := h
+	for l != h {
+		m := (l + h) / 2
+		if at < s.scheduled[m].at {
+			h = m
+		} else {
+			l = m + 1
+		}
+	}
+	s.scheduled = append(s.scheduled, event{})
+	copy(s.scheduled[l+1:], s.scheduled[l:ll])
+	s.scheduled[l] = event{do: do, at: at}
+}

les/backend.go

@@ -25,6 +25,7 @@ import (
 	"github.com/ethereum/go-ethereum/accounts"
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/common/hexutil"
+	"github.com/ethereum/go-ethereum/common/mclock"
 	"github.com/ethereum/go-ethereum/consensus"
 	"github.com/ethereum/go-ethereum/core"
 	"github.com/ethereum/go-ethereum/core/bloombits"
@@ -100,7 +101,7 @@ func New(ctx *node.ServiceContext, config *eth.Config) (*LightEthereum, error) {
 		chainDb:          chainDb,
 		eventMux:         ctx.EventMux,
 		peers:            peers,
-		reqDist:          newRequestDistributor(peers, quitSync),
+		reqDist:          newRequestDistributor(peers, quitSync, &mclock.MonotonicClock{}),
 		accountManager:   ctx.AccountManager,
 		engine:           eth.CreateConsensusEngine(ctx, &config.Ethash, chainConfig, chainDb),
 		shutdownChan:     make(chan bool),

les/distributor.go

@@ -23,6 +23,8 @@ import (
 	"errors"
 	"sync"
 	"time"
+
+	"github.com/ethereum/go-ethereum/common/mclock"
 )
 
 // ErrNoPeers is returned if no peers capable of serving a queued request are available
@@ -32,6 +34,7 @@ var ErrNoPeers = errors.New("no suitable peers available")
 // suitable peers, obeying flow control rules and prioritizing them in creation
 // order (even when a resend is necessary).
 type requestDistributor struct {
+	clock            mclock.Clock
 	reqQueue         *list.List
 	lastReqOrder     uint64
 	peers            map[distPeer]struct{}
@@ -71,8 +74,9 @@ type distReq struct {
 }
 
 // newRequestDistributor creates a new request distributor
-func newRequestDistributor(peers *peerSet, stopChn chan struct{}) *requestDistributor {
+func newRequestDistributor(peers *peerSet, stopChn chan struct{}, clock mclock.Clock) *requestDistributor {
 	d := &requestDistributor{
+		clock:    clock,
 		reqQueue: list.New(),
 		loopChn:  make(chan struct{}, 2),
 		stopChn:  stopChn,
@@ -150,7 +154,7 @@ func (d *requestDistributor) loop() {
 						wait = distMaxWait
 					}
 					go func() {
-						time.Sleep(wait)
+						d.clock.Sleep(wait)
 						d.loopChn <- struct{}{}
 					}()
 					break loop

les/distributor_test.go

@@ -23,6 +23,8 @@ import (
 	"sync"
 	"testing"
 	"time"
+
+	"github.com/ethereum/go-ethereum/common/mclock"
 )
 
 type testDistReq struct {
@@ -121,7 +123,7 @@ func testRequestDistributor(t *testing.T, resend bool) {
 	stop := make(chan struct{})
 	defer close(stop)
 
-	dist := newRequestDistributor(nil, stop)
+	dist := newRequestDistributor(nil, stop, &mclock.MonotonicClock{})
 	var peers [testDistPeerCount]*testDistPeer
 	for i := range peers {
 		peers[i] = &testDistPeer{}

les/fetcher.go

@@ -474,7 +474,7 @@ func (f *lightFetcher) nextRequest() (*distReq, uint64) {
 				f.lock.Unlock()
 
 				cost := p.GetRequestCost(GetBlockHeadersMsg, int(bestAmount))
-				p.fcServer.QueueRequest(reqID, cost)
+				p.fcServer.QueuedRequest(reqID, cost)
 				f.reqMu.Lock()
 				f.requested[reqID] = fetchRequest{hash: bestHash, amount: bestAmount, peer: p, sent: mclock.Now()}
 				f.reqMu.Unlock()

les/flowcontrol/control.go

@@ -24,34 +24,44 @@ import (
 	"github.com/ethereum/go-ethereum/common/mclock"
 )
 
+// fcTimeConst is the time constant applied for MinRecharge during linear
+// buffer recharge period
 const fcTimeConst = time.Millisecond
 
+// ServerParams are the flow control parameters specified by a server for a client
+//
+// Note: a server can assign different amounts of bandwidth to each client by giving
+// different parameters to them.
 type ServerParams struct {
 	BufLimit, MinRecharge uint64
 }
 
+// ClientNode is the flow control system's representation of a client
+// (used in server mode only)
 type ClientNode struct {
 	params   *ServerParams
 	bufValue uint64
 	lastTime mclock.AbsTime
 	lock     sync.Mutex
 	cm       *ClientManager
-	cmNode   *cmNode
+	cmNodeFields
 }
 
+// NewClientNode returns a new ClientNode
 func NewClientNode(cm *ClientManager, params *ServerParams) *ClientNode {
 	node := &ClientNode{
 		cm:       cm,
 		params:   params,
 		bufValue: params.BufLimit,
-		lastTime: mclock.Now(),
+		lastTime: cm.clock.Now(),
 	}
-	node.cmNode = cm.addNode(node)
+	cm.addNode(node)
 	return node
 }
 
-func (peer *ClientNode) Remove(cm *ClientManager) {
+// Remove removes the client from the client manager
-	cm.removeNode(peer.cmNode)
+func (peer *ClientNode) Remove() {
+	peer.cm.removeNode(peer)
 }
 
 func (peer *ClientNode) recalcBV(time mclock.AbsTime) {
@@ -66,35 +76,56 @@ func (peer *ClientNode) recalcBV(time mclock.AbsTime) {
 	peer.lastTime = time
 }
 
-func (peer *ClientNode) AcceptRequest() (uint64, bool) {
+// AcceptRequest returns whether a new request can be accepted and the missing
+// buffer amount if it was rejected due to a buffer underrun. If accepted, maxCost
+// is deducted from the flow control buffer.
+func (peer *ClientNode) AcceptRequest(maxCost uint64) (bool, uint64) {
 	peer.lock.Lock()
-	defer peer.lock.Unlock()
 
-	time := mclock.Now()
+	time := peer.cm.clock.Now()
 	peer.recalcBV(time)
-	return peer.bufValue, peer.cm.accept(peer.cmNode, time)
+	if maxCost > peer.bufValue {
+		return false, maxCost - peer.bufValue
+	}
+	peer.bufValue -= maxCost
+	ch := peer.cm.accept(peer, maxCost, time)
+
+	peer.lock.Unlock()
+	ok := true
+	if ch != nil {
+		ok = <-ch
+		if ok {
+			peer.lock.Lock()
+			peer.cm.started(peer, maxCost, time)
+			peer.lock.Unlock()
+		}
+	}
+	return ok, 0
 }
 
-func (peer *ClientNode) RequestProcessed(cost uint64) (bv, realCost uint64) {
+// RequestProcessed should be called when the request has been processed
+func (peer *ClientNode) RequestProcessed() (bv, realCost uint64) {
 	peer.lock.Lock()
 	defer peer.lock.Unlock()
 
-	time := mclock.Now()
+	time := peer.cm.clock.Now()
 	peer.recalcBV(time)
-	peer.bufValue -= cost
+	rcost := peer.cm.processed(peer, time)
-	peer.recalcBV(time)
-	rcValue, rcost := peer.cm.processed(peer.cmNode, time)
-	if rcValue < peer.params.BufLimit {
-		bv := peer.params.BufLimit - rcValue
-		if bv > peer.bufValue {
-			peer.bufValue = bv
-		}
-	}
 	return peer.bufValue, rcost
 }
 
+// WaitOrStop blocks while request processing is disabled and returns true if
+// it should be cancelled because the client has been disconnected.
+func (peer *ClientNode) WaitOrStop() bool {
+	return peer.cm.waitOrStop(peer)
+}
+
+// ServerNode is the flow control system's representation of a server
+// (used in client mode only)
 type ServerNode struct {
+	clock       mclock.Clock
 	bufEstimate uint64
+	bufRecharge bool
 	lastTime    mclock.AbsTime
 	params      *ServerParams
 	sumCost     uint64            // sum of req costs sent to this server
@@ -102,23 +133,29 @@ type ServerNode struct {
 	lock        sync.RWMutex
 }
 
-func NewServerNode(params *ServerParams) *ServerNode {
+// NewServerNode returns a new ServerNode
+func NewServerNode(params *ServerParams, clock mclock.Clock) *ServerNode {
 	return &ServerNode{
+		clock:       clock,
 		bufEstimate: params.BufLimit,
-		lastTime:    mclock.Now(),
+		bufRecharge: false,
+		lastTime:    clock.Now(),
 		params:      params,
 		pending:     make(map[uint64]uint64),
 	}
 }
 
 func (peer *ServerNode) recalcBLE(time mclock.AbsTime) {
-	dt := uint64(time - peer.lastTime)
 	if time < peer.lastTime {
-		dt = 0
+		return
 	}
+	if peer.bufRecharge {
+		dt := uint64(time - peer.lastTime)
 		peer.bufEstimate += peer.params.MinRecharge * dt / uint64(fcTimeConst)
-	if peer.bufEstimate > peer.params.BufLimit {
+		if peer.bufEstimate >= peer.params.BufLimit {
 			peer.bufEstimate = peer.params.BufLimit
+			peer.bufRecharge = false
+		}
 	}
 	peer.lastTime = time
 }
@@ -127,7 +164,7 @@ func (peer *ServerNode) recalcBLE(time mclock.AbsTime) {
 const safetyMargin = time.Millisecond
 
 func (peer *ServerNode) canSend(maxCost uint64) (time.Duration, float64) {
-	peer.recalcBLE(mclock.Now())
+	peer.recalcBLE(peer.clock.Now())
 	maxCost += uint64(safetyMargin) * peer.params.MinRecharge / uint64(fcTimeConst)
 	if maxCost > peer.params.BufLimit {
 		maxCost = peer.params.BufLimit
@@ -148,25 +185,29 @@ func (peer *ServerNode) CanSend(maxCost uint64) (time.Duration, float64) {
 	return peer.canSend(maxCost)
 }
 
-// QueueRequest should be called when the request has been assigned to the given
+// QueuedRequest should be called when the request has been assigned to the given
 // server node, before putting it in the send queue. It is mandatory that requests
-// are sent in the same order as the QueueRequest calls are made.
+// are sent in the same order as the QueuedRequest calls are made.
-func (peer *ServerNode) QueueRequest(reqID, maxCost uint64) {
+func (peer *ServerNode) QueuedRequest(reqID, maxCost uint64) {
 	peer.lock.Lock()
 	defer peer.lock.Unlock()
 
+	peer.recalcBLE(peer.clock.Now())
+	// Note: we do not know when requests actually arrive to the server so bufRecharge
+	// is not turned on here if buffer was full; in this case it is going to be turned
+	// on by the first reply's bufValue feedback
 	peer.bufEstimate -= maxCost
 	peer.sumCost += maxCost
 	peer.pending[reqID] = peer.sumCost
 }
 
-// GotReply adjusts estimated buffer value according to the value included in
+// ReceivedReply adjusts estimated buffer value according to the value included in
 // the latest request reply.
-func (peer *ServerNode) GotReply(reqID, bv uint64) {
+func (peer *ServerNode) ReceivedReply(reqID, bv uint64) {
-
 	peer.lock.Lock()
 	defer peer.lock.Unlock()
 
+	peer.recalcBLE(peer.clock.Now())
 	if bv > peer.params.BufLimit {
 		bv = peer.params.BufLimit
 	}
@@ -180,5 +221,6 @@ func (peer *ServerNode) GotReply(reqID, bv uint64) {
 	if bv > cc {
 		peer.bufEstimate = bv - cc
 	}
-	peer.lastTime = mclock.Now()
+	peer.bufRecharge = peer.bufEstimate < peer.params.BufLimit
+	peer.lastTime = peer.clock.Now()
 }

les/flowcontrol/manager.go

@@ -19,206 +19,338 @@ package flowcontrol
 
 import (
 	"sync"
-	"time"
 
 	"github.com/ethereum/go-ethereum/common/mclock"
+	"github.com/ethereum/go-ethereum/les/flowcontrol/prque"
 )
 
-const rcConst = 1000000
+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
+)
 
-type cmNode struct {
+// cmNodeFields are ClientNode fields used by the client manager
+// Note: these fields are locked by the client manager's mutex
+type cmNodeFields struct {
+	servingStarted                 mclock.AbsTime
+	servingMaxCost                 uint64
+	corrBufValue                   int64
+	rcLastUpdate                   mclock.AbsTime
+	rcLastIntValue, rcNextIntValue int64
+}
+
+// rcQueueItem represents an integrator threshold value where a certain client's buffer is recharged
+type rcQueueItem struct {
 	node     *ClientNode
-	lastUpdate                   mclock.AbsTime
+	intValue int64
-	serving, recharging          bool
-	rcWeight                     uint64
-	rcValue, rcDelta, startValue int64
-	finishRecharge               mclock.AbsTime
 }
 
-func (node *cmNode) update(time mclock.AbsTime) {
+// Before implements prque.item
-	dt := int64(time - node.lastUpdate)
+//
-	node.rcValue += node.rcDelta * dt / rcConst
+// Note: intValue is interpreted as mod 2^64, the difference between the highest
-	node.lastUpdate = time
+// and lowest value at any moment is always less than 2^63.
-	if node.recharging && time >= node.finishRecharge {
+func (rcq rcQueueItem) Before(j interface{}) bool {
-		node.recharging = false
+	return (j.(rcQueueItem).intValue - rcq.intValue) > 0
-		node.rcDelta = 0
-		node.rcValue = 0
-	}
 }
 
-func (node *cmNode) set(serving bool, simReqCnt, sumWeight uint64) {
+// Note: valid is called under client manager mutex lock
-	if node.serving && !serving {
+func (rcq rcQueueItem) valid() bool {
-		node.recharging = true
+	return rcq.intValue == rcq.node.rcNextIntValue
-		sumWeight += node.rcWeight
-	}
-	node.serving = serving
-	if node.recharging && serving {
-		node.recharging = false
-		sumWeight -= node.rcWeight
-	}
-
-	node.rcDelta = 0
-	if serving {
-		node.rcDelta = int64(rcConst / simReqCnt)
-	}
-	if node.recharging {
-		node.rcDelta = -int64(node.node.cm.rcRecharge * node.rcWeight / sumWeight)
-		node.finishRecharge = node.lastUpdate + mclock.AbsTime(node.rcValue*rcConst/(-node.rcDelta))
-	}
 }
 
+// servingQueueItem represents a queued request (prioritized by BufValue/BufLimit)
+type servingQueueItem struct {
+	start    func() bool
+	priority float64
+}
+
+// Before implements prque.item
+func (sq servingQueueItem) Before(j interface{}) bool {
+	return sq.priority > j.(servingQueueItem).priority
+}
+
+// ClientManager controls the bandwidth assigned to the clients of a server.
+// Since ServerParams guarantee a safe lower estimate for processable requests
+// even in case of all clients being active, ClientManager calculates a
+// corrigated buffer value and usually allows a higher remaining buffer value
+// to be returned with each reply.
 type ClientManager struct {
-	lock                             sync.Mutex
+	clock     mclock.Clock
-	nodes                            map[*cmNode]struct{}
+	child     *ClientManager
-	simReqCnt, sumWeight, rcSumValue uint64
+	lock      sync.RWMutex
-	maxSimReq, maxRcSum              uint64
+	nodes     map[*ClientNode]struct{}
-	rcRecharge                       uint64
+	enabledCh chan struct{}
-	resumeQueue                      chan chan bool
+
-	time                             mclock.AbsTime
+	parallelReqs, maxParallelReqs int
+	targetParallelReqs            float64
+	servingQueue                  *prque.Prque
+
+	mode                             int
+	totalRecharge                    float64
+	forceMinRecharge, bufCorrEnabled bool
+
+	sumRecharge    uint64
+	rcLastUpdate   mclock.AbsTime
+	rcLastIntValue int64 // normalized to MRR=1000000
+	rcQueue        *prque.Prque
 }
 
-func NewClientManager(rcTarget, maxSimReq, maxRcSum uint64) *ClientManager {
+// NewClientManager returns a new client manager. Multiple client managers can
+// be chained to realize priority levels. Each level has its own manager, the
+// parent has the higher priority (while the parent is processing a request
+// the child is disabled).
+func NewClientManager(maxParallelReqs int, targetParallelReqs float64, clock mclock.Clock, child *ClientManager) *ClientManager {
 	cm := &ClientManager{
-		nodes:       make(map[*cmNode]struct{}),
+		clock:        clock,
-		resumeQueue: make(chan chan bool),
+		nodes:        make(map[*ClientNode]struct{}),
-		rcRecharge:  rcConst * rcConst / (100*rcConst/rcTarget - rcConst),
+		child:        child,
-		maxSimReq:   maxSimReq,
+		servingQueue: prque.New(),
-		maxRcSum:    maxRcSum,
+		rcQueue:      prque.New(),
+
+		maxParallelReqs:    maxParallelReqs,
+		targetParallelReqs: targetParallelReqs,
 	}
-	go cm.queueProc()
+	cm.SetMode(cmNormal)
 	return cm
 }
 
-func (self *ClientManager) Stop() {
+// SetMode changes the operating mode of the manager and its children. When
-	self.lock.Lock()
+// multiple priority levels are used, mode should be changed at the manager
-	defer self.lock.Unlock()
+// of the highest level.
+func (cm *ClientManager) SetMode(newMode int) {
+	cm.lock.Lock()
+	defer cm.lock.Unlock()
 
-	// signal any waiting accept routines to return false
+	if newMode == cm.mode {
-	self.nodes = make(map[*cmNode]struct{})
-	close(self.resumeQueue)
-}
-
-func (self *ClientManager) addNode(cnode *ClientNode) *cmNode {
-	time := mclock.Now()
-	node := &cmNode{
-		node:           cnode,
-		lastUpdate:     time,
-		finishRecharge: time,
-		rcWeight:       1,
-	}
-	self.lock.Lock()
-	defer self.lock.Unlock()
-
-	self.nodes[node] = struct{}{}
-	self.update(mclock.Now())
-	return node
-}
-
-func (self *ClientManager) removeNode(node *cmNode) {
-	self.lock.Lock()
-	defer self.lock.Unlock()
-
-	time := mclock.Now()
-	self.stop(node, time)
-	delete(self.nodes, node)
-	self.update(time)
-}
-
-// recalc sumWeight
-func (self *ClientManager) updateNodes(time mclock.AbsTime) (rce bool) {
-	var sumWeight, rcSum uint64
-	for node := range self.nodes {
-		rc := node.recharging
-		node.update(time)
-		if rc && !node.recharging {
-			rce = true
-		}
-		if node.recharging {
-			sumWeight += node.rcWeight
-		}
-		rcSum += uint64(node.rcValue)
-	}
-	self.sumWeight = sumWeight
-	self.rcSumValue = rcSum
 		return
-}
+	}
+	cm.updateRecharge(cm.clock.Now())
 
-func (self *ClientManager) update(time mclock.AbsTime) {
+	enabled := cm.mode != cmDisabled
-	for {
+	newEnabled := cm.mode != cmDisabled
-		firstTime := time
+	if !enabled && newEnabled && cm.enabledCh != nil {
-		for node := range self.nodes {
+		close(cm.enabledCh)
-			if node.recharging && node.finishRecharge < firstTime {
+		cm.enabledCh = nil
-				firstTime = node.finishRecharge
 	}
+	if enabled && !newEnabled {
+		cm.enabledCh = make(chan struct{})
 	}
-		if self.updateNodes(firstTime) {
+
-			for node := range self.nodes {
+	switch newMode {
-				if node.recharging {
+	case cmDisabled:
-					node.set(node.serving, self.simReqCnt, self.sumWeight)
+		cm.totalRecharge = 0
-				}
+		cm.bufCorrEnabled = false
+		cm.forceMinRecharge = false
+	case cmNormal:
+		cm.totalRecharge = cm.targetParallelReqs * 1000000
+		cm.bufCorrEnabled = true
+		cm.forceMinRecharge = false
+	case cmBlockProcessing:
+		cm.totalRecharge = 0
+		cm.bufCorrEnabled = false
+		cm.forceMinRecharge = true
 	}
+
+	cm.mode = newMode
+
+	if cm.child != nil {
+		if cm.parallelReqs == 0 {
+			cm.child.SetMode(newMode)
 		} else {
-			self.time = time
+			cm.child.SetMode(cmDisabled)
+		}
+	}
+}
+
+func (cm *ClientManager) setParallelReqs(p int, time mclock.AbsTime) {
+	if p == cm.parallelReqs {
+		return
+	}
+	if cm.child != nil && cm.mode != cmDisabled {
+		if cm.parallelReqs == 0 {
+			cm.child.SetMode(cmDisabled)
+		}
+		if p == 0 {
+			cm.child.SetMode(cm.mode)
+		}
+	}
+	cm.parallelReqs = p
+}
+
+func (cm *ClientManager) updateRecharge(time mclock.AbsTime) {
+	lastUpdate := cm.rcLastUpdate
+	cm.rcLastUpdate = time
+	if cm.totalRecharge == 0 {
+		return
+	}
+	for cm.sumRecharge > 0 {
+		var slope float64
+		if cm.forceMinRecharge {
+			slope = 1
+		} else {
+			slope = cm.totalRecharge / float64(cm.sumRecharge)
+		}
+		dt := time - lastUpdate
+		q := cm.rcQueue.Pop()
+		for q != nil && !q.(rcQueueItem).valid() {
+			q = cm.rcQueue.Pop()
+		}
+		if q == nil {
+			cm.rcLastIntValue += int64(slope * float64(dt))
+			return
+		}
+		rcqItem := q.(rcQueueItem)
+		dtNext := mclock.AbsTime(float64(rcqItem.intValue-cm.rcLastIntValue) / slope)
+		if dt < dtNext {
+			cm.rcQueue.Push(q)
+			cm.rcLastIntValue += int64(slope * float64(dt))
+			return
+		}
+		if rcqItem.node.corrBufValue < int64(rcqItem.node.params.BufLimit) {
+			rcqItem.node.corrBufValue = int64(rcqItem.node.params.BufLimit)
+			cm.sumRecharge -= rcqItem.node.params.MinRecharge
+		}
+		lastUpdate += dtNext
+		cm.rcLastIntValue = rcqItem.intValue
+	}
+}
+
+func (cm *ClientManager) updateNodeRc(node *ClientNode, bvc int64, time mclock.AbsTime) {
+	cm.updateRecharge(time)
+	wasFull := true
+	if node.corrBufValue != int64(node.params.BufLimit) {
+		wasFull = false
+		node.corrBufValue += (cm.rcLastIntValue - node.rcLastIntValue) * int64(node.params.MinRecharge) / 1000000
+		if node.corrBufValue > int64(node.params.BufLimit) {
+			node.corrBufValue = int64(node.params.BufLimit)
+		}
+		node.rcLastIntValue = cm.rcLastIntValue
+	}
+	node.corrBufValue += bvc
+	if node.corrBufValue < 0 {
+		node.corrBufValue = 0
+	}
+	isFull := false
+	if node.corrBufValue >= int64(node.params.BufLimit) {
+		node.corrBufValue = int64(node.params.BufLimit)
+		isFull = true
+	}
+	if wasFull && !isFull {
+		cm.sumRecharge += node.params.MinRecharge
+	}
+	if !wasFull && isFull {
+		cm.sumRecharge -= node.params.MinRecharge
+	}
+	if !isFull {
+		node.rcLastIntValue = cm.rcLastIntValue
+		node.rcNextIntValue = cm.rcLastIntValue + (int64(node.params.BufLimit)-node.corrBufValue)*1000000/int64(node.params.MinRecharge)
+		cm.rcQueue.Push(rcQueueItem{node: node, intValue: node.rcNextIntValue})
+	}
+}
+
+// waitOrStop blocks while request processing is disabled and returns true if
+// it should be cancelled because the client has been disconnected.
+func (cm *ClientManager) waitOrStop(node *ClientNode) bool {
+	cm.lock.RLock()
+	_, ok := cm.nodes[node]
+	stop := !ok
+	ch := cm.enabledCh
+	cm.lock.RUnlock()
+
+	if !stop && ch != nil {
+		<-ch
+		cm.lock.RLock()
+		_, ok = cm.nodes[node]
+		stop = !ok
+		cm.lock.RUnlock()
+	}
+
+	return stop
+}
+
+func (cm *ClientManager) Stop() {
+	cm.lock.Lock()
+	defer cm.lock.Unlock()
+
+	cm.nodes = nil
+}
+
+func (cm *ClientManager) addNode(node *ClientNode) {
+	cm.lock.Lock()
+	defer cm.lock.Unlock()
+
+	node.corrBufValue = int64(node.params.BufLimit)
+	node.rcLastIntValue = cm.rcLastIntValue
+
+	if cm.nodes != nil {
+		cm.nodes[node] = struct{}{}
+	}
+}
+
+func (cm *ClientManager) removeNode(node *ClientNode) {
+	cm.lock.Lock()
+	defer cm.lock.Unlock()
+
+	if cm.nodes != nil {
+		delete(cm.nodes, node)
+	}
+}
+
+func (cm *ClientManager) accept(node *ClientNode, maxCost uint64, time mclock.AbsTime) chan bool {
+	cm.lock.Lock()
+	defer cm.lock.Unlock()
+
+	if cm.parallelReqs == cm.maxParallelReqs {
+		ch := make(chan bool, 1)
+		start := func() bool {
+			// always called while client manager lock is held
+			_, started := cm.nodes[node]
+			ch <- started
+			return started
+		}
+		cm.servingQueue.Push(servingQueueItem{start, float64(node.bufValue) / float64(node.params.BufLimit)})
+		return ch
+	}
+
+	cm.setParallelReqs(cm.parallelReqs+1, time)
+	node.servingStarted = time
+	node.servingMaxCost = maxCost
+	cm.updateNodeRc(node, -int64(maxCost), time)
+	return nil
+}
+
+func (cm *ClientManager) started(node *ClientNode, maxCost uint64, time mclock.AbsTime) {
+	cm.lock.Lock()
+	defer cm.lock.Unlock()
+
+	node.servingStarted = time
+	node.servingMaxCost = maxCost
+	cm.updateNodeRc(node, -int64(maxCost), time)
+}
+
+func (cm *ClientManager) processed(node *ClientNode, time mclock.AbsTime) (realCost uint64) {
+	cm.lock.Lock()
+	defer cm.lock.Unlock()
+
+	realCost = uint64(time - node.servingStarted)
+	if realCost > node.servingMaxCost {
+		realCost = node.servingMaxCost
+	}
+	if !cm.forceMinRecharge {
+		cm.updateNodeRc(node, int64(node.servingMaxCost-realCost), time)
+	}
+	if cm.bufCorrEnabled {
+		if uint64(node.corrBufValue) > node.bufValue {
+			node.bufValue = uint64(node.corrBufValue)
+		}
+	}
+
+	for !cm.servingQueue.Empty() {
+		if cm.servingQueue.Pop().(servingQueueItem).start() {
 			return
 		}
 	}
-}
+	cm.setParallelReqs(cm.parallelReqs-1, time)
-
+	return
-func (self *ClientManager) canStartReq() bool {
-	return self.simReqCnt < self.maxSimReq && self.rcSumValue < self.maxRcSum
-}
-
-func (self *ClientManager) queueProc() {
-	for rc := range self.resumeQueue {
-		for {
-			time.Sleep(time.Millisecond * 10)
-			self.lock.Lock()
-			self.update(mclock.Now())
-			cs := self.canStartReq()
-			self.lock.Unlock()
-			if cs {
-				break
-			}
-		}
-		close(rc)
-	}
-}
-
-func (self *ClientManager) accept(node *cmNode, time mclock.AbsTime) bool {
-	self.lock.Lock()
-	defer self.lock.Unlock()
-
-	self.update(time)
-	if !self.canStartReq() {
-		resume := make(chan bool)
-		self.lock.Unlock()
-		self.resumeQueue <- resume
-		<-resume
-		self.lock.Lock()
-		if _, ok := self.nodes[node]; !ok {
-			return false // reject if node has been removed or manager has been stopped
-		}
-	}
-	self.simReqCnt++
-	node.set(true, self.simReqCnt, self.sumWeight)
-	node.startValue = node.rcValue
-	self.update(self.time)
-	return true
-}
-
-func (self *ClientManager) stop(node *cmNode, time mclock.AbsTime) {
-	if node.serving {
-		self.update(time)
-		self.simReqCnt--
-		node.set(false, self.simReqCnt, self.sumWeight)
-		self.update(time)
-	}
-}
-
-func (self *ClientManager) processed(node *cmNode, time mclock.AbsTime) (rcValue, rcCost uint64) {
-	self.lock.Lock()
-	defer self.lock.Unlock()
-
-	self.stop(node, time)
-	return uint64(node.rcValue), uint64(node.rcValue - node.startValue)
 }

les/flowcontrol/prque/prque.go

@@ -0,0 +1,43 @@
+// This is a duplicated and slightly modified version of "gopkg.in/karalabe/cookiejar.v2/collections/prque".
+
+package prque
+
+import (
+	"container/heap"
+)
+
+// Priority queue data structure.
+type Prque struct {
+	cont *sstack
+}
+
+// Creates a new priority queue.
+func New() *Prque {
+	return &Prque{newSstack()}
+}
+
+// Pushes a value with a given priority into the queue, expanding if necessary.
+func (p *Prque) Push(i item) {
+	heap.Push(p.cont, i)
+}
+
+// Pops the value with the greates priority off the stack and returns it.
+// Currently no shrinking is done.
+func (p *Prque) Pop() item {
+	return heap.Pop(p.cont).(item)
+}
+
+// Checks whether the priority queue is empty.
+func (p *Prque) Empty() bool {
+	return p.cont.Len() == 0
+}
+
+// Returns the number of element in the priority queue.
+func (p *Prque) Size() int {
+	return p.cont.Len()
+}
+
+// Clears the contents of the priority queue.
+func (p *Prque) Reset() {
+	*p = *New()
+}

les/flowcontrol/prque/sstack.go

@@ -0,0 +1,84 @@
+// This is a duplicated and slightly modified version of "gopkg.in/karalabe/cookiejar.v2/collections/prque".
+
+package prque
+
+// The size of a block of data
+const blockSize = 4096
+
+// A prioritized item in the sorted stack.
+type item interface {
+	Before(interface{}) bool
+}
+
+// Internal sortable stack data structure. Implements the Push and Pop ops for
+// the stack (heap) functionality and the Len, Less and Swap methods for the
+// sortability requirements of the heaps.
+type sstack struct {
+	size     int
+	capacity int
+	offset   int
+
+	blocks [][]item
+	active []item
+}
+
+// Creates a new, empty stack.
+func newSstack() *sstack {
+	result := new(sstack)
+	result.active = make([]item, blockSize)
+	result.blocks = [][]item{result.active}
+	result.capacity = blockSize
+	return result
+}
+
+// Pushes a value onto the stack, expanding it if necessary. Required by
+// heap.Interface.
+func (s *sstack) Push(data interface{}) {
+	if s.size == s.capacity {
+		s.active = make([]item, blockSize)
+		s.blocks = append(s.blocks, s.active)
+		s.capacity += blockSize
+		s.offset = 0
+	} else if s.offset == blockSize {
+		s.active = s.blocks[s.size/blockSize]
+		s.offset = 0
+	}
+	s.active[s.offset] = data.(item)
+	s.offset++
+	s.size++
+}
+
+// Pops a value off the stack and returns it. Currently no shrinking is done.
+// Required by heap.Interface.
+func (s *sstack) Pop() (res interface{}) {
+	s.size--
+	s.offset--
+	if s.offset < 0 {
+		s.offset = blockSize - 1
+		s.active = s.blocks[s.size/blockSize]
+	}
+	res, s.active[s.offset] = s.active[s.offset], nil
+	return
+}
+
+// Returns the length of the stack. Required by sort.Interface.
+func (s *sstack) Len() int {
+	return s.size
+}
+
+// Compares the priority of two elements of the stack (higher is first).
+// Required by sort.Interface.
+func (s *sstack) Less(i, j int) bool {
+	return (s.blocks[i/blockSize][i%blockSize].Before(s.blocks[j/blockSize][j%blockSize]))
+}
+
+// Swaps two elements in the stack. Required by sort.Interface.
+func (s *sstack) Swap(i, j int) {
+	ib, io, jb, jo := i/blockSize, i%blockSize, j/blockSize, j%blockSize
+	s.blocks[ib][io], s.blocks[jb][jo] = s.blocks[jb][jo], s.blocks[ib][io]
+}
+
+// Resets the stack, effectively clearing its contents.
+func (s *sstack) Reset() {
+	*s = *newSstack()
+}

les/handler.go

@@ -292,7 +292,7 @@ func (pm *ProtocolManager) handle(p *peer) error {
 	}
 	defer func() {
 		if pm.server != nil && pm.server.fcManager != nil && p.fcClient != nil {
-			p.fcClient.Remove(pm.server.fcManager)
+			p.fcClient.Remove()
 		}
 		pm.removePeer(p.id)
 	}()
@@ -345,19 +345,20 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 	}
 	p.Log().Trace("Light Ethereum message arrived", "code", msg.Code, "bytes", msg.Size)
 
-	costs := p.fcCosts[msg.Code]
 	reject := func(reqCnt, maxCnt uint64) bool {
 		if p.fcClient == nil || reqCnt > maxCnt {
 			return true
 		}
-		bufValue, _ := p.fcClient.AcceptRequest()
+		costs := p.fcCosts[msg.Code]
 		cost := costs.baseCost + reqCnt*costs.reqCost
 		if cost > pm.server.defParams.BufLimit {
 			cost = pm.server.defParams.BufLimit
 		}
-		if cost > bufValue {
+
-			recharge := time.Duration((cost - bufValue) * 1000000 / pm.server.defParams.MinRecharge)
+		if accepted, bufShort := p.fcClient.AcceptRequest(cost); !accepted {
-			p.Log().Error("Request came too early", "recharge", common.PrettyDuration(recharge))
+			if bufShort > 0 {
+				p.Log().Error("Request came too early", "remaining", common.PrettyDuration(time.Duration(bufShort*1000000/pm.server.defParams.MinRecharge)))
+			}
 			return true
 		}
 		return false
@@ -429,6 +430,9 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 			unknown bool
 		)
 		for !unknown && len(headers) < int(query.Amount) && bytes < softResponseLimit {
+			if p.fcClient.WaitOrStop() {
+				return nil
+			}
 			// Retrieve the next header satisfying the query
 			var origin *types.Header
 			if hashMode {
@@ -498,7 +502,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 			}
 		}
 
-		bv, rcost := p.fcClient.RequestProcessed(costs.baseCost + query.Amount*costs.reqCost)
+		bv, rcost := p.fcClient.RequestProcessed()
 		pm.server.fcCostStats.update(msg.Code, query.Amount, rcost)
 		return p.SendBlockHeaders(req.ReqID, bv, headers)
 
@@ -516,7 +520,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 		if err := msg.Decode(&resp); err != nil {
 			return errResp(ErrDecode, "msg %v: %v", msg, err)
 		}
-		p.fcServer.GotReply(resp.ReqID, resp.BV)
+		p.fcServer.ReceivedReply(resp.ReqID, resp.BV)
 		if pm.fetcher != nil && pm.fetcher.requestedID(resp.ReqID) {
 			pm.fetcher.deliverHeaders(p, resp.ReqID, resp.Headers)
 		} else {
@@ -549,6 +553,9 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 			if bytes >= softResponseLimit {
 				break
 			}
+			if p.fcClient.WaitOrStop() {
+				return nil
+			}
 			// Retrieve the requested block body, stopping if enough was found
 			if number := rawdb.ReadHeaderNumber(pm.chainDb, hash); number != nil {
 				if data := rawdb.ReadBodyRLP(pm.chainDb, hash, *number); len(data) != 0 {
@@ -557,7 +564,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 				}
 			}
 		}
-		bv, rcost := p.fcClient.RequestProcessed(costs.baseCost + uint64(reqCnt)*costs.reqCost)
+		bv, rcost := p.fcClient.RequestProcessed()
 		pm.server.fcCostStats.update(msg.Code, uint64(reqCnt), rcost)
 		return p.SendBlockBodiesRLP(req.ReqID, bv, bodies)
 
@@ -575,7 +582,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 		if err := msg.Decode(&resp); err != nil {
 			return errResp(ErrDecode, "msg %v: %v", msg, err)
 		}
-		p.fcServer.GotReply(resp.ReqID, resp.BV)
+		p.fcServer.ReceivedReply(resp.ReqID, resp.BV)
 		deliverMsg = &Msg{
 			MsgType: MsgBlockBodies,
 			ReqID:   resp.ReqID,
@@ -602,6 +609,9 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 			return errResp(ErrRequestRejected, "")
 		}
 		for _, req := range req.Reqs {
+			if p.fcClient.WaitOrStop() {
+				return nil
+			}
 			// Retrieve the requested state entry, stopping if enough was found
 			if number := rawdb.ReadHeaderNumber(pm.chainDb, req.BHash); number != nil {
 				if header := rawdb.ReadHeader(pm.chainDb, req.BHash, *number); header != nil {
@@ -622,7 +632,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 				}
 			}
 		}
-		bv, rcost := p.fcClient.RequestProcessed(costs.baseCost + uint64(reqCnt)*costs.reqCost)
+		bv, rcost := p.fcClient.RequestProcessed()
 		pm.server.fcCostStats.update(msg.Code, uint64(reqCnt), rcost)
 		return p.SendCode(req.ReqID, bv, data)
 
@@ -640,7 +650,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 		if err := msg.Decode(&resp); err != nil {
 			return errResp(ErrDecode, "msg %v: %v", msg, err)
 		}
-		p.fcServer.GotReply(resp.ReqID, resp.BV)
+		p.fcServer.ReceivedReply(resp.ReqID, resp.BV)
 		deliverMsg = &Msg{
 			MsgType: MsgCode,
 			ReqID:   resp.ReqID,
@@ -670,6 +680,9 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 			if bytes >= softResponseLimit {
 				break
 			}
+			if p.fcClient.WaitOrStop() {
+				return nil
+			}
 			// Retrieve the requested block's receipts, skipping if unknown to us
 			var results types.Receipts
 			if number := rawdb.ReadHeaderNumber(pm.chainDb, hash); number != nil {
@@ -688,7 +701,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 				bytes += len(encoded)
 			}
 		}
-		bv, rcost := p.fcClient.RequestProcessed(costs.baseCost + uint64(reqCnt)*costs.reqCost)
+		bv, rcost := p.fcClient.RequestProcessed()
 		pm.server.fcCostStats.update(msg.Code, uint64(reqCnt), rcost)
 		return p.SendReceiptsRLP(req.ReqID, bv, receipts)
 
@@ -706,7 +719,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 		if err := msg.Decode(&resp); err != nil {
 			return errResp(ErrDecode, "msg %v: %v", msg, err)
 		}
-		p.fcServer.GotReply(resp.ReqID, resp.BV)
+		p.fcServer.ReceivedReply(resp.ReqID, resp.BV)
 		deliverMsg = &Msg{
 			MsgType: MsgReceipts,
 			ReqID:   resp.ReqID,
@@ -733,6 +746,9 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 			return errResp(ErrRequestRejected, "")
 		}
 		for _, req := range req.Reqs {
+			if p.fcClient.WaitOrStop() {
+				return nil
+			}
 			// Retrieve the requested state entry, stopping if enough was found
 			if number := rawdb.ReadHeaderNumber(pm.chainDb, req.BHash); number != nil {
 				if header := rawdb.ReadHeader(pm.chainDb, req.BHash, *number); header != nil {
@@ -762,7 +778,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 				}
 			}
 		}
-		bv, rcost := p.fcClient.RequestProcessed(costs.baseCost + uint64(reqCnt)*costs.reqCost)
+		bv, rcost := p.fcClient.RequestProcessed()
 		pm.server.fcCostStats.update(msg.Code, uint64(reqCnt), rcost)
 		return p.SendProofs(req.ReqID, bv, proofs)
 
@@ -790,6 +806,9 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 		nodes := light.NewNodeSet()
 
 		for _, req := range req.Reqs {
+			if p.fcClient.WaitOrStop() {
+				return nil
+			}
 			// Look up the state belonging to the request
 			if statedb == nil || req.BHash != lastBHash {
 				statedb, root, lastBHash = nil, common.Hash{}, req.BHash
@@ -824,7 +843,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 				break
 			}
 		}
-		bv, rcost := p.fcClient.RequestProcessed(costs.baseCost + uint64(reqCnt)*costs.reqCost)
+		bv, rcost := p.fcClient.RequestProcessed()
 		pm.server.fcCostStats.update(msg.Code, uint64(reqCnt), rcost)
 		return p.SendProofsV2(req.ReqID, bv, nodes.NodeList())
 
@@ -842,7 +861,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 		if err := msg.Decode(&resp); err != nil {
 			return errResp(ErrDecode, "msg %v: %v", msg, err)
 		}
-		p.fcServer.GotReply(resp.ReqID, resp.BV)
+		p.fcServer.ReceivedReply(resp.ReqID, resp.BV)
 		deliverMsg = &Msg{
 			MsgType: MsgProofsV1,
 			ReqID:   resp.ReqID,
@@ -863,7 +882,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 		if err := msg.Decode(&resp); err != nil {
 			return errResp(ErrDecode, "msg %v: %v", msg, err)
 		}
-		p.fcServer.GotReply(resp.ReqID, resp.BV)
+		p.fcServer.ReceivedReply(resp.ReqID, resp.BV)
 		deliverMsg = &Msg{
 			MsgType: MsgProofsV2,
 			ReqID:   resp.ReqID,
@@ -891,6 +910,9 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 		}
 		trieDb := trie.NewDatabase(ethdb.NewTable(pm.chainDb, light.ChtTablePrefix))
 		for _, req := range req.Reqs {
+			if p.fcClient.WaitOrStop() {
+				return nil
+			}
 			if header := pm.blockchain.GetHeaderByNumber(req.BlockNum); header != nil {
 				sectionHead := rawdb.ReadCanonicalHash(pm.chainDb, req.ChtNum*light.CHTFrequencyServer-1)
 				if root := light.GetChtRoot(pm.chainDb, req.ChtNum-1, sectionHead); root != (common.Hash{}) {
@@ -911,7 +933,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 				}
 			}
 		}
-		bv, rcost := p.fcClient.RequestProcessed(costs.baseCost + uint64(reqCnt)*costs.reqCost)
+		bv, rcost := p.fcClient.RequestProcessed()
 		pm.server.fcCostStats.update(msg.Code, uint64(reqCnt), rcost)
 		return p.SendHeaderProofs(req.ReqID, bv, proofs)
 
@@ -943,6 +965,9 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 		)
 		nodes := light.NewNodeSet()
 		for _, req := range req.Reqs {
+			if p.fcClient.WaitOrStop() {
+				return nil
+			}
 			if auxTrie == nil || req.Type != lastType || req.TrieIdx != lastIdx {
 				auxTrie, lastType, lastIdx = nil, req.Type, req.TrieIdx
 
@@ -972,7 +997,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 				break
 			}
 		}
-		bv, rcost := p.fcClient.RequestProcessed(costs.baseCost + uint64(reqCnt)*costs.reqCost)
+		bv, rcost := p.fcClient.RequestProcessed()
 		pm.server.fcCostStats.update(msg.Code, uint64(reqCnt), rcost)
 		return p.SendHelperTrieProofs(req.ReqID, bv, HelperTrieResps{Proofs: nodes.NodeList(), AuxData: auxData})
 
@@ -989,7 +1014,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 		if err := msg.Decode(&resp); err != nil {
 			return errResp(ErrDecode, "msg %v: %v", msg, err)
 		}
-		p.fcServer.GotReply(resp.ReqID, resp.BV)
+		p.fcServer.ReceivedReply(resp.ReqID, resp.BV)
 		deliverMsg = &Msg{
 			MsgType: MsgHeaderProofs,
 			ReqID:   resp.ReqID,
@@ -1010,7 +1035,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 			return errResp(ErrDecode, "msg %v: %v", msg, err)
 		}
 
-		p.fcServer.GotReply(resp.ReqID, resp.BV)
+		p.fcServer.ReceivedReply(resp.ReqID, resp.BV)
 		deliverMsg = &Msg{
 			MsgType: MsgHelperTrieProofs,
 			ReqID:   resp.ReqID,
@@ -1032,7 +1057,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 		}
 		pm.txpool.AddRemotes(txs)
 
-		_, rcost := p.fcClient.RequestProcessed(costs.baseCost + uint64(reqCnt)*costs.reqCost)
+		_, rcost := p.fcClient.RequestProcessed()
 		pm.server.fcCostStats.update(msg.Code, uint64(reqCnt), rcost)
 
 	case SendTxV2Msg:
@@ -1058,6 +1083,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 		}
 		stats := pm.txStatus(hashes)
 		for i, stat := range stats {
+			p.fcClient.WaitOrStop() // do not return; txs can be added even if client disconnected
 			if stat.Status == core.TxStatusUnknown {
 				if errs := pm.txpool.AddRemotes([]*types.Transaction{req.Txs[i]}); errs[0] != nil {
 					stats[i].Error = errs[0].Error()
@@ -1067,7 +1093,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 			}
 		}
 
-		bv, rcost := p.fcClient.RequestProcessed(costs.baseCost + uint64(reqCnt)*costs.reqCost)
+		bv, rcost := p.fcClient.RequestProcessed()
 		pm.server.fcCostStats.update(msg.Code, uint64(reqCnt), rcost)
 
 		return p.SendTxStatus(req.ReqID, bv, stats)
@@ -1088,7 +1114,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 		if reject(uint64(reqCnt), MaxTxStatus) {
 			return errResp(ErrRequestRejected, "")
 		}
-		bv, rcost := p.fcClient.RequestProcessed(costs.baseCost + uint64(reqCnt)*costs.reqCost)
+		bv, rcost := p.fcClient.RequestProcessed()
 		pm.server.fcCostStats.update(msg.Code, uint64(reqCnt), rcost)
 
 		return p.SendTxStatus(req.ReqID, bv, pm.txStatus(req.Hashes))
@@ -1107,7 +1133,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
 			return errResp(ErrDecode, "msg %v: %v", msg, err)
 		}
 
-		p.fcServer.GotReply(resp.ReqID, resp.BV)
+		p.fcServer.ReceivedReply(resp.ReqID, resp.BV)
 
 	default:
 		p.Log().Trace("Received unknown message", "code", msg.Code)
@@ -1233,7 +1259,7 @@ func (pc *peerConnection) RequestHeadersByHash(origin common.Hash, amount int, s
 		request: func(dp distPeer) func() {
 			peer := dp.(*peer)
 			cost := peer.GetRequestCost(GetBlockHeadersMsg, amount)
-			peer.fcServer.QueueRequest(reqID, cost)
+			peer.fcServer.QueuedRequest(reqID, cost)
 			return func() { peer.RequestHeadersByHash(reqID, cost, origin, amount, skip, reverse) }
 		},
 	}
@@ -1257,7 +1283,7 @@ func (pc *peerConnection) RequestHeadersByNumber(origin uint64, amount int, skip
 		request: func(dp distPeer) func() {
 			peer := dp.(*peer)
 			cost := peer.GetRequestCost(GetBlockHeadersMsg, amount)
-			peer.fcServer.QueueRequest(reqID, cost)
+			peer.fcServer.QueuedRequest(reqID, cost)
 			return func() { peer.RequestHeadersByNumber(reqID, cost, origin, amount, skip, reverse) }
 		},
 	}

les/helper_test.go

@@ -26,6 +26,7 @@ import (
 	"testing"
 
 	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/common/mclock"
 	"github.com/ethereum/go-ethereum/consensus/ethash"
 	"github.com/ethereum/go-ethereum/core"
 	"github.com/ethereum/go-ethereum/core/types"
@@ -191,7 +192,8 @@ func newTestProtocolManager(lightSync bool, blocks int, generator func(int, *cor
 			MinRecharge: 1,
 		}
 
-		srv.fcManager = flowcontrol.NewClientManager(50, 10, 1000000000)
+		//srv.fcManager = flowcontrol.NewClientManager(50, 10, 1000000000)
+		srv.fcManager = flowcontrol.NewClientManager(16, 4, &mclock.MonotonicClock{}, nil)
 		srv.fcCostStats = newCostStats(nil)
 	}
 	pm.Start(1000)

les/load_test.go

@@ -0,0 +1,422 @@
+// Copyright 2017 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+// Package light implements on-demand retrieval capable state and chain objects
+// for the Ethereum Light Client.
+package les
+
+import (
+	"math/rand"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/ethereum/go-ethereum/common/mclock"
+	"github.com/ethereum/go-ethereum/les/flowcontrol"
+)
+
+type testLoadPeer struct {
+	fcClient *flowcontrol.ClientNode
+	fcServer *flowcontrol.ServerNode
+	serveCh  chan uint64
+}
+
+func newTestLoadPeer(t *testing.T, client *testLoadClient, server *testLoadServer, params *flowcontrol.ServerParams, free bool, quit chan struct{}, clock mclock.Clock) *testLoadPeer {
+	cm := server.fcManager
+	if free {
+		cm = server.fcManagerFree
+	}
+
+	peer := &testLoadPeer{
+		fcClient: flowcontrol.NewClientNode(cm, params),
+		fcServer: flowcontrol.NewServerNode(params, clock),
+		serveCh:  make(chan uint64, 1000),
+	}
+	client.dist.registerTestPeer(peer)
+
+	type reply struct {
+		time      mclock.AbsTime
+		reqID, bv uint64
+	}
+	replyCh := make(chan reply, 1000)
+
+	go func() {
+		avgServeTime := time.Millisecond
+		randMax := avgServeTime * 2
+
+		for {
+			select {
+			case reqID := <-peer.serveCh:
+				ok, bufShort := peer.fcClient.AcceptRequest(testRequestCost)
+				if !ok {
+					recharge := time.Duration(bufShort * 1000000 / params.MinRecharge)
+					t.Errorf("Request came too early (%v)", recharge)
+				}
+				//testClock.Sleep(time.Microsecond * 500)
+				serveTime := time.Duration(rand.Int63n(int64(randMax)))
+				randMax += (avgServeTime - serveTime) / 16
+				clock.Sleep(serveTime)
+				bvAfter, _ := peer.fcClient.RequestProcessed()
+				replyCh <- reply{clock.Now(), reqID, bvAfter}
+				atomic.AddUint64(&server.count, 1)
+			case <-quit:
+				return
+			}
+		}
+	}()
+
+	go func() {
+		for {
+			select {
+			case reply := <-replyCh:
+				wait := time.Duration(reply.time-clock.Now()) + testMessageDelay
+				if wait > 0 {
+					clock.Sleep(wait)
+				}
+				peer.fcServer.ReceivedReply(reply.reqID, reply.bv)
+			case <-quit:
+				return
+			}
+		}
+	}()
+
+	return peer
+}
+
+func (p *testLoadPeer) waitBefore(maxCost uint64) (time.Duration, float64) {
+	return p.fcServer.CanSend(maxCost)
+}
+
+func (p *testLoadPeer) canQueue() bool {
+	return true
+}
+
+func (p *testLoadPeer) queueSend(f func()) {
+	f()
+}
+
+type testLoadTask struct {
+	procTime time.Duration
+	finished chan struct{}
+}
+
+type testLoadServer struct {
+	fcManager, fcManagerFree *flowcontrol.ClientManager
+	count                    uint64
+}
+
+func newTestLoadServer(capacity int, quit chan struct{}, clock mclock.Clock) *testLoadServer {
+	f := flowcontrol.NewClientManager(16, float64(capacity)/1000, clock, nil)
+	s := &testLoadServer{
+		fcManager:     flowcontrol.NewClientManager(16, float64(capacity)/1000, clock, f),
+		fcManagerFree: f,
+	}
+	go func() {
+		<-quit
+		s.fcManager.Stop()
+		s.fcManagerFree.Stop()
+	}()
+	return s
+}
+
+func (s *testLoadServer) requestsProcessed() uint64 {
+	return atomic.LoadUint64(&s.count)
+}
+
+type testLoadClient struct {
+	dist  *requestDistributor
+	quit  chan struct{}
+	count uint64
+}
+
+func newTestLoadClient(quit chan struct{}, clock mclock.Clock) *testLoadClient {
+	return &testLoadClient{
+		dist: newRequestDistributor(nil, quit, clock),
+		quit: quit,
+	}
+}
+
+func (c *testLoadClient) sendRequests(send bool, sw chan bool) {
+	expCh := make(chan struct{}, 100)
+	for {
+		if send {
+			select {
+			case send = <-sw:
+			case expCh <- struct{}{}:
+				reqID := genReqID()
+				rq := &distReq{
+					getCost: func(dp distPeer) uint64 {
+						return testRequestCost
+					},
+					canSend: func(dp distPeer) bool {
+						return true
+					},
+					request: func(dp distPeer) func() {
+						peer := dp.(*testLoadPeer)
+						peer.fcServer.QueuedRequest(reqID, testRequestCost)
+						return func() {
+							peer.serveCh <- reqID
+						}
+					},
+				}
+
+				sentCh := c.dist.queue(rq)
+				go func() {
+					<-sentCh
+					atomic.AddUint64(&c.count, 1)
+					<-expCh
+				}()
+			case <-c.quit:
+				return
+			}
+		} else {
+			select {
+			case send = <-sw:
+			case <-c.quit:
+				return
+			}
+		}
+	}
+}
+
+func (c *testLoadClient) requestsSent() uint64 {
+	return atomic.LoadUint64(&c.count)
+}
+
+const (
+	testRequestCost = 3000000
+	/*testClientCount  = 2
+	testServerCount  = 2*/
+	testMessageDelay = time.Millisecond * 200
+	defaultTolerance = 5 // percent
+)
+
+type testServerPeriod struct {
+	mode                  int
+	measureOff, measureOn int // duration in milliseconds
+	expResult, tolerance  uint64
+}
+
+type testConnection struct {
+	minCapacity int // request per second guaranteed by MRR
+	free        bool
+}
+
+type testServerParams struct {
+	capacity int // processing request per second
+	periods  []testServerPeriod
+}
+
+type testClientPeriod struct {
+	sendOff, measureOff, measureOn int // duration in milliseconds
+	expResult, tolerance           uint64
+}
+
+type testClientParams struct {
+	periods []testClientPeriod
+	servers []testConnection
+}
+
+func testLoad(t *testing.T, serverParams []testServerParams, clientParams []testClientParams) {
+	quit := make(chan struct{})
+	defer close(quit)
+
+	//clock := &mclock.MonotonicClock{}
+	clock := mclock.NewSimulatedClock()
+	defer clock.Stop()
+
+	var wg sync.WaitGroup
+
+	servers := make([]*testLoadServer, len(serverParams))
+	for i, params := range serverParams {
+		i, params := i, params
+		servers[i] = newTestLoadServer(params.capacity, quit, clock)
+		wg.Add(1)
+		go func() {
+			for k, p := range params.periods {
+				servers[i].fcManager.SetMode(p.mode)
+				clock.Sleep(time.Millisecond * time.Duration(p.measureOff))
+				start := servers[i].requestsProcessed()
+				clock.Sleep(time.Millisecond * time.Duration(p.measureOn))
+				result := servers[i].requestsProcessed() - start
+				relTol := p.tolerance
+				if relTol == 0 {
+					relTol = defaultTolerance
+				}
+				tolerance := p.expResult * relTol / 100
+				expMin := p.expResult - tolerance
+				expMax := p.expResult + tolerance
+				if result < expMin || result > expMax {
+					t.Errorf("servers[%d].periods[%d] processed count mismatch (processed %d, expected between %d and %d)", i, k, result, expMin, expMax)
+				}
+			}
+			wg.Done()
+		}()
+	}
+
+	clients := make([]*testLoadClient, len(clientParams))
+	for i, params := range clientParams {
+		i, params := i, params
+		clients[i] = newTestLoadClient(quit, clock)
+		for j, conn := range params.servers {
+			if conn.minCapacity > 0 {
+				params := &flowcontrol.ServerParams{
+					BufLimit:    18000000 * uint64(conn.minCapacity),
+					MinRecharge: 3000 * uint64(conn.minCapacity),
+				}
+				newTestLoadPeer(t, clients[i], servers[j], params, conn.free, quit, clock)
+			}
+		}
+		sw := make(chan bool)
+		go clients[i].sendRequests(false, sw)
+		wg.Add(1)
+		go func() {
+			for k, p := range params.periods {
+				clock.Sleep(time.Millisecond * time.Duration(p.sendOff))
+				sw <- true
+				clock.Sleep(time.Millisecond * time.Duration(p.measureOff))
+				start := clients[i].requestsSent()
+				clock.Sleep(time.Millisecond * time.Duration(p.measureOn))
+				result := clients[i].requestsSent() - start
+				relTol := p.tolerance
+				if relTol == 0 {
+					relTol = defaultTolerance
+				}
+				tolerance := p.expResult * relTol / 100
+				expMin := p.expResult - tolerance
+				expMax := p.expResult + tolerance
+				if result < expMin || result > expMax {
+					t.Errorf("clients[%d].periods[%d] sent count mismatch (sent %d, expected between %d and %d)", i, k, result, expMin, expMax)
+				}
+				sw <- false
+			}
+			wg.Done()
+		}()
+	}
+
+	wg.Wait()
+}
+
+func TestLoadBalance(t *testing.T) {
+	testLoad(t,
+		[]testServerParams{
+			{capacity: 1000, periods: []testServerPeriod{{mode: 1, measureOff: 3000, measureOn: 5000, expResult: 5000}}},
+		},
+		[]testClientParams{
+			{[]testClientPeriod{{sendOff: 0, measureOff: 3000, measureOn: 5000, expResult: 1000}}, []testConnection{{minCapacity: 30}}},
+			{[]testClientPeriod{{sendOff: 0, measureOff: 3000, measureOn: 5000, expResult: 1000}}, []testConnection{{minCapacity: 30}}},
+			{[]testClientPeriod{{sendOff: 0, measureOff: 3000, measureOn: 5000, expResult: 1000}}, []testConnection{{minCapacity: 30}}},
+			{[]testClientPeriod{{sendOff: 0, measureOff: 3000, measureOn: 5000, expResult: 2000}}, []testConnection{{minCapacity: 60}}},
+		})
+}
+
+func TestLoadBalanceMultiServer1(t *testing.T) {
+	testLoad(t,
+		[]testServerParams{
+			{capacity: 200, periods: []testServerPeriod{{mode: 1, measureOff: 3000, measureOn: 5000, expResult: 1000}}},
+			{capacity: 200, periods: []testServerPeriod{{mode: 1, measureOff: 3000, measureOn: 5000, expResult: 1000}}},
+			{capacity: 200, periods: []testServerPeriod{{mode: 1, measureOff: 3000, measureOn: 5000, expResult: 1000}}},
+			{capacity: 400, periods: []testServerPeriod{{mode: 1, measureOff: 3000, measureOn: 5000, expResult: 2000}}},
+		},
+		[]testClientParams{
+			{[]testClientPeriod{{sendOff: 0, measureOff: 3000, measureOn: 5000, expResult: 1000}}, []testConnection{{minCapacity: 30}, {minCapacity: 30}, {minCapacity: 30}, {minCapacity: 30}}},
+			{[]testClientPeriod{{sendOff: 0, measureOff: 3000, measureOn: 5000, expResult: 1000}}, []testConnection{{minCapacity: 30}, {minCapacity: 30}, {minCapacity: 30}, {minCapacity: 30}}},
+			{[]testClientPeriod{{sendOff: 0, measureOff: 3000, measureOn: 5000, expResult: 1000}}, []testConnection{{minCapacity: 30}, {minCapacity: 30}, {minCapacity: 30}, {minCapacity: 30}}},
+			{[]testClientPeriod{{sendOff: 0, measureOff: 3000, measureOn: 5000, expResult: 2000}}, []testConnection{{minCapacity: 60}, {minCapacity: 60}, {minCapacity: 60}, {minCapacity: 60}}},
+		})
+}
+
+func TestLoadBalanceMultiServer2(t *testing.T) {
+	testLoad(t,
+		[]testServerParams{
+			{capacity: 250, periods: []testServerPeriod{{mode: 1, measureOff: 3000, measureOn: 5000, expResult: 1250}}},
+			{capacity: 250, periods: []testServerPeriod{{mode: 1, measureOff: 3000, measureOn: 5000, expResult: 1250}}},
+			{capacity: 250, periods: []testServerPeriod{{mode: 1, measureOff: 3000, measureOn: 5000, expResult: 1250}}},
+			{capacity: 250, periods: []testServerPeriod{{mode: 1, measureOff: 3000, measureOn: 5000, expResult: 1250}}},
+		},
+		[]testClientParams{
+			{[]testClientPeriod{{sendOff: 0, measureOff: 3000, measureOn: 5000, expResult: 1000}}, []testConnection{{minCapacity: 60}, {minCapacity: 60}, {}, {}}},
+			{[]testClientPeriod{{sendOff: 0, measureOff: 3000, measureOn: 5000, expResult: 1000}}, []testConnection{{}, {}, {minCapacity: 60}, {minCapacity: 60}}},
+			{[]testClientPeriod{{sendOff: 0, measureOff: 3000, measureOn: 5000, expResult: 1000}}, []testConnection{{minCapacity: 10}, {minCapacity: 10}, {minCapacity: 10}, {minCapacity: 90}}},
+			{[]testClientPeriod{{sendOff: 0, measureOff: 3000, measureOn: 5000, expResult: 2000}}, []testConnection{{minCapacity: 80}, {minCapacity: 80}, {minCapacity: 80}, {}}},
+		})
+}
+
+func TestLoadSingle1(t *testing.T) {
+	testLoad(t,
+		[]testServerParams{
+			{capacity: 2000, periods: []testServerPeriod{{mode: 1, measureOff: 3000, measureOn: 5000, expResult: 4500, tolerance: 20}}},
+		},
+		[]testClientParams{
+			{[]testClientPeriod{{sendOff: 0, measureOff: 3000, measureOn: 5000, expResult: 4500, tolerance: 20}}, []testConnection{{minCapacity: 30}}},
+		})
+}
+
+func TestLoadSingle2(t *testing.T) {
+	testLoad(t,
+		[]testServerParams{
+			{capacity: 2000, periods: []testServerPeriod{{mode: 1, measureOff: 3000, measureOn: 5000, expResult: 5000}}},
+		},
+		[]testClientParams{
+			{[]testClientPeriod{{sendOff: 0, measureOff: 3000, measureOn: 5000, expResult: 5000}}, []testConnection{{minCapacity: 100}}},
+		})
+}
+
+func TestLoadSingle3(t *testing.T) {
+	testLoad(t,
+		[]testServerParams{
+			{capacity: 500, periods: []testServerPeriod{{mode: 1, measureOff: 3000, measureOn: 5000, expResult: 2500}}},
+		},
+		[]testClientParams{
+			{[]testClientPeriod{{sendOff: 0, measureOff: 3000, measureOn: 5000, expResult: 2500}}, []testConnection{{minCapacity: 30}}},
+		})
+}
+
+func TestLoadPriority(t *testing.T) {
+	testLoad(t,
+		[]testServerParams{
+			{capacity: 500, periods: []testServerPeriod{
+				{mode: 1, measureOff: 3000, measureOn: 5000, expResult: 2500},
+				{mode: 1, measureOff: 3000, measureOn: 5000, expResult: 2500},
+				{mode: 1, measureOff: 3000, measureOn: 5000, expResult: 3750},
+				{mode: 1, measureOff: 3000, measureOn: 5000, expResult: 2500},
+			}},
+		},
+		[]testClientParams{
+			// paying client
+			{[]testClientPeriod{
+				{sendOff: 0, measureOff: 3000, measureOn: 5000, expResult: 2500},
+				{sendOff: 8000, measureOff: 3000, measureOn: 5000, expResult: 2500},
+			}, []testConnection{{minCapacity: 30}}},
+			// free client
+			{[]testClientPeriod{
+				{sendOff: 8000, measureOff: 3000, measureOn: 5000, expResult: 2500},
+				{sendOff: 0, measureOff: 3000, measureOn: 5000, expResult: 1250},
+				{sendOff: 0, measureOff: 3000, measureOn: 5000, expResult: 2500},
+			}, []testConnection{{minCapacity: 30, free: true}}},
+		})
+}
+
+func TestLoadMinRecharge(t *testing.T) {
+	testLoad(t,
+		[]testServerParams{
+			{capacity: 500, periods: []testServerPeriod{{mode: 2, measureOff: 3000, measureOn: 5000, expResult: 150}}},
+		},
+		[]testClientParams{
+			{[]testClientPeriod{{sendOff: 0, measureOff: 3000, measureOn: 5000, expResult: 150}}, []testConnection{{minCapacity: 30}}},
+		})
+}

les/odr.go

@@ -103,7 +103,7 @@ func (odr *LesOdr) Retrieve(ctx context.Context, req light.OdrRequest) (err erro
 		request: func(dp distPeer) func() {
 			p := dp.(*peer)
 			cost := lreq.GetCost(p)
-			p.fcServer.QueueRequest(reqID, cost)
+			p.fcServer.QueuedRequest(reqID, cost)
 			return func() { lreq.Request(reqID, p) }
 		},
 	}

les/odr_test.go

@@ -25,6 +25,7 @@ import (
 
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/common/math"
+	"github.com/ethereum/go-ethereum/common/mclock"
 	"github.com/ethereum/go-ethereum/core"
 	"github.com/ethereum/go-ethereum/core/rawdb"
 	"github.com/ethereum/go-ethereum/core/state"
@@ -163,7 +164,7 @@ func odrContractCall(ctx context.Context, db ethdb.Database, config *params.Chai
 func testOdr(t *testing.T, protocol int, expFail uint64, fn odrTestFn) {
 	// Assemble the test environment
 	peers := newPeerSet()
-	dist := newRequestDistributor(peers, make(chan struct{}))
+	dist := newRequestDistributor(peers, make(chan struct{}), &mclock.MonotonicClock{})
 	rm := newRetrieveManager(peers, dist, nil)
 	db := ethdb.NewMemDatabase()
 	ldb := ethdb.NewMemDatabase()

les/peer.go

@@ -27,6 +27,7 @@ import (
 	"time"
 
 	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/common/mclock"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/eth"
 	"github.com/ethereum/go-ethereum/les/flowcontrol"
@@ -487,7 +488,7 @@ func (p *peer) Handshake(td *big.Int, head common.Hash, headNum uint64, genesis
 			return err
 		}
 		p.fcServerParams = params
-		p.fcServer = flowcontrol.NewServerNode(params)
+		p.fcServer = flowcontrol.NewServerNode(params, &mclock.MonotonicClock{})
 		p.fcCosts = MRC.decode()
 	}
 

les/request_test.go

@@ -22,6 +22,7 @@ import (
 	"time"
 
 	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/common/mclock"
 	"github.com/ethereum/go-ethereum/core/rawdb"
 	"github.com/ethereum/go-ethereum/crypto"
 	"github.com/ethereum/go-ethereum/eth"
@@ -85,7 +86,7 @@ func tfCodeAccess(db ethdb.Database, bhash common.Hash, num uint64) light.OdrReq
 func testAccess(t *testing.T, protocol int, fn accessTestFn) {
 	// Assemble the test environment
 	peers := newPeerSet()
-	dist := newRequestDistributor(peers, make(chan struct{}))
+	dist := newRequestDistributor(peers, make(chan struct{}), &mclock.MonotonicClock{})
 	rm := newRetrieveManager(peers, dist, nil)
 	db := ethdb.NewMemDatabase()
 	ldb := ethdb.NewMemDatabase()

les/server.go

@@ -24,6 +24,7 @@ import (
 	"sync"
 
 	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/common/mclock"
 	"github.com/ethereum/go-ethereum/core"
 	"github.com/ethereum/go-ethereum/core/rawdb"
 	"github.com/ethereum/go-ethereum/core/types"
@@ -98,7 +99,12 @@ func NewLesServer(eth *eth.Ethereum, config *eth.Config) (*LesServer, error) {
 		BufLimit:    300000000,
 		MinRecharge: 50000,
 	}
-	srv.fcManager = flowcontrol.NewClientManager(uint64(config.LightServ), 10, 1000000000)
+	tpr := float64(config.LightServ) / 100
+	mpr := int(tpr * 4)
+	if mpr < 4 {
+		mpr = 4
+	}
+	srv.fcManager = flowcontrol.NewClientManager(mpr, tpr, &mclock.MonotonicClock{}, nil)
 	srv.fcCostStats = newCostStats(eth.ChainDb())
 	return srv, nil
 }

les/txrelay.go

@@ -126,7 +126,7 @@ func (self *LesTxRelay) send(txs types.Transactions, count int) {
 			request: func(dp distPeer) func() {
 				peer := dp.(*peer)
 				cost := peer.GetRequestCost(SendTxMsg, len(ll))
-				peer.fcServer.QueueRequest(reqID, cost)
+				peer.fcServer.QueuedRequest(reqID, cost)
 				return func() { peer.SendTxs(reqID, cost, ll) }
 			},
 		}