From 9381aad1deef41b15c59203861da798ec764f808 Mon Sep 17 00:00:00 2001 From: Zsolt Felfoldi Date: Sun, 18 Mar 2018 13:56:30 +0100 Subject: [PATCH] les, les/flowcontrol: improved client manager and load tests --- cmd/utils/flags.go | 2 +- common/mclock/mclock.go | 26 ++ common/mclock/simclock.go | 136 +++++++++ les/backend.go | 3 +- les/distributor.go | 8 +- les/distributor_test.go | 4 +- les/fetcher.go | 2 +- les/flowcontrol/control.go | 118 +++++--- les/flowcontrol/manager.go | 480 ++++++++++++++++++++------------ les/flowcontrol/prque/prque.go | 43 +++ les/flowcontrol/prque/sstack.go | 84 ++++++ les/handler.go | 82 ++++-- les/helper_test.go | 4 +- les/load_test.go | 422 ++++++++++++++++++++++++++++ les/odr.go | 2 +- les/odr_test.go | 3 +- les/peer.go | 3 +- les/request_test.go | 3 +- les/server.go | 8 +- les/txrelay.go | 2 +- 20 files changed, 1182 insertions(+), 253 deletions(-) create mode 100644 common/mclock/simclock.go create mode 100755 les/flowcontrol/prque/prque.go create mode 100755 les/flowcontrol/prque/sstack.go create mode 100644 les/load_test.go diff --git a/cmd/utils/flags.go b/cmd/utils/flags.go index 41a1ac35fe..2daf1ddf79 100644 --- a/cmd/utils/flags.go +++ b/cmd/utils/flags.go @@ -177,7 +177,7 @@ var ( } LightServFlag = cli.IntFlag{ Name: "lightserv", - Usage: "Maximum percentage of time allowed for serving LES requests (0-90)", + Usage: "Maximum percentage of time allowed for serving LES requests (multi-threaded processing allows values over 100)", Value: 0, } LightPeersFlag = cli.IntFlag{ diff --git a/common/mclock/mclock.go b/common/mclock/mclock.go index 02608d17b0..76d1e5cdb7 100644 --- a/common/mclock/mclock.go +++ b/common/mclock/mclock.go @@ -30,3 +30,29 @@ type AbsTime time.Duration func Now() AbsTime { return AbsTime(monotime.Now()) } + +// Clock interface makes it possible to replace the monotonic system clock with +// a simulated clock +type Clock interface { + Now() AbsTime + Sleep(time.Duration) + After(time.Duration) <-chan time.Time +} + +// MonotonicClock implements Clock using the system clock +type MonotonicClock struct{} + +// Now implements Clock +func (MonotonicClock) Now() AbsTime { + return AbsTime(monotime.Now()) +} + +// Sleep implements Clock +func (MonotonicClock) Sleep(d time.Duration) { + time.Sleep(d) +} + +// After implements Clock +func (MonotonicClock) After(d time.Duration) <-chan time.Time { + return time.After(d) +} diff --git a/common/mclock/simclock.go b/common/mclock/simclock.go new file mode 100644 index 0000000000..1b3611f6c1 --- /dev/null +++ b/common/mclock/simclock.go @@ -0,0 +1,136 @@ +// Copyright 2016 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +// package mclock is a wrapper for a monotonic clock source +package mclock + +import ( + "runtime" + "sync" + "time" +) + +type event struct { + do func() + at AbsTime +} + +// SimulatedClock implements a virtual Clock for reproducible time-sensitive tests. +// It simulates a scheduler on a virtual timescale where actual processing takes zero time. +// +// Note: since there is no way in Go to know when all goroutines have reached a waiting +// state (which should theoretically happen in each virtual moment), the algorithm runs +// GoSched a fixed number of times after each step and limits time steps in order to +// minimize precision loss (see maxStep and goSchedCount). +type SimulatedClock struct { + now AbsTime + scheduled []event + stop bool + lock sync.RWMutex +} + +const ( + maxStep = time.Microsecond * 10 + goSchedCount = 10 +) + +// NewSimulatedClock creates a new simulated clock +func NewSimulatedClock() *SimulatedClock { + s := &SimulatedClock{scheduled: make([]event, 0, 100)} + + go func() { + lastScheduled := 0 + for { + for i := 0; i < goSchedCount; i++ { + runtime.Gosched() + } + //time.Sleep(time.Microsecond * 10) + s.lock.Lock() + if s.stop { + s.lock.Unlock() + return + } + scheduled := len(s.scheduled) + if scheduled > 0 && scheduled == lastScheduled { + ev := s.scheduled[0] + if ev.at <= s.now+AbsTime(maxStep) { + s.scheduled = s.scheduled[1:] + s.now = ev.at + ev.do() + } else { + s.now += AbsTime(maxStep) + } + } + lastScheduled = scheduled + s.lock.Unlock() + } + }() + + return s +} + +// Stop stops the clock (Sleeps and Afters will never return after this) +func (s *SimulatedClock) Stop() { + s.lock.Lock() + s.stop = true + s.lock.Unlock() +} + +// Now implements Clock +func (s *SimulatedClock) Now() AbsTime { + s.lock.RLock() + defer s.lock.RUnlock() + + return s.now +} + +// Sleep implements Clock +func (s *SimulatedClock) Sleep(d time.Duration) { + done := make(chan struct{}) + s.insert(d, func() { + close(done) + }) + <-done +} + +// After implements Clock +func (s *SimulatedClock) After(d time.Duration) <-chan time.Time { + after := make(chan time.Time, 1) + s.insert(d, func() { + after <- time.Unix(0, int64(s.now)) + }) + return after +} + +func (s *SimulatedClock) insert(d time.Duration, do func()) { + s.lock.Lock() + defer s.lock.Unlock() + + at := s.now + AbsTime(d) + l, h := 0, len(s.scheduled) + ll := h + for l != h { + m := (l + h) / 2 + if at < s.scheduled[m].at { + h = m + } else { + l = m + 1 + } + } + s.scheduled = append(s.scheduled, event{}) + copy(s.scheduled[l+1:], s.scheduled[l:ll]) + s.scheduled[l] = event{do: do, at: at} +} diff --git a/les/backend.go b/les/backend.go index 35f67f29f8..e30b114eb0 100644 --- a/les/backend.go +++ b/les/backend.go @@ -25,6 +25,7 @@ import ( "github.com/ethereum/go-ethereum/accounts" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common/hexutil" + "github.com/ethereum/go-ethereum/common/mclock" "github.com/ethereum/go-ethereum/consensus" "github.com/ethereum/go-ethereum/core" "github.com/ethereum/go-ethereum/core/bloombits" @@ -100,7 +101,7 @@ func New(ctx *node.ServiceContext, config *eth.Config) (*LightEthereum, error) { chainDb: chainDb, eventMux: ctx.EventMux, peers: peers, - reqDist: newRequestDistributor(peers, quitSync), + reqDist: newRequestDistributor(peers, quitSync, &mclock.MonotonicClock{}), accountManager: ctx.AccountManager, engine: eth.CreateConsensusEngine(ctx, &config.Ethash, chainConfig, chainDb), shutdownChan: make(chan bool), diff --git a/les/distributor.go b/les/distributor.go index 159fa4c73f..149f9e2851 100644 --- a/les/distributor.go +++ b/les/distributor.go @@ -23,6 +23,8 @@ import ( "errors" "sync" "time" + + "github.com/ethereum/go-ethereum/common/mclock" ) // ErrNoPeers is returned if no peers capable of serving a queued request are available @@ -32,6 +34,7 @@ var ErrNoPeers = errors.New("no suitable peers available") // suitable peers, obeying flow control rules and prioritizing them in creation // order (even when a resend is necessary). type requestDistributor struct { + clock mclock.Clock reqQueue *list.List lastReqOrder uint64 peers map[distPeer]struct{} @@ -71,8 +74,9 @@ type distReq struct { } // newRequestDistributor creates a new request distributor -func newRequestDistributor(peers *peerSet, stopChn chan struct{}) *requestDistributor { +func newRequestDistributor(peers *peerSet, stopChn chan struct{}, clock mclock.Clock) *requestDistributor { d := &requestDistributor{ + clock: clock, reqQueue: list.New(), loopChn: make(chan struct{}, 2), stopChn: stopChn, @@ -150,7 +154,7 @@ func (d *requestDistributor) loop() { wait = distMaxWait } go func() { - time.Sleep(wait) + d.clock.Sleep(wait) d.loopChn <- struct{}{} }() break loop diff --git a/les/distributor_test.go b/les/distributor_test.go index 2891bcab49..d3a1c901bc 100644 --- a/les/distributor_test.go +++ b/les/distributor_test.go @@ -23,6 +23,8 @@ import ( "sync" "testing" "time" + + "github.com/ethereum/go-ethereum/common/mclock" ) type testDistReq struct { @@ -121,7 +123,7 @@ func testRequestDistributor(t *testing.T, resend bool) { stop := make(chan struct{}) defer close(stop) - dist := newRequestDistributor(nil, stop) + dist := newRequestDistributor(nil, stop, &mclock.MonotonicClock{}) var peers [testDistPeerCount]*testDistPeer for i := range peers { peers[i] = &testDistPeer{} diff --git a/les/fetcher.go b/les/fetcher.go index 59d3a2aa3c..d0118d349b 100644 --- a/les/fetcher.go +++ b/les/fetcher.go @@ -474,7 +474,7 @@ func (f *lightFetcher) nextRequest() (*distReq, uint64) { f.lock.Unlock() cost := p.GetRequestCost(GetBlockHeadersMsg, int(bestAmount)) - p.fcServer.QueueRequest(reqID, cost) + p.fcServer.QueuedRequest(reqID, cost) f.reqMu.Lock() f.requested[reqID] = fetchRequest{hash: bestHash, amount: bestAmount, peer: p, sent: mclock.Now()} f.reqMu.Unlock() diff --git a/les/flowcontrol/control.go b/les/flowcontrol/control.go index d50eb809cc..7abac30243 100644 --- a/les/flowcontrol/control.go +++ b/les/flowcontrol/control.go @@ -24,34 +24,44 @@ import ( "github.com/ethereum/go-ethereum/common/mclock" ) +// fcTimeConst is the time constant applied for MinRecharge during linear +// buffer recharge period const fcTimeConst = time.Millisecond +// ServerParams are the flow control parameters specified by a server for a client +// +// Note: a server can assign different amounts of bandwidth to each client by giving +// different parameters to them. type ServerParams struct { BufLimit, MinRecharge uint64 } +// ClientNode is the flow control system's representation of a client +// (used in server mode only) type ClientNode struct { params *ServerParams bufValue uint64 lastTime mclock.AbsTime lock sync.Mutex cm *ClientManager - cmNode *cmNode + cmNodeFields } +// NewClientNode returns a new ClientNode func NewClientNode(cm *ClientManager, params *ServerParams) *ClientNode { node := &ClientNode{ cm: cm, params: params, bufValue: params.BufLimit, - lastTime: mclock.Now(), + lastTime: cm.clock.Now(), } - node.cmNode = cm.addNode(node) + cm.addNode(node) return node } -func (peer *ClientNode) Remove(cm *ClientManager) { - cm.removeNode(peer.cmNode) +// Remove removes the client from the client manager +func (peer *ClientNode) Remove() { + peer.cm.removeNode(peer) } func (peer *ClientNode) recalcBV(time mclock.AbsTime) { @@ -66,35 +76,56 @@ func (peer *ClientNode) recalcBV(time mclock.AbsTime) { peer.lastTime = time } -func (peer *ClientNode) AcceptRequest() (uint64, bool) { +// AcceptRequest returns whether a new request can be accepted and the missing +// buffer amount if it was rejected due to a buffer underrun. If accepted, maxCost +// is deducted from the flow control buffer. +func (peer *ClientNode) AcceptRequest(maxCost uint64) (bool, uint64) { peer.lock.Lock() - defer peer.lock.Unlock() - time := mclock.Now() + time := peer.cm.clock.Now() peer.recalcBV(time) - return peer.bufValue, peer.cm.accept(peer.cmNode, time) -} + if maxCost > peer.bufValue { + return false, maxCost - peer.bufValue + } + peer.bufValue -= maxCost + ch := peer.cm.accept(peer, maxCost, time) -func (peer *ClientNode) RequestProcessed(cost uint64) (bv, realCost uint64) { - peer.lock.Lock() - defer peer.lock.Unlock() - - time := mclock.Now() - peer.recalcBV(time) - peer.bufValue -= cost - peer.recalcBV(time) - rcValue, rcost := peer.cm.processed(peer.cmNode, time) - if rcValue < peer.params.BufLimit { - bv := peer.params.BufLimit - rcValue - if bv > peer.bufValue { - peer.bufValue = bv + peer.lock.Unlock() + ok := true + if ch != nil { + ok = <-ch + if ok { + peer.lock.Lock() + peer.cm.started(peer, maxCost, time) + peer.lock.Unlock() } } + return ok, 0 +} + +// RequestProcessed should be called when the request has been processed +func (peer *ClientNode) RequestProcessed() (bv, realCost uint64) { + peer.lock.Lock() + defer peer.lock.Unlock() + + time := peer.cm.clock.Now() + peer.recalcBV(time) + rcost := peer.cm.processed(peer, time) return peer.bufValue, rcost } +// WaitOrStop blocks while request processing is disabled and returns true if +// it should be cancelled because the client has been disconnected. +func (peer *ClientNode) WaitOrStop() bool { + return peer.cm.waitOrStop(peer) +} + +// ServerNode is the flow control system's representation of a server +// (used in client mode only) type ServerNode struct { + clock mclock.Clock bufEstimate uint64 + bufRecharge bool lastTime mclock.AbsTime params *ServerParams sumCost uint64 // sum of req costs sent to this server @@ -102,23 +133,29 @@ type ServerNode struct { lock sync.RWMutex } -func NewServerNode(params *ServerParams) *ServerNode { +// NewServerNode returns a new ServerNode +func NewServerNode(params *ServerParams, clock mclock.Clock) *ServerNode { return &ServerNode{ + clock: clock, bufEstimate: params.BufLimit, - lastTime: mclock.Now(), + bufRecharge: false, + lastTime: clock.Now(), params: params, pending: make(map[uint64]uint64), } } func (peer *ServerNode) recalcBLE(time mclock.AbsTime) { - dt := uint64(time - peer.lastTime) if time < peer.lastTime { - dt = 0 + return } - peer.bufEstimate += peer.params.MinRecharge * dt / uint64(fcTimeConst) - if peer.bufEstimate > peer.params.BufLimit { - peer.bufEstimate = peer.params.BufLimit + if peer.bufRecharge { + dt := uint64(time - peer.lastTime) + peer.bufEstimate += peer.params.MinRecharge * dt / uint64(fcTimeConst) + if peer.bufEstimate >= peer.params.BufLimit { + peer.bufEstimate = peer.params.BufLimit + peer.bufRecharge = false + } } peer.lastTime = time } @@ -127,7 +164,7 @@ func (peer *ServerNode) recalcBLE(time mclock.AbsTime) { const safetyMargin = time.Millisecond func (peer *ServerNode) canSend(maxCost uint64) (time.Duration, float64) { - peer.recalcBLE(mclock.Now()) + peer.recalcBLE(peer.clock.Now()) maxCost += uint64(safetyMargin) * peer.params.MinRecharge / uint64(fcTimeConst) if maxCost > peer.params.BufLimit { maxCost = peer.params.BufLimit @@ -148,25 +185,29 @@ func (peer *ServerNode) CanSend(maxCost uint64) (time.Duration, float64) { return peer.canSend(maxCost) } -// QueueRequest should be called when the request has been assigned to the given +// QueuedRequest should be called when the request has been assigned to the given // server node, before putting it in the send queue. It is mandatory that requests -// are sent in the same order as the QueueRequest calls are made. -func (peer *ServerNode) QueueRequest(reqID, maxCost uint64) { +// are sent in the same order as the QueuedRequest calls are made. +func (peer *ServerNode) QueuedRequest(reqID, maxCost uint64) { peer.lock.Lock() defer peer.lock.Unlock() + peer.recalcBLE(peer.clock.Now()) + // Note: we do not know when requests actually arrive to the server so bufRecharge + // is not turned on here if buffer was full; in this case it is going to be turned + // on by the first reply's bufValue feedback peer.bufEstimate -= maxCost peer.sumCost += maxCost peer.pending[reqID] = peer.sumCost } -// GotReply adjusts estimated buffer value according to the value included in +// ReceivedReply adjusts estimated buffer value according to the value included in // the latest request reply. -func (peer *ServerNode) GotReply(reqID, bv uint64) { - +func (peer *ServerNode) ReceivedReply(reqID, bv uint64) { peer.lock.Lock() defer peer.lock.Unlock() + peer.recalcBLE(peer.clock.Now()) if bv > peer.params.BufLimit { bv = peer.params.BufLimit } @@ -180,5 +221,6 @@ func (peer *ServerNode) GotReply(reqID, bv uint64) { if bv > cc { peer.bufEstimate = bv - cc } - peer.lastTime = mclock.Now() + peer.bufRecharge = peer.bufEstimate < peer.params.BufLimit + peer.lastTime = peer.clock.Now() } diff --git a/les/flowcontrol/manager.go b/les/flowcontrol/manager.go index 28cc6f0fe7..8b9501bb1b 100644 --- a/les/flowcontrol/manager.go +++ b/les/flowcontrol/manager.go @@ -19,206 +19,338 @@ package flowcontrol import ( "sync" - "time" "github.com/ethereum/go-ethereum/common/mclock" + "github.com/ethereum/go-ethereum/les/flowcontrol/prque" ) -const rcConst = 1000000 +const ( + cmDisabled = iota // client manager is disabled, no requests are accepted + cmNormal // normal operation, maximum available bandwidth can be allocated + cmBlockProcessing // requests are accepted but the buffers are only recharged with the guaranteed minimum rate +) -type cmNode struct { - node *ClientNode - lastUpdate mclock.AbsTime - serving, recharging bool - rcWeight uint64 - rcValue, rcDelta, startValue int64 - finishRecharge mclock.AbsTime +// cmNodeFields are ClientNode fields used by the client manager +// Note: these fields are locked by the client manager's mutex +type cmNodeFields struct { + servingStarted mclock.AbsTime + servingMaxCost uint64 + corrBufValue int64 + rcLastUpdate mclock.AbsTime + rcLastIntValue, rcNextIntValue int64 } -func (node *cmNode) update(time mclock.AbsTime) { - dt := int64(time - node.lastUpdate) - node.rcValue += node.rcDelta * dt / rcConst - node.lastUpdate = time - if node.recharging && time >= node.finishRecharge { - node.recharging = false - node.rcDelta = 0 - node.rcValue = 0 - } +// rcQueueItem represents an integrator threshold value where a certain client's buffer is recharged +type rcQueueItem struct { + node *ClientNode + intValue int64 } -func (node *cmNode) set(serving bool, simReqCnt, sumWeight uint64) { - if node.serving && !serving { - node.recharging = true - sumWeight += node.rcWeight - } - node.serving = serving - if node.recharging && serving { - node.recharging = false - sumWeight -= node.rcWeight - } - - node.rcDelta = 0 - if serving { - node.rcDelta = int64(rcConst / simReqCnt) - } - if node.recharging { - node.rcDelta = -int64(node.node.cm.rcRecharge * node.rcWeight / sumWeight) - node.finishRecharge = node.lastUpdate + mclock.AbsTime(node.rcValue*rcConst/(-node.rcDelta)) - } +// Before implements prque.item +// +// Note: intValue is interpreted as mod 2^64, the difference between the highest +// and lowest value at any moment is always less than 2^63. +func (rcq rcQueueItem) Before(j interface{}) bool { + return (j.(rcQueueItem).intValue - rcq.intValue) > 0 } +// Note: valid is called under client manager mutex lock +func (rcq rcQueueItem) valid() bool { + return rcq.intValue == rcq.node.rcNextIntValue +} + +// servingQueueItem represents a queued request (prioritized by BufValue/BufLimit) +type servingQueueItem struct { + start func() bool + priority float64 +} + +// Before implements prque.item +func (sq servingQueueItem) Before(j interface{}) bool { + return sq.priority > j.(servingQueueItem).priority +} + +// ClientManager controls the bandwidth assigned to the clients of a server. +// Since ServerParams guarantee a safe lower estimate for processable requests +// even in case of all clients being active, ClientManager calculates a +// corrigated buffer value and usually allows a higher remaining buffer value +// to be returned with each reply. type ClientManager struct { - lock sync.Mutex - nodes map[*cmNode]struct{} - simReqCnt, sumWeight, rcSumValue uint64 - maxSimReq, maxRcSum uint64 - rcRecharge uint64 - resumeQueue chan chan bool - time mclock.AbsTime + clock mclock.Clock + child *ClientManager + lock sync.RWMutex + nodes map[*ClientNode]struct{} + enabledCh chan struct{} + + parallelReqs, maxParallelReqs int + targetParallelReqs float64 + servingQueue *prque.Prque + + mode int + totalRecharge float64 + forceMinRecharge, bufCorrEnabled bool + + sumRecharge uint64 + rcLastUpdate mclock.AbsTime + rcLastIntValue int64 // normalized to MRR=1000000 + rcQueue *prque.Prque } -func NewClientManager(rcTarget, maxSimReq, maxRcSum uint64) *ClientManager { +// NewClientManager returns a new client manager. Multiple client managers can +// be chained to realize priority levels. Each level has its own manager, the +// parent has the higher priority (while the parent is processing a request +// the child is disabled). +func NewClientManager(maxParallelReqs int, targetParallelReqs float64, clock mclock.Clock, child *ClientManager) *ClientManager { cm := &ClientManager{ - nodes: make(map[*cmNode]struct{}), - resumeQueue: make(chan chan bool), - rcRecharge: rcConst * rcConst / (100*rcConst/rcTarget - rcConst), - maxSimReq: maxSimReq, - maxRcSum: maxRcSum, + clock: clock, + nodes: make(map[*ClientNode]struct{}), + child: child, + servingQueue: prque.New(), + rcQueue: prque.New(), + + maxParallelReqs: maxParallelReqs, + targetParallelReqs: targetParallelReqs, } - go cm.queueProc() + cm.SetMode(cmNormal) return cm } -func (self *ClientManager) Stop() { - self.lock.Lock() - defer self.lock.Unlock() +// SetMode changes the operating mode of the manager and its children. When +// multiple priority levels are used, mode should be changed at the manager +// of the highest level. +func (cm *ClientManager) SetMode(newMode int) { + cm.lock.Lock() + defer cm.lock.Unlock() - // signal any waiting accept routines to return false - self.nodes = make(map[*cmNode]struct{}) - close(self.resumeQueue) -} - -func (self *ClientManager) addNode(cnode *ClientNode) *cmNode { - time := mclock.Now() - node := &cmNode{ - node: cnode, - lastUpdate: time, - finishRecharge: time, - rcWeight: 1, + if newMode == cm.mode { + return } - self.lock.Lock() - defer self.lock.Unlock() + cm.updateRecharge(cm.clock.Now()) - self.nodes[node] = struct{}{} - self.update(mclock.Now()) - return node -} - -func (self *ClientManager) removeNode(node *cmNode) { - self.lock.Lock() - defer self.lock.Unlock() - - time := mclock.Now() - self.stop(node, time) - delete(self.nodes, node) - self.update(time) -} - -// recalc sumWeight -func (self *ClientManager) updateNodes(time mclock.AbsTime) (rce bool) { - var sumWeight, rcSum uint64 - for node := range self.nodes { - rc := node.recharging - node.update(time) - if rc && !node.recharging { - rce = true - } - if node.recharging { - sumWeight += node.rcWeight - } - rcSum += uint64(node.rcValue) + enabled := cm.mode != cmDisabled + newEnabled := cm.mode != cmDisabled + if !enabled && newEnabled && cm.enabledCh != nil { + close(cm.enabledCh) + cm.enabledCh = nil + } + if enabled && !newEnabled { + cm.enabledCh = make(chan struct{}) } - self.sumWeight = sumWeight - self.rcSumValue = rcSum - return -} -func (self *ClientManager) update(time mclock.AbsTime) { - for { - firstTime := time - for node := range self.nodes { - if node.recharging && node.finishRecharge < firstTime { - firstTime = node.finishRecharge - } - } - if self.updateNodes(firstTime) { - for node := range self.nodes { - if node.recharging { - node.set(node.serving, self.simReqCnt, self.sumWeight) - } - } + switch newMode { + case cmDisabled: + cm.totalRecharge = 0 + cm.bufCorrEnabled = false + cm.forceMinRecharge = false + case cmNormal: + cm.totalRecharge = cm.targetParallelReqs * 1000000 + cm.bufCorrEnabled = true + cm.forceMinRecharge = false + case cmBlockProcessing: + cm.totalRecharge = 0 + cm.bufCorrEnabled = false + cm.forceMinRecharge = true + } + + cm.mode = newMode + + if cm.child != nil { + if cm.parallelReqs == 0 { + cm.child.SetMode(newMode) } else { - self.time = time + cm.child.SetMode(cmDisabled) + } + } +} + +func (cm *ClientManager) setParallelReqs(p int, time mclock.AbsTime) { + if p == cm.parallelReqs { + return + } + if cm.child != nil && cm.mode != cmDisabled { + if cm.parallelReqs == 0 { + cm.child.SetMode(cmDisabled) + } + if p == 0 { + cm.child.SetMode(cm.mode) + } + } + cm.parallelReqs = p +} + +func (cm *ClientManager) updateRecharge(time mclock.AbsTime) { + lastUpdate := cm.rcLastUpdate + cm.rcLastUpdate = time + if cm.totalRecharge == 0 { + return + } + for cm.sumRecharge > 0 { + var slope float64 + if cm.forceMinRecharge { + slope = 1 + } else { + slope = cm.totalRecharge / float64(cm.sumRecharge) + } + dt := time - lastUpdate + q := cm.rcQueue.Pop() + for q != nil && !q.(rcQueueItem).valid() { + q = cm.rcQueue.Pop() + } + if q == nil { + cm.rcLastIntValue += int64(slope * float64(dt)) + return + } + rcqItem := q.(rcQueueItem) + dtNext := mclock.AbsTime(float64(rcqItem.intValue-cm.rcLastIntValue) / slope) + if dt < dtNext { + cm.rcQueue.Push(q) + cm.rcLastIntValue += int64(slope * float64(dt)) + return + } + if rcqItem.node.corrBufValue < int64(rcqItem.node.params.BufLimit) { + rcqItem.node.corrBufValue = int64(rcqItem.node.params.BufLimit) + cm.sumRecharge -= rcqItem.node.params.MinRecharge + } + lastUpdate += dtNext + cm.rcLastIntValue = rcqItem.intValue + } +} + +func (cm *ClientManager) updateNodeRc(node *ClientNode, bvc int64, time mclock.AbsTime) { + cm.updateRecharge(time) + wasFull := true + if node.corrBufValue != int64(node.params.BufLimit) { + wasFull = false + node.corrBufValue += (cm.rcLastIntValue - node.rcLastIntValue) * int64(node.params.MinRecharge) / 1000000 + if node.corrBufValue > int64(node.params.BufLimit) { + node.corrBufValue = int64(node.params.BufLimit) + } + node.rcLastIntValue = cm.rcLastIntValue + } + node.corrBufValue += bvc + if node.corrBufValue < 0 { + node.corrBufValue = 0 + } + isFull := false + if node.corrBufValue >= int64(node.params.BufLimit) { + node.corrBufValue = int64(node.params.BufLimit) + isFull = true + } + if wasFull && !isFull { + cm.sumRecharge += node.params.MinRecharge + } + if !wasFull && isFull { + cm.sumRecharge -= node.params.MinRecharge + } + if !isFull { + node.rcLastIntValue = cm.rcLastIntValue + node.rcNextIntValue = cm.rcLastIntValue + (int64(node.params.BufLimit)-node.corrBufValue)*1000000/int64(node.params.MinRecharge) + cm.rcQueue.Push(rcQueueItem{node: node, intValue: node.rcNextIntValue}) + } +} + +// waitOrStop blocks while request processing is disabled and returns true if +// it should be cancelled because the client has been disconnected. +func (cm *ClientManager) waitOrStop(node *ClientNode) bool { + cm.lock.RLock() + _, ok := cm.nodes[node] + stop := !ok + ch := cm.enabledCh + cm.lock.RUnlock() + + if !stop && ch != nil { + <-ch + cm.lock.RLock() + _, ok = cm.nodes[node] + stop = !ok + cm.lock.RUnlock() + } + + return stop +} + +func (cm *ClientManager) Stop() { + cm.lock.Lock() + defer cm.lock.Unlock() + + cm.nodes = nil +} + +func (cm *ClientManager) addNode(node *ClientNode) { + cm.lock.Lock() + defer cm.lock.Unlock() + + node.corrBufValue = int64(node.params.BufLimit) + node.rcLastIntValue = cm.rcLastIntValue + + if cm.nodes != nil { + cm.nodes[node] = struct{}{} + } +} + +func (cm *ClientManager) removeNode(node *ClientNode) { + cm.lock.Lock() + defer cm.lock.Unlock() + + if cm.nodes != nil { + delete(cm.nodes, node) + } +} + +func (cm *ClientManager) accept(node *ClientNode, maxCost uint64, time mclock.AbsTime) chan bool { + cm.lock.Lock() + defer cm.lock.Unlock() + + if cm.parallelReqs == cm.maxParallelReqs { + ch := make(chan bool, 1) + start := func() bool { + // always called while client manager lock is held + _, started := cm.nodes[node] + ch <- started + return started + } + cm.servingQueue.Push(servingQueueItem{start, float64(node.bufValue) / float64(node.params.BufLimit)}) + return ch + } + + cm.setParallelReqs(cm.parallelReqs+1, time) + node.servingStarted = time + node.servingMaxCost = maxCost + cm.updateNodeRc(node, -int64(maxCost), time) + return nil +} + +func (cm *ClientManager) started(node *ClientNode, maxCost uint64, time mclock.AbsTime) { + cm.lock.Lock() + defer cm.lock.Unlock() + + node.servingStarted = time + node.servingMaxCost = maxCost + cm.updateNodeRc(node, -int64(maxCost), time) +} + +func (cm *ClientManager) processed(node *ClientNode, time mclock.AbsTime) (realCost uint64) { + cm.lock.Lock() + defer cm.lock.Unlock() + + realCost = uint64(time - node.servingStarted) + if realCost > node.servingMaxCost { + realCost = node.servingMaxCost + } + if !cm.forceMinRecharge { + cm.updateNodeRc(node, int64(node.servingMaxCost-realCost), time) + } + if cm.bufCorrEnabled { + if uint64(node.corrBufValue) > node.bufValue { + node.bufValue = uint64(node.corrBufValue) + } + } + + for !cm.servingQueue.Empty() { + if cm.servingQueue.Pop().(servingQueueItem).start() { return } } -} - -func (self *ClientManager) canStartReq() bool { - return self.simReqCnt < self.maxSimReq && self.rcSumValue < self.maxRcSum -} - -func (self *ClientManager) queueProc() { - for rc := range self.resumeQueue { - for { - time.Sleep(time.Millisecond * 10) - self.lock.Lock() - self.update(mclock.Now()) - cs := self.canStartReq() - self.lock.Unlock() - if cs { - break - } - } - close(rc) - } -} - -func (self *ClientManager) accept(node *cmNode, time mclock.AbsTime) bool { - self.lock.Lock() - defer self.lock.Unlock() - - self.update(time) - if !self.canStartReq() { - resume := make(chan bool) - self.lock.Unlock() - self.resumeQueue <- resume - <-resume - self.lock.Lock() - if _, ok := self.nodes[node]; !ok { - return false // reject if node has been removed or manager has been stopped - } - } - self.simReqCnt++ - node.set(true, self.simReqCnt, self.sumWeight) - node.startValue = node.rcValue - self.update(self.time) - return true -} - -func (self *ClientManager) stop(node *cmNode, time mclock.AbsTime) { - if node.serving { - self.update(time) - self.simReqCnt-- - node.set(false, self.simReqCnt, self.sumWeight) - self.update(time) - } -} - -func (self *ClientManager) processed(node *cmNode, time mclock.AbsTime) (rcValue, rcCost uint64) { - self.lock.Lock() - defer self.lock.Unlock() - - self.stop(node, time) - return uint64(node.rcValue), uint64(node.rcValue - node.startValue) + cm.setParallelReqs(cm.parallelReqs-1, time) + return } diff --git a/les/flowcontrol/prque/prque.go b/les/flowcontrol/prque/prque.go new file mode 100755 index 0000000000..542e59f98c --- /dev/null +++ b/les/flowcontrol/prque/prque.go @@ -0,0 +1,43 @@ +// This is a duplicated and slightly modified version of "gopkg.in/karalabe/cookiejar.v2/collections/prque". + +package prque + +import ( + "container/heap" +) + +// Priority queue data structure. +type Prque struct { + cont *sstack +} + +// Creates a new priority queue. +func New() *Prque { + return &Prque{newSstack()} +} + +// Pushes a value with a given priority into the queue, expanding if necessary. +func (p *Prque) Push(i item) { + heap.Push(p.cont, i) +} + +// Pops the value with the greates priority off the stack and returns it. +// Currently no shrinking is done. +func (p *Prque) Pop() item { + return heap.Pop(p.cont).(item) +} + +// Checks whether the priority queue is empty. +func (p *Prque) Empty() bool { + return p.cont.Len() == 0 +} + +// Returns the number of element in the priority queue. +func (p *Prque) Size() int { + return p.cont.Len() +} + +// Clears the contents of the priority queue. +func (p *Prque) Reset() { + *p = *New() +} diff --git a/les/flowcontrol/prque/sstack.go b/les/flowcontrol/prque/sstack.go new file mode 100755 index 0000000000..026ddeffe1 --- /dev/null +++ b/les/flowcontrol/prque/sstack.go @@ -0,0 +1,84 @@ +// This is a duplicated and slightly modified version of "gopkg.in/karalabe/cookiejar.v2/collections/prque". + +package prque + +// The size of a block of data +const blockSize = 4096 + +// A prioritized item in the sorted stack. +type item interface { + Before(interface{}) bool +} + +// Internal sortable stack data structure. Implements the Push and Pop ops for +// the stack (heap) functionality and the Len, Less and Swap methods for the +// sortability requirements of the heaps. +type sstack struct { + size int + capacity int + offset int + + blocks [][]item + active []item +} + +// Creates a new, empty stack. +func newSstack() *sstack { + result := new(sstack) + result.active = make([]item, blockSize) + result.blocks = [][]item{result.active} + result.capacity = blockSize + return result +} + +// Pushes a value onto the stack, expanding it if necessary. Required by +// heap.Interface. +func (s *sstack) Push(data interface{}) { + if s.size == s.capacity { + s.active = make([]item, blockSize) + s.blocks = append(s.blocks, s.active) + s.capacity += blockSize + s.offset = 0 + } else if s.offset == blockSize { + s.active = s.blocks[s.size/blockSize] + s.offset = 0 + } + s.active[s.offset] = data.(item) + s.offset++ + s.size++ +} + +// Pops a value off the stack and returns it. Currently no shrinking is done. +// Required by heap.Interface. +func (s *sstack) Pop() (res interface{}) { + s.size-- + s.offset-- + if s.offset < 0 { + s.offset = blockSize - 1 + s.active = s.blocks[s.size/blockSize] + } + res, s.active[s.offset] = s.active[s.offset], nil + return +} + +// Returns the length of the stack. Required by sort.Interface. +func (s *sstack) Len() int { + return s.size +} + +// Compares the priority of two elements of the stack (higher is first). +// Required by sort.Interface. +func (s *sstack) Less(i, j int) bool { + return (s.blocks[i/blockSize][i%blockSize].Before(s.blocks[j/blockSize][j%blockSize])) +} + +// Swaps two elements in the stack. Required by sort.Interface. +func (s *sstack) Swap(i, j int) { + ib, io, jb, jo := i/blockSize, i%blockSize, j/blockSize, j%blockSize + s.blocks[ib][io], s.blocks[jb][jo] = s.blocks[jb][jo], s.blocks[ib][io] +} + +// Resets the stack, effectively clearing its contents. +func (s *sstack) Reset() { + *s = *newSstack() +} diff --git a/les/handler.go b/les/handler.go index a1c16cb875..e38caad396 100644 --- a/les/handler.go +++ b/les/handler.go @@ -292,7 +292,7 @@ func (pm *ProtocolManager) handle(p *peer) error { } defer func() { if pm.server != nil && pm.server.fcManager != nil && p.fcClient != nil { - p.fcClient.Remove(pm.server.fcManager) + p.fcClient.Remove() } pm.removePeer(p.id) }() @@ -345,19 +345,20 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { } p.Log().Trace("Light Ethereum message arrived", "code", msg.Code, "bytes", msg.Size) - costs := p.fcCosts[msg.Code] reject := func(reqCnt, maxCnt uint64) bool { if p.fcClient == nil || reqCnt > maxCnt { return true } - bufValue, _ := p.fcClient.AcceptRequest() + costs := p.fcCosts[msg.Code] cost := costs.baseCost + reqCnt*costs.reqCost if cost > pm.server.defParams.BufLimit { cost = pm.server.defParams.BufLimit } - if cost > bufValue { - recharge := time.Duration((cost - bufValue) * 1000000 / pm.server.defParams.MinRecharge) - p.Log().Error("Request came too early", "recharge", common.PrettyDuration(recharge)) + + if accepted, bufShort := p.fcClient.AcceptRequest(cost); !accepted { + if bufShort > 0 { + p.Log().Error("Request came too early", "remaining", common.PrettyDuration(time.Duration(bufShort*1000000/pm.server.defParams.MinRecharge))) + } return true } return false @@ -429,6 +430,9 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { unknown bool ) for !unknown && len(headers) < int(query.Amount) && bytes < softResponseLimit { + if p.fcClient.WaitOrStop() { + return nil + } // Retrieve the next header satisfying the query var origin *types.Header if hashMode { @@ -498,7 +502,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { } } - bv, rcost := p.fcClient.RequestProcessed(costs.baseCost + query.Amount*costs.reqCost) + bv, rcost := p.fcClient.RequestProcessed() pm.server.fcCostStats.update(msg.Code, query.Amount, rcost) return p.SendBlockHeaders(req.ReqID, bv, headers) @@ -516,7 +520,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { if err := msg.Decode(&resp); err != nil { return errResp(ErrDecode, "msg %v: %v", msg, err) } - p.fcServer.GotReply(resp.ReqID, resp.BV) + p.fcServer.ReceivedReply(resp.ReqID, resp.BV) if pm.fetcher != nil && pm.fetcher.requestedID(resp.ReqID) { pm.fetcher.deliverHeaders(p, resp.ReqID, resp.Headers) } else { @@ -549,6 +553,9 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { if bytes >= softResponseLimit { break } + if p.fcClient.WaitOrStop() { + return nil + } // Retrieve the requested block body, stopping if enough was found if number := rawdb.ReadHeaderNumber(pm.chainDb, hash); number != nil { if data := rawdb.ReadBodyRLP(pm.chainDb, hash, *number); len(data) != 0 { @@ -557,7 +564,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { } } } - bv, rcost := p.fcClient.RequestProcessed(costs.baseCost + uint64(reqCnt)*costs.reqCost) + bv, rcost := p.fcClient.RequestProcessed() pm.server.fcCostStats.update(msg.Code, uint64(reqCnt), rcost) return p.SendBlockBodiesRLP(req.ReqID, bv, bodies) @@ -575,7 +582,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { if err := msg.Decode(&resp); err != nil { return errResp(ErrDecode, "msg %v: %v", msg, err) } - p.fcServer.GotReply(resp.ReqID, resp.BV) + p.fcServer.ReceivedReply(resp.ReqID, resp.BV) deliverMsg = &Msg{ MsgType: MsgBlockBodies, ReqID: resp.ReqID, @@ -602,6 +609,9 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { return errResp(ErrRequestRejected, "") } for _, req := range req.Reqs { + if p.fcClient.WaitOrStop() { + return nil + } // Retrieve the requested state entry, stopping if enough was found if number := rawdb.ReadHeaderNumber(pm.chainDb, req.BHash); number != nil { if header := rawdb.ReadHeader(pm.chainDb, req.BHash, *number); header != nil { @@ -622,7 +632,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { } } } - bv, rcost := p.fcClient.RequestProcessed(costs.baseCost + uint64(reqCnt)*costs.reqCost) + bv, rcost := p.fcClient.RequestProcessed() pm.server.fcCostStats.update(msg.Code, uint64(reqCnt), rcost) return p.SendCode(req.ReqID, bv, data) @@ -640,7 +650,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { if err := msg.Decode(&resp); err != nil { return errResp(ErrDecode, "msg %v: %v", msg, err) } - p.fcServer.GotReply(resp.ReqID, resp.BV) + p.fcServer.ReceivedReply(resp.ReqID, resp.BV) deliverMsg = &Msg{ MsgType: MsgCode, ReqID: resp.ReqID, @@ -670,6 +680,9 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { if bytes >= softResponseLimit { break } + if p.fcClient.WaitOrStop() { + return nil + } // Retrieve the requested block's receipts, skipping if unknown to us var results types.Receipts if number := rawdb.ReadHeaderNumber(pm.chainDb, hash); number != nil { @@ -688,7 +701,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { bytes += len(encoded) } } - bv, rcost := p.fcClient.RequestProcessed(costs.baseCost + uint64(reqCnt)*costs.reqCost) + bv, rcost := p.fcClient.RequestProcessed() pm.server.fcCostStats.update(msg.Code, uint64(reqCnt), rcost) return p.SendReceiptsRLP(req.ReqID, bv, receipts) @@ -706,7 +719,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { if err := msg.Decode(&resp); err != nil { return errResp(ErrDecode, "msg %v: %v", msg, err) } - p.fcServer.GotReply(resp.ReqID, resp.BV) + p.fcServer.ReceivedReply(resp.ReqID, resp.BV) deliverMsg = &Msg{ MsgType: MsgReceipts, ReqID: resp.ReqID, @@ -733,6 +746,9 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { return errResp(ErrRequestRejected, "") } for _, req := range req.Reqs { + if p.fcClient.WaitOrStop() { + return nil + } // Retrieve the requested state entry, stopping if enough was found if number := rawdb.ReadHeaderNumber(pm.chainDb, req.BHash); number != nil { if header := rawdb.ReadHeader(pm.chainDb, req.BHash, *number); header != nil { @@ -762,7 +778,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { } } } - bv, rcost := p.fcClient.RequestProcessed(costs.baseCost + uint64(reqCnt)*costs.reqCost) + bv, rcost := p.fcClient.RequestProcessed() pm.server.fcCostStats.update(msg.Code, uint64(reqCnt), rcost) return p.SendProofs(req.ReqID, bv, proofs) @@ -790,6 +806,9 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { nodes := light.NewNodeSet() for _, req := range req.Reqs { + if p.fcClient.WaitOrStop() { + return nil + } // Look up the state belonging to the request if statedb == nil || req.BHash != lastBHash { statedb, root, lastBHash = nil, common.Hash{}, req.BHash @@ -824,7 +843,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { break } } - bv, rcost := p.fcClient.RequestProcessed(costs.baseCost + uint64(reqCnt)*costs.reqCost) + bv, rcost := p.fcClient.RequestProcessed() pm.server.fcCostStats.update(msg.Code, uint64(reqCnt), rcost) return p.SendProofsV2(req.ReqID, bv, nodes.NodeList()) @@ -842,7 +861,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { if err := msg.Decode(&resp); err != nil { return errResp(ErrDecode, "msg %v: %v", msg, err) } - p.fcServer.GotReply(resp.ReqID, resp.BV) + p.fcServer.ReceivedReply(resp.ReqID, resp.BV) deliverMsg = &Msg{ MsgType: MsgProofsV1, ReqID: resp.ReqID, @@ -863,7 +882,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { if err := msg.Decode(&resp); err != nil { return errResp(ErrDecode, "msg %v: %v", msg, err) } - p.fcServer.GotReply(resp.ReqID, resp.BV) + p.fcServer.ReceivedReply(resp.ReqID, resp.BV) deliverMsg = &Msg{ MsgType: MsgProofsV2, ReqID: resp.ReqID, @@ -891,6 +910,9 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { } trieDb := trie.NewDatabase(ethdb.NewTable(pm.chainDb, light.ChtTablePrefix)) for _, req := range req.Reqs { + if p.fcClient.WaitOrStop() { + return nil + } if header := pm.blockchain.GetHeaderByNumber(req.BlockNum); header != nil { sectionHead := rawdb.ReadCanonicalHash(pm.chainDb, req.ChtNum*light.CHTFrequencyServer-1) if root := light.GetChtRoot(pm.chainDb, req.ChtNum-1, sectionHead); root != (common.Hash{}) { @@ -911,7 +933,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { } } } - bv, rcost := p.fcClient.RequestProcessed(costs.baseCost + uint64(reqCnt)*costs.reqCost) + bv, rcost := p.fcClient.RequestProcessed() pm.server.fcCostStats.update(msg.Code, uint64(reqCnt), rcost) return p.SendHeaderProofs(req.ReqID, bv, proofs) @@ -943,6 +965,9 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { ) nodes := light.NewNodeSet() for _, req := range req.Reqs { + if p.fcClient.WaitOrStop() { + return nil + } if auxTrie == nil || req.Type != lastType || req.TrieIdx != lastIdx { auxTrie, lastType, lastIdx = nil, req.Type, req.TrieIdx @@ -972,7 +997,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { break } } - bv, rcost := p.fcClient.RequestProcessed(costs.baseCost + uint64(reqCnt)*costs.reqCost) + bv, rcost := p.fcClient.RequestProcessed() pm.server.fcCostStats.update(msg.Code, uint64(reqCnt), rcost) return p.SendHelperTrieProofs(req.ReqID, bv, HelperTrieResps{Proofs: nodes.NodeList(), AuxData: auxData}) @@ -989,7 +1014,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { if err := msg.Decode(&resp); err != nil { return errResp(ErrDecode, "msg %v: %v", msg, err) } - p.fcServer.GotReply(resp.ReqID, resp.BV) + p.fcServer.ReceivedReply(resp.ReqID, resp.BV) deliverMsg = &Msg{ MsgType: MsgHeaderProofs, ReqID: resp.ReqID, @@ -1010,7 +1035,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { return errResp(ErrDecode, "msg %v: %v", msg, err) } - p.fcServer.GotReply(resp.ReqID, resp.BV) + p.fcServer.ReceivedReply(resp.ReqID, resp.BV) deliverMsg = &Msg{ MsgType: MsgHelperTrieProofs, ReqID: resp.ReqID, @@ -1032,7 +1057,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { } pm.txpool.AddRemotes(txs) - _, rcost := p.fcClient.RequestProcessed(costs.baseCost + uint64(reqCnt)*costs.reqCost) + _, rcost := p.fcClient.RequestProcessed() pm.server.fcCostStats.update(msg.Code, uint64(reqCnt), rcost) case SendTxV2Msg: @@ -1058,6 +1083,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { } stats := pm.txStatus(hashes) for i, stat := range stats { + p.fcClient.WaitOrStop() // do not return; txs can be added even if client disconnected if stat.Status == core.TxStatusUnknown { if errs := pm.txpool.AddRemotes([]*types.Transaction{req.Txs[i]}); errs[0] != nil { stats[i].Error = errs[0].Error() @@ -1067,7 +1093,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { } } - bv, rcost := p.fcClient.RequestProcessed(costs.baseCost + uint64(reqCnt)*costs.reqCost) + bv, rcost := p.fcClient.RequestProcessed() pm.server.fcCostStats.update(msg.Code, uint64(reqCnt), rcost) return p.SendTxStatus(req.ReqID, bv, stats) @@ -1088,7 +1114,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { if reject(uint64(reqCnt), MaxTxStatus) { return errResp(ErrRequestRejected, "") } - bv, rcost := p.fcClient.RequestProcessed(costs.baseCost + uint64(reqCnt)*costs.reqCost) + bv, rcost := p.fcClient.RequestProcessed() pm.server.fcCostStats.update(msg.Code, uint64(reqCnt), rcost) return p.SendTxStatus(req.ReqID, bv, pm.txStatus(req.Hashes)) @@ -1107,7 +1133,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { return errResp(ErrDecode, "msg %v: %v", msg, err) } - p.fcServer.GotReply(resp.ReqID, resp.BV) + p.fcServer.ReceivedReply(resp.ReqID, resp.BV) default: p.Log().Trace("Received unknown message", "code", msg.Code) @@ -1233,7 +1259,7 @@ func (pc *peerConnection) RequestHeadersByHash(origin common.Hash, amount int, s request: func(dp distPeer) func() { peer := dp.(*peer) cost := peer.GetRequestCost(GetBlockHeadersMsg, amount) - peer.fcServer.QueueRequest(reqID, cost) + peer.fcServer.QueuedRequest(reqID, cost) return func() { peer.RequestHeadersByHash(reqID, cost, origin, amount, skip, reverse) } }, } @@ -1257,7 +1283,7 @@ func (pc *peerConnection) RequestHeadersByNumber(origin uint64, amount int, skip request: func(dp distPeer) func() { peer := dp.(*peer) cost := peer.GetRequestCost(GetBlockHeadersMsg, amount) - peer.fcServer.QueueRequest(reqID, cost) + peer.fcServer.QueuedRequest(reqID, cost) return func() { peer.RequestHeadersByNumber(reqID, cost, origin, amount, skip, reverse) } }, } diff --git a/les/helper_test.go b/les/helper_test.go index 8fd01a39e0..5ab8128ac0 100644 --- a/les/helper_test.go +++ b/les/helper_test.go @@ -26,6 +26,7 @@ import ( "testing" "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/common/mclock" "github.com/ethereum/go-ethereum/consensus/ethash" "github.com/ethereum/go-ethereum/core" "github.com/ethereum/go-ethereum/core/types" @@ -191,7 +192,8 @@ func newTestProtocolManager(lightSync bool, blocks int, generator func(int, *cor MinRecharge: 1, } - srv.fcManager = flowcontrol.NewClientManager(50, 10, 1000000000) + //srv.fcManager = flowcontrol.NewClientManager(50, 10, 1000000000) + srv.fcManager = flowcontrol.NewClientManager(16, 4, &mclock.MonotonicClock{}, nil) srv.fcCostStats = newCostStats(nil) } pm.Start(1000) diff --git a/les/load_test.go b/les/load_test.go new file mode 100644 index 0000000000..901824367d --- /dev/null +++ b/les/load_test.go @@ -0,0 +1,422 @@ +// Copyright 2017 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +// Package light implements on-demand retrieval capable state and chain objects +// for the Ethereum Light Client. +package les + +import ( + "math/rand" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/ethereum/go-ethereum/common/mclock" + "github.com/ethereum/go-ethereum/les/flowcontrol" +) + +type testLoadPeer struct { + fcClient *flowcontrol.ClientNode + fcServer *flowcontrol.ServerNode + serveCh chan uint64 +} + +func newTestLoadPeer(t *testing.T, client *testLoadClient, server *testLoadServer, params *flowcontrol.ServerParams, free bool, quit chan struct{}, clock mclock.Clock) *testLoadPeer { + cm := server.fcManager + if free { + cm = server.fcManagerFree + } + + peer := &testLoadPeer{ + fcClient: flowcontrol.NewClientNode(cm, params), + fcServer: flowcontrol.NewServerNode(params, clock), + serveCh: make(chan uint64, 1000), + } + client.dist.registerTestPeer(peer) + + type reply struct { + time mclock.AbsTime + reqID, bv uint64 + } + replyCh := make(chan reply, 1000) + + go func() { + avgServeTime := time.Millisecond + randMax := avgServeTime * 2 + + for { + select { + case reqID := <-peer.serveCh: + ok, bufShort := peer.fcClient.AcceptRequest(testRequestCost) + if !ok { + recharge := time.Duration(bufShort * 1000000 / params.MinRecharge) + t.Errorf("Request came too early (%v)", recharge) + } + //testClock.Sleep(time.Microsecond * 500) + serveTime := time.Duration(rand.Int63n(int64(randMax))) + randMax += (avgServeTime - serveTime) / 16 + clock.Sleep(serveTime) + bvAfter, _ := peer.fcClient.RequestProcessed() + replyCh <- reply{clock.Now(), reqID, bvAfter} + atomic.AddUint64(&server.count, 1) + case <-quit: + return + } + } + }() + + go func() { + for { + select { + case reply := <-replyCh: + wait := time.Duration(reply.time-clock.Now()) + testMessageDelay + if wait > 0 { + clock.Sleep(wait) + } + peer.fcServer.ReceivedReply(reply.reqID, reply.bv) + case <-quit: + return + } + } + }() + + return peer +} + +func (p *testLoadPeer) waitBefore(maxCost uint64) (time.Duration, float64) { + return p.fcServer.CanSend(maxCost) +} + +func (p *testLoadPeer) canQueue() bool { + return true +} + +func (p *testLoadPeer) queueSend(f func()) { + f() +} + +type testLoadTask struct { + procTime time.Duration + finished chan struct{} +} + +type testLoadServer struct { + fcManager, fcManagerFree *flowcontrol.ClientManager + count uint64 +} + +func newTestLoadServer(capacity int, quit chan struct{}, clock mclock.Clock) *testLoadServer { + f := flowcontrol.NewClientManager(16, float64(capacity)/1000, clock, nil) + s := &testLoadServer{ + fcManager: flowcontrol.NewClientManager(16, float64(capacity)/1000, clock, f), + fcManagerFree: f, + } + go func() { + <-quit + s.fcManager.Stop() + s.fcManagerFree.Stop() + }() + return s +} + +func (s *testLoadServer) requestsProcessed() uint64 { + return atomic.LoadUint64(&s.count) +} + +type testLoadClient struct { + dist *requestDistributor + quit chan struct{} + count uint64 +} + +func newTestLoadClient(quit chan struct{}, clock mclock.Clock) *testLoadClient { + return &testLoadClient{ + dist: newRequestDistributor(nil, quit, clock), + quit: quit, + } +} + +func (c *testLoadClient) sendRequests(send bool, sw chan bool) { + expCh := make(chan struct{}, 100) + for { + if send { + select { + case send = <-sw: + case expCh <- struct{}{}: + reqID := genReqID() + rq := &distReq{ + getCost: func(dp distPeer) uint64 { + return testRequestCost + }, + canSend: func(dp distPeer) bool { + return true + }, + request: func(dp distPeer) func() { + peer := dp.(*testLoadPeer) + peer.fcServer.QueuedRequest(reqID, testRequestCost) + return func() { + peer.serveCh <- reqID + } + }, + } + + sentCh := c.dist.queue(rq) + go func() { + <-sentCh + atomic.AddUint64(&c.count, 1) + <-expCh + }() + case <-c.quit: + return + } + } else { + select { + case send = <-sw: + case <-c.quit: + return + } + } + } +} + +func (c *testLoadClient) requestsSent() uint64 { + return atomic.LoadUint64(&c.count) +} + +const ( + testRequestCost = 3000000 + /*testClientCount = 2 + testServerCount = 2*/ + testMessageDelay = time.Millisecond * 200 + defaultTolerance = 5 // percent +) + +type testServerPeriod struct { + mode int + measureOff, measureOn int // duration in milliseconds + expResult, tolerance uint64 +} + +type testConnection struct { + minCapacity int // request per second guaranteed by MRR + free bool +} + +type testServerParams struct { + capacity int // processing request per second + periods []testServerPeriod +} + +type testClientPeriod struct { + sendOff, measureOff, measureOn int // duration in milliseconds + expResult, tolerance uint64 +} + +type testClientParams struct { + periods []testClientPeriod + servers []testConnection +} + +func testLoad(t *testing.T, serverParams []testServerParams, clientParams []testClientParams) { + quit := make(chan struct{}) + defer close(quit) + + //clock := &mclock.MonotonicClock{} + clock := mclock.NewSimulatedClock() + defer clock.Stop() + + var wg sync.WaitGroup + + servers := make([]*testLoadServer, len(serverParams)) + for i, params := range serverParams { + i, params := i, params + servers[i] = newTestLoadServer(params.capacity, quit, clock) + wg.Add(1) + go func() { + for k, p := range params.periods { + servers[i].fcManager.SetMode(p.mode) + clock.Sleep(time.Millisecond * time.Duration(p.measureOff)) + start := servers[i].requestsProcessed() + clock.Sleep(time.Millisecond * time.Duration(p.measureOn)) + result := servers[i].requestsProcessed() - start + relTol := p.tolerance + if relTol == 0 { + relTol = defaultTolerance + } + tolerance := p.expResult * relTol / 100 + expMin := p.expResult - tolerance + expMax := p.expResult + tolerance + if result < expMin || result > expMax { + t.Errorf("servers[%d].periods[%d] processed count mismatch (processed %d, expected between %d and %d)", i, k, result, expMin, expMax) + } + } + wg.Done() + }() + } + + clients := make([]*testLoadClient, len(clientParams)) + for i, params := range clientParams { + i, params := i, params + clients[i] = newTestLoadClient(quit, clock) + for j, conn := range params.servers { + if conn.minCapacity > 0 { + params := &flowcontrol.ServerParams{ + BufLimit: 18000000 * uint64(conn.minCapacity), + MinRecharge: 3000 * uint64(conn.minCapacity), + } + newTestLoadPeer(t, clients[i], servers[j], params, conn.free, quit, clock) + } + } + sw := make(chan bool) + go clients[i].sendRequests(false, sw) + wg.Add(1) + go func() { + for k, p := range params.periods { + clock.Sleep(time.Millisecond * time.Duration(p.sendOff)) + sw <- true + clock.Sleep(time.Millisecond * time.Duration(p.measureOff)) + start := clients[i].requestsSent() + clock.Sleep(time.Millisecond * time.Duration(p.measureOn)) + result := clients[i].requestsSent() - start + relTol := p.tolerance + if relTol == 0 { + relTol = defaultTolerance + } + tolerance := p.expResult * relTol / 100 + expMin := p.expResult - tolerance + expMax := p.expResult + tolerance + if result < expMin || result > expMax { + t.Errorf("clients[%d].periods[%d] sent count mismatch (sent %d, expected between %d and %d)", i, k, result, expMin, expMax) + } + sw <- false + } + wg.Done() + }() + } + + wg.Wait() +} + +func TestLoadBalance(t *testing.T) { + testLoad(t, + []testServerParams{ + {capacity: 1000, periods: []testServerPeriod{{mode: 1, measureOff: 3000, measureOn: 5000, expResult: 5000}}}, + }, + []testClientParams{ + {[]testClientPeriod{{sendOff: 0, measureOff: 3000, measureOn: 5000, expResult: 1000}}, []testConnection{{minCapacity: 30}}}, + {[]testClientPeriod{{sendOff: 0, measureOff: 3000, measureOn: 5000, expResult: 1000}}, []testConnection{{minCapacity: 30}}}, + {[]testClientPeriod{{sendOff: 0, measureOff: 3000, measureOn: 5000, expResult: 1000}}, []testConnection{{minCapacity: 30}}}, + {[]testClientPeriod{{sendOff: 0, measureOff: 3000, measureOn: 5000, expResult: 2000}}, []testConnection{{minCapacity: 60}}}, + }) +} + +func TestLoadBalanceMultiServer1(t *testing.T) { + testLoad(t, + []testServerParams{ + {capacity: 200, periods: []testServerPeriod{{mode: 1, measureOff: 3000, measureOn: 5000, expResult: 1000}}}, + {capacity: 200, periods: []testServerPeriod{{mode: 1, measureOff: 3000, measureOn: 5000, expResult: 1000}}}, + {capacity: 200, periods: []testServerPeriod{{mode: 1, measureOff: 3000, measureOn: 5000, expResult: 1000}}}, + {capacity: 400, periods: []testServerPeriod{{mode: 1, measureOff: 3000, measureOn: 5000, expResult: 2000}}}, + }, + []testClientParams{ + {[]testClientPeriod{{sendOff: 0, measureOff: 3000, measureOn: 5000, expResult: 1000}}, []testConnection{{minCapacity: 30}, {minCapacity: 30}, {minCapacity: 30}, {minCapacity: 30}}}, + {[]testClientPeriod{{sendOff: 0, measureOff: 3000, measureOn: 5000, expResult: 1000}}, []testConnection{{minCapacity: 30}, {minCapacity: 30}, {minCapacity: 30}, {minCapacity: 30}}}, + {[]testClientPeriod{{sendOff: 0, measureOff: 3000, measureOn: 5000, expResult: 1000}}, []testConnection{{minCapacity: 30}, {minCapacity: 30}, {minCapacity: 30}, {minCapacity: 30}}}, + {[]testClientPeriod{{sendOff: 0, measureOff: 3000, measureOn: 5000, expResult: 2000}}, []testConnection{{minCapacity: 60}, {minCapacity: 60}, {minCapacity: 60}, {minCapacity: 60}}}, + }) +} + +func TestLoadBalanceMultiServer2(t *testing.T) { + testLoad(t, + []testServerParams{ + {capacity: 250, periods: []testServerPeriod{{mode: 1, measureOff: 3000, measureOn: 5000, expResult: 1250}}}, + {capacity: 250, periods: []testServerPeriod{{mode: 1, measureOff: 3000, measureOn: 5000, expResult: 1250}}}, + {capacity: 250, periods: []testServerPeriod{{mode: 1, measureOff: 3000, measureOn: 5000, expResult: 1250}}}, + {capacity: 250, periods: []testServerPeriod{{mode: 1, measureOff: 3000, measureOn: 5000, expResult: 1250}}}, + }, + []testClientParams{ + {[]testClientPeriod{{sendOff: 0, measureOff: 3000, measureOn: 5000, expResult: 1000}}, []testConnection{{minCapacity: 60}, {minCapacity: 60}, {}, {}}}, + {[]testClientPeriod{{sendOff: 0, measureOff: 3000, measureOn: 5000, expResult: 1000}}, []testConnection{{}, {}, {minCapacity: 60}, {minCapacity: 60}}}, + {[]testClientPeriod{{sendOff: 0, measureOff: 3000, measureOn: 5000, expResult: 1000}}, []testConnection{{minCapacity: 10}, {minCapacity: 10}, {minCapacity: 10}, {minCapacity: 90}}}, + {[]testClientPeriod{{sendOff: 0, measureOff: 3000, measureOn: 5000, expResult: 2000}}, []testConnection{{minCapacity: 80}, {minCapacity: 80}, {minCapacity: 80}, {}}}, + }) +} + +func TestLoadSingle1(t *testing.T) { + testLoad(t, + []testServerParams{ + {capacity: 2000, periods: []testServerPeriod{{mode: 1, measureOff: 3000, measureOn: 5000, expResult: 4500, tolerance: 20}}}, + }, + []testClientParams{ + {[]testClientPeriod{{sendOff: 0, measureOff: 3000, measureOn: 5000, expResult: 4500, tolerance: 20}}, []testConnection{{minCapacity: 30}}}, + }) +} + +func TestLoadSingle2(t *testing.T) { + testLoad(t, + []testServerParams{ + {capacity: 2000, periods: []testServerPeriod{{mode: 1, measureOff: 3000, measureOn: 5000, expResult: 5000}}}, + }, + []testClientParams{ + {[]testClientPeriod{{sendOff: 0, measureOff: 3000, measureOn: 5000, expResult: 5000}}, []testConnection{{minCapacity: 100}}}, + }) +} + +func TestLoadSingle3(t *testing.T) { + testLoad(t, + []testServerParams{ + {capacity: 500, periods: []testServerPeriod{{mode: 1, measureOff: 3000, measureOn: 5000, expResult: 2500}}}, + }, + []testClientParams{ + {[]testClientPeriod{{sendOff: 0, measureOff: 3000, measureOn: 5000, expResult: 2500}}, []testConnection{{minCapacity: 30}}}, + }) +} + +func TestLoadPriority(t *testing.T) { + testLoad(t, + []testServerParams{ + {capacity: 500, periods: []testServerPeriod{ + {mode: 1, measureOff: 3000, measureOn: 5000, expResult: 2500}, + {mode: 1, measureOff: 3000, measureOn: 5000, expResult: 2500}, + {mode: 1, measureOff: 3000, measureOn: 5000, expResult: 3750}, + {mode: 1, measureOff: 3000, measureOn: 5000, expResult: 2500}, + }}, + }, + []testClientParams{ + // paying client + {[]testClientPeriod{ + {sendOff: 0, measureOff: 3000, measureOn: 5000, expResult: 2500}, + {sendOff: 8000, measureOff: 3000, measureOn: 5000, expResult: 2500}, + }, []testConnection{{minCapacity: 30}}}, + // free client + {[]testClientPeriod{ + {sendOff: 8000, measureOff: 3000, measureOn: 5000, expResult: 2500}, + {sendOff: 0, measureOff: 3000, measureOn: 5000, expResult: 1250}, + {sendOff: 0, measureOff: 3000, measureOn: 5000, expResult: 2500}, + }, []testConnection{{minCapacity: 30, free: true}}}, + }) +} + +func TestLoadMinRecharge(t *testing.T) { + testLoad(t, + []testServerParams{ + {capacity: 500, periods: []testServerPeriod{{mode: 2, measureOff: 3000, measureOn: 5000, expResult: 150}}}, + }, + []testClientParams{ + {[]testClientPeriod{{sendOff: 0, measureOff: 3000, measureOn: 5000, expResult: 150}}, []testConnection{{minCapacity: 30}}}, + }) +} diff --git a/les/odr.go b/les/odr.go index f8412aaad7..e224df3484 100644 --- a/les/odr.go +++ b/les/odr.go @@ -103,7 +103,7 @@ func (odr *LesOdr) Retrieve(ctx context.Context, req light.OdrRequest) (err erro request: func(dp distPeer) func() { p := dp.(*peer) cost := lreq.GetCost(p) - p.fcServer.QueueRequest(reqID, cost) + p.fcServer.QueuedRequest(reqID, cost) return func() { lreq.Request(reqID, p) } }, } diff --git a/les/odr_test.go b/les/odr_test.go index 983f7262b0..bac143c978 100644 --- a/les/odr_test.go +++ b/les/odr_test.go @@ -25,6 +25,7 @@ import ( "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common/math" + "github.com/ethereum/go-ethereum/common/mclock" "github.com/ethereum/go-ethereum/core" "github.com/ethereum/go-ethereum/core/rawdb" "github.com/ethereum/go-ethereum/core/state" @@ -163,7 +164,7 @@ func odrContractCall(ctx context.Context, db ethdb.Database, config *params.Chai func testOdr(t *testing.T, protocol int, expFail uint64, fn odrTestFn) { // Assemble the test environment peers := newPeerSet() - dist := newRequestDistributor(peers, make(chan struct{})) + dist := newRequestDistributor(peers, make(chan struct{}), &mclock.MonotonicClock{}) rm := newRetrieveManager(peers, dist, nil) db := ethdb.NewMemDatabase() ldb := ethdb.NewMemDatabase() diff --git a/les/peer.go b/les/peer.go index eb7452e276..20fbc2b2cc 100644 --- a/les/peer.go +++ b/les/peer.go @@ -27,6 +27,7 @@ import ( "time" "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/common/mclock" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/eth" "github.com/ethereum/go-ethereum/les/flowcontrol" @@ -487,7 +488,7 @@ func (p *peer) Handshake(td *big.Int, head common.Hash, headNum uint64, genesis return err } p.fcServerParams = params - p.fcServer = flowcontrol.NewServerNode(params) + p.fcServer = flowcontrol.NewServerNode(params, &mclock.MonotonicClock{}) p.fcCosts = MRC.decode() } diff --git a/les/request_test.go b/les/request_test.go index ba2f603d8b..30fff2b0ea 100644 --- a/les/request_test.go +++ b/les/request_test.go @@ -22,6 +22,7 @@ import ( "time" "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/common/mclock" "github.com/ethereum/go-ethereum/core/rawdb" "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/eth" @@ -85,7 +86,7 @@ func tfCodeAccess(db ethdb.Database, bhash common.Hash, num uint64) light.OdrReq func testAccess(t *testing.T, protocol int, fn accessTestFn) { // Assemble the test environment peers := newPeerSet() - dist := newRequestDistributor(peers, make(chan struct{})) + dist := newRequestDistributor(peers, make(chan struct{}), &mclock.MonotonicClock{}) rm := newRetrieveManager(peers, dist, nil) db := ethdb.NewMemDatabase() ldb := ethdb.NewMemDatabase() diff --git a/les/server.go b/les/server.go index fca6124c9c..9e0dcc96bc 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" @@ -98,7 +99,12 @@ func NewLesServer(eth *eth.Ethereum, config *eth.Config) (*LesServer, error) { BufLimit: 300000000, MinRecharge: 50000, } - srv.fcManager = flowcontrol.NewClientManager(uint64(config.LightServ), 10, 1000000000) + tpr := float64(config.LightServ) / 100 + mpr := int(tpr * 4) + if mpr < 4 { + mpr = 4 + } + srv.fcManager = flowcontrol.NewClientManager(mpr, tpr, &mclock.MonotonicClock{}, nil) srv.fcCostStats = newCostStats(eth.ChainDb()) return srv, nil } 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) } }, }