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

Merge b3d815e31d into e07e507d1a

Browse source
This commit is contained in:
Felföldi Zsolt 2018-08-13 17:33:48 +00:00 committed by GitHub
commit da3ebaad90
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
22 changed files with 1265 additions and 281 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
cmd/utils/flags.go
View file

@ -178,7 +178,7 @@ var (
}
LightServFlag = cli.IntFlag{
Name: "lightserv",
Usage: "Maximum percentage of time allowed for serving LES requests (0-90)",
Usage: "Maximum percentage of time allowed for serving LES requests (multi-threaded processing allows values over 100)",
Value: 0,
}
LightPeersFlag = cli.IntFlag{

26
common/mclock/mclock.go
View file

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

136
common/mclock/simclock.go Normal file
View file

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

36
core/blockchain.go
View file

@ -123,10 +123,11 @@ type BlockChain struct {
procInterrupt int32 // interrupt signaler for block processing
wg sync.WaitGroup // chain processing wait group for shutting down
engine consensus.Engine
processor Processor // block processor interface
validator Validator // block and state validator interface
vmConfig vm.Config
engine consensus.Engine
processor Processor // block processor interface
validator Validator // block and state validator interface
vmConfig vm.Config
procFeedback chan bool
badBlocks *lru.Cache // Bad block cache
}
@ -348,6 +349,14 @@ func (bc *BlockChain) CurrentFastBlock() *types.Block {
return bc.currentFastBlock.Load().(*types.Block)
}
// SetProcFeedback adds a feedback channel where true is sent each time block
// processing begins and false is sent when it is finished.
func (bc *BlockChain) SetProcFeedback(procFeedback chan bool) {
bc.procmu.Lock()
defer bc.procmu.Unlock()
bc.procFeedback = procFeedback
}
// SetProcessor sets the processor required for making state modifications.
func (bc *BlockChain) SetProcessor(processor Processor) {
bc.procmu.Lock()
@ -1013,6 +1022,25 @@ func (bc *BlockChain) insertChain(chain types.Blocks) (int, []interface{}, []*ty
if len(chain) == 0 {
return 0, nil, nil, nil
}
// send block processing feedback if needed
bc.procmu.RLock()
procFeedback := bc.procFeedback
bc.procmu.RUnlock()
if procFeedback != nil {
select {
case procFeedback <- true:
default:
}
defer func() {
select {
case procFeedback <- false:
default:
}
}()
}
// Do a sanity check that the provided chain is actually ordered and linked
for i := 1; i < len(chain); i++ {
if chain[i].NumberU64() != chain[i-1].NumberU64()+1 || chain[i].ParentHash() != chain[i-1].Hash() {

3
les/backend.go
View file

@ -25,6 +25,7 @@ import (
"github.com/ethereum/go-ethereum/accounts"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/common/hexutil"
"github.com/ethereum/go-ethereum/common/mclock"
"github.com/ethereum/go-ethereum/consensus"
"github.com/ethereum/go-ethereum/core"
"github.com/ethereum/go-ethereum/core/bloombits"
@ -100,7 +101,7 @@ func New(ctx *node.ServiceContext, config *eth.Config) (*LightEthereum, error) {
chainDb: chainDb,
eventMux: ctx.EventMux,
peers: peers,
reqDist: newRequestDistributor(peers, quitSync),
reqDist: newRequestDistributor(peers, quitSync, &mclock.MonotonicClock{}),
accountManager: ctx.AccountManager,
engine: eth.CreateConsensusEngine(ctx, chainConfig, &config.Ethash, nil, chainDb),
shutdownChan: make(chan bool),

8
les/distributor.go
View file

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

4
les/distributor_test.go
View file

@ -23,6 +23,8 @@ import (
"sync"
"testing"
"time"
"github.com/ethereum/go-ethereum/common/mclock"
)
type testDistReq struct {
@ -121,7 +123,7 @@ func testRequestDistributor(t *testing.T, resend bool) {
stop := make(chan struct{})
defer close(stop)
dist := newRequestDistributor(nil, stop)
dist := newRequestDistributor(nil, stop, &mclock.MonotonicClock{})
var peers [testDistPeerCount]*testDistPeer
for i := range peers {
peers[i] = &testDistPeer{}

2
les/fetcher.go
View file

@ -481,7 +481,7 @@ func (f *lightFetcher) nextRequest() (*distReq, uint64) {
f.lock.Unlock()
cost := p.GetRequestCost(GetBlockHeadersMsg, int(bestAmount))
p.fcServer.QueueRequest(reqID, cost)
p.fcServer.QueuedRequest(reqID, cost)
f.reqMu.Lock()
f.requested[reqID] = fetchRequest{hash: bestHash, amount: bestAmount, peer: p, sent: mclock.Now()}
f.reqMu.Unlock()

118
les/flowcontrol/control.go
View file

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

480
les/flowcontrol/manager.go
View file

@ -19,206 +19,338 @@ package flowcontrol
import (
"sync"
"time"
"github.com/ethereum/go-ethereum/common/mclock"
"github.com/ethereum/go-ethereum/les/flowcontrol/prque"
)
const rcConst = 1000000
const (
CmDisabled = iota // client manager is disabled, no requests are accepted
CmNormal // normal operation, maximum available bandwidth can be allocated
CmBlockProcessing // requests are accepted but the buffers are only recharged with the guaranteed minimum rate
)
type cmNode struct {
node *ClientNode
lastUpdate mclock.AbsTime
serving, recharging bool
rcWeight uint64
rcValue, rcDelta, startValue int64
finishRecharge mclock.AbsTime
// cmNodeFields are ClientNode fields used by the client manager
// Note: these fields are locked by the client manager's mutex
type cmNodeFields struct {
servingStarted mclock.AbsTime
servingMaxCost uint64
corrBufValue int64
rcLastUpdate mclock.AbsTime
rcLastIntValue, rcNextIntValue int64
}
func (node *cmNode) update(time mclock.AbsTime) {
dt := int64(time - node.lastUpdate)
node.rcValue += node.rcDelta * dt / rcConst
node.lastUpdate = time
if node.recharging && time >= node.finishRecharge {
node.recharging = false
node.rcDelta = 0
node.rcValue = 0
}
// rcQueueItem represents an integrator threshold value where a certain client's buffer is recharged
type rcQueueItem struct {
node *ClientNode
intValue int64
}
func (node *cmNode) set(serving bool, simReqCnt, sumWeight uint64) {
if node.serving && !serving {
node.recharging = true
sumWeight += node.rcWeight
}
node.serving = serving
if node.recharging && serving {
node.recharging = false
sumWeight -= node.rcWeight
}
node.rcDelta = 0
if serving {
node.rcDelta = int64(rcConst / simReqCnt)
}
if node.recharging {
node.rcDelta = -int64(node.node.cm.rcRecharge * node.rcWeight / sumWeight)
node.finishRecharge = node.lastUpdate + mclock.AbsTime(node.rcValue*rcConst/(-node.rcDelta))
}
// Before implements prque.item
//
// Note: intValue is interpreted as mod 2^64, the difference between the highest
// and lowest value at any moment is always less than 2^63.
func rcQueueCompare(i, j interface{}) bool {
return (j.(rcQueueItem).intValue - i.(rcQueueItem).intValue) > 0
}
// Note: valid is called under client manager mutex lock
func (rcq rcQueueItem) valid() bool {
return rcq.intValue == rcq.node.rcNextIntValue
}
// servingQueueItem represents a queued request (prioritized by BufValue/BufLimit)
type servingQueueItem struct {
start func() bool
priority float64
}
// Before implements prque.item
func servingQueueCompare(i, j interface{}) bool {
return i.(servingQueueItem).priority > j.(servingQueueItem).priority
}
// ClientManager controls the bandwidth assigned to the clients of a server.
// Since ServerParams guarantee a safe lower estimate for processable requests
// even in case of all clients being active, ClientManager calculates a
// corrigated buffer value and usually allows a higher remaining buffer value
// to be returned with each reply.
type ClientManager struct {
lock sync.Mutex
nodes map[*cmNode]struct{}
simReqCnt, sumWeight, rcSumValue uint64
maxSimReq, maxRcSum uint64
rcRecharge uint64
resumeQueue chan chan bool
time mclock.AbsTime
clock mclock.Clock
child *ClientManager
lock sync.RWMutex
nodes map[*ClientNode]struct{}
enabledCh chan struct{}
parallelReqs, maxParallelReqs int
targetParallelReqs float64
servingQueue *prque.Prque
mode int
totalRecharge float64
forceMinRecharge, bufCorrEnabled bool
sumRecharge uint64
rcLastUpdate mclock.AbsTime
rcLastIntValue int64 // normalized to MRR=1000000
rcQueue *prque.Prque
}
func NewClientManager(rcTarget, maxSimReq, maxRcSum uint64) *ClientManager {
// NewClientManager returns a new client manager. Multiple client managers can
// be chained to realize priority levels. Each level has its own manager, the
// parent has the higher priority (while the parent is processing a request
// the child is disabled).
func NewClientManager(maxParallelReqs int, targetParallelReqs float64, clock mclock.Clock, child *ClientManager) *ClientManager {
cm := &ClientManager{
nodes: make(map[*cmNode]struct{}),
resumeQueue: make(chan chan bool),
rcRecharge: rcConst * rcConst / (100*rcConst/rcTarget - rcConst),
maxSimReq: maxSimReq,
maxRcSum: maxRcSum,
clock: clock,
nodes: make(map[*ClientNode]struct{}),
child: child,
servingQueue: prque.New(servingQueueCompare),
rcQueue: prque.New(rcQueueCompare),
maxParallelReqs: maxParallelReqs,
targetParallelReqs: targetParallelReqs,
}
go cm.queueProc()
cm.SetMode(CmNormal)
return cm
}
func (self *ClientManager) Stop() {
self.lock.Lock()
defer self.lock.Unlock()
// SetMode changes the operating mode of the manager and its children. When
// multiple priority levels are used, mode should be changed at the manager
// of the highest level.
func (cm *ClientManager) SetMode(newMode int) {
cm.lock.Lock()
defer cm.lock.Unlock()
// signal any waiting accept routines to return false
self.nodes = make(map[*cmNode]struct{})
close(self.resumeQueue)
}
func (self *ClientManager) addNode(cnode *ClientNode) *cmNode {
time := mclock.Now()
node := &cmNode{
node: cnode,
lastUpdate: time,
finishRecharge: time,
rcWeight: 1,
if newMode == cm.mode {
return
}
self.lock.Lock()
defer self.lock.Unlock()
cm.updateRecharge(cm.clock.Now())
self.nodes[node] = struct{}{}
self.update(mclock.Now())
return node
}
func (self *ClientManager) removeNode(node *cmNode) {
self.lock.Lock()
defer self.lock.Unlock()
time := mclock.Now()
self.stop(node, time)
delete(self.nodes, node)
self.update(time)
}
// recalc sumWeight
func (self *ClientManager) updateNodes(time mclock.AbsTime) (rce bool) {
var sumWeight, rcSum uint64
for node := range self.nodes {
rc := node.recharging
node.update(time)
if rc && !node.recharging {
rce = true
}
if node.recharging {
sumWeight += node.rcWeight
}
rcSum += uint64(node.rcValue)
enabled := cm.mode != CmDisabled
newEnabled := cm.mode != CmDisabled
if !enabled && newEnabled && cm.enabledCh != nil {
close(cm.enabledCh)
cm.enabledCh = nil
}
if enabled && !newEnabled {
cm.enabledCh = make(chan struct{})
}
self.sumWeight = sumWeight
self.rcSumValue = rcSum
return
}
func (self *ClientManager) update(time mclock.AbsTime) {
for {
firstTime := time
for node := range self.nodes {
if node.recharging && node.finishRecharge < firstTime {
firstTime = node.finishRecharge
}
}
if self.updateNodes(firstTime) {
for node := range self.nodes {
if node.recharging {
node.set(node.serving, self.simReqCnt, self.sumWeight)
}
}
switch newMode {
case CmDisabled:
cm.totalRecharge = 0
cm.bufCorrEnabled = false
cm.forceMinRecharge = false
case CmNormal:
cm.totalRecharge = cm.targetParallelReqs * 1000000
cm.bufCorrEnabled = true
cm.forceMinRecharge = false
case CmBlockProcessing:
cm.totalRecharge = 0
cm.bufCorrEnabled = false
cm.forceMinRecharge = true
}
cm.mode = newMode
if cm.child != nil {
if cm.parallelReqs == 0 {
cm.child.SetMode(newMode)
} else {
self.time = time
cm.child.SetMode(CmDisabled)
}
}
}
func (cm *ClientManager) setParallelReqs(p int, time mclock.AbsTime) {
if p == cm.parallelReqs {
return
}
if cm.child != nil && cm.mode != CmDisabled {
if cm.parallelReqs == 0 {
cm.child.SetMode(CmDisabled)
}
if p == 0 {
cm.child.SetMode(cm.mode)
}
}
cm.parallelReqs = p
}
func (cm *ClientManager) updateRecharge(time mclock.AbsTime) {
lastUpdate := cm.rcLastUpdate
cm.rcLastUpdate = time
if cm.totalRecharge == 0 {
return
}
for cm.sumRecharge > 0 {
var slope float64
if cm.forceMinRecharge {
slope = 1
} else {
slope = cm.totalRecharge / float64(cm.sumRecharge)
}
dt := time - lastUpdate
q := cm.rcQueue.Pop()
for q != nil && !q.(rcQueueItem).valid() {
q = cm.rcQueue.Pop()
}
if q == nil {
cm.rcLastIntValue += int64(slope * float64(dt))
return
}
rcqItem := q.(rcQueueItem)
dtNext := mclock.AbsTime(float64(rcqItem.intValue-cm.rcLastIntValue) / slope)
if dt < dtNext {
cm.rcQueue.Push(q)
cm.rcLastIntValue += int64(slope * float64(dt))
return
}
if rcqItem.node.corrBufValue < int64(rcqItem.node.params.BufLimit) {
rcqItem.node.corrBufValue = int64(rcqItem.node.params.BufLimit)
cm.sumRecharge -= rcqItem.node.params.MinRecharge
}
lastUpdate += dtNext
cm.rcLastIntValue = rcqItem.intValue
}
}
func (cm *ClientManager) updateNodeRc(node *ClientNode, bvc int64, time mclock.AbsTime) {
cm.updateRecharge(time)
wasFull := true
if node.corrBufValue != int64(node.params.BufLimit) {
wasFull = false
node.corrBufValue += (cm.rcLastIntValue - node.rcLastIntValue) * int64(node.params.MinRecharge) / 1000000
if node.corrBufValue > int64(node.params.BufLimit) {
node.corrBufValue = int64(node.params.BufLimit)
}
node.rcLastIntValue = cm.rcLastIntValue
}
node.corrBufValue += bvc
if node.corrBufValue < 0 {
node.corrBufValue = 0
}
isFull := false
if node.corrBufValue >= int64(node.params.BufLimit) {
node.corrBufValue = int64(node.params.BufLimit)
isFull = true
}
if wasFull && !isFull {
cm.sumRecharge += node.params.MinRecharge
}
if !wasFull && isFull {
cm.sumRecharge -= node.params.MinRecharge
}
if !isFull {
node.rcLastIntValue = cm.rcLastIntValue
node.rcNextIntValue = cm.rcLastIntValue + (int64(node.params.BufLimit)-node.corrBufValue)*1000000/int64(node.params.MinRecharge)
cm.rcQueue.Push(rcQueueItem{node: node, intValue: node.rcNextIntValue})
}
}
// waitOrStop blocks while request processing is disabled and returns true if
// it should be cancelled because the client has been disconnected.
func (cm *ClientManager) waitOrStop(node *ClientNode) bool {
cm.lock.RLock()
_, ok := cm.nodes[node]
stop := !ok
ch := cm.enabledCh
cm.lock.RUnlock()
if !stop && ch != nil {
<-ch
cm.lock.RLock()
_, ok = cm.nodes[node]
stop = !ok
cm.lock.RUnlock()
}
return stop
}
func (cm *ClientManager) Stop() {
cm.lock.Lock()
defer cm.lock.Unlock()
cm.nodes = nil
}
func (cm *ClientManager) addNode(node *ClientNode) {
cm.lock.Lock()
defer cm.lock.Unlock()
node.corrBufValue = int64(node.params.BufLimit)
node.rcLastIntValue = cm.rcLastIntValue
if cm.nodes != nil {
cm.nodes[node] = struct{}{}
}
}
func (cm *ClientManager) removeNode(node *ClientNode) {
cm.lock.Lock()
defer cm.lock.Unlock()
if cm.nodes != nil {
delete(cm.nodes, node)
}
}
func (cm *ClientManager) accept(node *ClientNode, maxCost uint64, time mclock.AbsTime) chan bool {
cm.lock.Lock()
defer cm.lock.Unlock()
if cm.parallelReqs == cm.maxParallelReqs {
ch := make(chan bool, 1)
start := func() bool {
// always called while client manager lock is held
_, started := cm.nodes[node]
ch <- started
return started
}
cm.servingQueue.Push(servingQueueItem{start, float64(node.bufValue) / float64(node.params.BufLimit)})
return ch
}
cm.setParallelReqs(cm.parallelReqs+1, time)
node.servingStarted = time
node.servingMaxCost = maxCost
cm.updateNodeRc(node, -int64(maxCost), time)
return nil
}
func (cm *ClientManager) started(node *ClientNode, maxCost uint64, time mclock.AbsTime) {
cm.lock.Lock()
defer cm.lock.Unlock()
node.servingStarted = time
node.servingMaxCost = maxCost
cm.updateNodeRc(node, -int64(maxCost), time)
}
func (cm *ClientManager) processed(node *ClientNode, time mclock.AbsTime) (realCost uint64) {
cm.lock.Lock()
defer cm.lock.Unlock()
realCost = uint64(time - node.servingStarted)
if realCost > node.servingMaxCost {
realCost = node.servingMaxCost
}
if !cm.forceMinRecharge {
cm.updateNodeRc(node, int64(node.servingMaxCost-realCost), time)
}
if cm.bufCorrEnabled {
if uint64(node.corrBufValue) > node.bufValue {
node.bufValue = uint64(node.corrBufValue)
}
}
for !cm.servingQueue.Empty() {
if cm.servingQueue.Pop().(servingQueueItem).start() {
return
}
}
}
func (self *ClientManager) canStartReq() bool {
return self.simReqCnt < self.maxSimReq && self.rcSumValue < self.maxRcSum
}
func (self *ClientManager) queueProc() {
for rc := range self.resumeQueue {
for {
time.Sleep(time.Millisecond * 10)
self.lock.Lock()
self.update(mclock.Now())
cs := self.canStartReq()
self.lock.Unlock()
if cs {
break
}
}
close(rc)
}
}
func (self *ClientManager) accept(node *cmNode, time mclock.AbsTime) bool {
self.lock.Lock()
defer self.lock.Unlock()
self.update(time)
if !self.canStartReq() {
resume := make(chan bool)
self.lock.Unlock()
self.resumeQueue <- resume
<-resume
self.lock.Lock()
if _, ok := self.nodes[node]; !ok {
return false // reject if node has been removed or manager has been stopped
}
}
self.simReqCnt++
node.set(true, self.simReqCnt, self.sumWeight)
node.startValue = node.rcValue
self.update(self.time)
return true
}
func (self *ClientManager) stop(node *cmNode, time mclock.AbsTime) {
if node.serving {
self.update(time)
self.simReqCnt--
node.set(false, self.simReqCnt, self.sumWeight)
self.update(time)
}
}
func (self *ClientManager) processed(node *cmNode, time mclock.AbsTime) (rcValue, rcCost uint64) {
self.lock.Lock()
defer self.lock.Unlock()
self.stop(node, time)
return uint64(node.rcValue), uint64(node.rcValue - node.startValue)
cm.setParallelReqs(cm.parallelReqs-1, time)
return
}

43
les/flowcontrol/prque/prque.go Executable file
View file

@ -0,0 +1,43 @@
// This is a duplicated and slightly modified version of "gopkg.in/karalabe/cookiejar.v2/collections/prque".
package prque
import (
"container/heap"
)
// Priority queue data structure.
type Prque struct {
cont *sstack
}
// Creates a new priority queue.
func New(compare compareFn) *Prque {
return &Prque{newSstack(compare)}
}
// Pushes a value with a given priority into the queue, expanding if necessary.
func (p *Prque) Push(i interface{}) {
heap.Push(p.cont, i)
}
// Pops the value with the greates priority off the stack and returns it.
// Currently no shrinking is done.
func (p *Prque) Pop() interface{} {
return heap.Pop(p.cont)
}
// Checks whether the priority queue is empty.
func (p *Prque) Empty() bool {
return p.cont.Len() == 0
}
// Returns the number of element in the priority queue.
func (p *Prque) Size() int {
return p.cont.Len()
}
// Clears the contents of the priority queue.
func (p *Prque) Reset() {
*p = *New(p.cont.compare)
}

84
les/flowcontrol/prque/sstack.go Executable file
View file

@ -0,0 +1,84 @@
// This is a duplicated and slightly modified version of "gopkg.in/karalabe/cookiejar.v2/collections/prque".
package prque
// The size of a block of data
const blockSize = 4096
// returns true if a comes before b
type compareFn func(a, b interface{}) bool
// Internal sortable stack data structure. Implements the Push and Pop ops for
// the stack (heap) functionality and the Len, Less and Swap methods for the
// sortability requirements of the heaps.
type sstack struct {
compare compareFn
size int
capacity int
offset int
blocks [][]interface{}
active []interface{}
}
// Creates a new, empty stack.
func newSstack(compare compareFn) *sstack {
result := new(sstack)
result.compare = compare
result.active = make([]interface{}, blockSize)
result.blocks = [][]interface{}{result.active}
result.capacity = blockSize
return result
}
// Pushes a value onto the stack, expanding it if necessary. Required by
// heap.Interface.
func (s *sstack) Push(data interface{}) {
if s.size == s.capacity {
s.active = make([]interface{}, blockSize)
s.blocks = append(s.blocks, s.active)
s.capacity += blockSize
s.offset = 0
} else if s.offset == blockSize {
s.active = s.blocks[s.size/blockSize]
s.offset = 0
}
s.active[s.offset] = data
s.offset++
s.size++
}
// Pops a value off the stack and returns it. Currently no shrinking is done.
// Required by heap.Interface.
func (s *sstack) Pop() (res interface{}) {
s.size--
s.offset--
if s.offset < 0 {
s.offset = blockSize - 1
s.active = s.blocks[s.size/blockSize]
}
res, s.active[s.offset] = s.active[s.offset], nil
return
}
// Returns the length of the stack. Required by sort.Interface.
func (s *sstack) Len() int {
return s.size
}
// Compares the priority of two elements of the stack (higher is first).
// Required by sort.Interface.
func (s *sstack) Less(i, j int) bool {
return s.compare(s.blocks[i/blockSize][i%blockSize], s.blocks[j/blockSize][j%blockSize])
}
// Swaps two elements in the stack. Required by sort.Interface.
func (s *sstack) Swap(i, j int) {
ib, io, jb, jo := i/blockSize, i%blockSize, j/blockSize, j%blockSize
s.blocks[ib][io], s.blocks[jb][jo] = s.blocks[jb][jo], s.blocks[ib][io]
}
// Resets the stack, effectively clearing its contents.
func (s *sstack) Reset() {
*s = *newSstack(s.compare)
}

82
les/handler.go
View file

@ -292,7 +292,7 @@ func (pm *ProtocolManager) handle(p *peer) error {
}
defer func() {
if pm.server != nil && pm.server.fcManager != nil && p.fcClient != nil {
p.fcClient.Remove(pm.server.fcManager)
p.fcClient.Remove()
}
pm.removePeer(p.id)
}()
@ -345,19 +345,20 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
}
p.Log().Trace("Light Ethereum message arrived", "code", msg.Code, "bytes", msg.Size)
costs := p.fcCosts[msg.Code]
reject := func(reqCnt, maxCnt uint64) bool {
if p.fcClient == nil || reqCnt > maxCnt {
return true
}
bufValue, _ := p.fcClient.AcceptRequest()
costs := p.fcCosts[msg.Code]
cost := costs.baseCost + reqCnt*costs.reqCost
if cost > pm.server.defParams.BufLimit {
cost = pm.server.defParams.BufLimit
}
if cost > bufValue {
recharge := time.Duration((cost - bufValue) * 1000000 / pm.server.defParams.MinRecharge)
p.Log().Error("Request came too early", "recharge", common.PrettyDuration(recharge))
if accepted, bufShort := p.fcClient.AcceptRequest(cost); !accepted {
if bufShort > 0 {
p.Log().Error("Request came too early", "remaining", common.PrettyDuration(time.Duration(bufShort*1000000/pm.server.defParams.MinRecharge)))
}
return true
}
return false
@ -429,6 +430,9 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
unknown bool
)
for !unknown && len(headers) < int(query.Amount) && bytes < softResponseLimit {
if p.fcClient.WaitOrStop() {
return nil
}
// Retrieve the next header satisfying the query
var origin *types.Header
if hashMode {
@ -498,7 +502,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
}
}
bv, rcost := p.fcClient.RequestProcessed(costs.baseCost + query.Amount*costs.reqCost)
bv, rcost := p.fcClient.RequestProcessed()
pm.server.fcCostStats.update(msg.Code, query.Amount, rcost)
return p.SendBlockHeaders(req.ReqID, bv, headers)
@ -516,7 +520,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
if err := msg.Decode(&resp); err != nil {
return errResp(ErrDecode, "msg %v: %v", msg, err)
}
p.fcServer.GotReply(resp.ReqID, resp.BV)
p.fcServer.ReceivedReply(resp.ReqID, resp.BV)
if pm.fetcher != nil && pm.fetcher.requestedID(resp.ReqID) {
pm.fetcher.deliverHeaders(p, resp.ReqID, resp.Headers)
} else {
@ -549,6 +553,9 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
if bytes >= softResponseLimit {
break
}
if p.fcClient.WaitOrStop() {
return nil
}
// Retrieve the requested block body, stopping if enough was found
if number := rawdb.ReadHeaderNumber(pm.chainDb, hash); number != nil {
if data := rawdb.ReadBodyRLP(pm.chainDb, hash, *number); len(data) != 0 {
@ -557,7 +564,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
}
}
}
bv, rcost := p.fcClient.RequestProcessed(costs.baseCost + uint64(reqCnt)*costs.reqCost)
bv, rcost := p.fcClient.RequestProcessed()
pm.server.fcCostStats.update(msg.Code, uint64(reqCnt), rcost)
return p.SendBlockBodiesRLP(req.ReqID, bv, bodies)
@ -575,7 +582,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
if err := msg.Decode(&resp); err != nil {
return errResp(ErrDecode, "msg %v: %v", msg, err)
}
p.fcServer.GotReply(resp.ReqID, resp.BV)
p.fcServer.ReceivedReply(resp.ReqID, resp.BV)
deliverMsg = &Msg{
MsgType: MsgBlockBodies,
ReqID: resp.ReqID,
@ -602,6 +609,9 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
return errResp(ErrRequestRejected, "")
}
for _, req := range req.Reqs {
if p.fcClient.WaitOrStop() {
return nil
}
// Retrieve the requested state entry, stopping if enough was found
if number := rawdb.ReadHeaderNumber(pm.chainDb, req.BHash); number != nil {
if header := rawdb.ReadHeader(pm.chainDb, req.BHash, *number); header != nil {
@ -622,7 +632,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
}
}
}
bv, rcost := p.fcClient.RequestProcessed(costs.baseCost + uint64(reqCnt)*costs.reqCost)
bv, rcost := p.fcClient.RequestProcessed()
pm.server.fcCostStats.update(msg.Code, uint64(reqCnt), rcost)
return p.SendCode(req.ReqID, bv, data)
@ -640,7 +650,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
if err := msg.Decode(&resp); err != nil {
return errResp(ErrDecode, "msg %v: %v", msg, err)
}
p.fcServer.GotReply(resp.ReqID, resp.BV)
p.fcServer.ReceivedReply(resp.ReqID, resp.BV)
deliverMsg = &Msg{
MsgType: MsgCode,
ReqID: resp.ReqID,
@ -670,6 +680,9 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
if bytes >= softResponseLimit {
break
}
if p.fcClient.WaitOrStop() {
return nil
}
// Retrieve the requested block's receipts, skipping if unknown to us
var results types.Receipts
if number := rawdb.ReadHeaderNumber(pm.chainDb, hash); number != nil {
@ -688,7 +701,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
bytes += len(encoded)
}
}
bv, rcost := p.fcClient.RequestProcessed(costs.baseCost + uint64(reqCnt)*costs.reqCost)
bv, rcost := p.fcClient.RequestProcessed()
pm.server.fcCostStats.update(msg.Code, uint64(reqCnt), rcost)
return p.SendReceiptsRLP(req.ReqID, bv, receipts)
@ -706,7 +719,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
if err := msg.Decode(&resp); err != nil {
return errResp(ErrDecode, "msg %v: %v", msg, err)
}
p.fcServer.GotReply(resp.ReqID, resp.BV)
p.fcServer.ReceivedReply(resp.ReqID, resp.BV)
deliverMsg = &Msg{
MsgType: MsgReceipts,
ReqID: resp.ReqID,
@ -733,6 +746,9 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
return errResp(ErrRequestRejected, "")
}
for _, req := range req.Reqs {
if p.fcClient.WaitOrStop() {
return nil
}
// Retrieve the requested state entry, stopping if enough was found
if number := rawdb.ReadHeaderNumber(pm.chainDb, req.BHash); number != nil {
if header := rawdb.ReadHeader(pm.chainDb, req.BHash, *number); header != nil {
@ -762,7 +778,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
}
}
}
bv, rcost := p.fcClient.RequestProcessed(costs.baseCost + uint64(reqCnt)*costs.reqCost)
bv, rcost := p.fcClient.RequestProcessed()
pm.server.fcCostStats.update(msg.Code, uint64(reqCnt), rcost)
return p.SendProofs(req.ReqID, bv, proofs)
@ -790,6 +806,9 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
nodes := light.NewNodeSet()
for _, req := range req.Reqs {
if p.fcClient.WaitOrStop() {
return nil
}
// Look up the state belonging to the request
if statedb == nil || req.BHash != lastBHash {
statedb, root, lastBHash = nil, common.Hash{}, req.BHash
@ -824,7 +843,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
break
}
}
bv, rcost := p.fcClient.RequestProcessed(costs.baseCost + uint64(reqCnt)*costs.reqCost)
bv, rcost := p.fcClient.RequestProcessed()
pm.server.fcCostStats.update(msg.Code, uint64(reqCnt), rcost)
return p.SendProofsV2(req.ReqID, bv, nodes.NodeList())
@ -842,7 +861,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
if err := msg.Decode(&resp); err != nil {
return errResp(ErrDecode, "msg %v: %v", msg, err)
}
p.fcServer.GotReply(resp.ReqID, resp.BV)
p.fcServer.ReceivedReply(resp.ReqID, resp.BV)
deliverMsg = &Msg{
MsgType: MsgProofsV1,
ReqID: resp.ReqID,
@ -863,7 +882,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
if err := msg.Decode(&resp); err != nil {
return errResp(ErrDecode, "msg %v: %v", msg, err)
}
p.fcServer.GotReply(resp.ReqID, resp.BV)
p.fcServer.ReceivedReply(resp.ReqID, resp.BV)
deliverMsg = &Msg{
MsgType: MsgProofsV2,
ReqID: resp.ReqID,
@ -891,6 +910,9 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
}
trieDb := trie.NewDatabase(ethdb.NewTable(pm.chainDb, light.ChtTablePrefix))
for _, req := range req.Reqs {
if p.fcClient.WaitOrStop() {
return nil
}
if header := pm.blockchain.GetHeaderByNumber(req.BlockNum); header != nil {
sectionHead := rawdb.ReadCanonicalHash(pm.chainDb, req.ChtNum*light.CHTFrequencyServer-1)
if root := light.GetChtRoot(pm.chainDb, req.ChtNum-1, sectionHead); root != (common.Hash{}) {
@ -911,7 +933,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
}
}
}
bv, rcost := p.fcClient.RequestProcessed(costs.baseCost + uint64(reqCnt)*costs.reqCost)
bv, rcost := p.fcClient.RequestProcessed()
pm.server.fcCostStats.update(msg.Code, uint64(reqCnt), rcost)
return p.SendHeaderProofs(req.ReqID, bv, proofs)
@ -943,6 +965,9 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
)
nodes := light.NewNodeSet()
for _, req := range req.Reqs {
if p.fcClient.WaitOrStop() {
return nil
}
if auxTrie == nil || req.Type != lastType || req.TrieIdx != lastIdx {
auxTrie, lastType, lastIdx = nil, req.Type, req.TrieIdx
@ -972,7 +997,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
break
}
}
bv, rcost := p.fcClient.RequestProcessed(costs.baseCost + uint64(reqCnt)*costs.reqCost)
bv, rcost := p.fcClient.RequestProcessed()
pm.server.fcCostStats.update(msg.Code, uint64(reqCnt), rcost)
return p.SendHelperTrieProofs(req.ReqID, bv, HelperTrieResps{Proofs: nodes.NodeList(), AuxData: auxData})
@ -989,7 +1014,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
if err := msg.Decode(&resp); err != nil {
return errResp(ErrDecode, "msg %v: %v", msg, err)
}
p.fcServer.GotReply(resp.ReqID, resp.BV)
p.fcServer.ReceivedReply(resp.ReqID, resp.BV)
deliverMsg = &Msg{
MsgType: MsgHeaderProofs,
ReqID: resp.ReqID,
@ -1010,7 +1035,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
return errResp(ErrDecode, "msg %v: %v", msg, err)
}
p.fcServer.GotReply(resp.ReqID, resp.BV)
p.fcServer.ReceivedReply(resp.ReqID, resp.BV)
deliverMsg = &Msg{
MsgType: MsgHelperTrieProofs,
ReqID: resp.ReqID,
@ -1032,7 +1057,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
}
pm.txpool.AddRemotes(txs)
_, rcost := p.fcClient.RequestProcessed(costs.baseCost + uint64(reqCnt)*costs.reqCost)
_, rcost := p.fcClient.RequestProcessed()
pm.server.fcCostStats.update(msg.Code, uint64(reqCnt), rcost)
case SendTxV2Msg:
@ -1058,6 +1083,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
}
stats := pm.txStatus(hashes)
for i, stat := range stats {
p.fcClient.WaitOrStop() // do not return; txs can be added even if client disconnected
if stat.Status == core.TxStatusUnknown {
if errs := pm.txpool.AddRemotes([]*types.Transaction{req.Txs[i]}); errs[0] != nil {
stats[i].Error = errs[0].Error()
@ -1067,7 +1093,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
}
}
bv, rcost := p.fcClient.RequestProcessed(costs.baseCost + uint64(reqCnt)*costs.reqCost)
bv, rcost := p.fcClient.RequestProcessed()
pm.server.fcCostStats.update(msg.Code, uint64(reqCnt), rcost)
return p.SendTxStatus(req.ReqID, bv, stats)
@ -1088,7 +1114,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
if reject(uint64(reqCnt), MaxTxStatus) {
return errResp(ErrRequestRejected, "")
}
bv, rcost := p.fcClient.RequestProcessed(costs.baseCost + uint64(reqCnt)*costs.reqCost)
bv, rcost := p.fcClient.RequestProcessed()
pm.server.fcCostStats.update(msg.Code, uint64(reqCnt), rcost)
return p.SendTxStatus(req.ReqID, bv, pm.txStatus(req.Hashes))
@ -1107,7 +1133,7 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
return errResp(ErrDecode, "msg %v: %v", msg, err)
}
p.fcServer.GotReply(resp.ReqID, resp.BV)
p.fcServer.ReceivedReply(resp.ReqID, resp.BV)
default:
p.Log().Trace("Received unknown message", "code", msg.Code)
@ -1232,7 +1258,7 @@ func (pc *peerConnection) RequestHeadersByHash(origin common.Hash, amount int, s
request: func(dp distPeer) func() {
peer := dp.(*peer)
cost := peer.GetRequestCost(GetBlockHeadersMsg, amount)
peer.fcServer.QueueRequest(reqID, cost)
peer.fcServer.QueuedRequest(reqID, cost)
return func() { peer.RequestHeadersByHash(reqID, cost, origin, amount, skip, reverse) }
},
}
@ -1256,7 +1282,7 @@ func (pc *peerConnection) RequestHeadersByNumber(origin uint64, amount int, skip
request: func(dp distPeer) func() {
peer := dp.(*peer)
cost := peer.GetRequestCost(GetBlockHeadersMsg, amount)
peer.fcServer.QueueRequest(reqID, cost)
peer.fcServer.QueuedRequest(reqID, cost)
return func() { peer.RequestHeadersByNumber(reqID, cost, origin, amount, skip, reverse) }
},
}

24
les/handler_test.go
View file

@ -51,7 +51,7 @@ func TestGetBlockHeadersLes1(t *testing.T) { testGetBlockHeaders(t, 1) }
func TestGetBlockHeadersLes2(t *testing.T) { testGetBlockHeaders(t, 2) }
func testGetBlockHeaders(t *testing.T, protocol int) {
pm := newTestProtocolManagerMust(t, false, downloader.MaxHashFetch+15, nil, nil, nil, ethdb.NewMemDatabase())
pm := newTestProtocolManagerMust(t, false, downloader.MaxHashFetch+15, 0, nil, nil, nil, ethdb.NewMemDatabase())
bc := pm.blockchain.(*core.BlockChain)
peer, _ := newTestPeer(t, "peer", protocol, pm, true)
defer peer.close()
@ -180,7 +180,7 @@ func TestGetBlockBodiesLes1(t *testing.T) { testGetBlockBodies(t, 1) }
func TestGetBlockBodiesLes2(t *testing.T) { testGetBlockBodies(t, 2) }
func testGetBlockBodies(t *testing.T, protocol int) {
pm := newTestProtocolManagerMust(t, false, downloader.MaxBlockFetch+15, nil, nil, nil, ethdb.NewMemDatabase())
pm := newTestProtocolManagerMust(t, false, downloader.MaxBlockFetch+15, 0, nil, nil, nil, ethdb.NewMemDatabase())
bc := pm.blockchain.(*core.BlockChain)
peer, _ := newTestPeer(t, "peer", protocol, pm, true)
defer peer.close()
@ -257,7 +257,7 @@ func TestGetCodeLes2(t *testing.T) { testGetCode(t, 2) }
func testGetCode(t *testing.T, protocol int) {
// Assemble the test environment
pm := newTestProtocolManagerMust(t, false, 4, testChainGen, nil, nil, ethdb.NewMemDatabase())
pm := newTestProtocolManagerMust(t, false, 4, 0, testChainGen, nil, nil, ethdb.NewMemDatabase())
bc := pm.blockchain.(*core.BlockChain)
peer, _ := newTestPeer(t, "peer", protocol, pm, true)
defer peer.close()
@ -291,7 +291,7 @@ func TestGetReceiptLes2(t *testing.T) { testGetReceipt(t, 2) }
func testGetReceipt(t *testing.T, protocol int) {
// Assemble the test environment
db := ethdb.NewMemDatabase()
pm := newTestProtocolManagerMust(t, false, 4, testChainGen, nil, nil, db)
pm := newTestProtocolManagerMust(t, false, 4, 0, testChainGen, nil, nil, db)
bc := pm.blockchain.(*core.BlockChain)
peer, _ := newTestPeer(t, "peer", protocol, pm, true)
defer peer.close()
@ -319,7 +319,7 @@ func TestGetProofsLes2(t *testing.T) { testGetProofs(t, 2) }
func testGetProofs(t *testing.T, protocol int) {
// Assemble the test environment
db := ethdb.NewMemDatabase()
pm := newTestProtocolManagerMust(t, false, 4, testChainGen, nil, nil, db)
pm := newTestProtocolManagerMust(t, false, 4, 0, testChainGen, nil, nil, db)
bc := pm.blockchain.(*core.BlockChain)
peer, _ := newTestPeer(t, "peer", protocol, pm, true)
defer peer.close()
@ -382,15 +382,11 @@ func testGetCHTProofs(t *testing.T, protocol int) {
}
// Assemble the test environment
db := ethdb.NewMemDatabase()
pm := newTestProtocolManagerMust(t, false, int(frequency)+light.HelperTrieProcessConfirmations, testChainGen, nil, nil, db)
pm := newTestProtocolManagerMust(t, false, int(frequency)+light.HelperTrieProcessConfirmations, 1, testChainGen, nil, nil, db)
bc := pm.blockchain.(*core.BlockChain)
peer, _ := newTestPeer(t, "peer", protocol, pm, true)
defer peer.close()
// Wait a while for the CHT indexer to process the new headers
time.Sleep(100 * time.Millisecond * time.Duration(frequency/light.CHTFrequencyServer)) // Chain indexer throttling
time.Sleep(250 * time.Millisecond) // CI tester slack
// Assemble the proofs from the different protocols
header := bc.GetHeaderByNumber(frequency)
rlp, _ := rlp.EncodeToBytes(header)
@ -450,15 +446,11 @@ func testGetCHTProofs(t *testing.T, protocol int) {
func TestGetBloombitsProofs(t *testing.T) {
// Assemble the test environment
db := ethdb.NewMemDatabase()
pm := newTestProtocolManagerMust(t, false, light.BloomTrieFrequency+256, testChainGen, nil, nil, db)
pm := newTestProtocolManagerMust(t, false, light.BloomTrieFrequency+256, 1, testChainGen, nil, nil, db)
bc := pm.blockchain.(*core.BlockChain)
peer, _ := newTestPeer(t, "peer", 2, pm, true)
defer peer.close()
// Wait a while for the bloombits indexer to process the new headers
time.Sleep(100 * time.Millisecond * time.Duration(light.BloomTrieFrequency/4096)) // Chain indexer throttling
time.Sleep(250 * time.Millisecond) // CI tester slack
// Request and verify each bit of the bloom bits proofs
for bit := 0; bit < 2048; bit++ {
// Assemble therequest and proofs for the bloombits
@ -489,7 +481,7 @@ func TestGetBloombitsProofs(t *testing.T) {
func TestTransactionStatusLes2(t *testing.T) {
db := ethdb.NewMemDatabase()
pm := newTestProtocolManagerMust(t, false, 0, nil, nil, nil, db)
pm := newTestProtocolManagerMust(t, false, 0, 0, nil, nil, nil, db)
chain := pm.blockchain.(*core.BlockChain)
config := core.DefaultTxPoolConfig
config.Journal = ""

23
les/helper_test.go
View file

@ -24,8 +24,10 @@ import (
"math/big"
"sync"
"testing"
"time"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/common/mclock"
"github.com/ethereum/go-ethereum/consensus/ethash"
"github.com/ethereum/go-ethereum/core"
"github.com/ethereum/go-ethereum/core/types"
@ -136,7 +138,7 @@ func testRCL() RequestCostList {
// newTestProtocolManager creates a new protocol manager for testing purposes,
// with the given number of blocks already known, and potential notification
// channels for different events.
func newTestProtocolManager(lightSync bool, blocks int, generator func(int, *core.BlockGen), peers *peerSet, odr *LesOdr, db ethdb.Database) (*ProtocolManager, error) {
func newTestProtocolManager(lightSync bool, blocks int, processSections uint64, generator func(int, *core.BlockGen), peers *peerSet, odr *LesOdr, db ethdb.Database) (*ProtocolManager, error) {
var (
evmux = new(event.TypeMux)
engine = ethash.NewFaker()
@ -169,6 +171,18 @@ func newTestProtocolManager(lightSync bool, blocks int, generator func(int, *cor
if _, err := blockchain.InsertChain(gchain); err != nil {
panic(err)
}
if processSections > 0 {
for {
cs, _, _ := chtIndexer.Sections()
bs, _, _ := bloomIndexer.Sections()
if cs >= processSections && bs >= processSections {
break
}
time.Sleep(time.Millisecond * 10)
}
}
chain = blockchain
}
@ -191,7 +205,8 @@ func newTestProtocolManager(lightSync bool, blocks int, generator func(int, *cor
MinRecharge: 1,
}
srv.fcManager = flowcontrol.NewClientManager(50, 10, 1000000000)
//srv.fcManager = flowcontrol.NewClientManager(50, 10, 1000000000)
srv.fcManager = flowcontrol.NewClientManager(16, 4, &mclock.MonotonicClock{}, nil)
srv.fcCostStats = newCostStats(nil)
}
pm.Start(1000)
@ -202,8 +217,8 @@ func newTestProtocolManager(lightSync bool, blocks int, generator func(int, *cor
// with the given number of blocks already known, and potential notification
// channels for different events. In case of an error, the constructor force-
// fails the test.
func newTestProtocolManagerMust(t *testing.T, lightSync bool, blocks int, generator func(int, *core.BlockGen), peers *peerSet, odr *LesOdr, db ethdb.Database) *ProtocolManager {
pm, err := newTestProtocolManager(lightSync, blocks, generator, peers, odr, db)
func newTestProtocolManagerMust(t *testing.T, lightSync bool, blocks int, processSections uint64, generator func(int, *core.BlockGen), peers *peerSet, odr *LesOdr, db ethdb.Database) *ProtocolManager {
pm, err := newTestProtocolManager(lightSync, blocks, processSections, generator, peers, odr, db)
if err != nil {
t.Fatalf("Failed to create protocol manager: %v", err)
}

422
les/load_test.go Normal file
View file

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

2
les/odr.go
View file

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

7
les/odr_test.go
View file

@ -25,6 +25,7 @@ import (
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/common/math"
"github.com/ethereum/go-ethereum/common/mclock"
"github.com/ethereum/go-ethereum/core"
"github.com/ethereum/go-ethereum/core/rawdb"
"github.com/ethereum/go-ethereum/core/state"
@ -163,13 +164,13 @@ func odrContractCall(ctx context.Context, db ethdb.Database, config *params.Chai
func testOdr(t *testing.T, protocol int, expFail uint64, fn odrTestFn) {
// Assemble the test environment
peers := newPeerSet()
dist := newRequestDistributor(peers, make(chan struct{}))
dist := newRequestDistributor(peers, make(chan struct{}), &mclock.MonotonicClock{})
rm := newRetrieveManager(peers, dist, nil)
db := ethdb.NewMemDatabase()
ldb := ethdb.NewMemDatabase()
odr := NewLesOdr(ldb, light.NewChtIndexer(db, true), light.NewBloomTrieIndexer(db, true), eth.NewBloomIndexer(db, light.BloomTrieFrequency), rm)
pm := newTestProtocolManagerMust(t, false, 4, testChainGen, nil, nil, db)
lpm := newTestProtocolManagerMust(t, true, 0, nil, peers, odr, ldb)
pm := newTestProtocolManagerMust(t, false, 4, 0, testChainGen, nil, nil, db)
lpm := newTestProtocolManagerMust(t, true, 0, 0, nil, peers, odr, ldb)
_, err1, lpeer, err2 := newTestPeerPair("peer", protocol, pm, lpm)
select {
case <-time.After(time.Millisecond * 100):

3
les/peer.go
View file

@ -27,6 +27,7 @@ import (
"time"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/common/mclock"
"github.com/ethereum/go-ethereum/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()
}

7
les/request_test.go
View file

@ -22,6 +22,7 @@ import (
"time"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/common/mclock"
"github.com/ethereum/go-ethereum/core/rawdb"
"github.com/ethereum/go-ethereum/crypto"
"github.com/ethereum/go-ethereum/eth"
@ -85,14 +86,14 @@ func tfCodeAccess(db ethdb.Database, bhash common.Hash, num uint64) light.OdrReq
func testAccess(t *testing.T, protocol int, fn accessTestFn) {
// Assemble the test environment
peers := newPeerSet()
dist := newRequestDistributor(peers, make(chan struct{}))
dist := newRequestDistributor(peers, make(chan struct{}), &mclock.MonotonicClock{})
rm := newRetrieveManager(peers, dist, nil)
db := ethdb.NewMemDatabase()
ldb := ethdb.NewMemDatabase()
odr := NewLesOdr(ldb, light.NewChtIndexer(db, true), light.NewBloomTrieIndexer(db, true), eth.NewBloomIndexer(db, light.BloomTrieFrequency), rm)
pm := newTestProtocolManagerMust(t, false, 4, testChainGen, nil, nil, db)
lpm := newTestProtocolManagerMust(t, true, 0, nil, peers, odr, ldb)
pm := newTestProtocolManagerMust(t, false, 4, 0, testChainGen, nil, nil, db)
lpm := newTestProtocolManagerMust(t, true, 0, 0, nil, peers, odr, ldb)
_, err1, lpeer, err2 := newTestPeerPair("peer", protocol, pm, lpm)
select {
case <-time.After(time.Millisecond * 100):

32
les/server.go
View file

@ -24,6 +24,7 @@ import (
"sync"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/common/mclock"
"github.com/ethereum/go-ethereum/core"
"github.com/ethereum/go-ethereum/core/rawdb"
"github.com/ethereum/go-ethereum/core/types"
@ -40,7 +41,7 @@ import (
type LesServer struct {
config *eth.Config
protocolManager *ProtocolManager
fcManager *flowcontrol.ClientManager // nil if our node is client only
fcManager *flowcontrol.ClientManager
fcCostStats *requestCostStats
defParams *flowcontrol.ServerParams
lesTopics []discv5.Topic
@ -98,11 +99,38 @@ func NewLesServer(eth *eth.Ethereum, config *eth.Config) (*LesServer, error) {
BufLimit: 300000000,
MinRecharge: 50000,
}
srv.fcManager = flowcontrol.NewClientManager(uint64(config.LightServ), 10, 1000000000)
tpr := float64(config.LightServ) / 100
mpr := int(tpr * 4)
if mpr < 4 {
mpr = 4
}
srv.fcManager = flowcontrol.NewClientManager(mpr, tpr, &mclock.MonotonicClock{}, nil)
pm.blockProcLoop(srv.fcManager)
srv.fcCostStats = newCostStats(eth.ChainDb())
return srv, nil
}
func (pm *ProtocolManager) blockProcLoop(cm *flowcontrol.ClientManager) {
pm.wg.Add(1)
procFeedback := make(chan bool, 10)
pm.blockchain.(*core.BlockChain).SetProcFeedback(procFeedback)
go func() {
for {
select {
case processing := <-procFeedback:
if processing {
cm.SetMode(flowcontrol.CmBlockProcessing)
} else {
cm.SetMode(flowcontrol.CmNormal)
}
case <-pm.quitSync:
pm.wg.Done()
return
}
}
}()
}
func (s *LesServer) Protocols() []p2p.Protocol {
return s.protocolManager.SubProtocols
}

2
les/txrelay.go
View file

@ -126,7 +126,7 @@ func (self *LesTxRelay) send(txs types.Transactions, count int) {
request: func(dp distPeer) func() {
peer := dp.(*peer)
cost := peer.GetRequestCost(SendTxMsg, len(ll))
peer.fcServer.QueueRequest(reqID, cost)
peer.fcServer.QueuedRequest(reqID, cost)
return func() { peer.SendTxs(reqID, cost, ll) }
},
}