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

les, les/flowcontrol: implement client freeze feature

Browse source
This commit is contained in:
Zsolt Felfoldi 2019-03-18 00:44:18 +01:00
parent dfe14f9f3b
commit 9bf2d758f3
16 changed files with 901 additions and 472 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
les/api.go
View file

@ -168,14 +168,14 @@ type priorityClientInfo struct {
}
// newPriorityClientPool creates a new priority client pool
func newPriorityClientPool(freeClientCap uint64, ps *peerSet, child clientPool, logger *csvlogger.Logger) *priorityClientPool {
func newPriorityClientPool(freeClientCap uint64, ps *peerSet, child clientPool, metricsLogger, eventLogger *csvlogger.Logger) *priorityClientPool {
return &priorityClientPool{
clients: make(map[enode.ID]priorityClientInfo),
freeClientCap: freeClientCap,
ps: ps,
child: child,
logger: logger,
logTotalPriConn: logger.NewChannel("totalPriConn", 0),
logger: eventLogger,
logTotalPriConn: metricsLogger.NewChannel("totalPriConn", 0),
}
}

14
les/costtracker.go
View file

@ -212,6 +212,13 @@ func (ct *costTracker) gfLoop() {
if ct.logRelCost != nil && r.avgTime > 1e-20 {
ct.logRelCost.Update(max / r.avgTime)
}
if r.servingTime > 1000000000 {
ct.logger.Event(fmt.Sprintf("Very long servingTime = %f avgTime = %f costFactor = %f", r.servingTime, r.avgTime, gf))
}
if max > r.avgTime*maxCostFactor {
max = r.avgTime * maxCostFactor
r.servingTime = max / gf
}
if r.avgTime > max {
max = r.avgTime
}
@ -221,9 +228,6 @@ func (ct *costTracker) gfLoop() {
totalRecharge := ct.utilTarget * gf
ct.logRecentUsage.Update(gfUsage)
ct.logTotalRecharge.Update(totalRecharge)
if r.servingTime > 1000000000 {
ct.logger.Event(fmt.Sprintf("Very long servingTime = %f avgTime = %f costFactor = %f", r.servingTime, r.avgTime, gf))
}
if gfUsage >= gfUsageThreshold*totalRecharge {
gfSum += r.avgTime
@ -356,8 +360,8 @@ type (
}
)
// getCost calculates the estimated cost for a given request type and amount
func (table requestCostTable) getCost(code, amount uint64) uint64 {
// getMaxCost calculates the estimated cost for a given request type and amount
func (table requestCostTable) getMaxCost(code, amount uint64) uint64 {
costs := table[code]
return costs.baseCost + amount*costs.reqCost
}

23
les/csvlogger/csvlogger.go
View file

@ -26,6 +26,7 @@ import (
"github.com/ethereum/go-ethereum/log"
)
// Logger is a metrics/events logger that writes logged values and events into a comma separated file
type Logger struct {
file *os.File
started mclock.AbsTime
@ -36,7 +37,11 @@ type Logger struct {
eventHeader string
}
func NewLogger(fileName string, period time.Duration, eventHeader string) *Logger {
// NewLogger creates a new Logger
func NewLogger(fileName string, updatePeriod time.Duration, eventHeader string) *Logger {
if fileName == "" {
return nil
}
f, err := os.Create(fileName)
if err != nil {
log.Error("Error creating log file", "name", fileName, "error", err)
@ -44,13 +49,16 @@ func NewLogger(fileName string, period time.Duration, eventHeader string) *Logge
}
return &Logger{
file: f,
period: period,
period: updatePeriod,
stopCh: make(chan struct{}),
storeCh: make(chan string, 1),
eventHeader: eventHeader,
}
}
// NewChannel creates a new value logger channel that writes values in a single
// column. If the relative change of the value is bigger than the given threshold
// then a new line is added immediately (threshold can also be 0).
func (l *Logger) NewChannel(name string, threshold float64) *Channel {
if l == nil {
return nil
@ -64,6 +72,11 @@ func (l *Logger) NewChannel(name string, threshold float64) *Channel {
return c
}
// NewMinMaxChannel creates a new value logger channel that writes the minimum and
// maximum of the tracked value in two columns. It never triggers adding a new line.
// If zeroDefault is true then 0 is written to both min and max columns if no update
// was given during the last period. If it is false then the last update will appear
// in both columns.
func (l *Logger) NewMinMaxChannel(name string, zeroDefault bool) *Channel {
if l == nil {
return nil
@ -89,6 +102,7 @@ func (l *Logger) store(event string) {
l.file.WriteString(s + "\n")
}
// Start writes the header line and starts the logger
func (l *Logger) Start() {
if l == nil {
return
@ -102,7 +116,6 @@ func (l *Logger) Start() {
s += ", " + l.eventHeader
}
l.file.WriteString(s + "\n")
fmt.Println(s)
go func() {
timer := time.NewTimer(l.period)
for {
@ -124,6 +137,7 @@ func (l *Logger) Start() {
}()
}
// Stop stops the logger and closes the file
func (l *Logger) Stop() {
if l == nil {
return
@ -134,6 +148,7 @@ func (l *Logger) Stop() {
l.file.Close()
}
// Event immediately adds a new line and adds the given event string in the last column
func (l *Logger) Event(event string) {
if l == nil {
return
@ -144,6 +159,7 @@ func (l *Logger) Event(event string) {
}
}
// Channel represents a logger channel tracking a single value
type Channel struct {
logger *Logger
lock sync.Mutex
@ -152,6 +168,7 @@ type Channel struct {
minmax, mmSet, mmZeroDefault bool
}
// Update updates the tracked value
func (lc *Channel) Update(value float64) {
if lc == nil {
return

9
les/execqueue.go
View file

@ -44,7 +44,7 @@ func (q *execQueue) loop() {
func (q *execQueue) waitNext(drop bool) (f func()) {
q.mu.Lock()
if drop {
if drop && len(q.funcs) > 0 {
// Remove the function that just executed. We do this here instead of when
// dequeuing so len(q.funcs) includes the function that is running.
q.funcs = append(q.funcs[:0], q.funcs[1:]...)
@ -84,6 +84,13 @@ func (q *execQueue) queue(f func()) bool {
return ok
}
// clear drops all queued functions
func (q *execQueue) clear() {
q.mu.Lock()
q.funcs = q.funcs[:0]
q.mu.Unlock()
}
// quit stops the exec queue.
// quit waits for the current execution to finish before returning.
func (q *execQueue) quit() {

89
les/flowcontrol/control.go
View file

@ -56,11 +56,12 @@ type scheduledUpdate struct {
// (used in server mode only)
type ClientNode struct {
params ServerParams
bufValue uint64
bufValue int64
lastTime mclock.AbsTime
updateSchedule []scheduledUpdate
sumCost uint64 // sum of req costs received from this client
accepted map[uint64]uint64 // value = sumCost after accepting the given req
connected bool
lock sync.Mutex
cm *ClientManager
log *logger
@ -70,11 +71,12 @@ type ClientNode struct {
// NewClientNode returns a new ClientNode
func NewClientNode(cm *ClientManager, params ServerParams) *ClientNode {
node := &ClientNode{
cm: cm,
params: params,
bufValue: params.BufLimit,
lastTime: cm.clock.Now(),
accepted: make(map[uint64]uint64),
cm: cm,
params: params,
bufValue: int64(params.BufLimit),
lastTime: cm.clock.Now(),
accepted: make(map[uint64]uint64),
connected: true,
}
if keepLogs > 0 {
node.log = newLogger(keepLogs)
@ -85,9 +87,55 @@ func NewClientNode(cm *ClientManager, params ServerParams) *ClientNode {
// Disconnect should be called when a client is disconnected
func (node *ClientNode) Disconnect() {
node.lock.Lock()
defer node.lock.Unlock()
node.connected = false
node.cm.disconnect(node)
}
// BufferStatus returns the current buffer value and limit
func (node *ClientNode) BufferStatus() (uint64, uint64) {
node.lock.Lock()
defer node.lock.Unlock()
if !node.connected {
return 0, 0
}
now := node.cm.clock.Now()
node.update(now)
node.cm.updateBuffer(node, 0, now)
bv := node.bufValue
if bv < 0 {
bv = 0
}
return uint64(bv), node.params.BufLimit
}
// OneTimeCost subtracts the given amount from the node's buffer.
//
// Note: this call can take the buffer into the negative region internally.
// In this case zero buffer value is returned by exported calls and no requests
// are accepted.
func (node *ClientNode) OneTimeCost(cost uint64) {
node.lock.Lock()
defer node.lock.Unlock()
now := node.cm.clock.Now()
node.update(now)
node.bufValue -= int64(cost)
node.cm.updateBuffer(node, -int64(cost), now)
}
// Freeze notifies the client manager about a client freeze event in which case
// the total capacity allowance is slightly reduced.
func (node *ClientNode) Freeze() {
node.lock.Lock()
frozenCap := node.params.MinRecharge
node.lock.Unlock()
node.cm.reduceTotalCap(frozenCap)
}
// update recalculates the buffer value at a specified time while also performing
// scheduled flow control parameter updates if necessary
func (node *ClientNode) update(now mclock.AbsTime) {
@ -105,9 +153,9 @@ func (node *ClientNode) recalcBV(now mclock.AbsTime) {
if now < node.lastTime {
dt = 0
}
node.bufValue += node.params.MinRecharge * dt / uint64(fcTimeConst)
if node.bufValue > node.params.BufLimit {
node.bufValue = node.params.BufLimit
node.bufValue += int64(node.params.MinRecharge * dt / uint64(fcTimeConst))
if node.bufValue > int64(node.params.BufLimit) {
node.bufValue = int64(node.params.BufLimit)
}
if node.log != nil {
node.log.add(now, fmt.Sprintf("updated bv=%d MRR=%d BufLimit=%d", node.bufValue, node.params.MinRecharge, node.params.BufLimit))
@ -139,11 +187,11 @@ func (node *ClientNode) UpdateParams(params ServerParams) {
// updateParams updates the flow control parameters of the node
func (node *ClientNode) updateParams(params ServerParams, now mclock.AbsTime) {
diff := params.BufLimit - node.params.BufLimit
if int64(diff) > 0 {
diff := int64(params.BufLimit - node.params.BufLimit)
if diff > 0 {
node.bufValue += diff
} else if node.bufValue > params.BufLimit {
node.bufValue = params.BufLimit
} else if node.bufValue > int64(params.BufLimit) {
node.bufValue = int64(params.BufLimit)
}
node.cm.updateParams(node, params, now)
}
@ -157,14 +205,14 @@ func (node *ClientNode) AcceptRequest(reqID, index, maxCost uint64) (accepted bo
now := node.cm.clock.Now()
node.update(now)
if maxCost > node.bufValue {
if int64(maxCost) > node.bufValue {
if node.log != nil {
node.log.add(now, fmt.Sprintf("rejected reqID=%d bv=%d maxCost=%d", reqID, node.bufValue, maxCost))
node.log.dump(now)
}
return false, maxCost - node.bufValue, 0
return false, maxCost - uint64(node.bufValue), 0
}
node.bufValue -= maxCost
node.bufValue -= int64(maxCost)
node.sumCost += maxCost
if node.log != nil {
node.log.add(now, fmt.Sprintf("accepted reqID=%d bv=%d maxCost=%d sumCost=%d", reqID, node.bufValue, maxCost, node.sumCost))
@ -174,19 +222,22 @@ func (node *ClientNode) AcceptRequest(reqID, index, maxCost uint64) (accepted bo
}
// RequestProcessed should be called when the request has been processed
func (node *ClientNode) RequestProcessed(reqID, index, maxCost, realCost uint64) (bv uint64) {
func (node *ClientNode) RequestProcessed(reqID, index, maxCost, realCost uint64) uint64 {
node.lock.Lock()
defer node.lock.Unlock()
now := node.cm.clock.Now()
node.update(now)
node.cm.processed(node, maxCost, realCost, now)
bv = node.bufValue + node.sumCost - node.accepted[index]
bv := node.bufValue + int64(node.sumCost-node.accepted[index])
if node.log != nil {
node.log.add(now, fmt.Sprintf("processed reqID=%d bv=%d maxCost=%d realCost=%d sumCost=%d oldSumCost=%d reportedBV=%d", reqID, node.bufValue, maxCost, realCost, node.sumCost, node.accepted[index], bv))
}
delete(node.accepted, index)
return
if bv < 0 {
return 0
}
return uint64(bv)
}
// ServerNode is the flow control system's representation of a server

141
les/flowcontrol/manager.go
View file

@ -24,7 +24,6 @@ import (
"github.com/ethereum/go-ethereum/common/mclock"
"github.com/ethereum/go-ethereum/common/prque"
"github.com/ethereum/go-ethereum/les/csvlogger"
)
// cmNodeFields are ClientNode fields used by the client manager
@ -48,9 +47,11 @@ type cmNodeFields struct {
const FixedPointMultiplier = 1000000
var (
capFactorDropTC = 1 / float64(time.Second*10) // time constant for dropping the capacity factor
capFactorRaiseTC = 1 / float64(time.Hour) // time constant for raising the capacity factor
capFactorRaiseThreshold = 0.75 // connected / total capacity ratio threshold for raising the capacity factor
capFactorDrop = 0.1
capFactorRaiseTC = 10 / float64(time.Hour) // time constant for raising the capacity factor
capFactorRaiseThresholdRatio = 1.125 // total/connected capacity ratio threshold for raising the capacity factor
minCapLogFactor = math.Log(0.75) // lower limit for capacity adjustment
maxCapLogFactor = math.Log(3) // upper limit for capacity adjustment
)
// ClientManager controls the capacity assigned to the clients of a server.
@ -66,11 +67,11 @@ type ClientManager struct {
curve PieceWiseLinear
sumRecharge, totalRecharge, totalConnected uint64
capLogFactor, totalCapacity float64
capLogFactorRaiseLimit float64
capFactorRaiseThreshold uint64
capLastUpdate mclock.AbsTime
totalCapacityCh chan uint64
logTotalCap *csvlogger.Channel
// recharge integrator is increasing in each moment with a rate of
// (totalRecharge / sumRecharge)*FixedPointMultiplier or 0 if sumRecharge==0
rcLastUpdate mclock.AbsTime // last time the recharge integrator was updated
@ -104,12 +105,11 @@ type ClientManager struct {
// starting from zero in order to not let a single low-priority client use up
// the entire server capacity and thus ensure quick availability for others at
// any moment.
func NewClientManager(curve PieceWiseLinear, clock mclock.Clock, logger *csvlogger.Logger) *ClientManager {
func NewClientManager(curve PieceWiseLinear, clock mclock.Clock) *ClientManager {
cm := &ClientManager{
clock: clock,
rcQueue: prque.New(func(a interface{}, i int) { a.(*ClientNode).queueIndex = i }),
capLastUpdate: clock.Now(),
logTotalCap: logger.NewChannel("totalCapacity", 0.01),
}
if curve != nil {
cm.SetRechargeCurve(curve)
@ -134,6 +134,14 @@ func (cm *ClientManager) SetRechargeCurve(curve PieceWiseLinear) {
cm.refreshCapacity()
}
// SetCapFactorRaiseThreshold sets a threshold value used for raising capFactor.
// Either if the difference between total allowed and connected capacity is less
// than this threshold or if their ratio is less than capFactorRaiseThresholdRatio
// then capFactor is allowed to slowly raise.
func (cm *ClientManager) SetCapFactorRaiseThreshold(c uint64) {
cm.capFactorRaiseThreshold = c
}
// connect should be called when a client is connected, before passing it to any
// other ClientManager function
func (cm *ClientManager) connect(node *ClientNode) {
@ -147,6 +155,7 @@ func (cm *ClientManager) connect(node *ClientNode) {
node.queueIndex = -1
cm.updateCapFactor(now, true)
cm.totalConnected += node.params.MinRecharge
cm.updateRaiseLimit()
}
// disconnect should be called when a client is disconnected
@ -158,6 +167,7 @@ func (cm *ClientManager) disconnect(node *ClientNode) {
cm.updateRecharge(cm.clock.Now())
cm.updateCapFactor(now, true)
cm.totalConnected -= node.params.MinRecharge
cm.updateRaiseLimit()
}
// accepted is called when a request with given maximum cost is accepted.
@ -178,18 +188,24 @@ func (cm *ClientManager) accepted(node *ClientNode, maxCost uint64, now mclock.A
//
// Note: processed should always be called for all accepted requests
func (cm *ClientManager) processed(node *ClientNode, maxCost, realCost uint64, now mclock.AbsTime) {
cm.lock.Lock()
defer cm.lock.Unlock()
if realCost > maxCost {
realCost = maxCost
}
cm.updateNodeRc(node, int64(maxCost-realCost), &node.params, now)
if uint64(node.corrBufValue) > node.bufValue {
cm.updateBuffer(node, int64(maxCost-realCost), now)
}
// updateBuffer recalulates the corrected buffer value, adds the given value to it
// and updates the node's actual buffer value if possible
func (cm *ClientManager) updateBuffer(node *ClientNode, add int64, now mclock.AbsTime) {
cm.lock.Lock()
defer cm.lock.Unlock()
cm.updateNodeRc(node, add, &node.params, now)
if node.corrBufValue > node.bufValue {
if node.log != nil {
node.log.add(now, fmt.Sprintf("corrected bv=%d oldBv=%d", node.corrBufValue, node.bufValue))
}
node.bufValue = uint64(node.corrBufValue)
node.bufValue = node.corrBufValue
}
}
@ -201,9 +217,29 @@ func (cm *ClientManager) updateParams(node *ClientNode, params ServerParams, now
cm.updateRecharge(now)
cm.updateCapFactor(now, true)
cm.totalConnected += params.MinRecharge - node.params.MinRecharge
cm.updateRaiseLimit()
cm.updateNodeRc(node, 0, &params, now)
}
// updateRaiseLimit recalculates the limiting value until which capLogFactor
// can be raised when no client freeze events occur
func (cm *ClientManager) updateRaiseLimit() {
if cm.capFactorRaiseThreshold == 0 {
cm.capLogFactorRaiseLimit = 0
return
}
limit := float64(cm.totalConnected + cm.capFactorRaiseThreshold)
limit2 := float64(cm.totalConnected) * capFactorRaiseThresholdRatio
if limit2 > limit {
limit = limit2
}
if limit <= float64(cm.totalRecharge) || cm.totalRecharge == 0 {
cm.capLogFactorRaiseLimit = 0
return
}
cm.capLogFactorRaiseLimit = math.Log(limit / float64(cm.totalRecharge))
}
// updateRecharge updates the recharge integrator and checks the recharge queue
// for nodes with recently filled buffers
func (cm *ClientManager) updateRecharge(now mclock.AbsTime) {
@ -212,7 +248,11 @@ func (cm *ClientManager) updateRecharge(now mclock.AbsTime) {
// updating is done in multiple steps if node buffers are filled and sumRecharge
// is decreased before the given target time
for cm.sumRecharge > 0 {
bonusRatio := cm.curve.ValueAt(cm.sumRecharge) / float64(cm.sumRecharge)
sumRecharge := cm.sumRecharge
if sumRecharge > cm.totalRecharge {
sumRecharge = cm.totalRecharge
}
bonusRatio := cm.curve.ValueAt(sumRecharge) / float64(sumRecharge)
if bonusRatio < 1 {
bonusRatio = 1
}
@ -232,7 +272,6 @@ func (cm *ClientManager) updateRecharge(now mclock.AbsTime) {
// finished recharging, update corrBufValue and sumRecharge if necessary and do next step
if rcqNode.corrBufValue < int64(rcqNode.params.BufLimit) {
rcqNode.corrBufValue = int64(rcqNode.params.BufLimit)
cm.updateCapFactor(lastUpdate, true)
cm.sumRecharge -= rcqNode.params.MinRecharge
}
cm.rcLastIntValue = rcqNode.rcFullIntValue
@ -253,9 +292,6 @@ func (cm *ClientManager) updateNodeRc(node *ClientNode, bvc int64, params *Serve
node.rcLastIntValue = cm.rcLastIntValue
}
node.corrBufValue += bvc
if node.corrBufValue < 0 {
node.corrBufValue = 0
}
diff := int64(params.BufLimit - node.params.BufLimit)
if diff > 0 {
node.corrBufValue += diff
@ -285,52 +321,48 @@ func (cm *ClientManager) updateNodeRc(node *ClientNode, bvc int64, params *Serve
cm.updateCapFactor(now, true)
cm.sumRecharge = sumRecharge
}
}
// reduceTotalCap reduces the total capacity allowance in case of a client freeze event
func (cm *ClientManager) reduceTotalCap(frozenCap uint64) {
cm.lock.Lock()
defer cm.lock.Unlock()
f := float64(frozenCap)
if f >= cm.totalCapacity {
return
}
now := cm.clock.Now()
cm.updateCapFactor(now, false)
cm.capLogFactor -= capFactorDrop * f / cm.totalCapacity
if cm.capLogFactor < minCapLogFactor {
cm.capLogFactor = minCapLogFactor
}
cm.updateCapFactor(now, true)
}
// updateCapFactor updates the total capacity factor. The capacity factor allows
// the total capacity of the system to go over the allowed total recharge value
// if the sum of momentarily recharging clients only exceeds the total recharge
// allowance in a very small fraction of time.
// The capacity factor is dropped quickly (with a small time constant) if sumRecharge
// exceeds totalRecharge. It is raised slowly (with a large time constant) if most
// of the total capacity is used by connected clients (totalConnected is larger than
// totalCapacity*capFactorRaiseThreshold) and sumRecharge stays under
// totalRecharge*totalConnected/totalCapacity.
// if clients go to frozen state sufficiently rarely.
// The capacity factor is dropped instantly by a small amount if a clients is frozen.
// It is raised slowly (with a large time constant) if the total connected capacity
// is close to the total allowed amount and no clients are frozen.
func (cm *ClientManager) updateCapFactor(now mclock.AbsTime, refresh bool) {
if cm.totalRecharge == 0 {
return
}
dt := now - cm.capLastUpdate
cm.capLastUpdate = now
var d float64
if cm.sumRecharge > cm.totalRecharge {
d = (1 - float64(cm.sumRecharge)/float64(cm.totalRecharge)) * capFactorDropTC
} else {
totalConnected := float64(cm.totalConnected)
var connRatio float64
if totalConnected < cm.totalCapacity {
connRatio = totalConnected / cm.totalCapacity
} else {
connRatio = 1
}
if connRatio > capFactorRaiseThreshold {
sumRecharge := float64(cm.sumRecharge)
limit := float64(cm.totalRecharge) * connRatio
if sumRecharge < limit {
d = (1 - sumRecharge/limit) * (connRatio - capFactorRaiseThreshold) * (1 / (1 - capFactorRaiseThreshold)) * capFactorRaiseTC
}
if cm.capLogFactor < cm.capLogFactorRaiseLimit {
cm.capLogFactor += capFactorRaiseTC * float64(dt)
if cm.capLogFactor > cm.capLogFactorRaiseLimit {
cm.capLogFactor = cm.capLogFactorRaiseLimit
}
}
if d != 0 {
cm.capLogFactor += d * float64(dt)
if cm.capLogFactor < 0 {
cm.capLogFactor = 0
}
if refresh {
cm.refreshCapacity()
}
if cm.capLogFactor > maxCapLogFactor {
cm.capLogFactor = maxCapLogFactor
}
if refresh {
cm.refreshCapacity()
}
}
@ -341,7 +373,6 @@ func (cm *ClientManager) refreshCapacity() {
if totalCapacity >= cm.totalCapacity*0.999 && totalCapacity <= cm.totalCapacity*1.001 {
return
}
cm.logTotalCap.Update(totalCapacity)
cm.totalCapacity = totalCapacity
if cm.totalCapacityCh != nil {
select {

12
les/flowcontrol/manager_test.go
View file

@ -63,7 +63,7 @@ func testConstantTotalCapacity(t *testing.T, nodeCount, maxCapacityNodes, random
}
m := NewClientManager(PieceWiseLinear{{0, totalCapacity}}, clock)
for _, n := range nodes {
n.bufLimit = n.capacity * 6000 //uint64(2000+rand.Intn(10000))
n.bufLimit = n.capacity * 6000
n.node = NewClientNode(m, ServerParams{BufLimit: n.bufLimit, MinRecharge: n.capacity})
}
maxNodes := make([]int, maxCapacityNodes)
@ -73,6 +73,7 @@ func testConstantTotalCapacity(t *testing.T, nodeCount, maxCapacityNodes, random
maxNodes[i] = rand.Intn(nodeCount)
}
var sendCount int
for i := 0; i < testLength; i++ {
now := clock.Now()
for _, idx := range maxNodes {
@ -83,13 +84,15 @@ func testConstantTotalCapacity(t *testing.T, nodeCount, maxCapacityNodes, random
maxNodes[rand.Intn(maxCapacityNodes)] = rand.Intn(nodeCount)
}
sendCount := randomSend
for sendCount > 0 {
sendCount += randomSend
failCount := randomSend * 10
for sendCount > 0 && failCount > 0 {
if nodes[rand.Intn(nodeCount)].send(t, now) {
sendCount--
} else {
failCount--
}
}
clock.Run(time.Millisecond)
}
@ -117,7 +120,6 @@ func (n *testNode) send(t *testing.T, now mclock.AbsTime) bool {
if bv < testMaxCost {
n.waitUntil = now + mclock.AbsTime((testMaxCost-bv)*1001000/n.capacity)
}
//n.waitUntil = now + mclock.AbsTime(float64(testMaxCost)*1001000/float64(n.capacity)*(1-float64(bv)/float64(n.bufLimit)))
n.totalCost += rcost
return true
}

6
les/freeclient.go
View file

@ -69,7 +69,7 @@ const (
)
// newFreeClientPool creates a new free client pool
func newFreeClientPool(db ethdb.Database, freeClientCap uint64, totalLimit int, clock mclock.Clock, removePeer func(string), logger *csvlogger.Logger) *freeClientPool {
func newFreeClientPool(db ethdb.Database, freeClientCap uint64, totalLimit int, clock mclock.Clock, removePeer func(string), metricsLogger, eventLogger *csvlogger.Logger) *freeClientPool {
pool := &freeClientPool{
db: db,
clock: clock,
@ -78,8 +78,8 @@ func newFreeClientPool(db ethdb.Database, freeClientCap uint64, totalLimit int,
disconnPool: prque.New(poolSetIndex),
freeClientCap: freeClientCap,
totalLimit: totalLimit,
logger: logger,
logTotalFreeConn: logger.NewChannel("totalFreeConn", 0),
logger: eventLogger,
logTotalFreeConn: metricsLogger.NewChannel("totalFreeConn", 0),
removePeer: removePeer,
}
pool.loadFromDb()

4
les/freeclient_test.go
View file

@ -61,7 +61,7 @@ func testFreeClientPool(t *testing.T, connLimit, clientCount int) {
}
disconnCh <- i
}
pool = newFreeClientPool(db, 1, 10000, &clock, disconnFn, nil)
pool = newFreeClientPool(db, 1, 10000, &clock, disconnFn, nil, nil)
)
pool.setLimits(connLimit, uint64(connLimit))
@ -130,7 +130,7 @@ func testFreeClientPool(t *testing.T, connLimit, clientCount int) {
// close and restart pool
pool.stop()
pool = newFreeClientPool(db, 1, 10000, &clock, disconnFn, nil)
pool = newFreeClientPool(db, 1, 10000, &clock, disconnFn, nil, nil)
pool.setLimits(connLimit, uint64(connLimit))
// try connecting all known peers (connLimit should be filled up)

650
les/handler.go
View file

@ -167,8 +167,6 @@ func NewProtocolManager(
if odr != nil {
manager.retriever = odr.retriever
manager.reqDist = odr.retriever.dist
} else {
manager.servingQueue = newServingQueue(int64(time.Millisecond * 10))
}
if ulcConfig != nil {
@ -366,23 +364,40 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
)
accept := func(reqID, reqCnt, maxCnt uint64) bool {
if reqCnt == 0 {
inSizeCost := func() uint64 {
if pm.server.costTracker != nil {
return pm.server.costTracker.realCost(0, msg.Size, 0)
}
return 0
}
if p.isFrozen() || reqCnt == 0 || p.fcClient == nil || reqCnt > maxCnt {
p.fcClient.OneTimeCost(inSizeCost())
return false
}
if p.fcClient == nil || reqCnt > maxCnt {
return false
maxCost = p.fcCosts.getMaxCost(msg.Code, reqCnt)
gf := float64(1)
if pm.server.costTracker != nil {
gf = pm.server.costTracker.globalFactor()
if gf < 0.001 {
p.Log().Error("Invalid global cost factor", "globalFactor", gf)
gf = 1
}
}
maxCost = p.fcCosts.getCost(msg.Code, reqCnt)
maxTime := uint64(float64(maxCost) / gf)
if accepted, bufShort, servingPriority := p.fcClient.AcceptRequest(reqID, responseCount, maxCost); !accepted {
if bufShort > 0 {
p.Log().Error("Request came too early", "remaining", common.PrettyDuration(time.Duration(bufShort*1000000/p.fcParams.MinRecharge)))
}
p.freezeClient()
p.Log().Warn("Request came too early", "remaining", common.PrettyDuration(time.Duration(bufShort*1000000/p.fcParams.MinRecharge)))
p.fcClient.OneTimeCost(inSizeCost())
return false
} else {
task = pm.servingQueue.newTask(servingPriority)
task = pm.servingQueue.newTask(p, maxTime, servingPriority)
}
return task.start()
if task.start() {
return true
}
p.fcClient.RequestProcessed(reqID, responseCount, maxCost, inSizeCost())
return false
}
if msg.Size > ProtocolMaxMsgSize {
@ -396,6 +411,10 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
p.responseLock.Lock()
defer p.responseLock.Unlock()
if p.isFrozen() {
amount = 0
reply = nil
}
var replySize uint32
if reply != nil {
replySize = reply.size()
@ -403,7 +422,9 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
var realCost uint64
if pm.server.costTracker != nil {
realCost = pm.server.costTracker.realCost(servingTime, msg.Size, replySize)
pm.server.costTracker.updateStats(msg.Code, amount, servingTime, realCost)
if amount != 0 {
pm.server.costTracker.updateStats(msg.Code, amount, servingTime, realCost)
}
} else {
realCost = maxCost
}
@ -471,94 +492,94 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
}
query := req.Query
if !accept(req.ReqID, query.Amount, MaxHeaderFetch) {
return errResp(ErrRequestRejected, "")
}
go func() {
hashMode := query.Origin.Hash != (common.Hash{})
first := true
maxNonCanonical := uint64(100)
if accept(req.ReqID, query.Amount, MaxHeaderFetch) {
go func() {
hashMode := query.Origin.Hash != (common.Hash{})
first := true
maxNonCanonical := uint64(100)
// Gather headers until the fetch or network limits is reached
var (
bytes common.StorageSize
headers []*types.Header
unknown bool
)
for !unknown && len(headers) < int(query.Amount) && bytes < softResponseLimit {
if !first && !task.waitOrStop() {
return
}
// Retrieve the next header satisfying the query
var origin *types.Header
if hashMode {
if first {
origin = pm.blockchain.GetHeaderByHash(query.Origin.Hash)
if origin != nil {
query.Origin.Number = origin.Number.Uint64()
// Gather headers until the fetch or network limits is reached
var (
bytes common.StorageSize
headers []*types.Header
unknown bool
)
for !unknown && len(headers) < int(query.Amount) && bytes < softResponseLimit {
if !first && !task.waitOrStop() {
sendResponse(req.ReqID, 0, nil, task.servingTime)
return
}
// Retrieve the next header satisfying the query
var origin *types.Header
if hashMode {
if first {
origin = pm.blockchain.GetHeaderByHash(query.Origin.Hash)
if origin != nil {
query.Origin.Number = origin.Number.Uint64()
}
} else {
origin = pm.blockchain.GetHeader(query.Origin.Hash, query.Origin.Number)
}
} else {
origin = pm.blockchain.GetHeader(query.Origin.Hash, query.Origin.Number)
origin = pm.blockchain.GetHeaderByNumber(query.Origin.Number)
}
} else {
origin = pm.blockchain.GetHeaderByNumber(query.Origin.Number)
}
if origin == nil {
break
}
headers = append(headers, origin)
bytes += estHeaderRlpSize
if origin == nil {
break
}
headers = append(headers, origin)
bytes += estHeaderRlpSize
// Advance to the next header of the query
switch {
case hashMode && query.Reverse:
// Hash based traversal towards the genesis block
ancestor := query.Skip + 1
if ancestor == 0 {
unknown = true
} else {
query.Origin.Hash, query.Origin.Number = pm.blockchain.GetAncestor(query.Origin.Hash, query.Origin.Number, ancestor, &maxNonCanonical)
unknown = (query.Origin.Hash == common.Hash{})
}
case hashMode && !query.Reverse:
// Hash based traversal towards the leaf block
var (
current = origin.Number.Uint64()
next = current + query.Skip + 1
)
if next <= current {
infos, _ := json.MarshalIndent(p.Peer.Info(), "", " ")
p.Log().Warn("GetBlockHeaders skip overflow attack", "current", current, "skip", query.Skip, "next", next, "attacker", infos)
unknown = true
} else {
if header := pm.blockchain.GetHeaderByNumber(next); header != nil {
nextHash := header.Hash()
expOldHash, _ := pm.blockchain.GetAncestor(nextHash, next, query.Skip+1, &maxNonCanonical)
if expOldHash == query.Origin.Hash {
query.Origin.Hash, query.Origin.Number = nextHash, next
// Advance to the next header of the query
switch {
case hashMode && query.Reverse:
// Hash based traversal towards the genesis block
ancestor := query.Skip + 1
if ancestor == 0 {
unknown = true
} else {
query.Origin.Hash, query.Origin.Number = pm.blockchain.GetAncestor(query.Origin.Hash, query.Origin.Number, ancestor, &maxNonCanonical)
unknown = (query.Origin.Hash == common.Hash{})
}
case hashMode && !query.Reverse:
// Hash based traversal towards the leaf block
var (
current = origin.Number.Uint64()
next = current + query.Skip + 1
)
if next <= current {
infos, _ := json.MarshalIndent(p.Peer.Info(), "", " ")
p.Log().Warn("GetBlockHeaders skip overflow attack", "current", current, "skip", query.Skip, "next", next, "attacker", infos)
unknown = true
} else {
if header := pm.blockchain.GetHeaderByNumber(next); header != nil {
nextHash := header.Hash()
expOldHash, _ := pm.blockchain.GetAncestor(nextHash, next, query.Skip+1, &maxNonCanonical)
if expOldHash == query.Origin.Hash {
query.Origin.Hash, query.Origin.Number = nextHash, next
} else {
unknown = true
}
} else {
unknown = true
}
}
case query.Reverse:
// Number based traversal towards the genesis block
if query.Origin.Number >= query.Skip+1 {
query.Origin.Number -= query.Skip + 1
} else {
unknown = true
}
}
case query.Reverse:
// Number based traversal towards the genesis block
if query.Origin.Number >= query.Skip+1 {
query.Origin.Number -= query.Skip + 1
} else {
unknown = true
}
case !query.Reverse:
// Number based traversal towards the leaf block
query.Origin.Number += query.Skip + 1
case !query.Reverse:
// Number based traversal towards the leaf block
query.Origin.Number += query.Skip + 1
}
first = false
}
first = false
}
sendResponse(req.ReqID, query.Amount, p.ReplyBlockHeaders(req.ReqID, headers), task.done())
}()
sendResponse(req.ReqID, query.Amount, p.ReplyBlockHeaders(req.ReqID, headers), task.done())
}()
}
case BlockHeadersMsg:
if pm.downloader == nil {
@ -600,27 +621,27 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
bodies []rlp.RawValue
)
reqCnt := len(req.Hashes)
if !accept(req.ReqID, uint64(reqCnt), MaxBodyFetch) {
return errResp(ErrRequestRejected, "")
}
go func() {
for i, hash := range req.Hashes {
if i != 0 && !task.waitOrStop() {
return
}
if bytes >= softResponseLimit {
break
}
// Retrieve the requested block body, stopping if enough was found
if number := rawdb.ReadHeaderNumber(pm.chainDb, hash); number != nil {
if data := rawdb.ReadBodyRLP(pm.chainDb, hash, *number); len(data) != 0 {
bodies = append(bodies, data)
bytes += len(data)
if accept(req.ReqID, uint64(reqCnt), MaxBodyFetch) {
go func() {
for i, hash := range req.Hashes {
if i != 0 && !task.waitOrStop() {
sendResponse(req.ReqID, 0, nil, task.servingTime)
return
}
if bytes >= softResponseLimit {
break
}
// Retrieve the requested block body, stopping if enough was found
if number := rawdb.ReadHeaderNumber(pm.chainDb, hash); number != nil {
if data := rawdb.ReadBodyRLP(pm.chainDb, hash, *number); len(data) != 0 {
bodies = append(bodies, data)
bytes += len(data)
}
}
}
}
sendResponse(req.ReqID, uint64(reqCnt), p.ReplyBlockBodiesRLP(req.ReqID, bodies), task.done())
}()
sendResponse(req.ReqID, uint64(reqCnt), p.ReplyBlockBodiesRLP(req.ReqID, bodies), task.done())
}()
}
case BlockBodiesMsg:
if pm.odr == nil {
@ -659,45 +680,45 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
data [][]byte
)
reqCnt := len(req.Reqs)
if !accept(req.ReqID, uint64(reqCnt), MaxCodeFetch) {
return errResp(ErrRequestRejected, "")
}
go func() {
for i, req := range req.Reqs {
if i != 0 && !task.waitOrStop() {
return
}
// Look up the root hash belonging to the request
number := rawdb.ReadHeaderNumber(pm.chainDb, req.BHash)
if number == nil {
p.Log().Warn("Failed to retrieve block num for code", "hash", req.BHash)
continue
}
header := rawdb.ReadHeader(pm.chainDb, req.BHash, *number)
if header == nil {
p.Log().Warn("Failed to retrieve header for code", "block", *number, "hash", req.BHash)
continue
}
triedb := pm.blockchain.StateCache().TrieDB()
if accept(req.ReqID, uint64(reqCnt), MaxCodeFetch) {
go func() {
for i, request := range req.Reqs {
if i != 0 && !task.waitOrStop() {
sendResponse(req.ReqID, 0, nil, task.servingTime)
return
}
// Look up the root hash belonging to the request
number := rawdb.ReadHeaderNumber(pm.chainDb, request.BHash)
if number == nil {
p.Log().Warn("Failed to retrieve block num for code", "hash", request.BHash)
continue
}
header := rawdb.ReadHeader(pm.chainDb, request.BHash, *number)
if header == nil {
p.Log().Warn("Failed to retrieve header for code", "block", *number, "hash", request.BHash)
continue
}
triedb := pm.blockchain.StateCache().TrieDB()
account, err := pm.getAccount(triedb, header.Root, common.BytesToHash(req.AccKey))
if err != nil {
p.Log().Warn("Failed to retrieve account for code", "block", header.Number, "hash", header.Hash(), "account", common.BytesToHash(req.AccKey), "err", err)
continue
account, err := pm.getAccount(triedb, header.Root, common.BytesToHash(request.AccKey))
if err != nil {
p.Log().Warn("Failed to retrieve account for code", "block", header.Number, "hash", header.Hash(), "account", common.BytesToHash(request.AccKey), "err", err)
continue
}
code, err := triedb.Node(common.BytesToHash(account.CodeHash))
if err != nil {
p.Log().Warn("Failed to retrieve account code", "block", header.Number, "hash", header.Hash(), "account", common.BytesToHash(request.AccKey), "codehash", common.BytesToHash(account.CodeHash), "err", err)
continue
}
// Accumulate the code and abort if enough data was retrieved
data = append(data, code)
if bytes += len(code); bytes >= softResponseLimit {
break
}
}
code, err := triedb.Node(common.BytesToHash(account.CodeHash))
if err != nil {
p.Log().Warn("Failed to retrieve account code", "block", header.Number, "hash", header.Hash(), "account", common.BytesToHash(req.AccKey), "codehash", common.BytesToHash(account.CodeHash), "err", err)
continue
}
// Accumulate the code and abort if enough data was retrieved
data = append(data, code)
if bytes += len(code); bytes >= softResponseLimit {
break
}
}
sendResponse(req.ReqID, uint64(reqCnt), p.ReplyCode(req.ReqID, data), task.done())
}()
sendResponse(req.ReqID, uint64(reqCnt), p.ReplyCode(req.ReqID, data), task.done())
}()
}
case CodeMsg:
if pm.odr == nil {
@ -736,37 +757,37 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
receipts []rlp.RawValue
)
reqCnt := len(req.Hashes)
if !accept(req.ReqID, uint64(reqCnt), MaxReceiptFetch) {
return errResp(ErrRequestRejected, "")
}
go func() {
for i, hash := range req.Hashes {
if i != 0 && !task.waitOrStop() {
return
}
if bytes >= softResponseLimit {
break
}
// Retrieve the requested block's receipts, skipping if unknown to us
var results types.Receipts
if number := rawdb.ReadHeaderNumber(pm.chainDb, hash); number != nil {
results = rawdb.ReadRawReceipts(pm.chainDb, hash, *number)
}
if results == nil {
if header := pm.blockchain.GetHeaderByHash(hash); header == nil || header.ReceiptHash != types.EmptyRootHash {
continue
if accept(req.ReqID, uint64(reqCnt), MaxReceiptFetch) {
go func() {
for i, hash := range req.Hashes {
if i != 0 && !task.waitOrStop() {
sendResponse(req.ReqID, 0, nil, task.servingTime)
return
}
if bytes >= softResponseLimit {
break
}
// Retrieve the requested block's receipts, skipping if unknown to us
var results types.Receipts
if number := rawdb.ReadHeaderNumber(pm.chainDb, hash); number != nil {
results = rawdb.ReadRawReceipts(pm.chainDb, hash, *number)
}
if results == nil {
if header := pm.blockchain.GetHeaderByHash(hash); header == nil || header.ReceiptHash != types.EmptyRootHash {
continue
}
}
// If known, encode and queue for response packet
if encoded, err := rlp.EncodeToBytes(results); err != nil {
log.Error("Failed to encode receipt", "err", err)
} else {
receipts = append(receipts, encoded)
bytes += len(encoded)
}
}
// If known, encode and queue for response packet
if encoded, err := rlp.EncodeToBytes(results); err != nil {
log.Error("Failed to encode receipt", "err", err)
} else {
receipts = append(receipts, encoded)
bytes += len(encoded)
}
}
sendResponse(req.ReqID, uint64(reqCnt), p.ReplyReceiptsRLP(req.ReqID, receipts), task.done())
}()
sendResponse(req.ReqID, uint64(reqCnt), p.ReplyReceiptsRLP(req.ReqID, receipts), task.done())
}()
}
case ReceiptsMsg:
if pm.odr == nil {
@ -805,70 +826,70 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
root common.Hash
)
reqCnt := len(req.Reqs)
if !accept(req.ReqID, uint64(reqCnt), MaxProofsFetch) {
return errResp(ErrRequestRejected, "")
if accept(req.ReqID, uint64(reqCnt), MaxProofsFetch) {
go func() {
nodes := light.NewNodeSet()
for i, request := range req.Reqs {
if i != 0 && !task.waitOrStop() {
sendResponse(req.ReqID, 0, nil, task.servingTime)
return
}
// Look up the root hash belonging to the request
var (
number *uint64
header *types.Header
trie state.Trie
)
if request.BHash != lastBHash {
root, lastBHash = common.Hash{}, request.BHash
if number = rawdb.ReadHeaderNumber(pm.chainDb, request.BHash); number == nil {
p.Log().Warn("Failed to retrieve block num for proof", "hash", request.BHash)
continue
}
if header = rawdb.ReadHeader(pm.chainDb, request.BHash, *number); header == nil {
p.Log().Warn("Failed to retrieve header for proof", "block", *number, "hash", request.BHash)
continue
}
root = header.Root
}
// Open the account or storage trie for the request
statedb := pm.blockchain.StateCache()
switch len(request.AccKey) {
case 0:
// No account key specified, open an account trie
trie, err = statedb.OpenTrie(root)
if trie == nil || err != nil {
p.Log().Warn("Failed to open storage trie for proof", "block", header.Number, "hash", header.Hash(), "root", root, "err", err)
continue
}
default:
// Account key specified, open a storage trie
account, err := pm.getAccount(statedb.TrieDB(), root, common.BytesToHash(request.AccKey))
if err != nil {
p.Log().Warn("Failed to retrieve account for proof", "block", header.Number, "hash", header.Hash(), "account", common.BytesToHash(request.AccKey), "err", err)
continue
}
trie, err = statedb.OpenStorageTrie(common.BytesToHash(request.AccKey), account.Root)
if trie == nil || err != nil {
p.Log().Warn("Failed to open storage trie for proof", "block", header.Number, "hash", header.Hash(), "account", common.BytesToHash(request.AccKey), "root", account.Root, "err", err)
continue
}
}
// Prove the user's request from the account or stroage trie
if err := trie.Prove(request.Key, request.FromLevel, nodes); err != nil {
p.Log().Warn("Failed to prove state request", "block", header.Number, "hash", header.Hash(), "err", err)
continue
}
if nodes.DataSize() >= softResponseLimit {
break
}
}
sendResponse(req.ReqID, uint64(reqCnt), p.ReplyProofsV2(req.ReqID, nodes.NodeList()), task.done())
}()
}
go func() {
nodes := light.NewNodeSet()
for i, req := range req.Reqs {
if i != 0 && !task.waitOrStop() {
return
}
// Look up the root hash belonging to the request
var (
number *uint64
header *types.Header
trie state.Trie
)
if req.BHash != lastBHash {
root, lastBHash = common.Hash{}, req.BHash
if number = rawdb.ReadHeaderNumber(pm.chainDb, req.BHash); number == nil {
p.Log().Warn("Failed to retrieve block num for proof", "hash", req.BHash)
continue
}
if header = rawdb.ReadHeader(pm.chainDb, req.BHash, *number); header == nil {
p.Log().Warn("Failed to retrieve header for proof", "block", *number, "hash", req.BHash)
continue
}
root = header.Root
}
// Open the account or storage trie for the request
statedb := pm.blockchain.StateCache()
switch len(req.AccKey) {
case 0:
// No account key specified, open an account trie
trie, err = statedb.OpenTrie(root)
if trie == nil || err != nil {
p.Log().Warn("Failed to open storage trie for proof", "block", header.Number, "hash", header.Hash(), "root", root, "err", err)
continue
}
default:
// Account key specified, open a storage trie
account, err := pm.getAccount(statedb.TrieDB(), root, common.BytesToHash(req.AccKey))
if err != nil {
p.Log().Warn("Failed to retrieve account for proof", "block", header.Number, "hash", header.Hash(), "account", common.BytesToHash(req.AccKey), "err", err)
continue
}
trie, err = statedb.OpenStorageTrie(common.BytesToHash(req.AccKey), account.Root)
if trie == nil || err != nil {
p.Log().Warn("Failed to open storage trie for proof", "block", header.Number, "hash", header.Hash(), "account", common.BytesToHash(req.AccKey), "root", account.Root, "err", err)
continue
}
}
// Prove the user's request from the account or stroage trie
if err := trie.Prove(req.Key, req.FromLevel, nodes); err != nil {
p.Log().Warn("Failed to prove state request", "block", header.Number, "hash", header.Hash(), "err", err)
continue
}
if nodes.DataSize() >= softResponseLimit {
break
}
}
sendResponse(req.ReqID, uint64(reqCnt), p.ReplyProofsV2(req.ReqID, nodes.NodeList()), task.done())
}()
case ProofsV2Msg:
if pm.odr == nil {
@ -907,53 +928,53 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
auxData [][]byte
)
reqCnt := len(req.Reqs)
if !accept(req.ReqID, uint64(reqCnt), MaxHelperTrieProofsFetch) {
return errResp(ErrRequestRejected, "")
}
go func() {
if accept(req.ReqID, uint64(reqCnt), MaxHelperTrieProofsFetch) {
go func() {
var (
lastIdx uint64
lastType uint
root common.Hash
auxTrie *trie.Trie
)
nodes := light.NewNodeSet()
for i, req := range req.Reqs {
if i != 0 && !task.waitOrStop() {
return
}
if auxTrie == nil || req.Type != lastType || req.TrieIdx != lastIdx {
auxTrie, lastType, lastIdx = nil, req.Type, req.TrieIdx
var (
lastIdx uint64
lastType uint
root common.Hash
auxTrie *trie.Trie
)
nodes := light.NewNodeSet()
for i, request := range req.Reqs {
if i != 0 && !task.waitOrStop() {
sendResponse(req.ReqID, 0, nil, task.servingTime)
return
}
if auxTrie == nil || request.Type != lastType || request.TrieIdx != lastIdx {
auxTrie, lastType, lastIdx = nil, request.Type, request.TrieIdx
var prefix string
if root, prefix = pm.getHelperTrie(req.Type, req.TrieIdx); root != (common.Hash{}) {
auxTrie, _ = trie.New(root, trie.NewDatabase(rawdb.NewTable(pm.chainDb, prefix)))
var prefix string
if root, prefix = pm.getHelperTrie(request.Type, request.TrieIdx); root != (common.Hash{}) {
auxTrie, _ = trie.New(root, trie.NewDatabase(rawdb.NewTable(pm.chainDb, prefix)))
}
}
}
if req.AuxReq == auxRoot {
var data []byte
if root != (common.Hash{}) {
data = root[:]
}
auxData = append(auxData, data)
auxBytes += len(data)
} else {
if auxTrie != nil {
auxTrie.Prove(req.Key, req.FromLevel, nodes)
}
if req.AuxReq != 0 {
data := pm.getHelperTrieAuxData(req)
if request.AuxReq == auxRoot {
var data []byte
if root != (common.Hash{}) {
data = root[:]
}
auxData = append(auxData, data)
auxBytes += len(data)
} else {
if auxTrie != nil {
auxTrie.Prove(request.Key, request.FromLevel, nodes)
}
if request.AuxReq != 0 {
data := pm.getHelperTrieAuxData(request)
auxData = append(auxData, data)
auxBytes += len(data)
}
}
if nodes.DataSize()+auxBytes >= softResponseLimit {
break
}
}
if nodes.DataSize()+auxBytes >= softResponseLimit {
break
}
}
sendResponse(req.ReqID, uint64(reqCnt), p.ReplyHelperTrieProofs(req.ReqID, HelperTrieResps{Proofs: nodes.NodeList(), AuxData: auxData}), task.done())
}()
sendResponse(req.ReqID, uint64(reqCnt), p.ReplyHelperTrieProofs(req.ReqID, HelperTrieResps{Proofs: nodes.NodeList(), AuxData: auxData}), task.done())
}()
}
case HelperTrieProofsMsg:
if pm.odr == nil {
@ -989,27 +1010,27 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
return errResp(ErrDecode, "msg %v: %v", msg, err)
}
reqCnt := len(req.Txs)
if !accept(req.ReqID, uint64(reqCnt), MaxTxSend) {
return errResp(ErrRequestRejected, "")
}
go func() {
stats := make([]light.TxStatus, len(req.Txs))
for i, tx := range req.Txs {
if i != 0 && !task.waitOrStop() {
return
}
hash := tx.Hash()
stats[i] = pm.txStatus(hash)
if stats[i].Status == core.TxStatusUnknown {
if errs := pm.txpool.AddRemotes([]*types.Transaction{tx}); errs[0] != nil {
stats[i].Error = errs[0].Error()
continue
if accept(req.ReqID, uint64(reqCnt), MaxTxSend) {
go func() {
stats := make([]light.TxStatus, len(req.Txs))
for i, tx := range req.Txs {
if i != 0 && !task.waitOrStop() {
sendResponse(req.ReqID, 0, nil, task.servingTime)
return
}
hash := tx.Hash()
stats[i] = pm.txStatus(hash)
if stats[i].Status == core.TxStatusUnknown {
if errs := pm.txpool.AddRemotes([]*types.Transaction{tx}); errs[0] != nil {
stats[i].Error = errs[0].Error()
continue
}
stats[i] = pm.txStatus(hash)
}
}
}
sendResponse(req.ReqID, uint64(reqCnt), p.ReplyTxStatus(req.ReqID, stats), task.done())
}()
sendResponse(req.ReqID, uint64(reqCnt), p.ReplyTxStatus(req.ReqID, stats), task.done())
}()
}
case GetTxStatusMsg:
if pm.txpool == nil {
@ -1024,19 +1045,19 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
return errResp(ErrDecode, "msg %v: %v", msg, err)
}
reqCnt := len(req.Hashes)
if !accept(req.ReqID, uint64(reqCnt), MaxTxStatus) {
return errResp(ErrRequestRejected, "")
}
go func() {
stats := make([]light.TxStatus, len(req.Hashes))
for i, hash := range req.Hashes {
if i != 0 && !task.waitOrStop() {
return
if accept(req.ReqID, uint64(reqCnt), MaxTxStatus) {
go func() {
stats := make([]light.TxStatus, len(req.Hashes))
for i, hash := range req.Hashes {
if i != 0 && !task.waitOrStop() {
sendResponse(req.ReqID, 0, nil, task.servingTime)
return
}
stats[i] = pm.txStatus(hash)
}
stats[i] = pm.txStatus(hash)
}
sendResponse(req.ReqID, uint64(reqCnt), p.ReplyTxStatus(req.ReqID, stats), task.done())
}()
sendResponse(req.ReqID, uint64(reqCnt), p.ReplyTxStatus(req.ReqID, stats), task.done())
}()
}
case TxStatusMsg:
if pm.odr == nil {
@ -1061,6 +1082,25 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
Obj: resp.Status,
}
case StopMsg:
if pm.odr == nil {
return errResp(ErrUnexpectedResponse, "")
}
p.freezeServer(true)
pm.retriever.frozen(p)
p.Log().Warn("Service stopped")
case ResumeMsg:
if pm.odr == nil {
return errResp(ErrUnexpectedResponse, "")
}
var bv uint64
if err := msg.Decode(&bv); err != nil {
return errResp(ErrDecode, "msg %v: %v", msg, err)
}
p.freezeServer(false)
p.Log().Warn("Service resumed")
default:
p.Log().Trace("Received unknown message", "code", msg.Code)
return errResp(ErrInvalidMsgCode, "%v", msg.Code)

1
les/helper_test.go
View file

@ -177,6 +177,7 @@ func newTestProtocolManager(lightSync bool, blocks int, generator func(int, *cor
if !lightSync {
srv := &LesServer{lesCommons: lesCommons{protocolManager: pm}}
pm.server = srv
pm.servingQueue = newServingQueue(int64(time.Millisecond*10), 1, nil)
pm.servingQueue.setThreads(4)
srv.defParams = flowcontrol.ServerParams{

65
les/peer.go
View file

@ -20,7 +20,9 @@ import (
"errors"
"fmt"
"math/big"
"math/rand"
"sync"
"sync/atomic"
"time"
"github.com/ethereum/go-ethereum/common"
@ -47,6 +49,12 @@ const (
allowedUpdateRate = time.Millisecond * 10 // time constant for recharging one byte of allowance
)
const (
freezeTimeBase = time.Millisecond * 700 // fixed component of client freeze time
freezeTimeRandom = time.Millisecond * 600 // random component of client freeze time
freezeCheckPeriod = time.Millisecond * 100 // buffer value recheck period after initial freeze time has elapsed
)
// if the total encoded size of a sent transaction batch is over txSizeCostLimit
// per transaction then the request cost is calculated as proportional to the
// encoded size instead of the transaction count
@ -86,6 +94,7 @@ type peer struct {
responseErrors int
updateCounter uint64
updateTime mclock.AbsTime
frozen uint32 // 1 if client is in frozen state
fcClient *flowcontrol.ClientNode // nil if the peer is server only
fcServer *flowcontrol.ServerNode // nil if the peer is client only
@ -129,8 +138,52 @@ func (p *peer) rejectUpdate(size uint64) bool {
return p.updateCounter > allowedUpdateBytes
}
// freezeClient temporarily puts the client in a frozen state which means all
// unprocessed and subsequent requests are dropped. Unfreezing happens automatically
// after a short time if the client's buffer value is at least in the slightly positive
// region. The client is also notified about being frozen/unfrozen with a Stop/Resume
// message.
func (p *peer) freezeClient() {
if atomic.SwapUint32(&p.frozen, 1) == 0 {
go func() {
p.SendStop()
time.Sleep(freezeTimeBase + time.Duration(rand.Int63n(int64(freezeTimeRandom))))
for {
bufValue, bufLimit := p.fcClient.BufferStatus()
if bufLimit == 0 {
return
}
if bufValue <= bufLimit/8 {
time.Sleep(freezeCheckPeriod)
} else {
atomic.StoreUint32(&p.frozen, 0)
p.SendResume(bufValue)
break
}
}
}()
}
}
// freezeServer processes Stop/Resume messages from the given server
func (p *peer) freezeServer(frozen bool) {
var f uint32
if frozen {
f = 1
}
if atomic.SwapUint32(&p.frozen, f) != f && frozen {
p.sendQueue.clear()
}
}
// isFrozen returns true if the client is frozen or the server has put our
// client in frozen state
func (p *peer) isFrozen() bool {
return atomic.LoadUint32(&p.frozen) != 0
}
func (p *peer) canQueue() bool {
return p.sendQueue.canQueue()
return p.sendQueue.canQueue() && !p.isFrozen()
}
func (p *peer) queueSend(f func()) {
@ -277,6 +330,16 @@ func (p *peer) SendAnnounce(request announceData) error {
return p2p.Send(p.rw, AnnounceMsg, request)
}
// SendStop notifies the client about being in frozen state
func (p *peer) SendStop() error {
return p2p.Send(p.rw, StopMsg, struct{}{})
}
// SendResume notifies the client about getting out of frozen state
func (p *peer) SendResume(bv uint64) error {
return p2p.Send(p.rw, ResumeMsg, bv)
}
// ReplyBlockHeaders creates a reply with a batch of block headers
func (p *peer) ReplyBlockHeaders(reqID uint64, headers []*types.Header) *reply {
data, _ := rlp.EncodeToBytes(headers)

4
les/protocol.go
View file

@ -42,7 +42,7 @@ var (
)
// Number of implemented message corresponding to different protocol versions.
var ProtocolLengths = map[uint]uint64{lpv2: 22}
var ProtocolLengths = map[uint]uint64{lpv2: 24}
const (
NetworkId = 1
@ -70,6 +70,8 @@ const (
SendTxV2Msg = 0x13
GetTxStatusMsg = 0x14
TxStatusMsg = 0x15
StopMsg = 0x16
ResumeMsg = 0x17
)
type requestInfo struct {

70
les/retrieve.go
View file

@ -78,8 +78,8 @@ type sentReq struct {
// after which delivered is set to true, the validity of the response is sent on the
// valid channel and no more responses are accepted.
type sentReqToPeer struct {
delivered bool
valid chan bool
delivered, frozen bool
event chan int
}
// reqPeerEvent is sent by the request-from-peer goroutine (tryRequest) to the
@ -95,6 +95,7 @@ const (
rpHardTimeout
rpDeliveredValid
rpDeliveredInvalid
rpNotDelivered
)
// newRetrieveManager creates the retrieve manager
@ -149,7 +150,7 @@ func (rm *retrieveManager) sendReq(reqID uint64, req *distReq, val validatorFunc
req.request = func(p distPeer) func() {
// before actually sending the request, put an entry into the sentTo map
r.lock.Lock()
r.sentTo[p] = sentReqToPeer{false, make(chan bool, 1)}
r.sentTo[p] = sentReqToPeer{delivered: false, frozen: false, event: make(chan int, 1)}
r.lock.Unlock()
return request(p)
}
@ -173,6 +174,17 @@ func (rm *retrieveManager) deliver(peer distPeer, msg *Msg) error {
return errResp(ErrUnexpectedResponse, "reqID = %v", msg.ReqID)
}
// frozen is called by the LES protocol manager when a server has suspended its service and we
// should not expect an answer for the requests already sent there
func (rm *retrieveManager) frozen(peer distPeer) {
rm.lock.RLock()
defer rm.lock.RUnlock()
for _, req := range rm.sentReqs {
req.frozen(peer)
}
}
// reqStateFn represents a state of the retrieve loop state machine
type reqStateFn func() reqStateFn
@ -215,7 +227,7 @@ func (r *sentReq) stateRequesting() reqStateFn {
go r.tryRequest()
r.lastReqQueued = true
return r.stateRequesting
case rpDeliveredInvalid:
case rpDeliveredInvalid, rpNotDelivered:
// if it was the last sent request (set to nil by update) then start a new one
if !r.lastReqQueued && r.lastReqSentTo == nil {
go r.tryRequest()
@ -277,7 +289,7 @@ func (r *sentReq) update(ev reqPeerEvent) {
r.reqSrtoCount++
case rpHardTimeout:
r.reqSrtoCount--
case rpDeliveredValid, rpDeliveredInvalid:
case rpDeliveredValid, rpDeliveredInvalid, rpNotDelivered:
if ev.peer == r.lastReqSentTo {
r.lastReqSentTo = nil
} else {
@ -343,12 +355,13 @@ func (r *sentReq) tryRequest() {
}()
select {
case ok := <-s.valid:
if ok {
r.eventsCh <- reqPeerEvent{rpDeliveredValid, p}
} else {
r.eventsCh <- reqPeerEvent{rpDeliveredInvalid, p}
case event := <-s.event:
if event == rpNotDelivered {
r.lock.Lock()
delete(r.sentTo, p)
r.lock.Unlock()
}
r.eventsCh <- reqPeerEvent{event, p}
return
case <-time.After(softRequestTimeout):
srto = true
@ -356,12 +369,13 @@ func (r *sentReq) tryRequest() {
}
select {
case ok := <-s.valid:
if ok {
r.eventsCh <- reqPeerEvent{rpDeliveredValid, p}
} else {
r.eventsCh <- reqPeerEvent{rpDeliveredInvalid, p}
case event := <-s.event:
if event == rpNotDelivered {
r.lock.Lock()
delete(r.sentTo, p)
r.lock.Unlock()
}
r.eventsCh <- reqPeerEvent{event, p}
case <-time.After(hardRequestTimeout):
hrto = true
r.eventsCh <- reqPeerEvent{rpHardTimeout, p}
@ -377,15 +391,37 @@ func (r *sentReq) deliver(peer distPeer, msg *Msg) error {
if !ok || s.delivered {
return errResp(ErrUnexpectedResponse, "reqID = %v", msg.ReqID)
}
if s.frozen {
return nil
}
valid := r.validate(peer, msg) == nil
r.sentTo[peer] = sentReqToPeer{true, s.valid}
s.valid <- valid
r.sentTo[peer] = sentReqToPeer{delivered: true, frozen: false, event: s.event}
if valid {
s.event <- rpDeliveredValid
} else {
s.event <- rpDeliveredInvalid
}
if !valid {
return errResp(ErrInvalidResponse, "reqID = %v", msg.ReqID)
}
return nil
}
// frozen sends a "not delivered" event to the peer event channel belonging to the
// given peer if the request has been sent there, causing the state machine to not
// expect an answer and potentially even send the request to the same peer again
// when canSend allows it.
func (r *sentReq) frozen(peer distPeer) {
r.lock.Lock()
defer r.lock.Unlock()
s, ok := r.sentTo[peer]
if ok && !s.delivered && !s.frozen {
r.sentTo[peer] = sentReqToPeer{delivered: false, frozen: true, event: s.event}
s.event <- rpNotDelivered
}
}
// stop stops the retrieval process and sets an error code that will be returned
// by getError
func (r *sentReq) stop(err error) {

60
les/server.go
View file

@ -39,6 +39,15 @@ import (
const bufLimitRatio = 6000 // fixed bufLimit/MRR ratio
const (
logFileName = "" // csv log file name (disabled if empty)
logClientPoolMetrics = true // log client pool metrics
logClientPoolEvents = false // detailed client pool event logging
logRequestServing = true // log request serving metrics and events
logBlockProcEvents = true // log block processing events
logProtocolHandler = true // log protocol handler events
)
type LesServer struct {
lesCommons
@ -50,6 +59,7 @@ type LesServer struct {
quitSync chan struct{}
onlyAnnounce bool
csvLogger *csvlogger.Logger
logTotalCap *csvlogger.Channel
thcNormal, thcBlockProcessing int // serving thread count for normal operation and block processing mode
@ -60,6 +70,11 @@ type LesServer struct {
}
func NewLesServer(eth *eth.Ethereum, config *eth.Config) (*LesServer, error) {
var csvLogger *csvlogger.Logger
if logFileName != "" {
csvLogger = csvlogger.NewLogger(logFileName, time.Second*10, "event, peerId")
}
quitSync := make(chan struct{})
pm, err := NewProtocolManager(
eth.BlockChain().Config(),
@ -81,13 +96,19 @@ func NewLesServer(eth *eth.Ethereum, config *eth.Config) (*LesServer, error) {
if err != nil {
return nil, err
}
if logProtocolHandler {
pm.logger = csvLogger
}
requestLogger := csvLogger
if !logRequestServing {
requestLogger = nil
}
pm.servingQueue = newServingQueue(int64(time.Millisecond*10), float64(config.LightServ)/100, requestLogger)
lesTopics := make([]discv5.Topic, len(AdvertiseProtocolVersions))
for i, pv := range AdvertiseProtocolVersions {
lesTopics[i] = lesTopic(eth.BlockChain().Genesis().Hash(), pv)
}
var csvLogger *csvlogger.Logger
csvLogger = csvlogger.NewLogger("/tmp/server.csv", time.Second*10, "event, peerId")
srv := &LesServer{
lesCommons: lesCommons{
@ -98,22 +119,22 @@ func NewLesServer(eth *eth.Ethereum, config *eth.Config) (*LesServer, error) {
bloomTrieIndexer: light.NewBloomTrieIndexer(eth.ChainDb(), nil, params.BloomBitsBlocks, params.BloomTrieFrequency),
protocolManager: pm,
},
costTracker: newCostTracker(eth.ChainDb(), config, csvLogger),
costTracker: newCostTracker(eth.ChainDb(), config, requestLogger),
quitSync: quitSync,
lesTopics: lesTopics,
onlyAnnounce: config.OnlyAnnounce,
csvLogger: csvLogger,
logTotalCap: requestLogger.NewChannel("totalCapacity", 0.01),
}
logger := log.New()
pm.server = srv
pm.logger = csvLogger
srv.thcNormal = config.LightServ * 4 / 100
if srv.thcNormal < 4 {
srv.thcNormal = 4
}
srv.thcBlockProcessing = config.LightServ/100 + 1
srv.fcManager = flowcontrol.NewClientManager(nil, &mclock.System{}, csvLogger)
srv.fcManager = flowcontrol.NewClientManager(nil, &mclock.System{})
chtSectionCount, _, _ := srv.chtIndexer.Sections()
if chtSectionCount != 0 {
@ -151,7 +172,11 @@ func (s *LesServer) APIs() []rpc.API {
func (s *LesServer) startEventLoop() {
s.protocolManager.wg.Add(1)
var processing bool
blockProcLogger := s.csvLogger
if !logBlockProcEvents {
blockProcLogger = nil
}
var processing, procLast bool
blockProcFeed := make(chan bool, 100)
s.protocolManager.blockchain.(*core.BlockChain).SubscribeBlockProcessingEvent(blockProcFeed)
totalRechargeCh := make(chan uint64, 100)
@ -159,12 +184,19 @@ func (s *LesServer) startEventLoop() {
totalCapacityCh := make(chan uint64, 100)
updateRecharge := func() {
if processing {
if !procLast {
blockProcLogger.Event("block processing started")
}
s.protocolManager.servingQueue.setThreads(s.thcBlockProcessing)
s.fcManager.SetRechargeCurve(flowcontrol.PieceWiseLinear{{0, 0}, {totalRecharge, totalRecharge}})
} else {
if procLast {
blockProcLogger.Event("block processing finished")
}
s.protocolManager.servingQueue.setThreads(s.thcNormal)
s.fcManager.SetRechargeCurve(flowcontrol.PieceWiseLinear{{0, 0}, {totalRecharge / 10, totalRecharge}, {totalRecharge, totalRecharge}})
s.fcManager.SetRechargeCurve(flowcontrol.PieceWiseLinear{{0, 0}, {totalRecharge / 16, totalRecharge / 2}, {totalRecharge / 2, totalRecharge / 2}, {totalRecharge, totalRecharge}})
}
procLast = processing
}
updateRecharge()
totalCapacity := s.fcManager.SubscribeTotalCapacity(totalCapacityCh)
@ -178,6 +210,7 @@ func (s *LesServer) startEventLoop() {
case totalRecharge = <-totalRechargeCh:
updateRecharge()
case totalCapacity = <-totalCapacityCh:
s.logTotalCap.Update(float64(totalCapacity))
s.priorityClientPool.setLimits(s.maxPeers, totalCapacity)
case <-s.protocolManager.quitSync:
s.protocolManager.wg.Done()
@ -212,8 +245,17 @@ func (s *LesServer) Start(srvr *p2p.Server) {
log.Warn("Light peer count limited", "specified", s.maxPeers, "allowed", freePeers)
}
s.freeClientPool = newFreeClientPool(s.chainDb, s.freeClientCap, 10000, mclock.System{}, func(id string) { go s.protocolManager.removePeer(id) }, s.csvLogger)
s.priorityClientPool = newPriorityClientPool(s.freeClientCap, s.protocolManager.peers, s.freeClientPool, s.csvLogger)
s.fcManager.SetCapFactorRaiseThreshold(s.freeClientCap * 2)
poolMetricsLogger := s.csvLogger
if !logClientPoolMetrics {
poolMetricsLogger = nil
}
poolEventLogger := s.csvLogger
if !logClientPoolEvents {
poolEventLogger = nil
}
s.freeClientPool = newFreeClientPool(s.chainDb, s.freeClientCap, 10000, mclock.System{}, func(id string) { go s.protocolManager.removePeer(id) }, poolMetricsLogger, poolEventLogger)
s.priorityClientPool = newPriorityClientPool(s.freeClientCap, s.protocolManager.peers, s.freeClientPool, poolMetricsLogger, poolEventLogger)
s.protocolManager.peers.notify(s.priorityClientPool)
s.csvLogger.Start()

219
les/servingqueue.go
View file

@ -17,16 +17,24 @@
package les
import (
"fmt"
"sort"
"sync"
"sync/atomic"
"github.com/ethereum/go-ethereum/common/mclock"
"github.com/ethereum/go-ethereum/common/prque"
"github.com/ethereum/go-ethereum/les/csvlogger"
)
// servingQueue allows running tasks in a limited number of threads and puts the
// waiting tasks in a priority queue
type servingQueue struct {
tokenCh chan runToken
recentTime, queuedTime, servingTimeDiff uint64
burstLimit, burstDropLimit uint64
burstDecRate float64
lastUpdate mclock.AbsTime
queueAddCh, queueBestCh chan *servingTask
stopThreadCh, quit chan struct{}
setThreadsCh chan int
@ -36,6 +44,10 @@ type servingQueue struct {
queue *prque.Prque // priority queue for waiting or suspended tasks
best *servingTask // the highest priority task (not included in the queue)
suspendBias int64 // priority bias against suspending an already running task
logger *csvlogger.Logger
logRecentTime *csvlogger.Channel
logQueuedTime *csvlogger.Channel
}
// servingTask represents a request serving task. Tasks can be implemented to
@ -47,12 +59,13 @@ type servingQueue struct {
// - run: execute a single step; return true if finished
// - after: executed after run finishes or returns an error, receives the total serving time
type servingTask struct {
sq *servingQueue
servingTime uint64
priority int64
biasAdded bool
token runToken
tokenCh chan runToken
sq *servingQueue
servingTime, timeAdded, maxTime, expTime uint64
peer *peer
priority int64
biasAdded bool
token runToken
tokenCh chan runToken
}
// runToken received by servingTask.start allows the task to run. Closing the
@ -63,20 +76,19 @@ type runToken chan struct{}
// start blocks until the task can start and returns true if it is allowed to run.
// Returning false means that the task should be cancelled.
func (t *servingTask) start() bool {
if t.peer.isFrozen() {
return false
}
t.tokenCh = make(chan runToken, 1)
select {
case t.token = <-t.sq.tokenCh:
default:
t.tokenCh = make(chan runToken, 1)
select {
case t.sq.queueAddCh <- t:
case <-t.sq.quit:
return false
}
select {
case t.token = <-t.tokenCh:
case <-t.sq.quit:
return false
}
case t.sq.queueAddCh <- t:
case <-t.sq.quit:
return false
}
select {
case t.token = <-t.tokenCh:
case <-t.sq.quit:
return false
}
if t.token == nil {
return false
@ -90,6 +102,14 @@ func (t *servingTask) start() bool {
func (t *servingTask) done() uint64 {
t.servingTime += uint64(mclock.Now())
close(t.token)
diff := t.servingTime - t.timeAdded
t.timeAdded = t.servingTime
if t.expTime > diff {
t.expTime -= diff
atomic.AddUint64(&t.sq.servingTimeDiff, t.expTime)
} else {
t.expTime = 0
}
return t.servingTime
}
@ -107,16 +127,22 @@ func (t *servingTask) waitOrStop() bool {
}
// newServingQueue returns a new servingQueue
func newServingQueue(suspendBias int64) *servingQueue {
func newServingQueue(suspendBias int64, utilTarget float64, logger *csvlogger.Logger) *servingQueue {
sq := &servingQueue{
queue: prque.New(nil),
suspendBias: suspendBias,
tokenCh: make(chan runToken),
queueAddCh: make(chan *servingTask, 100),
queueBestCh: make(chan *servingTask),
stopThreadCh: make(chan struct{}),
quit: make(chan struct{}),
setThreadsCh: make(chan int, 10),
queue: prque.New(nil),
suspendBias: suspendBias,
queueAddCh: make(chan *servingTask, 100),
queueBestCh: make(chan *servingTask),
stopThreadCh: make(chan struct{}),
quit: make(chan struct{}),
setThreadsCh: make(chan int, 10),
burstLimit: uint64(utilTarget * bufLimitRatio * 1200000),
burstDropLimit: uint64(utilTarget * bufLimitRatio * 1000000),
burstDecRate: utilTarget,
lastUpdate: mclock.Now(),
logger: logger,
logRecentTime: logger.NewMinMaxChannel("recentTime", false),
logQueuedTime: logger.NewMinMaxChannel("queuedTime", false),
}
sq.wg.Add(2)
go sq.queueLoop()
@ -125,9 +151,12 @@ func newServingQueue(suspendBias int64) *servingQueue {
}
// newTask creates a new task with the given priority
func (sq *servingQueue) newTask(priority int64) *servingTask {
func (sq *servingQueue) newTask(peer *peer, maxTime uint64, priority int64) *servingTask {
return &servingTask{
sq: sq,
peer: peer,
maxTime: maxTime,
expTime: maxTime,
priority: priority,
}
}
@ -144,18 +173,12 @@ func (sq *servingQueue) threadController() {
select {
case best := <-sq.queueBestCh:
best.tokenCh <- token
default:
select {
case best := <-sq.queueBestCh:
best.tokenCh <- token
case sq.tokenCh <- token:
case <-sq.stopThreadCh:
sq.wg.Done()
return
case <-sq.quit:
sq.wg.Done()
return
}
case <-sq.stopThreadCh:
sq.wg.Done()
return
case <-sq.quit:
sq.wg.Done()
return
}
<-token
select {
@ -170,6 +193,100 @@ func (sq *servingQueue) threadController() {
}
}
type (
// peerTasks lists the tasks received from a given peer when selecting peers to freeze
peerTasks struct {
peer *peer
list []*servingTask
sumTime uint64
worstPriority int64
}
// peerList is a sortable list of peerTasks
peerList []*peerTasks
)
func (l peerList) Len() int {
return len(l)
}
func (l peerList) Less(i, j int) bool {
return (l[i].worstPriority - l[j].worstPriority) < 0
}
func (l peerList) Swap(i, j int) {
l[i], l[j] = l[j], l[i]
}
// freezePeers selects the peers with the worst priority queued tasks and freezes
// them until burstTime goes under burstDropLimit or all peers are frozen
func (sq *servingQueue) freezePeers() {
peerMap := make(map[*peer]*peerTasks)
var peerList peerList
if sq.best != nil {
sq.queue.Push(sq.best, sq.best.priority)
}
sq.best = nil
for sq.queue.Size() > 0 {
task := sq.queue.PopItem().(*servingTask)
tasks := peerMap[task.peer]
if tasks == nil {
tasks = &peerTasks{
peer: task.peer,
worstPriority: task.priority,
}
peerMap[task.peer] = tasks
peerList = append(peerList, tasks)
} else {
if tasks.worstPriority-task.priority > 0 {
tasks.worstPriority = task.priority
}
}
tasks.list = append(tasks.list, task)
tasks.sumTime += task.expTime
}
sort.Sort(peerList)
drop := true
sq.logger.Event("freezing peers")
for _, tasks := range peerList {
if drop {
tasks.peer.freezeClient()
tasks.peer.fcClient.Freeze()
sq.queuedTime -= tasks.sumTime
if sq.logQueuedTime != nil {
sq.logQueuedTime.Update(float64(sq.queuedTime) / 1000)
}
sq.logger.Event(fmt.Sprintf("frozen peer sumTime=%d, %v", tasks.sumTime, tasks.peer.id))
drop = sq.recentTime+sq.queuedTime > sq.burstDropLimit
for _, task := range tasks.list {
task.tokenCh <- nil
}
} else {
for _, task := range tasks.list {
sq.queue.Push(task, task.priority)
}
}
}
if sq.queue.Size() > 0 {
sq.best = sq.queue.PopItem().(*servingTask)
}
}
// updateRecentTime recalculates the recent serving time value
func (sq *servingQueue) updateRecentTime() {
subTime := atomic.SwapUint64(&sq.servingTimeDiff, 0)
now := mclock.Now()
dt := now - sq.lastUpdate
sq.lastUpdate = now
if dt > 0 {
subTime += uint64(float64(dt) * sq.burstDecRate)
}
if sq.recentTime > subTime {
sq.recentTime -= subTime
} else {
sq.recentTime = 0
}
}
// addTask inserts a task into the priority queue
func (sq *servingQueue) addTask(task *servingTask) {
if sq.best == nil {
@ -177,10 +294,18 @@ func (sq *servingQueue) addTask(task *servingTask) {
} else if task.priority > sq.best.priority {
sq.queue.Push(sq.best, sq.best.priority)
sq.best = task
return
} else {
sq.queue.Push(task, task.priority)
}
sq.updateRecentTime()
sq.queuedTime += task.expTime
if sq.logQueuedTime != nil {
sq.logRecentTime.Update(float64(sq.recentTime) / 1000)
sq.logQueuedTime.Update(float64(sq.queuedTime) / 1000)
}
if sq.recentTime+sq.queuedTime > sq.burstLimit {
sq.freezePeers()
}
}
// queueLoop is an event loop running in a goroutine. It receives tasks from queueAddCh
@ -189,10 +314,18 @@ func (sq *servingQueue) addTask(task *servingTask) {
func (sq *servingQueue) queueLoop() {
for {
if sq.best != nil {
expTime := sq.best.expTime
select {
case task := <-sq.queueAddCh:
sq.addTask(task)
case sq.queueBestCh <- sq.best:
sq.updateRecentTime()
sq.queuedTime -= expTime
sq.recentTime += expTime
if sq.logQueuedTime != nil {
sq.logRecentTime.Update(float64(sq.recentTime) / 1000)
sq.logQueuedTime.Update(float64(sq.queuedTime) / 1000)
}
if sq.queue.Size() == 0 {
sq.best = nil
} else {