import std/[tables, sets], chronicles, chronos, eth/[rlp, common], eth/p2p/[rlpx, private/p2p_types], ./private/les_types const maxSamples = 100000 rechargingScale = 1000000 lesStatsKey = "les.flow_control.stats" lesStatsVer = 0 logScope: topics = "les flow_control" # TODO: move this somewhere proc pop[A, B](t: var Table[A, B], key: A): B = result = t[key] t.del(key) when LesTime is SomeInteger: template `/`(lhs, rhs: LesTime): LesTime = lhs div rhs when defined(testing): var lesTime* = LesTime(0) template now(): LesTime = lesTime template advanceTime(t) = lesTime += LesTime(t) else: import times let startTime = epochTime() proc now(): LesTime = return LesTime((times.epochTime() - startTime) * 1000.0) proc addSample(ra: var StatsRunningAverage; x, y: float64) = if ra.count >= maxSamples: let decay = float64(ra.count + 1 - maxSamples) / maxSamples template applyDecay(x) = x -= x * decay applyDecay ra.sumX applyDecay ra.sumY applyDecay ra.sumXX applyDecay ra.sumXY ra.count = maxSamples - 1 inc ra.count ra.sumX += x ra.sumY += y ra.sumXX += x * x ra.sumXY += x * y proc calc(ra: StatsRunningAverage): tuple[m, b: float] = if ra.count == 0: return let count = float64(ra.count) let d = count * ra.sumXX - ra.sumX * ra.sumX if d < 0.001: return (m: ra.sumY / count, b: 0.0) result.m = (count * ra.sumXY - ra.sumX * ra.sumY) / d result.b = (ra.sumY / count) - (result.m * ra.sumX / count) proc currentRequestsCosts*(network: LesNetwork, les: ProtocolInfo): seq[ReqCostInfo] = # Make sure the message costs are already initialized doAssert network.messageStats.len > les.messages[^1].id, "Have you called `initFlowControl`" for msg in les.messages: var (m, b) = network.messageStats[msg.id].calc() if m < 0: b += m m = 0 if b < 0: b = 0 result.add ReqCostInfo(msgId: msg.id, baseCost: ReqCostInt(b * 2), reqCost: ReqCostInt(m * 2)) proc persistMessageStats*(network: LesNetwork) = # XXX: Because of the package_visible_types template magic, Nim complains # when we pass the messageStats expression directly to `encodeList` let stats = network.messageStats network.setSetting(lesStatsKey, rlp.encodeList(lesStatsVer, stats)) proc loadMessageStats*(network: LesNetwork, les: ProtocolInfo): bool = block readFromDB: var stats = network.getSetting(lesStatsKey) if stats.len == 0: notice "LES stats not present in the database" break readFromDB try: var statsRlp = rlpFromBytes(stats) if not statsRlp.enterList: notice "Found a corrupted LES stats record" break readFromDB let version = statsRlp.read(int) if version != lesStatsVer: notice "Found an outdated LES stats record" break readFromDB statsRlp >> network.messageStats if network.messageStats.len <= les.messages[^1].id: notice "Found an incomplete LES stats record" break readFromDB return true except RlpError as e: error "Error while loading LES message stats", err = e.msg newSeq(network.messageStats, les.messages[^1].id + 1) return false proc update(s: var FlowControlState, t: LesTime) = let dt = max(t - s.lastUpdate, LesTime(0)) s.bufValue = min( s.bufValue + s.minRecharge * dt, s.bufLimit) s.lastUpdate = t proc init(s: var FlowControlState, bufLimit: BufValueInt, minRecharge: int, t: LesTime) = s.bufValue = bufLimit s.bufLimit = bufLimit s.minRecharge = minRecharge s.lastUpdate = t func canMakeRequest(s: FlowControlState, maxCost: ReqCostInt): (LesTime, float64) = ## Returns the required waiting time before sending a request and ## the estimated buffer level afterwards (as a fraction of the limit) const safetyMargin = 50 var maxCost = min( maxCost + safetyMargin * s.minRecharge, s.bufLimit) if s.bufValue >= maxCost: result[1] = float64(s.bufValue - maxCost) / float64(s.bufLimit) else: result[0] = (maxCost - s.bufValue) / s.minRecharge func canServeRequest(srv: LesNetwork): bool = result = srv.reqCount < srv.maxReqCount and srv.reqCostSum < srv.maxReqCostSum proc rechargeReqCost(peer: LesPeer, t: LesTime) = let dt = t - peer.lastRechargeTime peer.reqCostVal += peer.reqCostGradient * dt / rechargingScale peer.lastRechargeTime = t if peer.isRecharging and t >= peer.rechargingEndsAt: peer.isRecharging = false peer.reqCostGradient = 0 peer.reqCostVal = 0 proc updateRechargingParams(peer: LesPeer, network: LesNetwork) = peer.reqCostGradient = 0 if peer.reqCount > 0: peer.reqCostGradient = rechargingScale / network.reqCount if peer.isRecharging: peer.reqCostGradient = (network.rechargingRate * (peer.rechargingPower / network.totalRechargingPower).int64).int peer.rechargingEndsAt = peer.lastRechargeTime + LesTime(peer.reqCostVal * rechargingScale / -peer.reqCostGradient ) proc trackRequests(network: LesNetwork, peer: LesPeer, reqCountChange: int) = peer.reqCount += reqCountChange network.reqCount += reqCountChange doAssert peer.reqCount >= 0 and network.reqCount >= 0 if peer.reqCount == 0: # All requests have been finished. Start recharging. peer.isRecharging = true network.totalRechargingPower += peer.rechargingPower elif peer.reqCount == reqCountChange and peer.isRecharging: # `peer.reqCount` must have been 0 for the condition above to hold. # This is a transition from recharging to serving state. peer.isRecharging = false network.totalRechargingPower -= peer.rechargingPower peer.startReqCostVal = peer.reqCostVal updateRechargingParams peer, network proc updateFlowControl(network: LesNetwork, t: LesTime) = while true: var firstTime = t for peer in network.peers: # TODO: perhaps use a bin heap here if peer.isRecharging and peer.rechargingEndsAt < firstTime: firstTime = peer.rechargingEndsAt let rechargingEndedForSomePeer = firstTime < t network.reqCostSum = 0 for peer in network.peers: peer.rechargeReqCost firstTime network.reqCostSum += peer.reqCostVal if rechargingEndedForSomePeer: for peer in network.peers: if peer.isRecharging: updateRechargingParams peer, network else: network.lastUpdate = t return proc endPendingRequest*(network: LesNetwork, peer: LesPeer, t: LesTime) = if peer.reqCount > 0: network.updateFlowControl t network.trackRequests peer, -1 network.updateFlowControl t proc enlistInFlowControl*(network: LesNetwork, peer: LesPeer, peerRechargingPower = 100) = let t = now() doAssert peer.isServer or peer.isClient # Each Peer must be potential communication partner for us. # There will be useless peers on the network, but the logic # should make sure to disconnect them earlier in `onPeerConnected`. if peer.isServer: peer.localFlowState.init network.bufferLimit, network.minRechargingRate, t peer.pendingReqs = initTable[int, ReqCostInt]() if peer.isClient: peer.remoteFlowState.init network.bufferLimit, network.minRechargingRate, t peer.lastRechargeTime = t peer.rechargingEndsAt = t peer.rechargingPower = peerRechargingPower network.updateFlowControl t proc delistFromFlowControl*(network: LesNetwork, peer: LesPeer) = let t = now() # XXX: perhaps this is not safe with our reqCount logic. # The original code may depend on the binarity of the `serving` flag. network.endPendingRequest peer, t network.updateFlowControl t proc initFlowControl*(network: LesNetwork, les: ProtocolInfo, maxReqCount, maxReqCostSum, reqCostTarget: int) = network.rechargingRate = rechargingScale * (rechargingScale / (100 * rechargingScale / reqCostTarget - rechargingScale)) network.maxReqCount = maxReqCount network.maxReqCostSum = maxReqCostSum if not network.loadMessageStats(les): warn "Failed to load persisted LES message stats. " & "Flow control will be re-initilized." proc canMakeRequest(peer: var LesPeer, maxCost: int): (LesTime, float64) = peer.localFlowState.update now() return peer.localFlowState.canMakeRequest(maxCost) template getRequestCost(peer: LesPeer, localOrRemote: untyped, msgId, costQuantity: int): ReqCostInt = let baseCost = peer.`localOrRemote ReqCosts`[msgId].baseCost reqCost = peer.`localOrRemote ReqCosts`[msgId].reqCost min(baseCost + reqCost * costQuantity, peer.`localOrRemote FlowState`.bufLimit) proc trackOutgoingRequest*(network: LesNetwork, peer: LesPeer, msgId, reqId, costQuantity: int) = let maxCost = peer.getRequestCost(local, msgId, costQuantity) peer.localFlowState.bufValue -= maxCost peer.pendingReqsCost += maxCost peer.pendingReqs[reqId] = peer.pendingReqsCost proc trackIncomingResponse*(peer: LesPeer, reqId: int, bv: BufValueInt) = let bv = min(bv, peer.localFlowState.bufLimit) if not peer.pendingReqs.hasKey(reqId): return let costsSumAtSending = peer.pendingReqs.pop(reqId) let costsSumChange = peer.pendingReqsCost - costsSumAtSending peer.localFlowState.bufValue = if bv > costsSumChange: bv - costsSumChange else: 0 peer.localFlowState.lastUpdate = now() proc acceptRequest*(network: LesNetwork, peer: LesPeer, msgId, costQuantity: int): Future[bool] {.async.} = let t = now() let reqCost = peer.getRequestCost(remote, msgId, costQuantity) peer.remoteFlowState.update t network.updateFlowControl t while not network.canServeRequest: await sleepAsync(chronos.milliseconds(10)) if peer notin network.peers: # The peer was disconnected or the network # was shut down while we waited return false network.trackRequests peer, +1 network.updateFlowControl network.lastUpdate if reqCost > peer.remoteFlowState.bufValue: error "LES peer sent request too early", recharge = (reqCost - peer.remoteFlowState.bufValue) * rechargingScale / peer.remoteFlowState.minRecharge return false return true proc bufValueAfterRequest*(network: LesNetwork, peer: LesPeer, msgId: int, quantity: int): BufValueInt = let t = now() let costs = peer.remoteReqCosts[msgId] var reqCost = costs.baseCost + quantity * costs.reqCost peer.remoteFlowState.update t peer.remoteFlowState.bufValue -= reqCost network.endPendingRequest peer, t let curReqCost = peer.reqCostVal if curReqCost < peer.remoteFlowState.bufLimit: let bv = peer.remoteFlowState.bufLimit - curReqCost if bv > peer.remoteFlowState.bufValue: peer.remoteFlowState.bufValue = bv network.messageStats[msgId].addSample(float64(quantity), float64(curReqCost - peer.startReqCostVal)) return peer.remoteFlowState.bufValue when defined(testing): import unittest2, random, ../../rlpx proc isMax(s: FlowControlState): bool = s.bufValue == s.bufLimit p2pProtocol dummyLes(version = 1, rlpxName = "abc"): proc a(p: Peer) proc b(p: Peer) proc c(p: Peer) proc d(p: Peer) proc e(p: Peer) template fequals(lhs, rhs: float64, epsilon = 0.0001): bool = abs(lhs-rhs) < epsilon proc tests* = randomize(3913631) suite "les flow control": suite "running averages": test "consistent costs": var s: StatsRunningAverage for i in 0..100: s.addSample(5.0, 100.0) let (cost, base) = s.calc check: fequals(cost, 100.0) fequals(base, 0.0) test "randomized averages": proc performTest(qBase, qRandom: int, cBase, cRandom: float64) = var s: StatsRunningAverage expectedFinalCost = cBase + cRandom / 2 error = expectedFinalCost for samples in [100, 1000, 10000]: for i in 0..samples: let q = float64(qBase + rand(10)) s.addSample(q, q * (cBase + rand(cRandom))) let (newCost, newBase) = s.calc # With more samples, our error should decrease, getting # closer and closer to the average (unless we are already close enough) let newError = abs(newCost - expectedFinalCost) # This check fails with Nim-1.6: # check newError < error error = newError # After enough samples we should be very close the the final result check error < (expectedFinalCost * 0.02) performTest(1, 10, 5.0, 100.0) performTest(1, 4, 200.0, 1000.0) suite "buffer value calculations": type TestReq = object peer: LesPeer msgId, quantity: int accepted: bool setup: var lesNetwork = new LesNetwork lesNetwork.peers = initHashSet[LesPeer]() lesNetwork.initFlowControl(dummyLes.protocolInfo, reqCostTarget = 300, maxReqCount = 5, maxReqCostSum = 1000) for i in 0 ..< lesNetwork.messageStats.len: lesNetwork.messageStats[i].addSample(1.0, float(i) * 100.0) var client = new LesPeer client.isClient = true var server = new LesPeer server.isServer = true var clientServer = new LesPeer clientServer.isClient = true clientServer.isServer = true var client2 = new LesPeer client2.isClient = true var client3 = new LesPeer client3.isClient = true var bv: BufValueInt template enlist(peer: LesPeer) {.dirty.} = let reqCosts = currentRequestsCosts(lesNetwork, dummyLes.protocolInfo) peer.remoteReqCosts = reqCosts peer.localReqCosts = reqCosts lesNetwork.peers.incl peer lesNetwork.enlistInFlowControl peer template startReq(p: LesPeer, msg, q: int): TestReq = var req: TestReq req.peer = p req.msgId = msg req.quantity = q req.accepted = waitFor lesNetwork.acceptRequest(p, msg, q) req template endReq(req: TestReq): BufValueInt = bufValueAfterRequest(lesNetwork, req.peer, req.msgId, req.quantity) test "single peer recharging": lesNetwork.bufferLimit = 1000 lesNetwork.minRechargingRate = 100 enlist client check: client.remoteFlowState.isMax client.rechargingPower > 0 advanceTime 100 let r1 = client.startReq(0, 100) check r1.accepted check client.isRecharging == false advanceTime 50 let r2 = client.startReq(1, 1) check r2.accepted check client.isRecharging == false advanceTime 25 bv = endReq r2 check client.isRecharging == false advanceTime 130 bv = endReq r1 check client.isRecharging == true advanceTime 300 lesNetwork.updateFlowControl now() check: client.isRecharging == false client.remoteFlowState.isMax