diff --git a/cmd/utils/flags.go b/cmd/utils/flags.go
index d6142f246c..52ea7968ac 100644
--- a/cmd/utils/flags.go
+++ b/cmd/utils/flags.go
@@ -178,7 +178,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{
diff --git a/common/mclock/mclock.go b/common/mclock/mclock.go
index 02608d17b0..76d1e5cdb7 100644
--- a/common/mclock/mclock.go
+++ b/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)
+}
diff --git a/common/mclock/simclock.go b/common/mclock/simclock.go
new file mode 100644
index 0000000000..1b3611f6c1
--- /dev/null
+++ b/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 .
+
+// 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}
+}
diff --git a/core/blockchain.go b/core/blockchain.go
index 62dc261250..c19a8109bb 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()
@@ -1013,6 +1022,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/backend.go b/les/backend.go
index 178bc1e0e4..ff278fa107 100644
--- a/les/backend.go
+++ b/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, chainConfig, &config.Ethash, nil, chainDb),
shutdownChan: make(chan bool),
diff --git a/les/distributor.go b/les/distributor.go
index 159fa4c73f..149f9e2851 100644
--- a/les/distributor.go
+++ b/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
diff --git a/les/distributor_test.go b/les/distributor_test.go
index 2891bcab49..d3a1c901bc 100644
--- a/les/distributor_test.go
+++ b/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{}
diff --git a/les/fetcher.go b/les/fetcher.go
index cc539c42bf..c6b46ed837 100644
--- a/les/fetcher.go
+++ b/les/fetcher.go
@@ -481,7 +481,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()
diff --git a/les/flowcontrol/control.go b/les/flowcontrol/control.go
index d50eb809cc..7abac30243 100644
--- a/les/flowcontrol/control.go
+++ b/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) {
- cm.removeNode(peer.cmNode)
+// Remove removes the client from the client manager
+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)
-func (peer *ClientNode) RequestProcessed(cost uint64) (bv, realCost uint64) {
- peer.lock.Lock()
- defer peer.lock.Unlock()
-
- time := mclock.Now()
- peer.recalcBV(time)
- peer.bufValue -= cost
- 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
+ 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
+}
+
+// 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 := peer.cm.clock.Now()
+ peer.recalcBV(time)
+ rcost := peer.cm.processed(peer, time)
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
}
- peer.bufEstimate += peer.params.MinRecharge * dt / uint64(fcTimeConst)
- if peer.bufEstimate > peer.params.BufLimit {
- peer.bufEstimate = peer.params.BufLimit
+ if peer.bufRecharge {
+ dt := uint64(time - peer.lastTime)
+ peer.bufEstimate += peer.params.MinRecharge * dt / uint64(fcTimeConst)
+ 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.
-func (peer *ServerNode) QueueRequest(reqID, maxCost uint64) {
+// are sent in the same order as the QueuedRequest calls are made.
+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()
}
diff --git a/les/flowcontrol/manager.go b/les/flowcontrol/manager.go
index 28cc6f0fe7..242b07ad2a 100644
--- a/les/flowcontrol/manager.go
+++ b/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 {
- node *ClientNode
- lastUpdate mclock.AbsTime
- serving, recharging bool
- rcWeight uint64
- rcValue, rcDelta, startValue int64
- finishRecharge mclock.AbsTime
+// 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
}
-func (node *cmNode) update(time mclock.AbsTime) {
- dt := int64(time - node.lastUpdate)
- node.rcValue += node.rcDelta * dt / rcConst
- node.lastUpdate = time
- if node.recharging && time >= node.finishRecharge {
- node.recharging = false
- node.rcDelta = 0
- node.rcValue = 0
- }
+// rcQueueItem represents an integrator threshold value where a certain client's buffer is recharged
+type rcQueueItem struct {
+ node *ClientNode
+ intValue int64
}
-func (node *cmNode) set(serving bool, simReqCnt, sumWeight uint64) {
- if node.serving && !serving {
- node.recharging = true
- 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))
- }
+// Before implements prque.item
+//
+// Note: intValue is interpreted as mod 2^64, the difference between the highest
+// and lowest value at any moment is always less than 2^63.
+func rcQueueCompare(i, j interface{}) bool {
+ return (j.(rcQueueItem).intValue - i.(rcQueueItem).intValue) > 0
}
+// Note: valid is called under client manager mutex lock
+func (rcq rcQueueItem) valid() bool {
+ return rcq.intValue == rcq.node.rcNextIntValue
+}
+
+// servingQueueItem represents a queued request (prioritized by BufValue/BufLimit)
+type servingQueueItem struct {
+ start func() bool
+ priority float64
+}
+
+// Before implements prque.item
+func servingQueueCompare(i, j interface{}) bool {
+ return i.(servingQueueItem).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
- nodes map[*cmNode]struct{}
- simReqCnt, sumWeight, rcSumValue uint64
- maxSimReq, maxRcSum uint64
- rcRecharge uint64
- resumeQueue chan chan bool
- time mclock.AbsTime
+ clock mclock.Clock
+ child *ClientManager
+ lock sync.RWMutex
+ nodes map[*ClientNode]struct{}
+ enabledCh chan struct{}
+
+ 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{}),
- resumeQueue: make(chan chan bool),
- rcRecharge: rcConst * rcConst / (100*rcConst/rcTarget - rcConst),
- maxSimReq: maxSimReq,
- maxRcSum: maxRcSum,
+ clock: clock,
+ nodes: make(map[*ClientNode]struct{}),
+ child: child,
+ servingQueue: prque.New(servingQueueCompare),
+ rcQueue: prque.New(rcQueueCompare),
+
+ maxParallelReqs: maxParallelReqs,
+ targetParallelReqs: targetParallelReqs,
}
- go cm.queueProc()
+ cm.SetMode(CmNormal)
return cm
}
-func (self *ClientManager) Stop() {
- self.lock.Lock()
- defer self.lock.Unlock()
+// SetMode changes the operating mode of the manager and its children. When
+// multiple priority levels are used, mode should be changed at the manager
+// 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
- 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,
+ if newMode == cm.mode {
+ return
}
- self.lock.Lock()
- defer self.lock.Unlock()
+ cm.updateRecharge(cm.clock.Now())
- 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)
+ enabled := cm.mode != CmDisabled
+ newEnabled := cm.mode != CmDisabled
+ if !enabled && newEnabled && cm.enabledCh != nil {
+ close(cm.enabledCh)
+ cm.enabledCh = nil
+ }
+ if enabled && !newEnabled {
+ cm.enabledCh = make(chan struct{})
}
- self.sumWeight = sumWeight
- self.rcSumValue = rcSum
- return
-}
-func (self *ClientManager) update(time mclock.AbsTime) {
- for {
- firstTime := time
- for node := range self.nodes {
- if node.recharging && node.finishRecharge < firstTime {
- firstTime = node.finishRecharge
- }
- }
- if self.updateNodes(firstTime) {
- for node := range self.nodes {
- if node.recharging {
- node.set(node.serving, self.simReqCnt, self.sumWeight)
- }
- }
+ switch newMode {
+ case CmDisabled:
+ 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
}
}
-}
-
-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)
+ cm.setParallelReqs(cm.parallelReqs-1, time)
+ return
}
diff --git a/les/flowcontrol/prque/prque.go b/les/flowcontrol/prque/prque.go
new file mode 100755
index 0000000000..acfa8bf9a3
--- /dev/null
+++ b/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(compare compareFn) *Prque {
+ return &Prque{newSstack(compare)}
+}
+
+// Pushes a value with a given priority into the queue, expanding if necessary.
+func (p *Prque) Push(i interface{}) {
+ 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() interface{} {
+ return heap.Pop(p.cont)
+}
+
+// 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(p.cont.compare)
+}
diff --git a/les/flowcontrol/prque/sstack.go b/les/flowcontrol/prque/sstack.go
new file mode 100755
index 0000000000..b018c735df
--- /dev/null
+++ b/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
+
+// returns true if a comes before b
+type compareFn func(a, b 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 {
+ compare compareFn
+ size int
+ capacity int
+ offset int
+
+ blocks [][]interface{}
+ active []interface{}
+}
+
+// Creates a new, empty stack.
+func newSstack(compare compareFn) *sstack {
+ result := new(sstack)
+ result.compare = compare
+ result.active = make([]interface{}, blockSize)
+ result.blocks = [][]interface{}{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([]interface{}, 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
+ 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.compare(s.blocks[i/blockSize][i%blockSize], 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(s.compare)
+}
diff --git a/les/handler.go b/les/handler.go
index 2fc4cde349..793b57011a 100644
--- a/les/handler.go
+++ b/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)
- p.Log().Error("Request came too early", "recharge", common.PrettyDuration(recharge))
+
+ if accepted, bufShort := p.fcClient.AcceptRequest(cost); !accepted {
+ 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)
@@ -1232,7 +1258,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) }
},
}
@@ -1256,7 +1282,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) }
},
}
diff --git a/les/handler_test.go b/les/handler_test.go
index 31aad3ed45..76d575e63d 100644
--- a/les/handler_test.go
+++ b/les/handler_test.go
@@ -51,7 +51,7 @@ func TestGetBlockHeadersLes1(t *testing.T) { testGetBlockHeaders(t, 1) }
func TestGetBlockHeadersLes2(t *testing.T) { testGetBlockHeaders(t, 2) }
func testGetBlockHeaders(t *testing.T, protocol int) {
- pm := newTestProtocolManagerMust(t, false, downloader.MaxHashFetch+15, nil, nil, nil, ethdb.NewMemDatabase())
+ pm := newTestProtocolManagerMust(t, false, downloader.MaxHashFetch+15, 0, nil, nil, nil, ethdb.NewMemDatabase())
bc := pm.blockchain.(*core.BlockChain)
peer, _ := newTestPeer(t, "peer", protocol, pm, true)
defer peer.close()
@@ -180,7 +180,7 @@ func TestGetBlockBodiesLes1(t *testing.T) { testGetBlockBodies(t, 1) }
func TestGetBlockBodiesLes2(t *testing.T) { testGetBlockBodies(t, 2) }
func testGetBlockBodies(t *testing.T, protocol int) {
- pm := newTestProtocolManagerMust(t, false, downloader.MaxBlockFetch+15, nil, nil, nil, ethdb.NewMemDatabase())
+ pm := newTestProtocolManagerMust(t, false, downloader.MaxBlockFetch+15, 0, nil, nil, nil, ethdb.NewMemDatabase())
bc := pm.blockchain.(*core.BlockChain)
peer, _ := newTestPeer(t, "peer", protocol, pm, true)
defer peer.close()
@@ -257,7 +257,7 @@ func TestGetCodeLes2(t *testing.T) { testGetCode(t, 2) }
func testGetCode(t *testing.T, protocol int) {
// Assemble the test environment
- pm := newTestProtocolManagerMust(t, false, 4, testChainGen, nil, nil, ethdb.NewMemDatabase())
+ pm := newTestProtocolManagerMust(t, false, 4, 0, testChainGen, nil, nil, ethdb.NewMemDatabase())
bc := pm.blockchain.(*core.BlockChain)
peer, _ := newTestPeer(t, "peer", protocol, pm, true)
defer peer.close()
@@ -291,7 +291,7 @@ func TestGetReceiptLes2(t *testing.T) { testGetReceipt(t, 2) }
func testGetReceipt(t *testing.T, protocol int) {
// Assemble the test environment
db := ethdb.NewMemDatabase()
- pm := newTestProtocolManagerMust(t, false, 4, testChainGen, nil, nil, db)
+ pm := newTestProtocolManagerMust(t, false, 4, 0, testChainGen, nil, nil, db)
bc := pm.blockchain.(*core.BlockChain)
peer, _ := newTestPeer(t, "peer", protocol, pm, true)
defer peer.close()
@@ -319,7 +319,7 @@ func TestGetProofsLes2(t *testing.T) { testGetProofs(t, 2) }
func testGetProofs(t *testing.T, protocol int) {
// Assemble the test environment
db := ethdb.NewMemDatabase()
- pm := newTestProtocolManagerMust(t, false, 4, testChainGen, nil, nil, db)
+ pm := newTestProtocolManagerMust(t, false, 4, 0, testChainGen, nil, nil, db)
bc := pm.blockchain.(*core.BlockChain)
peer, _ := newTestPeer(t, "peer", protocol, pm, true)
defer peer.close()
@@ -382,15 +382,11 @@ func testGetCHTProofs(t *testing.T, protocol int) {
}
// Assemble the test environment
db := ethdb.NewMemDatabase()
- pm := newTestProtocolManagerMust(t, false, int(frequency)+light.HelperTrieProcessConfirmations, testChainGen, nil, nil, db)
+ pm := newTestProtocolManagerMust(t, false, int(frequency)+light.HelperTrieProcessConfirmations, 1, testChainGen, nil, nil, db)
bc := pm.blockchain.(*core.BlockChain)
peer, _ := newTestPeer(t, "peer", protocol, pm, true)
defer peer.close()
- // Wait a while for the CHT indexer to process the new headers
- time.Sleep(100 * time.Millisecond * time.Duration(frequency/light.CHTFrequencyServer)) // Chain indexer throttling
- time.Sleep(250 * time.Millisecond) // CI tester slack
-
// Assemble the proofs from the different protocols
header := bc.GetHeaderByNumber(frequency)
rlp, _ := rlp.EncodeToBytes(header)
@@ -450,15 +446,11 @@ func testGetCHTProofs(t *testing.T, protocol int) {
func TestGetBloombitsProofs(t *testing.T) {
// Assemble the test environment
db := ethdb.NewMemDatabase()
- pm := newTestProtocolManagerMust(t, false, light.BloomTrieFrequency+256, testChainGen, nil, nil, db)
+ pm := newTestProtocolManagerMust(t, false, light.BloomTrieFrequency+256, 1, testChainGen, nil, nil, db)
bc := pm.blockchain.(*core.BlockChain)
peer, _ := newTestPeer(t, "peer", 2, pm, true)
defer peer.close()
- // Wait a while for the bloombits indexer to process the new headers
- time.Sleep(100 * time.Millisecond * time.Duration(light.BloomTrieFrequency/4096)) // Chain indexer throttling
- time.Sleep(250 * time.Millisecond) // CI tester slack
-
// Request and verify each bit of the bloom bits proofs
for bit := 0; bit < 2048; bit++ {
// Assemble therequest and proofs for the bloombits
@@ -489,7 +481,7 @@ func TestGetBloombitsProofs(t *testing.T) {
func TestTransactionStatusLes2(t *testing.T) {
db := ethdb.NewMemDatabase()
- pm := newTestProtocolManagerMust(t, false, 0, nil, nil, nil, db)
+ pm := newTestProtocolManagerMust(t, false, 0, 0, nil, nil, nil, db)
chain := pm.blockchain.(*core.BlockChain)
config := core.DefaultTxPoolConfig
config.Journal = ""
diff --git a/les/helper_test.go b/les/helper_test.go
index 8fd01a39e0..8af3f78a38 100644
--- a/les/helper_test.go
+++ b/les/helper_test.go
@@ -24,8 +24,10 @@ import (
"math/big"
"sync"
"testing"
+ "time"
"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"
@@ -136,7 +138,7 @@ func testRCL() RequestCostList {
// newTestProtocolManager creates a new protocol manager for testing purposes,
// with the given number of blocks already known, and potential notification
// channels for different events.
-func newTestProtocolManager(lightSync bool, blocks int, generator func(int, *core.BlockGen), peers *peerSet, odr *LesOdr, db ethdb.Database) (*ProtocolManager, error) {
+func newTestProtocolManager(lightSync bool, blocks int, processSections uint64, generator func(int, *core.BlockGen), peers *peerSet, odr *LesOdr, db ethdb.Database) (*ProtocolManager, error) {
var (
evmux = new(event.TypeMux)
engine = ethash.NewFaker()
@@ -169,6 +171,18 @@ func newTestProtocolManager(lightSync bool, blocks int, generator func(int, *cor
if _, err := blockchain.InsertChain(gchain); err != nil {
panic(err)
}
+
+ if processSections > 0 {
+ for {
+ cs, _, _ := chtIndexer.Sections()
+ bs, _, _ := bloomIndexer.Sections()
+ if cs >= processSections && bs >= processSections {
+ break
+ }
+ time.Sleep(time.Millisecond * 10)
+ }
+ }
+
chain = blockchain
}
@@ -191,7 +205,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)
@@ -202,8 +217,8 @@ func newTestProtocolManager(lightSync bool, blocks int, generator func(int, *cor
// with the given number of blocks already known, and potential notification
// channels for different events. In case of an error, the constructor force-
// fails the test.
-func newTestProtocolManagerMust(t *testing.T, lightSync bool, blocks int, generator func(int, *core.BlockGen), peers *peerSet, odr *LesOdr, db ethdb.Database) *ProtocolManager {
- pm, err := newTestProtocolManager(lightSync, blocks, generator, peers, odr, db)
+func newTestProtocolManagerMust(t *testing.T, lightSync bool, blocks int, processSections uint64, generator func(int, *core.BlockGen), peers *peerSet, odr *LesOdr, db ethdb.Database) *ProtocolManager {
+ pm, err := newTestProtocolManager(lightSync, blocks, processSections, generator, peers, odr, db)
if err != nil {
t.Fatalf("Failed to create protocol manager: %v", err)
}
diff --git a/les/load_test.go b/les/load_test.go
new file mode 100644
index 0000000000..901824367d
--- /dev/null
+++ b/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 .
+
+// 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}}},
+ })
+}
diff --git a/les/odr.go b/les/odr.go
index f8412aaad7..e224df3484 100644
--- a/les/odr.go
+++ b/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) }
},
}
diff --git a/les/odr_test.go b/les/odr_test.go
index 983f7262b0..dba9ee1f03 100644
--- a/les/odr_test.go
+++ b/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,13 +164,13 @@ 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()
odr := NewLesOdr(ldb, light.NewChtIndexer(db, true), light.NewBloomTrieIndexer(db, true), eth.NewBloomIndexer(db, light.BloomTrieFrequency), rm)
- pm := newTestProtocolManagerMust(t, false, 4, testChainGen, nil, nil, db)
- lpm := newTestProtocolManagerMust(t, true, 0, nil, peers, odr, ldb)
+ pm := newTestProtocolManagerMust(t, false, 4, 0, testChainGen, nil, nil, db)
+ lpm := newTestProtocolManagerMust(t, true, 0, 0, nil, peers, odr, ldb)
_, err1, lpeer, err2 := newTestPeerPair("peer", protocol, pm, lpm)
select {
case <-time.After(time.Millisecond * 100):
diff --git a/les/peer.go b/les/peer.go
index eb7452e276..20fbc2b2cc 100644
--- a/les/peer.go
+++ b/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()
}
diff --git a/les/request_test.go b/les/request_test.go
index ba2f603d8b..9caa633056 100644
--- a/les/request_test.go
+++ b/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,14 +86,14 @@ 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()
odr := NewLesOdr(ldb, light.NewChtIndexer(db, true), light.NewBloomTrieIndexer(db, true), eth.NewBloomIndexer(db, light.BloomTrieFrequency), rm)
- pm := newTestProtocolManagerMust(t, false, 4, testChainGen, nil, nil, db)
- lpm := newTestProtocolManagerMust(t, true, 0, nil, peers, odr, ldb)
+ pm := newTestProtocolManagerMust(t, false, 4, 0, testChainGen, nil, nil, db)
+ lpm := newTestProtocolManagerMust(t, true, 0, 0, nil, peers, odr, ldb)
_, err1, lpeer, err2 := newTestPeerPair("peer", protocol, pm, lpm)
select {
case <-time.After(time.Millisecond * 100):
diff --git a/les/server.go b/les/server.go
index fca6124c9c..18fead3f2b 100644
--- a/les/server.go
+++ b/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"
@@ -40,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
@@ -98,11 +99,38 @@ 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)
+ 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
}
diff --git a/les/txrelay.go b/les/txrelay.go
index 7a02cc837e..eaeb2cc22b 100644
--- a/les/txrelay.go
+++ b/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) }
},
}