From 891c8a0fc3c31e43bc0880de613c1d60afb3cff2 Mon Sep 17 00:00:00 2001 From: Zsolt Felfoldi Date: Sun, 18 Mar 2018 13:56:30 +0100 Subject: [PATCH] les, les/flowcontrol: parallel request serving and improved client manager --- cmd/utils/flags.go | 2 +- core/blockchain.go | 36 +- les/backend.go | 3 +- les/distributor.go | 8 +- les/distributor_test.go | 4 +- les/fetcher.go | 2 +- les/flowcontrol/control.go | 111 +++-- les/flowcontrol/manager.go | 380 +++++++++-------- les/flowcontrol/manager_test.go | 117 ++++++ les/handler.go | 693 ++++++++++++++++++-------------- les/helper_test.go | 6 +- les/odr.go | 2 +- les/peer.go | 7 +- les/server.go | 48 ++- les/servingqueue.go | 177 ++++++++ les/txrelay.go | 2 +- 16 files changed, 1073 insertions(+), 525 deletions(-) create mode 100644 les/flowcontrol/manager_test.go create mode 100644 les/servingqueue.go 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) } }, }