From 9bf2d758f36cb8a26e0e4010b77cb28d181bbdd1 Mon Sep 17 00:00:00 2001 From: Zsolt Felfoldi Date: Mon, 18 Mar 2019 00:44:18 +0100 Subject: [PATCH] les, les/flowcontrol: implement client freeze feature --- les/api.go | 6 +- les/costtracker.go | 14 +- les/csvlogger/csvlogger.go | 23 +- les/execqueue.go | 9 +- les/flowcontrol/control.go | 89 ++++- les/flowcontrol/manager.go | 141 ++++--- les/flowcontrol/manager_test.go | 12 +- les/freeclient.go | 6 +- les/freeclient_test.go | 4 +- les/handler.go | 650 +++++++++++++++++--------------- les/helper_test.go | 1 + les/peer.go | 65 +++- les/protocol.go | 4 +- les/retrieve.go | 70 +++- les/server.go | 60 ++- les/servingqueue.go | 219 ++++++++--- 16 files changed, 901 insertions(+), 472 deletions(-) diff --git a/les/api.go b/les/api.go index dbeb4a6317..24547f3da7 100644 --- a/les/api.go +++ b/les/api.go @@ -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), } } diff --git a/les/costtracker.go b/les/costtracker.go index 3b1327a2f4..6e822fc282 100644 --- a/les/costtracker.go +++ b/les/costtracker.go @@ -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 } diff --git a/les/csvlogger/csvlogger.go b/les/csvlogger/csvlogger.go index b82710aaf7..9a4093cb9a 100644 --- a/les/csvlogger/csvlogger.go +++ b/les/csvlogger/csvlogger.go @@ -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 diff --git a/les/execqueue.go b/les/execqueue.go index 614721bf0d..e0c88a990f 100644 --- a/les/execqueue.go +++ b/les/execqueue.go @@ -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() { diff --git a/les/flowcontrol/control.go b/les/flowcontrol/control.go index c03f673b28..fed8ae9a37 100644 --- a/les/flowcontrol/control.go +++ b/les/flowcontrol/control.go @@ -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 diff --git a/les/flowcontrol/manager.go b/les/flowcontrol/manager.go index b80bf666d5..d145316c24 100644 --- a/les/flowcontrol/manager.go +++ b/les/flowcontrol/manager.go @@ -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, ¶ms, 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 { diff --git a/les/flowcontrol/manager_test.go b/les/flowcontrol/manager_test.go index 4e7746d400..b32ec55990 100644 --- a/les/flowcontrol/manager_test.go +++ b/les/flowcontrol/manager_test.go @@ -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 } diff --git a/les/freeclient.go b/les/freeclient.go index f58bba86b3..958b7daa78 100644 --- a/les/freeclient.go +++ b/les/freeclient.go @@ -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() diff --git a/les/freeclient_test.go b/les/freeclient_test.go index e5f264604c..5a58a6c1c7 100644 --- a/les/freeclient_test.go +++ b/les/freeclient_test.go @@ -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) diff --git a/les/handler.go b/les/handler.go index d810987e44..a10104dc7d 100644 --- a/les/handler.go +++ b/les/handler.go @@ -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) diff --git a/les/helper_test.go b/les/helper_test.go index 878e44404c..84ca277d7a 100644 --- a/les/helper_test.go +++ b/les/helper_test.go @@ -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{ diff --git a/les/peer.go b/les/peer.go index 42c13ab7d1..e4a7d01e8b 100644 --- a/les/peer.go +++ b/les/peer.go @@ -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) diff --git a/les/protocol.go b/les/protocol.go index 24dc9bdeae..2cbc6ccf74 100644 --- a/les/protocol.go +++ b/les/protocol.go @@ -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 { diff --git a/les/retrieve.go b/les/retrieve.go index dd9d14598b..d17a02e1ae 100644 --- a/les/retrieve.go +++ b/les/retrieve.go @@ -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) { diff --git a/les/server.go b/les/server.go index f04e49db54..27d4ce091f 100644 --- a/les/server.go +++ b/les/server.go @@ -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() diff --git a/les/servingqueue.go b/les/servingqueue.go index 2438fdfe3c..8ec2fe0f51 100644 --- a/les/servingqueue.go +++ b/les/servingqueue.go @@ -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 {