diff --git a/cmd/utils/flags.go b/cmd/utils/flags.go
index 495bfe13e0..24551a30fb 100644
--- a/cmd/utils/flags.go
+++ b/cmd/utils/flags.go
@@ -170,7 +170,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/core/blockchain.go b/core/blockchain.go
index 0461da7fd9..f104f6be86 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()
@@ -1014,6 +1023,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 a3474a6830..12b516f339 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) {
chainConfig: chainConfig,
eventMux: ctx.EventMux,
peers: peers,
- reqDist: newRequestDistributor(peers, quitSync),
+ reqDist: newRequestDistributor(peers, quitSync, &mclock.System{}),
accountManager: ctx.AccountManager,
engine: eth.CreateConsensusEngine(ctx, chainConfig, &config.Ethash, nil, false, chainDb),
shutdownChan: make(chan bool),
diff --git a/les/distributor.go b/les/distributor.go
index d3f6b21d18..bc33922c50 100644
--- a/les/distributor.go
+++ b/les/distributor.go
@@ -22,12 +22,15 @@ import (
"container/list"
"sync"
"time"
+
+ "github.com/ethereum/go-ethereum/common/mclock"
)
// requestDistributor implements a mechanism that distributes requests to
// 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{}
@@ -67,8 +70,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,
@@ -146,7 +150,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..cd64c8ac0d 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.System{})
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..df1259bb27 100644
--- a/les/flowcontrol/control.go
+++ b/les/flowcontrol/control.go
@@ -24,36 +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
+ sumCost uint64 // sum of req costs received from this client
+ accepted map[uint64]uint64 // value = sumCost after accepting the given req
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(),
+ accepted: make(map[uint64]uint64),
}
- node.cmNode = cm.addNode(node)
+ cm.init(node)
return node
}
-func (peer *ClientNode) Remove(cm *ClientManager) {
- cm.removeNode(peer.cmNode)
-}
-
func (peer *ClientNode) recalcBV(time mclock.AbsTime) {
dt := uint64(time - peer.lastTime)
if time < peer.lastTime {
@@ -66,35 +74,43 @@ 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(index, maxCost uint64) (accepted bool, bufShort uint64, priority int64) {
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)
-}
-
-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
- }
+ if maxCost > peer.bufValue {
+ return false, maxCost - peer.bufValue, 0
}
- return peer.bufValue, rcost
+ peer.bufValue -= maxCost
+ peer.sumCost += maxCost
+ peer.accepted[index] = peer.sumCost
+ return true, 0, peer.cm.accepted(peer, maxCost, time)
}
+// RequestProcessed should be called when the request has been processed
+func (peer *ClientNode) RequestProcessed(index, maxCost, servingTime uint64) (bv, realCost uint64) {
+ peer.lock.Lock()
+ defer peer.lock.Unlock()
+
+ time := peer.cm.clock.Now()
+ peer.recalcBV(time)
+ realCost = peer.cm.processed(peer, maxCost, servingTime, time)
+ bv = peer.bufValue + peer.sumCost - peer.accepted[index]
+ delete(peer.accepted, index)
+ return
+}
+
+// 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 +118,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 +149,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 +170,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 +206,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..fbd671a72c 100644
--- a/les/flowcontrol/manager.go
+++ b/les/flowcontrol/manager.go
@@ -1,4 +1,4 @@
-// Copyright 2016 The go-ethereum Authors
+// Copyright 2018 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
@@ -19,206 +19,248 @@ package flowcontrol
import (
"sync"
- "time"
"github.com/ethereum/go-ethereum/common/mclock"
+ "github.com/ethereum/go-ethereum/common/prque"
)
-const rcConst = 1000000
-
-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 {
+ corrBufValue int64 // buffer value adjusted with the extra recharge amount
+ rcLastIntValue int64 // past recharge integrator value when corrBufValue was last updated
+ rcFullIntValue int64 // future recharge integrator value when corrBufValue will reach maximum
+ queueIndex int // position in the recharge queue (-1 if not queued)
}
-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
- }
-}
-
-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))
- }
-}
+// FixedPointMultiplier is applied to the recharge integrator and the recharge curve.
+//
+// Note: fixed point arithmetic is required for the integrator because it is a
+// constantly increasing value that can wrap around int64 limits (which behavior is
+// also supported by the priority queue). A floating point value would gradually lose
+// precision in this application.
+// The recharge curve and all recharge values are encoded as fixed point because
+// sumRecharge is frequently updated by adding or subtracting individual recharge
+// values and perfect precision is required.
+const FixedPointMultiplier = 1000000
+// 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
+ lock sync.Mutex
+ nodes map[*ClientNode]struct{}
+ enabledCh chan struct{}
+
+ curve PieceWiseLinear
+ sumRecharge uint64
+ // recharge integrator is increasing in each moment with a rate of
+ // (totalRecharge / sumRecharge)*FixedPointMultiplier or 0 if sumRecharge==0
+ rcLastUpdate mclock.AbsTime // last time the recharge integrator was updated
+ rcLastIntValue int64 // last updated value of the recharge integrator
+ // recharge queue is a priority queue with currently recharging client nodes
+ // as elements. The priority value is rcFullIntValue which allows to quickly
+ // determine which client will first finish recharge.
+ rcQueue *prque.Prque
}
-func NewClientManager(rcTarget, maxSimReq, maxRcSum uint64) *ClientManager {
+// NewClientManager returns a new client manager.
+// Client manager enhances flow control performance by allowing client buffers
+// to recharge quicker than the minimum guaranteed recharge rate if possible.
+// The sum of all minimum recharge rates (sumRecharge) is updated each time
+// a clients starts or finishes buffer recharging. Then an adjusted total
+// recharge rate is calculated using a piecewise linear recharge curve:
+//
+// totalRecharge = curve(sumRecharge)
+// (totalRecharge >= sumRecharge is enforced)
+//
+// Then the "bonus" buffer recharge is distributed between currently recharging
+// clients proportionally to their minimum recharge rates.
+//
+// Note: total recharge is proportional to the average number of parallel running
+// serving threads. A recharge value of 1000000 corresponds to one thread in average.
+// The maximum number of allowed serving threads should always be considerably
+// higher than the targeted average number.
+//
+// Note 2: although it is possible to specify a curve allowing the total target
+// recharge starting from zero sumRecharge, it makes sense to add a linear ramp
+// starting from zero in order to not let a single low-priority client use up
+// the entire server capacity and thus ensure quick availability for others at
+// any moment.
+func NewClientManager(curve PieceWiseLinear, clock mclock.Clock) *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{}),
+ rcQueue: prque.New(func(a interface{}, i int) { a.(*ClientNode).queueIndex = i }),
+ curve: curve,
}
- go cm.queueProc()
return cm
}
-func (self *ClientManager) Stop() {
- self.lock.Lock()
- defer self.lock.Unlock()
+// SetRechargeCurve updates the recharge curve
+func (cm *ClientManager) SetRechargeCurve(curve PieceWiseLinear) {
+ cm.lock.Lock()
+ defer cm.lock.Unlock()
- // signal any waiting accept routines to return false
- self.nodes = make(map[*cmNode]struct{})
- close(self.resumeQueue)
+ cm.updateRecharge(cm.clock.Now())
+ cm.curve = curve
}
-func (self *ClientManager) addNode(cnode *ClientNode) *cmNode {
- time := mclock.Now()
- node := &cmNode{
- node: cnode,
- lastUpdate: time,
- finishRecharge: time,
- rcWeight: 1,
+// init initializes the ClientManager specific fields of a ClientNode structure
+func (cm *ClientManager) init(node *ClientNode) {
+ cm.lock.Lock()
+ defer cm.lock.Unlock()
+
+ node.corrBufValue = int64(node.params.BufLimit)
+ node.rcLastIntValue = cm.rcLastIntValue
+ node.queueIndex = -1
+}
+
+// accepted deduces the upper estimate for request cost from the buffer and returns a priority
+// value based on current buffer status which is used by the serving queue.
+func (cm *ClientManager) accepted(node *ClientNode, maxCost uint64, now mclock.AbsTime) (priority int64) {
+ cm.lock.Lock()
+ defer cm.lock.Unlock()
+
+ cm.updateNodeRc(node, -int64(maxCost), now)
+ rcTime := (node.params.BufLimit - uint64(node.corrBufValue)) * FixedPointMultiplier / node.params.MinRecharge
+ return -int64(now) - int64(rcTime)
+}
+
+// processed updates the client buffer according to actual request cost after
+// serving has been finished.
+//
+// Note: processed should always be called for all accepted requests
+func (cm *ClientManager) processed(node *ClientNode, maxCost, servingTime uint64, now mclock.AbsTime) (realCost uint64) {
+ cm.lock.Lock()
+ defer cm.lock.Unlock()
+
+ realCost = servingTime
+ if realCost > maxCost {
+ realCost = maxCost
}
- 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)
+ cm.updateNodeRc(node, int64(maxCost-realCost), now)
+ if uint64(node.corrBufValue) > node.bufValue {
+ node.bufValue = uint64(node.corrBufValue)
}
- 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
- }
+// updateRecharge updates the recharge integrator and checks the recharge queue
+// for nodes with recently filled buffers
+func (cm *ClientManager) updateRecharge(time mclock.AbsTime) {
+ lastUpdate := cm.rcLastUpdate
+ cm.rcLastUpdate = time
+ // updating is done in multiple steps if node buffers are filled and sumRecharge
+ // is decreased before the given target time
+ for cm.sumRecharge > 0 {
+ bonusRatio := cm.curve.ValueAt(cm.sumRecharge) / float64(cm.sumRecharge)
+ if bonusRatio < 1 {
+ bonusRatio = 1
}
- if self.updateNodes(firstTime) {
- for node := range self.nodes {
- if node.recharging {
- node.set(node.serving, self.simReqCnt, self.sumWeight)
- }
- }
- } else {
- self.time = time
+ dt := time - lastUpdate
+ // fetch the client that finishes first
+ rcqNode := cm.rcQueue.PopItem().(*ClientNode) // if sumRecharge > 0 then the queue cannot be empty
+ // check whether it has already finished
+ dtNext := mclock.AbsTime(float64(rcqNode.rcFullIntValue-cm.rcLastIntValue) / bonusRatio)
+ if dt < dtNext {
+ // not finished yet, put it back, update integrator according
+ // to current bonusRatio and return
+ cm.rcQueue.Push(rcqNode, -rcqNode.rcFullIntValue)
+ cm.rcLastIntValue += int64(bonusRatio * float64(dt))
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
- }
+ // finished recharging, update corrBufValue and sumRecharge if necessary and do next step
+ if rcqNode.corrBufValue < int64(rcqNode.params.BufLimit) {
+ rcqNode.corrBufValue = int64(rcqNode.params.BufLimit)
+ cm.sumRecharge -= rcqNode.params.MinRecharge
}
- close(rc)
+ lastUpdate += dtNext
+ cm.rcLastIntValue = rcqNode.rcFullIntValue
}
}
-func (self *ClientManager) accept(node *cmNode, time mclock.AbsTime) bool {
- self.lock.Lock()
- defer self.lock.Unlock()
+// updateNodeRc updates a node's corrBufValue and adds an external correction value.
+// It also adds or removes the rcQueue entry and updates sumRecharge if necessary.
+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) / FixedPointMultiplier
+ 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 {
+ if node.queueIndex != -1 {
+ cm.rcQueue.Remove(node.queueIndex)
+ }
+ node.rcLastIntValue = cm.rcLastIntValue
+ node.rcFullIntValue = cm.rcLastIntValue + (int64(node.params.BufLimit)-node.corrBufValue)*FixedPointMultiplier/int64(node.params.MinRecharge)
+ cm.rcQueue.Push(node, -node.rcFullIntValue)
+ }
+}
- 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
+// PieceWiseLinear is used to describe recharge curves
+type PieceWiseLinear []struct{ X, Y uint64 }
+
+// ValueAt returns the curve's value at a given point
+func (pwl PieceWiseLinear) ValueAt(x uint64) float64 {
+ l := 0
+ h := len(pwl)
+ if h == 0 {
+ return 0
+ }
+ for h != l {
+ m := (l + h) / 2
+ if x > pwl[m].X {
+ l = m + 1
+ } else {
+ h = m
}
}
- self.simReqCnt++
- node.set(true, self.simReqCnt, self.sumWeight)
- node.startValue = node.rcValue
- self.update(self.time)
+ if l == 0 {
+ return float64(pwl[0].Y)
+ }
+ l--
+ if h == len(pwl) {
+ return float64(pwl[l].Y)
+ }
+ dx := pwl[h].X - pwl[l].X
+ if dx < 1 {
+ return float64(pwl[l].Y)
+ }
+ return float64(pwl[l].Y) + float64(pwl[h].Y-pwl[l].Y)*float64(x-pwl[l].X)/float64(dx)
+}
+
+// Valid returns true if the X coordinates of the curve points are non-strictly monotonic
+func (pwl PieceWiseLinear) Valid() bool {
+ var lastX uint64
+ for _, i := range pwl {
+ if i.X < lastX {
+ return false
+ }
+ lastX = i.X
+ }
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)
-}
diff --git a/les/flowcontrol/manager_test.go b/les/flowcontrol/manager_test.go
new file mode 100644
index 0000000000..77e636f659
--- /dev/null
+++ b/les/flowcontrol/manager_test.go
@@ -0,0 +1,117 @@
+// Copyright 2018 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 flowcontrol implements a client side flow control mechanism
+package flowcontrol
+
+import (
+ "math/rand"
+ "testing"
+ "time"
+
+ "github.com/ethereum/go-ethereum/common/mclock"
+)
+
+type testNode struct {
+ node *ClientNode
+ bufLimit, bandwidth uint64
+ waitUntil mclock.AbsTime
+ index, totalCost uint64
+}
+
+const (
+ testMaxCost = 1000000
+ testLength = 100000
+)
+
+func (n *testNode) send(t *testing.T, now mclock.AbsTime) bool {
+ if now < n.waitUntil {
+ return false
+ }
+ n.index++
+ if ok, _, _ := n.node.AcceptRequest(n.index, testMaxCost); !ok {
+ t.Fatalf("Rejected request after expected waiting time has passed")
+ }
+ rcost := uint64(rand.Int63n(testMaxCost))
+ bv, _ := n.node.RequestProcessed(n.index, testMaxCost, rcost)
+ if bv < testMaxCost {
+ n.waitUntil = now + mclock.AbsTime((testMaxCost-bv)*1001000/n.bandwidth)
+ }
+ //n.waitUntil = now + mclock.AbsTime(float64(testMaxCost)*1001000/float64(n.bandwidth)*(1-float64(bv)/float64(n.bufLimit)))
+ n.totalCost += rcost
+ return true
+}
+
+func TestConstantTotalBandwidth(t *testing.T) {
+ testConstantTotalBandwidth(t, 10, 1, 0)
+ testConstantTotalBandwidth(t, 10, 1, 1)
+ testConstantTotalBandwidth(t, 30, 1, 0)
+ testConstantTotalBandwidth(t, 30, 2, 3)
+ testConstantTotalBandwidth(t, 100, 1, 0)
+ testConstantTotalBandwidth(t, 100, 3, 5)
+ testConstantTotalBandwidth(t, 100, 5, 10)
+}
+
+func testConstantTotalBandwidth(t *testing.T, nodeCount, maxCapacityNodes, randomSend int) {
+ clock := &mclock.Simulated{}
+ nodes := make([]*testNode, nodeCount)
+ var totalBandwidth uint64
+ for i, _ := range nodes {
+ nodes[i] = &testNode{bandwidth: uint64(50000 + rand.Intn(100000))}
+ totalBandwidth += nodes[i].bandwidth
+ }
+ m := NewClientManager(PieceWiseLinear{{0, totalBandwidth}}, clock)
+ for _, n := range nodes {
+ n.bufLimit = n.bandwidth * 6000 //uint64(2000+rand.Intn(10000))
+ n.node = NewClientNode(m, &ServerParams{BufLimit: n.bufLimit, MinRecharge: n.bandwidth})
+ }
+ maxNodes := make([]int, maxCapacityNodes)
+ for i, _ := range maxNodes {
+ // we don't care if some indexes are selected multiple times
+ // in that case we have fewer max nodes
+ maxNodes[i] = rand.Intn(nodeCount)
+ }
+
+ for i := 0; i < testLength; i++ {
+ now := clock.Now()
+ for _, idx := range maxNodes {
+ for nodes[idx].send(t, now) {
+ }
+ }
+ if rand.Intn(testLength) < maxCapacityNodes*3 {
+ maxNodes[rand.Intn(maxCapacityNodes)] = rand.Intn(nodeCount)
+ }
+
+ sendCount := randomSend
+ for sendCount > 0 {
+ if nodes[rand.Intn(nodeCount)].send(t, now) {
+ sendCount--
+ }
+ }
+
+ clock.Run(time.Millisecond)
+ }
+
+ var totalCost uint64
+ for _, n := range nodes {
+ totalCost += n.totalCost
+ }
+ ratio := float64(totalCost) / float64(totalBandwidth) / testLength
+ if ratio < 0.98 || ratio > 1.02 {
+ t.Errorf("totalCost/totalBandwidth/testLength ratio incorrect (expected: 1, got: %f)", ratio)
+ }
+
+}
diff --git a/les/handler.go b/les/handler.go
index 243a6dabd4..f1ac93f22f 100644
--- a/les/handler.go
+++ b/les/handler.go
@@ -89,21 +89,22 @@ type txPool interface {
}
type ProtocolManager struct {
- lightSync bool
- txpool txPool
- txrelay *LesTxRelay
- networkId uint64
- chainConfig *params.ChainConfig
- iConfig *light.IndexerConfig
- blockchain BlockChain
- chainDb ethdb.Database
- odr *LesOdr
- server *LesServer
- serverPool *serverPool
- clientPool *freeClientPool
- lesTopic discv5.Topic
- reqDist *requestDistributor
- retriever *retrieveManager
+ lightSync bool
+ txpool txPool
+ txrelay *LesTxRelay
+ networkId uint64
+ chainConfig *params.ChainConfig
+ iConfig *light.IndexerConfig
+ blockchain BlockChain
+ chainDb ethdb.Database
+ odr *LesOdr
+ server *LesServer
+ serverPool *serverPool
+ clientPool *freeClientPool
+ lesTopic discv5.Topic
+ reqDist *requestDistributor
+ retriever *retrieveManager
+ servingQueue *servingQueue
downloader *downloader.Downloader
fetcher *lightFetcher
@@ -147,6 +148,8 @@ func NewProtocolManager(chainConfig *params.ChainConfig, indexerConfig *light.In
if odr != nil {
manager.retriever = odr.retriever
manager.reqDist = odr.retriever.dist
+ } else {
+ manager.servingQueue = newServingQueue(int64(time.Millisecond * 10))
}
removePeer := manager.removePeer
@@ -283,9 +286,6 @@ func (pm *ProtocolManager) handle(p *peer) error {
return err
}
defer func() {
- if pm.server != nil && pm.server.fcManager != nil && p.fcClient != nil {
- p.fcClient.Remove(pm.server.fcManager)
- }
pm.removePeer(p.id)
}()
// Register the peer in the downloader. If the downloader considers it banned, we disconnect
@@ -337,20 +337,33 @@ 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]
+ p.responseCount++
+ responseCount := p.responseCount
+ var (
+ maxCost uint64
+ priority int64
+ )
+
reject := func(reqCnt, maxCnt uint64) bool {
+ if reqCnt == 0 {
+ return true
+ }
if p.fcClient == nil || reqCnt > maxCnt {
return true
}
- bufValue, _ := p.fcClient.AcceptRequest()
- cost := costs.baseCost + reqCnt*costs.reqCost
- if cost > pm.server.defParams.BufLimit {
- cost = pm.server.defParams.BufLimit
+ costs := p.fcCosts[msg.Code]
+ maxCost = costs.baseCost + reqCnt*costs.reqCost
+ if maxCost > pm.server.defParams.BufLimit {
+ maxCost = 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, servingPriority := p.fcClient.AcceptRequest(responseCount, maxCost); !accepted {
+ if bufShort > 0 {
+ p.Log().Error("Request came too early", "remaining", common.PrettyDuration(time.Duration(bufShort*1000000/pm.server.defParams.MinRecharge)))
+ }
return true
+ } else {
+ priority = servingPriority
}
return false
}
@@ -362,6 +375,30 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
var deliverMsg *Msg
+ sendFunc := func(amount uint64, send func(bv uint64) error) func(servingTime uint64, err error) {
+ return func(servingTime uint64, err error) {
+ if err != nil {
+ p.errCh <- err
+ return
+ }
+
+ // responseLock ensures that responses are queued in the same order as
+ // RequestProcessed is called
+ p.responseLock.Lock()
+ defer p.responseLock.Unlock()
+
+ bv, rcost := p.fcClient.RequestProcessed(responseCount, maxCost, servingTime)
+ pm.server.fcCostStats.update(msg.Code, amount, rcost)
+ if send != nil {
+ p.queueSend(func() {
+ if err := send(bv); err != nil {
+ p.errCh <- err
+ }
+ })
+ }
+ }
+ }
+
// Handle the message depending on its contents
switch msg.Code {
case StatusMsg:
@@ -420,79 +457,85 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
headers []*types.Header
unknown bool
)
- for !unknown && len(headers) < int(query.Amount) && bytes < softResponseLimit {
- // Retrieve the next header satisfying the query
- var origin *types.Header
- if hashMode {
- if first {
- first = false
- origin = pm.blockchain.GetHeaderByHash(query.Origin.Hash)
- if origin != nil {
- query.Origin.Number = origin.Number.Uint64()
- }
- } else {
- origin = pm.blockchain.GetHeader(query.Origin.Hash, query.Origin.Number)
- }
- } else {
- origin = pm.blockchain.GetHeaderByNumber(query.Origin.Number)
- }
- if origin == nil {
- break
- }
- headers = append(headers, origin)
- bytes += estHeaderRlpSize
- // Advance to the next header of the query
- switch {
- case hashMode && query.Reverse:
- // Hash based traversal towards the genesis block
- ancestor := query.Skip + 1
- if ancestor == 0 {
- unknown = true
- } else {
- query.Origin.Hash, query.Origin.Number = pm.blockchain.GetAncestor(query.Origin.Hash, query.Origin.Number, ancestor, &maxNonCanonical)
- unknown = (query.Origin.Hash == common.Hash{})
- }
- case hashMode && !query.Reverse:
- // Hash based traversal towards the leaf block
- var (
- current = origin.Number.Uint64()
- next = current + query.Skip + 1
- )
- if next <= current {
- infos, _ := json.MarshalIndent(p.Peer.Info(), "", " ")
- p.Log().Warn("GetBlockHeaders skip overflow attack", "current", current, "skip", query.Skip, "next", next, "attacker", infos)
- unknown = true
- } else {
- if header := pm.blockchain.GetHeaderByNumber(next); header != nil {
- nextHash := header.Hash()
- expOldHash, _ := pm.blockchain.GetAncestor(nextHash, next, query.Skip+1, &maxNonCanonical)
- if expOldHash == query.Origin.Hash {
- query.Origin.Hash, query.Origin.Number = nextHash, next
+ pm.servingQueue.addTask(&servingTask{
+ priority: priority,
+ run: func() (bool, error) {
+ if !unknown && len(headers) < int(query.Amount) && bytes < softResponseLimit {
+ // Retrieve the next header satisfying the query
+ var origin *types.Header
+ if hashMode {
+ if first {
+ first = false
+ origin = pm.blockchain.GetHeaderByHash(query.Origin.Hash)
+ if origin != nil {
+ query.Origin.Number = origin.Number.Uint64()
+ }
+ } else {
+ origin = pm.blockchain.GetHeader(query.Origin.Hash, query.Origin.Number)
+ }
+ } else {
+ origin = pm.blockchain.GetHeaderByNumber(query.Origin.Number)
+ }
+ if origin == nil {
+ return true, nil
+ }
+ headers = append(headers, origin)
+ bytes += estHeaderRlpSize
+
+ // Advance to the next header of the query
+ switch {
+ case hashMode && query.Reverse:
+ // Hash based traversal towards the genesis block
+ ancestor := query.Skip + 1
+ if ancestor == 0 {
+ unknown = true
+ } else {
+ query.Origin.Hash, query.Origin.Number = pm.blockchain.GetAncestor(query.Origin.Hash, query.Origin.Number, ancestor, &maxNonCanonical)
+ unknown = (query.Origin.Hash == common.Hash{})
+ }
+ case hashMode && !query.Reverse:
+ // Hash based traversal towards the leaf block
+ var (
+ current = origin.Number.Uint64()
+ next = current + query.Skip + 1
+ )
+ if next <= current {
+ infos, _ := json.MarshalIndent(p.Peer.Info(), "", " ")
+ p.Log().Warn("GetBlockHeaders skip overflow attack", "current", current, "skip", query.Skip, "next", next, "attacker", infos)
+ unknown = true
+ } else {
+ if header := pm.blockchain.GetHeaderByNumber(next); header != nil {
+ nextHash := header.Hash()
+ expOldHash, _ := pm.blockchain.GetAncestor(nextHash, next, query.Skip+1, &maxNonCanonical)
+ if expOldHash == query.Origin.Hash {
+ query.Origin.Hash, query.Origin.Number = nextHash, next
+ } else {
+ unknown = true
+ }
+ } else {
+ unknown = true
+ }
+ }
+ case query.Reverse:
+ // Number based traversal towards the genesis block
+ if query.Origin.Number >= query.Skip+1 {
+ query.Origin.Number -= query.Skip + 1
} else {
unknown = true
}
- } else {
- unknown = true
+
+ case !query.Reverse:
+ // Number based traversal towards the leaf block
+ query.Origin.Number += query.Skip + 1
}
- }
- case query.Reverse:
- // Number based traversal towards the genesis block
- if query.Origin.Number >= query.Skip+1 {
- query.Origin.Number -= query.Skip + 1
+ return false, nil
} else {
- unknown = true
+ return true, nil
}
-
- case !query.Reverse:
- // Number based traversal towards the leaf block
- query.Origin.Number += query.Skip + 1
- }
- }
-
- bv, rcost := p.fcClient.RequestProcessed(costs.baseCost + query.Amount*costs.reqCost)
- pm.server.fcCostStats.update(msg.Code, query.Amount, rcost)
- return p.SendBlockHeaders(req.ReqID, bv, headers)
+ },
+ after: sendFunc(query.Amount, func(bv uint64) error { return p.SendBlockHeaders(req.ReqID, bv, headers) }),
+ })
case BlockHeadersMsg:
if pm.downloader == nil {
@@ -508,7 +551,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 {
@@ -537,21 +580,27 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
if reject(uint64(reqCnt), MaxBodyFetch) {
return errResp(ErrRequestRejected, "")
}
- for _, hash := range req.Hashes {
- if bytes >= softResponseLimit {
- break
- }
- // 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 {
- bodies = append(bodies, data)
- bytes += len(data)
+
+ index := 0
+ pm.servingQueue.addTask(&servingTask{
+ priority: priority,
+ run: func() (bool, error) {
+ hash := req.Hashes[index]
+ index++
+ if bytes >= softResponseLimit {
+ return true, nil
}
- }
- }
- bv, rcost := p.fcClient.RequestProcessed(costs.baseCost + uint64(reqCnt)*costs.reqCost)
- pm.server.fcCostStats.update(msg.Code, uint64(reqCnt), rcost)
- return p.SendBlockBodiesRLP(req.ReqID, bv, bodies)
+ // 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 {
+ bodies = append(bodies, data)
+ bytes += len(data)
+ }
+ }
+ return index == reqCnt, nil
+ },
+ after: sendFunc(uint64(reqCnt), func(bv uint64) error { return p.SendBlockBodiesRLP(req.ReqID, bv, bodies) }),
+ })
case BlockBodiesMsg:
if pm.odr == nil {
@@ -567,7 +616,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,
@@ -593,30 +642,35 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
if reject(uint64(reqCnt), MaxCodeFetch) {
return errResp(ErrRequestRejected, "")
}
- for _, req := range req.Reqs {
- // 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 {
- statedb, err := pm.blockchain.State()
- if err != nil {
- continue
- }
- account, err := pm.getAccount(statedb, header.Root, common.BytesToHash(req.AccKey))
- if err != nil {
- continue
- }
- code, _ := statedb.Database().TrieDB().Node(common.BytesToHash(account.CodeHash))
+ index := 0
+ pm.servingQueue.addTask(&servingTask{
+ priority: priority,
+ run: func() (bool, error) {
+ req := req.Reqs[index]
+ index++
+ // 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 {
+ statedb, err := pm.blockchain.State()
+ if err != nil {
+ return false, nil
+ }
+ account, err := pm.getAccount(statedb, header.Root, common.BytesToHash(req.AccKey))
+ if err != nil {
+ return false, nil
+ }
+ code, _ := statedb.Database().TrieDB().Node(common.BytesToHash(account.CodeHash))
- data = append(data, code)
- if bytes += len(code); bytes >= softResponseLimit {
- break
+ data = append(data, code)
+ if bytes += len(code); bytes >= softResponseLimit {
+ return true, nil
+ }
}
}
- }
- }
- bv, rcost := p.fcClient.RequestProcessed(costs.baseCost + uint64(reqCnt)*costs.reqCost)
- pm.server.fcCostStats.update(msg.Code, uint64(reqCnt), rcost)
- return p.SendCode(req.ReqID, bv, data)
+ return index == reqCnt, nil
+ },
+ after: sendFunc(uint64(reqCnt), func(bv uint64) error { return p.SendCode(req.ReqID, bv, data) }),
+ })
case CodeMsg:
if pm.odr == nil {
@@ -632,7 +686,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,
@@ -658,31 +712,37 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
if reject(uint64(reqCnt), MaxReceiptFetch) {
return errResp(ErrRequestRejected, "")
}
- for _, hash := range req.Hashes {
- if bytes >= softResponseLimit {
- break
- }
- // Retrieve the requested block's receipts, skipping if unknown to us
- var results types.Receipts
- if number := rawdb.ReadHeaderNumber(pm.chainDb, hash); number != nil {
- results = rawdb.ReadReceipts(pm.chainDb, hash, *number)
- }
- if results == nil {
- if header := pm.blockchain.GetHeaderByHash(hash); header == nil || header.ReceiptHash != types.EmptyRootHash {
- continue
+
+ index := 0
+ pm.servingQueue.addTask(&servingTask{
+ priority: priority,
+ run: func() (bool, error) {
+ hash := req.Hashes[index]
+ index++
+ if bytes >= softResponseLimit {
+ return true, nil
}
- }
- // If known, encode and queue for response packet
- if encoded, err := rlp.EncodeToBytes(results); err != nil {
- log.Error("Failed to encode receipt", "err", err)
- } else {
- receipts = append(receipts, encoded)
- bytes += len(encoded)
- }
- }
- bv, rcost := p.fcClient.RequestProcessed(costs.baseCost + uint64(reqCnt)*costs.reqCost)
- pm.server.fcCostStats.update(msg.Code, uint64(reqCnt), rcost)
- return p.SendReceiptsRLP(req.ReqID, bv, receipts)
+ // Retrieve the requested block's receipts, skipping if unknown to us
+ var results types.Receipts
+ if number := rawdb.ReadHeaderNumber(pm.chainDb, hash); number != nil {
+ results = rawdb.ReadReceipts(pm.chainDb, hash, *number)
+ }
+ if results == nil {
+ if header := pm.blockchain.GetHeaderByHash(hash); header == nil || header.ReceiptHash != types.EmptyRootHash {
+ return false, nil
+ }
+ }
+ // If known, encode and queue for response packet
+ if encoded, err := rlp.EncodeToBytes(results); err != nil {
+ log.Error("Failed to encode receipt", "err", err)
+ } else {
+ receipts = append(receipts, encoded)
+ bytes += len(encoded)
+ }
+ return index == reqCnt, nil
+ },
+ after: sendFunc(uint64(reqCnt), func(bv uint64) error { return p.SendReceiptsRLP(req.ReqID, bv, receipts) }),
+ })
case ReceiptsMsg:
if pm.odr == nil {
@@ -698,7 +758,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,
@@ -724,39 +784,45 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
if reject(uint64(reqCnt), MaxProofsFetch) {
return errResp(ErrRequestRejected, "")
}
- for _, req := range req.Reqs {
- // 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 {
- statedb, err := pm.blockchain.State()
- if err != nil {
- continue
- }
- var trie state.Trie
- if len(req.AccKey) > 0 {
- account, err := pm.getAccount(statedb, header.Root, common.BytesToHash(req.AccKey))
- if err != nil {
- continue
- }
- trie, _ = statedb.Database().OpenStorageTrie(common.BytesToHash(req.AccKey), account.Root)
- } else {
- trie, _ = statedb.Database().OpenTrie(header.Root)
- }
- if trie != nil {
- var proof light.NodeList
- trie.Prove(req.Key, 0, &proof)
- proofs = append(proofs, proof)
- if bytes += proof.DataSize(); bytes >= softResponseLimit {
- break
+ index := 0
+ pm.servingQueue.addTask(&servingTask{
+ priority: priority,
+ run: func() (bool, error) {
+ req := req.Reqs[index]
+ index++
+ // 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 {
+ statedb, err := pm.blockchain.State()
+ if err != nil {
+ return false, nil
+ }
+ var trie state.Trie
+ if len(req.AccKey) > 0 {
+ account, err := pm.getAccount(statedb, header.Root, common.BytesToHash(req.AccKey))
+ if err != nil {
+ return false, nil
+ }
+ trie, _ = statedb.Database().OpenStorageTrie(common.BytesToHash(req.AccKey), account.Root)
+ } else {
+ trie, _ = statedb.Database().OpenTrie(header.Root)
+ }
+ if trie != nil {
+ var proof light.NodeList
+ trie.Prove(req.Key, 0, &proof)
+
+ proofs = append(proofs, proof)
+ if bytes += proof.DataSize(); bytes >= softResponseLimit {
+ return true, nil
+ }
}
}
}
- }
- }
- bv, rcost := p.fcClient.RequestProcessed(costs.baseCost + uint64(reqCnt)*costs.reqCost)
- pm.server.fcCostStats.update(msg.Code, uint64(reqCnt), rcost)
- return p.SendProofs(req.ReqID, bv, proofs)
+ return index == reqCnt, nil
+ },
+ after: sendFunc(uint64(reqCnt), func(bv uint64) error { return p.SendProofs(req.ReqID, bv, proofs) }),
+ })
case GetProofsV2Msg:
p.Log().Trace("Received les/2 proofs request")
@@ -781,44 +847,49 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
nodes := light.NewNodeSet()
- for _, req := range req.Reqs {
- // Look up the state belonging to the request
- if statedb == nil || req.BHash != lastBHash {
- statedb, root, lastBHash = nil, common.Hash{}, req.BHash
+ index := 0
+ pm.servingQueue.addTask(&servingTask{
+ priority: priority,
+ run: func() (bool, error) {
+ req := req.Reqs[index]
+ index++
+ // Look up the state belonging to the request
+ if statedb == nil || req.BHash != lastBHash {
+ statedb, root, lastBHash = nil, common.Hash{}, req.BHash
- if number := rawdb.ReadHeaderNumber(pm.chainDb, req.BHash); number != nil {
- if header := rawdb.ReadHeader(pm.chainDb, req.BHash, *number); header != nil {
- statedb, _ = pm.blockchain.State()
- root = header.Root
+ if number := rawdb.ReadHeaderNumber(pm.chainDb, req.BHash); number != nil {
+ if header := rawdb.ReadHeader(pm.chainDb, req.BHash, *number); header != nil {
+ statedb, _ = pm.blockchain.State()
+ root = header.Root
+ }
}
}
- }
- if statedb == nil {
- continue
- }
- // Pull the account or storage trie of the request
- var trie state.Trie
- if len(req.AccKey) > 0 {
- account, err := pm.getAccount(statedb, root, common.BytesToHash(req.AccKey))
- if err != nil {
- continue
+ if statedb == nil {
+ return false, nil
}
- trie, _ = statedb.Database().OpenStorageTrie(common.BytesToHash(req.AccKey), account.Root)
- } else {
- trie, _ = statedb.Database().OpenTrie(root)
- }
- if trie == nil {
- continue
- }
- // Prove the user's request from the account or stroage trie
- trie.Prove(req.Key, req.FromLevel, nodes)
- if nodes.DataSize() >= softResponseLimit {
- break
- }
- }
- bv, rcost := p.fcClient.RequestProcessed(costs.baseCost + uint64(reqCnt)*costs.reqCost)
- pm.server.fcCostStats.update(msg.Code, uint64(reqCnt), rcost)
- return p.SendProofsV2(req.ReqID, bv, nodes.NodeList())
+ // Pull the account or storage trie of the request
+ var trie state.Trie
+ if len(req.AccKey) > 0 {
+ account, err := pm.getAccount(statedb, root, common.BytesToHash(req.AccKey))
+ if err != nil {
+ return false, nil
+ }
+ trie, _ = statedb.Database().OpenStorageTrie(common.BytesToHash(req.AccKey), account.Root)
+ } else {
+ trie, _ = statedb.Database().OpenTrie(root)
+ }
+ if trie == nil {
+ return false, nil
+ }
+ // Prove the user's request from the account or stroage trie
+ trie.Prove(req.Key, req.FromLevel, nodes)
+ if nodes.DataSize() >= softResponseLimit {
+ return true, nil
+ }
+ return index == reqCnt, nil
+ },
+ after: sendFunc(uint64(reqCnt), func(bv uint64) error { return p.SendProofsV2(req.ReqID, bv, nodes.NodeList()) }),
+ })
case ProofsV1Msg:
if pm.odr == nil {
@@ -834,7 +905,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,
@@ -855,7 +926,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,
@@ -882,30 +953,36 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
return errResp(ErrRequestRejected, "")
}
trieDb := trie.NewDatabase(ethdb.NewTable(pm.chainDb, light.ChtTablePrefix))
- for _, req := range req.Reqs {
- if header := pm.blockchain.GetHeaderByNumber(req.BlockNum); header != nil {
- sectionHead := rawdb.ReadCanonicalHash(pm.chainDb, req.ChtNum*pm.iConfig.ChtSize-1)
- if root := light.GetChtRoot(pm.chainDb, req.ChtNum-1, sectionHead); root != (common.Hash{}) {
- trie, err := trie.New(root, trieDb)
- if err != nil {
- continue
- }
- var encNumber [8]byte
- binary.BigEndian.PutUint64(encNumber[:], req.BlockNum)
- var proof light.NodeList
- trie.Prove(encNumber[:], 0, &proof)
+ index := 0
+ pm.servingQueue.addTask(&servingTask{
+ priority: priority,
+ run: func() (bool, error) {
+ req := req.Reqs[index]
+ index++
+ if header := pm.blockchain.GetHeaderByNumber(req.BlockNum); header != nil {
+ sectionHead := rawdb.ReadCanonicalHash(pm.chainDb, req.ChtNum*pm.iConfig.ChtSize-1)
+ if root := light.GetChtRoot(pm.chainDb, req.ChtNum-1, sectionHead); root != (common.Hash{}) {
+ trie, err := trie.New(root, trieDb)
+ if err != nil {
+ return false, nil
+ }
+ var encNumber [8]byte
+ binary.BigEndian.PutUint64(encNumber[:], req.BlockNum)
- proofs = append(proofs, ChtResp{Header: header, Proof: proof})
- if bytes += proof.DataSize() + estHeaderRlpSize; bytes >= softResponseLimit {
- break
+ var proof light.NodeList
+ trie.Prove(encNumber[:], 0, &proof)
+
+ proofs = append(proofs, ChtResp{Header: header, Proof: proof})
+ if bytes += proof.DataSize() + estHeaderRlpSize; bytes >= softResponseLimit {
+ return true, nil
+ }
}
}
- }
- }
- bv, rcost := p.fcClient.RequestProcessed(costs.baseCost + uint64(reqCnt)*costs.reqCost)
- pm.server.fcCostStats.update(msg.Code, uint64(reqCnt), rcost)
- return p.SendHeaderProofs(req.ReqID, bv, proofs)
+ return index == reqCnt, nil
+ },
+ after: sendFunc(uint64(reqCnt), func(bv uint64) error { return p.SendHeaderProofs(req.ReqID, bv, proofs) }),
+ })
case GetHelperTrieProofsMsg:
p.Log().Trace("Received helper trie proof request")
@@ -934,39 +1011,47 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
auxTrie *trie.Trie
)
nodes := light.NewNodeSet()
- for _, req := range req.Reqs {
- if auxTrie == nil || req.Type != lastType || req.TrieIdx != lastIdx {
- auxTrie, lastType, lastIdx = nil, req.Type, req.TrieIdx
- var prefix string
- if root, prefix = pm.getHelperTrie(req.Type, req.TrieIdx); root != (common.Hash{}) {
- auxTrie, _ = trie.New(root, trie.NewDatabase(ethdb.NewTable(pm.chainDb, prefix)))
+ index := 0
+ pm.servingQueue.addTask(&servingTask{
+ priority: priority,
+ run: func() (bool, error) {
+ req := req.Reqs[index]
+ index++
+ if auxTrie == nil || req.Type != lastType || req.TrieIdx != lastIdx {
+ auxTrie, lastType, lastIdx = nil, req.Type, req.TrieIdx
+
+ var prefix string
+ if root, prefix = pm.getHelperTrie(req.Type, req.TrieIdx); root != (common.Hash{}) {
+ auxTrie, _ = trie.New(root, trie.NewDatabase(ethdb.NewTable(pm.chainDb, prefix)))
+ }
}
- }
- if req.AuxReq == auxRoot {
- var data []byte
- if root != (common.Hash{}) {
- data = root[:]
- }
- auxData = append(auxData, data)
- auxBytes += len(data)
- } else {
- if auxTrie != nil {
- auxTrie.Prove(req.Key, req.FromLevel, nodes)
- }
- if req.AuxReq != 0 {
- data := pm.getHelperTrieAuxData(req)
+ if req.AuxReq == auxRoot {
+ var data []byte
+ if root != (common.Hash{}) {
+ data = root[:]
+ }
auxData = append(auxData, data)
auxBytes += len(data)
+ } else {
+ if auxTrie != nil {
+ auxTrie.Prove(req.Key, req.FromLevel, nodes)
+ }
+ if req.AuxReq != 0 {
+ data := pm.getHelperTrieAuxData(req)
+ auxData = append(auxData, data)
+ auxBytes += len(data)
+ }
}
- }
- if nodes.DataSize()+auxBytes >= softResponseLimit {
- break
- }
- }
- bv, rcost := p.fcClient.RequestProcessed(costs.baseCost + uint64(reqCnt)*costs.reqCost)
- pm.server.fcCostStats.update(msg.Code, uint64(reqCnt), rcost)
- return p.SendHelperTrieProofs(req.ReqID, bv, HelperTrieResps{Proofs: nodes.NodeList(), AuxData: auxData})
+ if nodes.DataSize()+auxBytes >= softResponseLimit {
+ return true, nil
+ }
+ return index == reqCnt, nil
+ },
+ after: sendFunc(uint64(reqCnt), func(bv uint64) error {
+ return p.SendHelperTrieProofs(req.ReqID, bv, HelperTrieResps{Proofs: nodes.NodeList(), AuxData: auxData})
+ }),
+ })
case HeaderProofsMsg:
if pm.odr == nil {
@@ -981,7 +1066,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,
@@ -1002,7 +1087,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,
@@ -1022,10 +1107,15 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
if reject(uint64(reqCnt), MaxTxSend) {
return errResp(ErrRequestRejected, "")
}
- pm.txpool.AddRemotes(txs)
- _, rcost := p.fcClient.RequestProcessed(costs.baseCost + uint64(reqCnt)*costs.reqCost)
- pm.server.fcCostStats.update(msg.Code, uint64(reqCnt), rcost)
+ pm.servingQueue.addTask(&servingTask{
+ priority: priority,
+ run: func() (bool, error) {
+ pm.txpool.AddRemotes(txs)
+ return true, nil
+ },
+ after: sendFunc(uint64(reqCnt), nil),
+ })
case SendTxV2Msg:
if pm.txpool == nil {
@@ -1044,25 +1134,28 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
return errResp(ErrRequestRejected, "")
}
- hashes := make([]common.Hash, len(req.Txs))
- for i, tx := range req.Txs {
- hashes[i] = tx.Hash()
- }
- stats := pm.txStatus(hashes)
- for i, stat := range stats {
- if stat.Status == core.TxStatusUnknown {
- if errs := pm.txpool.AddRemotes([]*types.Transaction{req.Txs[i]}); errs[0] != nil {
- stats[i].Error = errs[0].Error()
- continue
+ var stats []txStatus
+ pm.servingQueue.addTask(&servingTask{
+ priority: priority,
+ run: func() (bool, error) {
+ hashes := make([]common.Hash, len(req.Txs))
+ for i, tx := range req.Txs {
+ hashes[i] = tx.Hash()
}
- stats[i] = pm.txStatus([]common.Hash{hashes[i]})[0]
- }
- }
-
- bv, rcost := p.fcClient.RequestProcessed(costs.baseCost + uint64(reqCnt)*costs.reqCost)
- pm.server.fcCostStats.update(msg.Code, uint64(reqCnt), rcost)
-
- return p.SendTxStatus(req.ReqID, bv, stats)
+ stats = pm.txStatus(hashes)
+ for i, stat := range stats {
+ if stat.Status == core.TxStatusUnknown {
+ if errs := pm.txpool.AddRemotes([]*types.Transaction{req.Txs[i]}); errs[0] != nil {
+ stats[i].Error = errs[0].Error()
+ continue
+ }
+ stats[i] = pm.txStatus([]common.Hash{hashes[i]})[0]
+ }
+ }
+ return true, nil
+ },
+ after: sendFunc(uint64(reqCnt), func(bv uint64) error { return p.SendTxStatus(req.ReqID, bv, stats) }),
+ })
case GetTxStatusMsg:
if pm.txpool == nil {
@@ -1080,10 +1173,16 @@ 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)
- pm.server.fcCostStats.update(msg.Code, uint64(reqCnt), rcost)
- return p.SendTxStatus(req.ReqID, bv, pm.txStatus(req.Hashes))
+ var stats []txStatus
+ pm.servingQueue.addTask(&servingTask{
+ priority: priority,
+ run: func() (bool, error) {
+ stats = pm.txStatus(req.Hashes)
+ return true, nil
+ },
+ after: sendFunc(uint64(reqCnt), func(bv uint64) error { return p.SendTxStatus(req.ReqID, bv, stats) }),
+ })
case TxStatusMsg:
if pm.odr == nil {
@@ -1099,7 +1198,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)
@@ -1201,7 +1300,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) }
},
}
@@ -1225,7 +1324,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/helper_test.go b/les/helper_test.go
index 206ee2d920..f7e4bcf692 100644
--- a/les/helper_test.go
+++ b/les/helper_test.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/consensus/ethash"
"github.com/ethereum/go-ethereum/core"
"github.com/ethereum/go-ethereum/core/types"
@@ -183,13 +184,14 @@ func newTestProtocolManager(lightSync bool, blocks int, generator func(int, *cor
if !lightSync {
srv := &LesServer{lesCommons: lesCommons{protocolManager: pm}}
pm.server = srv
+ pm.servingQueue.setThreads(4)
srv.defParams = &flowcontrol.ServerParams{
BufLimit: testBufLimit,
MinRecharge: 1,
}
- srv.fcManager = flowcontrol.NewClientManager(50, 10, 1000000000)
+ srv.fcManager = flowcontrol.NewClientManager(nil, &mclock.System{})
srv.fcCostStats = newCostStats(nil)
}
pm.Start(1000)
@@ -375,7 +377,7 @@ func newClientServerEnv(t *testing.T, blocks int, protocol int, waitIndexers fun
db, ldb := ethdb.NewMemDatabase(), ethdb.NewMemDatabase()
peers, lPeers := newPeerSet(), newPeerSet()
- dist := newRequestDistributor(lPeers, make(chan struct{}))
+ dist := newRequestDistributor(lPeers, make(chan struct{}), &mclock.System{})
rm := newRetrieveManager(lPeers, dist, nil)
odr := NewLesOdr(ldb, light.TestClientIndexerConfig, rm)
diff --git a/les/odr.go b/les/odr.go
index 9def05a676..db8c5d4fd9 100644
--- a/les/odr.go
+++ b/les/odr.go
@@ -114,7 +114,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/peer.go b/les/peer.go
index 70c863c2ff..09e55fc719 100644
--- a/les/peer.go
+++ b/les/peer.go
@@ -26,6 +26,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"
@@ -68,6 +69,10 @@ type peer struct {
announceChn chan announceData
sendQueue *execQueue
+ errCh chan error
+ responseLock sync.Mutex
+ responseCount uint64
+
poolEntry *poolEntry
hasBlock func(common.Hash, uint64) bool
responseErrors int
@@ -487,7 +492,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.System{})
p.fcCosts = MRC.decode()
}
diff --git a/les/server.go b/les/server.go
index 2fa0456d69..41e6096f65 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"
@@ -47,6 +48,9 @@ type LesServer struct {
lesTopics []discv5.Topic
privateKey *ecdsa.PrivateKey
quitSync chan struct{}
+
+ bwcNormal, bwcBlockProcessing flowcontrol.PieceWiseLinear // bandwidth curve for normal operation and block processing mode
+ thcNormal, thcBlockProcessing int // serving thread count for normal operation and block processing mode
}
func NewLesServer(eth *eth.Ethereum, config *eth.Config) (*LesServer, error) {
@@ -102,11 +106,52 @@ func NewLesServer(eth *eth.Ethereum, config *eth.Config) (*LesServer, error) {
BufLimit: 300000000,
MinRecharge: 50000,
}
- srv.fcManager = flowcontrol.NewClientManager(uint64(config.LightServ), 10, 1000000000)
+ bwNormal := uint64(config.LightServ) * flowcontrol.FixedPointMultiplier / 100
+ srv.bwcNormal = flowcontrol.PieceWiseLinear{{0, 0}, {bwNormal / 10, bwNormal}, {bwNormal, bwNormal}}
+ // limit the serving thread count to at least 4 times the targeted average
+ // bandwidth, allowing more paralellization in short-term load spikes but
+ // still limiting the total thread count at a reasonable level
+ srv.thcNormal = int(bwNormal * 4 / flowcontrol.FixedPointMultiplier)
+ if srv.thcNormal < 4 {
+ srv.thcNormal = 4
+ }
+ // while processing blocks use half of the normal target bandwidth
+ bwBlockProcessing := bwNormal / 2
+ srv.bwcBlockProcessing = flowcontrol.PieceWiseLinear{{0, 0}, {bwBlockProcessing / 10, bwBlockProcessing}, {bwBlockProcessing, bwBlockProcessing}}
+ // limit the serving thread count just above the targeted average bandwidth,
+ // ensuring that block processing is minimally hindered
+ srv.thcBlockProcessing = int(bwBlockProcessing/flowcontrol.FixedPointMultiplier) + 1
+
+ pm.servingQueue.setThreads(srv.thcNormal)
+ srv.fcManager = flowcontrol.NewClientManager(srv.bwcNormal, &mclock.System{})
+ srv.blockProcLoop(pm)
srv.fcCostStats = newCostStats(eth.ChainDb())
return srv, nil
}
+func (s *LesServer) blockProcLoop(pm *ProtocolManager) {
+ pm.wg.Add(1)
+ procFeedback := make(chan bool, 10)
+ pm.blockchain.(*core.BlockChain).SetProcFeedback(procFeedback)
+ go func() {
+ for {
+ select {
+ case processing := <-procFeedback:
+ if processing {
+ pm.servingQueue.setThreads(s.thcBlockProcessing)
+ s.fcManager.SetRechargeCurve(s.bwcBlockProcessing)
+ } else {
+ pm.servingQueue.setThreads(s.thcNormal)
+ s.fcManager.SetRechargeCurve(s.bwcNormal)
+ }
+ case <-pm.quitSync:
+ pm.wg.Done()
+ return
+ }
+ }
+ }()
+}
+
func (s *LesServer) Protocols() []p2p.Protocol {
return s.makeProtocols(ServerProtocolVersions)
}
@@ -139,7 +184,6 @@ func (s *LesServer) Stop() {
s.chtIndexer.Close()
// bloom trie indexer is closed by parent bloombits indexer
s.fcCostStats.store()
- s.fcManager.Stop()
go func() {
<-s.protocolManager.noMorePeers
}()
diff --git a/les/servingqueue.go b/les/servingqueue.go
new file mode 100644
index 0000000000..0efb0faa84
--- /dev/null
+++ b/les/servingqueue.go
@@ -0,0 +1,177 @@
+// Copyright 2018 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 flowcontrol implements a client side flow control mechanism
+package les
+
+import (
+ "sync"
+
+ "github.com/ethereum/go-ethereum/common/mclock"
+ "github.com/ethereum/go-ethereum/common/prque"
+)
+
+// servingQueue runs serving tasks in a limited number of threads and puts the
+// waiting tasks in a priority queue
+type servingQueue struct {
+ lock sync.Mutex
+ threadCount int // number of currently running threads
+ stopCount int // number of threads to be stopped after they finish their current task
+ queue *prque.Prque // priority queue for waiting or suspended tasks
+ best *servingTask // either best == nil (queue empty) or waitingForTask is empty
+ waiting []chan *servingTask // threads waiting for a task
+ suspendBias int64 // priority bias against suspending an already running task
+}
+
+// servingTask represents a request serving task. Tasks can be implemented to
+// run in multiple steps, allowing the serving queue to suspend execution between
+// steps if higher priority tasks are entered. The creator of the task should
+// set the following fields:
+//
+// - priority: greater value means higher priority; values can wrap around the int64 range
+// - run: execute a single step; return true if finished
+// - after: executed after run finishes or returns an error, receives the total serving time
+type servingTask struct {
+ servingTime uint64
+ done bool
+ err error
+ priority int64
+ run func() (finished bool, err error)
+ after func(servingTime uint64, err error)
+}
+
+// newServingQueue returns a new servingQueue
+func newServingQueue(_suspendBias int64) *servingQueue {
+ return &servingQueue{
+ queue: prque.New(nil),
+ suspendBias: _suspendBias,
+ }
+}
+
+// addTask adds a new task, either starting it immediately or queueing it
+func (sq *servingQueue) addTask(task *servingTask) {
+ sq.lock.Lock()
+ defer sq.lock.Unlock()
+
+ if l := len(sq.waiting); l != 0 {
+ l--
+ sq.waiting[l] <- task
+ sq.waiting = sq.waiting[:l]
+ return
+ }
+
+ if sq.best == nil {
+ sq.best = task
+ return
+ }
+ if task.priority < sq.best.priority {
+ sq.queue.Push(sq.best, sq.best.priority)
+ sq.best = task
+ return
+ }
+ sq.queue.Push(task, task.priority)
+}
+
+// getNewTask selects a new task to be processed. If blocking == true then it waits
+// until a runnable task arrives or returns nil if the thread should be stopped.
+// if currentTask != nil then it returns immediately and only returns a new task
+// if the current one should be suspended.
+// Note: either blocking should be false or currentTask should be nil.
+func (sq *servingQueue) getNewTask(currentTask *servingTask, blocking bool) *servingTask {
+ sq.lock.Lock()
+ if sq.stopCount == 0 {
+ if sq.best != nil && (currentTask == nil || sq.best.priority <= currentTask.priority-sq.suspendBias) {
+ best := sq.best
+ sq.best, _ = sq.queue.PopItem().(*servingTask)
+ sq.lock.Unlock()
+ return best
+ }
+ if blocking {
+ ch := make(chan *servingTask)
+ sq.waiting = append(sq.waiting, ch)
+ sq.lock.Unlock()
+ return <-ch
+ }
+ } else {
+ sq.stopCount--
+ sq.threadCount--
+ }
+ sq.lock.Unlock()
+ return nil
+}
+
+// setThreads sets the processing thread count, suspending tasks as soon as
+// possible if necessary.
+func (sq *servingQueue) setThreads(threadCount int) {
+ sq.lock.Lock()
+ defer sq.lock.Unlock()
+
+ diff := threadCount - sq.threadCount + sq.stopCount
+ if diff > 0 {
+ // start more threads
+ if sq.stopCount >= diff {
+ sq.stopCount -= diff
+ } else {
+ diff -= sq.stopCount
+ sq.stopCount = 0
+ for ; diff > 0; diff-- {
+ go sq.servingThread()
+ }
+ }
+ }
+ if diff < 0 {
+ // stop some threads
+ lw := len(sq.waiting)
+ for diff < 0 && lw > 0 {
+ diff++
+ lw--
+ sq.waiting[lw] <- nil
+ }
+ sq.waiting = sq.waiting[:lw]
+ sq.stopCount += diff
+ }
+}
+
+// stop stops task processing as soon as possible
+func (sq *servingQueue) stop() {
+ sq.setThreads(0)
+}
+
+// servingThread implements a single serving thread
+func (sq *servingQueue) servingThread() {
+ for {
+ task := sq.getNewTask(nil, true)
+ if task == nil {
+ return
+ }
+ task.servingTime -= uint64(mclock.Now())
+ for {
+ task.done, task.err = task.run()
+ if task.done || task.err != nil {
+ task.servingTime += uint64(mclock.Now())
+ task.after(task.servingTime, task.err)
+ break
+ }
+ if newTask := sq.getNewTask(task, false); newTask != nil {
+ now := uint64(mclock.Now())
+ task.servingTime += now
+ sq.addTask(task)
+ task = newTask
+ task.servingTime -= now
+ }
+ }
+ }
+}
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) }
},
}