forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-19 10:22:23 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

les, les/flowcontrol: parallel request serving and improved client manager

Browse source
This commit is contained in:
Zsolt Felfoldi 2018-03-18 13:56:30 +01:00
parent 62e94895da
commit 891c8a0fc3
16 changed files with 1073 additions and 525 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
cmd/utils/flags.go
View file

@ -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{

28
core/blockchain.go
View file

@ -127,6 +127,7 @@ type BlockChain struct {
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() {

3
les/backend.go
View file

@ -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),

8
les/distributor.go
View file

@ -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

4
les/distributor_test.go
View file

@ -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{}

2
les/fetcher.go
View file

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

97
les/flowcontrol/control.go
View file

@ -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)
if maxCost > peer.bufValue {
return false, maxCost - peer.bufValue, 0
}
peer.bufValue -= maxCost
peer.sumCost += maxCost
peer.accepted[index] = peer.sumCost
return true, 0, peer.cm.accepted(peer, maxCost, time)
}
func (peer *ClientNode) RequestProcessed(cost uint64) (bv, realCost uint64) {
// 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 := mclock.Now()
time := peer.cm.clock.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
}
}
return peer.bufValue, rcost
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
}
if peer.bufRecharge {
dt := uint64(time - peer.lastTime)
peer.bufEstimate += peer.params.MinRecharge * dt / uint64(fcTimeConst)
if peer.bufEstimate > peer.params.BufLimit {
if peer.bufEstimate >= peer.params.BufLimit {
peer.bufEstimate = peer.params.BufLimit
peer.bufRecharge = false
}
}
peer.lastTime = time
}
@ -127,7 +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()
}

364
les/flowcontrol/manager.go
View file

@ -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 {
clock mclock.Clock
lock sync.Mutex
nodes map[*cmNode]struct{}
simReqCnt, sumWeight, rcSumValue uint64
maxSimReq, maxRcSum uint64
rcRecharge uint64
resumeQueue chan chan bool
time mclock.AbsTime
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,
}
self.lock.Lock()
defer self.lock.Unlock()
// init initializes the ClientManager specific fields of a ClientNode structure
func (cm *ClientManager) init(node *ClientNode) {
cm.lock.Lock()
defer cm.lock.Unlock()
self.nodes[node] = struct{}{}
self.update(mclock.Now())
return node
node.corrBufValue = int64(node.params.BufLimit)
node.rcLastIntValue = cm.rcLastIntValue
node.queueIndex = -1
}
func (self *ClientManager) removeNode(node *cmNode) {
self.lock.Lock()
defer self.lock.Unlock()
// 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()
time := mclock.Now()
self.stop(node, time)
delete(self.nodes, node)
self.update(time)
cm.updateNodeRc(node, -int64(maxCost), now)
rcTime := (node.params.BufLimit - uint64(node.corrBufValue)) * FixedPointMultiplier / node.params.MinRecharge
return -int64(now) - int64(rcTime)
}
// 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
// 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
}
if node.recharging {
sumWeight += node.rcWeight
cm.updateNodeRc(node, int64(maxCost-realCost), now)
if uint64(node.corrBufValue) > node.bufValue {
node.bufValue = uint64(node.corrBufValue)
}
rcSum += uint64(node.rcValue)
}
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
}
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
}
// 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
}
lastUpdate += dtNext
cm.rcLastIntValue = rcqNode.rcFullIntValue
}
}
if self.updateNodes(firstTime) {
for node := range self.nodes {
if node.recharging {
node.set(node.serving, self.simReqCnt, self.sumWeight)
// 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)
}
}
// 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 {
self.time = time
return
h = m
}
}
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)
}
func (self *ClientManager) canStartReq() bool {
return self.simReqCnt < self.maxSimReq && self.rcSumValue < self.maxRcSum
// 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
}
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
lastX = i.X
}
}
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)
}

117
les/flowcontrol/manager_test.go Normal file
View file

@ -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 <http://www.gnu.org/licenses/>.
// 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)
}
}

279
les/handler.go
View file

@ -104,6 +104,7 @@ type ProtocolManager struct {
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 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)))
}
if cost > bufValue {
recharge := time.Duration((cost - bufValue) * 1000000 / pm.server.defParams.MinRecharge)
p.Log().Error("Request came too early", "recharge", common.PrettyDuration(recharge))
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,7 +457,11 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
headers []*types.Header
unknown bool
)
for !unknown && len(headers) < int(query.Amount) && bytes < softResponseLimit {
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 {
@ -437,7 +478,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
origin = pm.blockchain.GetHeaderByNumber(query.Origin.Number)
}
if origin == nil {
break
return true, nil
}
headers = append(headers, origin)
bytes += estHeaderRlpSize
@ -488,11 +529,13 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
// Number based traversal towards the leaf block
query.Origin.Number += query.Skip + 1
}
return false, nil
} else {
return true, nil
}
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,9 +580,15 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
if reject(uint64(reqCnt), MaxBodyFetch) {
return errResp(ErrRequestRejected, "")
}
for _, hash := range req.Hashes {
index := 0
pm.servingQueue.addTask(&servingTask{
priority: priority,
run: func() (bool, error) {
hash := req.Hashes[index]
index++
if bytes >= softResponseLimit {
break
return true, nil
}
// Retrieve the requested block body, stopping if enough was found
if number := rawdb.ReadHeaderNumber(pm.chainDb, hash); number != nil {
@ -548,10 +597,10 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
bytes += len(data)
}
}
}
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)
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 {
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 {
continue
return false, nil
}
account, err := pm.getAccount(statedb, header.Root, common.BytesToHash(req.AccKey))
if err != nil {
continue
return false, nil
}
code, _ := statedb.Database().TrieDB().Node(common.BytesToHash(account.CodeHash))
data = append(data, code)
if bytes += len(code); bytes >= softResponseLimit {
break
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,9 +712,15 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
if reject(uint64(reqCnt), MaxReceiptFetch) {
return errResp(ErrRequestRejected, "")
}
for _, hash := range req.Hashes {
index := 0
pm.servingQueue.addTask(&servingTask{
priority: priority,
run: func() (bool, error) {
hash := req.Hashes[index]
index++
if bytes >= softResponseLimit {
break
return true, nil
}
// Retrieve the requested block's receipts, skipping if unknown to us
var results types.Receipts
@ -669,7 +729,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
}
if results == nil {
if header := pm.blockchain.GetHeaderByHash(hash); header == nil || header.ReceiptHash != types.EmptyRootHash {
continue
return false, nil
}
}
// If known, encode and queue for response packet
@ -679,10 +739,10 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
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)
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,19 +784,25 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
if reject(uint64(reqCnt), MaxProofsFetch) {
return errResp(ErrRequestRejected, "")
}
for _, req := range req.Reqs {
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 {
continue
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 {
continue
return false, nil
}
trie, _ = statedb.Database().OpenStorageTrie(common.BytesToHash(req.AccKey), account.Root)
} else {
@ -748,15 +814,15 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
proofs = append(proofs, proof)
if bytes += proof.DataSize(); bytes >= softResponseLimit {
break
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,7 +847,12 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
nodes := light.NewNodeSet()
for _, req := range req.Reqs {
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
@ -794,31 +865,31 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
}
}
if statedb == nil {
continue
return false, nil
}
// 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
return false, nil
}
trie, _ = statedb.Database().OpenStorageTrie(common.BytesToHash(req.AccKey), account.Root)
} else {
trie, _ = statedb.Database().OpenTrie(root)
}
if trie == nil {
continue
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 {
break
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.SendProofsV2(req.ReqID, bv, nodes.NodeList())
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,13 +953,19 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
return errResp(ErrRequestRejected, "")
}
trieDb := trie.NewDatabase(ethdb.NewTable(pm.chainDb, light.ChtTablePrefix))
for _, req := range req.Reqs {
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 {
continue
return false, nil
}
var encNumber [8]byte
binary.BigEndian.PutUint64(encNumber[:], req.BlockNum)
@ -898,14 +975,14 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
proofs = append(proofs, ChtResp{Header: header, Proof: proof})
if bytes += proof.DataSize() + estHeaderRlpSize; bytes >= softResponseLimit {
break
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,7 +1011,13 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
auxTrie *trie.Trie
)
nodes := light.NewNodeSet()
for _, req := range req.Reqs {
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
@ -961,12 +1044,14 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
}
}
if nodes.DataSize()+auxBytes >= softResponseLimit {
break
return true, nil
}
}
bv, rcost := p.fcClient.RequestProcessed(costs.baseCost + uint64(reqCnt)*costs.reqCost)
pm.server.fcCostStats.update(msg.Code, uint64(reqCnt), rcost)
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,11 +1134,15 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
return errResp(ErrRequestRejected, "")
}
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 := pm.txStatus(hashes)
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 {
@ -1058,11 +1152,10 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
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)
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) }
},
}

6
les/helper_test.go
View file

@ -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)

2
les/odr.go
View file

@ -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) }
},
}

7
les/peer.go
View file

@ -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()
}

48
les/server.go
View file

@ -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
}()

177
les/servingqueue.go Normal file
View file

@ -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 <http://www.gnu.org/licenses/>.
// 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
}
}
}
}

2
les/txrelay.go
View file

@ -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) }
},
}