123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517 |
- // Copyright 2019 The go-ethereum Authors
- // This file is part of the go-ethereum library.
- //
- // The go-ethereum library is free software: you can redistribute it and/or modify
- // it under the terms of the GNU Lesser General Public License as published by
- // the Free Software Foundation, either version 3 of the License, or
- // (at your option) any later version.
- //
- // The go-ethereum library is distributed in the hope that it will be useful,
- // but WITHOUT ANY WARRANTY; without even the implied warranty of
- // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
- // GNU Lesser General Public License for more details.
- //
- // You should have received a copy of the GNU Lesser General Public License
- // along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
- package les
- import (
- "encoding/binary"
- "math"
- "sync"
- "sync/atomic"
- "time"
- "github.com/ethereum/go-ethereum/common/mclock"
- "github.com/ethereum/go-ethereum/eth/ethconfig"
- "github.com/ethereum/go-ethereum/ethdb"
- "github.com/ethereum/go-ethereum/les/flowcontrol"
- "github.com/ethereum/go-ethereum/log"
- "github.com/ethereum/go-ethereum/metrics"
- )
- const makeCostStats = false // make request cost statistics during operation
- var (
- // average request cost estimates based on serving time
- reqAvgTimeCost = requestCostTable{
- GetBlockHeadersMsg: {150000, 30000},
- GetBlockBodiesMsg: {0, 700000},
- GetReceiptsMsg: {0, 1000000},
- GetCodeMsg: {0, 450000},
- GetProofsV2Msg: {0, 600000},
- GetHelperTrieProofsMsg: {0, 1000000},
- SendTxV2Msg: {0, 450000},
- GetTxStatusMsg: {0, 250000},
- }
- // maximum incoming message size estimates
- reqMaxInSize = requestCostTable{
- GetBlockHeadersMsg: {40, 0},
- GetBlockBodiesMsg: {0, 40},
- GetReceiptsMsg: {0, 40},
- GetCodeMsg: {0, 80},
- GetProofsV2Msg: {0, 80},
- GetHelperTrieProofsMsg: {0, 20},
- SendTxV2Msg: {0, 16500},
- GetTxStatusMsg: {0, 50},
- }
- // maximum outgoing message size estimates
- reqMaxOutSize = requestCostTable{
- GetBlockHeadersMsg: {0, 556},
- GetBlockBodiesMsg: {0, 100000},
- GetReceiptsMsg: {0, 200000},
- GetCodeMsg: {0, 50000},
- GetProofsV2Msg: {0, 4000},
- GetHelperTrieProofsMsg: {0, 4000},
- SendTxV2Msg: {0, 100},
- GetTxStatusMsg: {0, 100},
- }
- // request amounts that have to fit into the minimum buffer size minBufferMultiplier times
- minBufferReqAmount = map[uint64]uint64{
- GetBlockHeadersMsg: 192,
- GetBlockBodiesMsg: 1,
- GetReceiptsMsg: 1,
- GetCodeMsg: 1,
- GetProofsV2Msg: 1,
- GetHelperTrieProofsMsg: 16,
- SendTxV2Msg: 8,
- GetTxStatusMsg: 64,
- }
- minBufferMultiplier = 3
- )
- const (
- maxCostFactor = 2 // ratio of maximum and average cost estimates
- bufLimitRatio = 6000 // fixed bufLimit/MRR ratio
- gfUsageThreshold = 0.5
- gfUsageTC = time.Second
- gfRaiseTC = time.Second * 200
- gfDropTC = time.Second * 50
- gfDbKey = "_globalCostFactorV6"
- )
- // costTracker is responsible for calculating costs and cost estimates on the
- // server side. It continuously updates the global cost factor which is defined
- // as the number of cost units per nanosecond of serving time in a single thread.
- // It is based on statistics collected during serving requests in high-load periods
- // and practically acts as a one-dimension request price scaling factor over the
- // pre-defined cost estimate table.
- //
- // The reason for dynamically maintaining the global factor on the server side is:
- // the estimated time cost of the request is fixed(hardcoded) but the configuration
- // of the machine running the server is really different. Therefore, the request serving
- // time in different machine will vary greatly. And also, the request serving time
- // in same machine may vary greatly with different request pressure.
- //
- // In order to more effectively limit resources, we apply the global factor to serving
- // time to make the result as close as possible to the estimated time cost no matter
- // the server is slow or fast. And also we scale the totalRecharge with global factor
- // so that fast server can serve more requests than estimation and slow server can
- // reduce request pressure.
- //
- // Instead of scaling the cost values, the real value of cost units is changed by
- // applying the factor to the serving times. This is more convenient because the
- // changes in the cost factor can be applied immediately without always notifying
- // the clients about the changed cost tables.
- type costTracker struct {
- db ethdb.Database
- stopCh chan chan struct{}
- inSizeFactor float64
- outSizeFactor float64
- factor float64
- utilTarget float64
- minBufLimit uint64
- gfLock sync.RWMutex
- reqInfoCh chan reqInfo
- totalRechargeCh chan uint64
- stats map[uint64][]uint64 // Used for testing purpose.
- // TestHooks
- testing bool // Disable real cost evaluation for testing purpose.
- testCostList RequestCostList // Customized cost table for testing purpose.
- }
- // newCostTracker creates a cost tracker and loads the cost factor statistics from the database.
- // It also returns the minimum capacity that can be assigned to any peer.
- func newCostTracker(db ethdb.Database, config *ethconfig.Config) (*costTracker, uint64) {
- utilTarget := float64(config.LightServ) * flowcontrol.FixedPointMultiplier / 100
- ct := &costTracker{
- db: db,
- stopCh: make(chan chan struct{}),
- reqInfoCh: make(chan reqInfo, 100),
- utilTarget: utilTarget,
- }
- if config.LightIngress > 0 {
- ct.inSizeFactor = utilTarget / float64(config.LightIngress)
- }
- if config.LightEgress > 0 {
- ct.outSizeFactor = utilTarget / float64(config.LightEgress)
- }
- if makeCostStats {
- ct.stats = make(map[uint64][]uint64)
- for code := range reqAvgTimeCost {
- ct.stats[code] = make([]uint64, 10)
- }
- }
- ct.gfLoop()
- costList := ct.makeCostList(ct.globalFactor() * 1.25)
- for _, c := range costList {
- amount := minBufferReqAmount[c.MsgCode]
- cost := c.BaseCost + amount*c.ReqCost
- if cost > ct.minBufLimit {
- ct.minBufLimit = cost
- }
- }
- ct.minBufLimit *= uint64(minBufferMultiplier)
- return ct, (ct.minBufLimit-1)/bufLimitRatio + 1
- }
- // stop stops the cost tracker and saves the cost factor statistics to the database
- func (ct *costTracker) stop() {
- stopCh := make(chan struct{})
- ct.stopCh <- stopCh
- <-stopCh
- if makeCostStats {
- ct.printStats()
- }
- }
- // makeCostList returns upper cost estimates based on the hardcoded cost estimate
- // tables and the optionally specified incoming/outgoing bandwidth limits
- func (ct *costTracker) makeCostList(globalFactor float64) RequestCostList {
- maxCost := func(avgTimeCost, inSize, outSize uint64) uint64 {
- cost := avgTimeCost * maxCostFactor
- inSizeCost := uint64(float64(inSize) * ct.inSizeFactor * globalFactor)
- if inSizeCost > cost {
- cost = inSizeCost
- }
- outSizeCost := uint64(float64(outSize) * ct.outSizeFactor * globalFactor)
- if outSizeCost > cost {
- cost = outSizeCost
- }
- return cost
- }
- var list RequestCostList
- for code, data := range reqAvgTimeCost {
- baseCost := maxCost(data.baseCost, reqMaxInSize[code].baseCost, reqMaxOutSize[code].baseCost)
- reqCost := maxCost(data.reqCost, reqMaxInSize[code].reqCost, reqMaxOutSize[code].reqCost)
- if ct.minBufLimit != 0 {
- // if minBufLimit is set then always enforce maximum request cost <= minBufLimit
- maxCost := baseCost + reqCost*minBufferReqAmount[code]
- if maxCost > ct.minBufLimit {
- mul := 0.999 * float64(ct.minBufLimit) / float64(maxCost)
- baseCost = uint64(float64(baseCost) * mul)
- reqCost = uint64(float64(reqCost) * mul)
- }
- }
- list = append(list, requestCostListItem{
- MsgCode: code,
- BaseCost: baseCost,
- ReqCost: reqCost,
- })
- }
- return list
- }
- // reqInfo contains the estimated time cost and the actual request serving time
- // which acts as a feed source to update factor maintained by costTracker.
- type reqInfo struct {
- // avgTimeCost is the estimated time cost corresponding to maxCostTable.
- avgTimeCost float64
- // servingTime is the CPU time corresponding to the actual processing of
- // the request.
- servingTime float64
- // msgCode indicates the type of request.
- msgCode uint64
- }
- // gfLoop starts an event loop which updates the global cost factor which is
- // calculated as a weighted average of the average estimate / serving time ratio.
- // The applied weight equals the serving time if gfUsage is over a threshold,
- // zero otherwise. gfUsage is the recent average serving time per time unit in
- // an exponential moving window. This ensures that statistics are collected only
- // under high-load circumstances where the measured serving times are relevant.
- // The total recharge parameter of the flow control system which controls the
- // total allowed serving time per second but nominated in cost units, should
- // also be scaled with the cost factor and is also updated by this loop.
- func (ct *costTracker) gfLoop() {
- var (
- factor, totalRecharge float64
- gfLog, recentTime, recentAvg float64
- lastUpdate, expUpdate = mclock.Now(), mclock.Now()
- )
- // Load historical cost factor statistics from the database.
- data, _ := ct.db.Get([]byte(gfDbKey))
- if len(data) == 8 {
- gfLog = math.Float64frombits(binary.BigEndian.Uint64(data[:]))
- }
- ct.factor = math.Exp(gfLog)
- factor, totalRecharge = ct.factor, ct.utilTarget*ct.factor
- // In order to perform factor data statistics under the high request pressure,
- // we only adjust factor when recent factor usage beyond the threshold.
- threshold := gfUsageThreshold * float64(gfUsageTC) * ct.utilTarget / flowcontrol.FixedPointMultiplier
- go func() {
- saveCostFactor := func() {
- var data [8]byte
- binary.BigEndian.PutUint64(data[:], math.Float64bits(gfLog))
- ct.db.Put([]byte(gfDbKey), data[:])
- log.Debug("global cost factor saved", "value", factor)
- }
- saveTicker := time.NewTicker(time.Minute * 10)
- defer saveTicker.Stop()
- for {
- select {
- case r := <-ct.reqInfoCh:
- relCost := int64(factor * r.servingTime * 100 / r.avgTimeCost) // Convert the value to a percentage form
- // Record more metrics if we are debugging
- if metrics.EnabledExpensive {
- switch r.msgCode {
- case GetBlockHeadersMsg:
- relativeCostHeaderHistogram.Update(relCost)
- case GetBlockBodiesMsg:
- relativeCostBodyHistogram.Update(relCost)
- case GetReceiptsMsg:
- relativeCostReceiptHistogram.Update(relCost)
- case GetCodeMsg:
- relativeCostCodeHistogram.Update(relCost)
- case GetProofsV2Msg:
- relativeCostProofHistogram.Update(relCost)
- case GetHelperTrieProofsMsg:
- relativeCostHelperProofHistogram.Update(relCost)
- case SendTxV2Msg:
- relativeCostSendTxHistogram.Update(relCost)
- case GetTxStatusMsg:
- relativeCostTxStatusHistogram.Update(relCost)
- }
- }
- // SendTxV2 and GetTxStatus requests are two special cases.
- // All other requests will only put pressure on the database, and
- // the corresponding delay is relatively stable. While these two
- // requests involve txpool query, which is usually unstable.
- //
- // TODO(rjl493456442) fixes this.
- if r.msgCode == SendTxV2Msg || r.msgCode == GetTxStatusMsg {
- continue
- }
- requestServedMeter.Mark(int64(r.servingTime))
- requestServedTimer.Update(time.Duration(r.servingTime))
- requestEstimatedMeter.Mark(int64(r.avgTimeCost / factor))
- requestEstimatedTimer.Update(time.Duration(r.avgTimeCost / factor))
- relativeCostHistogram.Update(relCost)
- now := mclock.Now()
- dt := float64(now - expUpdate)
- expUpdate = now
- exp := math.Exp(-dt / float64(gfUsageTC))
- // calculate factor correction until now, based on previous values
- var gfCorr float64
- max := recentTime
- if recentAvg > max {
- max = recentAvg
- }
- // we apply continuous correction when MAX(recentTime, recentAvg) > threshold
- if max > threshold {
- // calculate correction time between last expUpdate and now
- if max*exp >= threshold {
- gfCorr = dt
- } else {
- gfCorr = math.Log(max/threshold) * float64(gfUsageTC)
- }
- // calculate log(factor) correction with the right direction and time constant
- if recentTime > recentAvg {
- // drop factor if actual serving times are larger than average estimates
- gfCorr /= -float64(gfDropTC)
- } else {
- // raise factor if actual serving times are smaller than average estimates
- gfCorr /= float64(gfRaiseTC)
- }
- }
- // update recent cost values with current request
- recentTime = recentTime*exp + r.servingTime
- recentAvg = recentAvg*exp + r.avgTimeCost/factor
- if gfCorr != 0 {
- // Apply the correction to factor
- gfLog += gfCorr
- factor = math.Exp(gfLog)
- // Notify outside modules the new factor and totalRecharge.
- if time.Duration(now-lastUpdate) > time.Second {
- totalRecharge, lastUpdate = ct.utilTarget*factor, now
- ct.gfLock.Lock()
- ct.factor = factor
- ch := ct.totalRechargeCh
- ct.gfLock.Unlock()
- if ch != nil {
- select {
- case ct.totalRechargeCh <- uint64(totalRecharge):
- default:
- }
- }
- globalFactorGauge.Update(int64(1000 * factor))
- log.Debug("global cost factor updated", "factor", factor)
- }
- }
- recentServedGauge.Update(int64(recentTime))
- recentEstimatedGauge.Update(int64(recentAvg))
- case <-saveTicker.C:
- saveCostFactor()
- case stopCh := <-ct.stopCh:
- saveCostFactor()
- close(stopCh)
- return
- }
- }
- }()
- }
- // globalFactor returns the current value of the global cost factor
- func (ct *costTracker) globalFactor() float64 {
- ct.gfLock.RLock()
- defer ct.gfLock.RUnlock()
- return ct.factor
- }
- // totalRecharge returns the current total recharge parameter which is used by
- // flowcontrol.ClientManager and is scaled by the global cost factor
- func (ct *costTracker) totalRecharge() uint64 {
- ct.gfLock.RLock()
- defer ct.gfLock.RUnlock()
- return uint64(ct.factor * ct.utilTarget)
- }
- // subscribeTotalRecharge returns all future updates to the total recharge value
- // through a channel and also returns the current value
- func (ct *costTracker) subscribeTotalRecharge(ch chan uint64) uint64 {
- ct.gfLock.Lock()
- defer ct.gfLock.Unlock()
- ct.totalRechargeCh = ch
- return uint64(ct.factor * ct.utilTarget)
- }
- // updateStats updates the global cost factor and (if enabled) the real cost vs.
- // average estimate statistics
- func (ct *costTracker) updateStats(code, amount, servingTime, realCost uint64) {
- avg := reqAvgTimeCost[code]
- avgTimeCost := avg.baseCost + amount*avg.reqCost
- select {
- case ct.reqInfoCh <- reqInfo{float64(avgTimeCost), float64(servingTime), code}:
- default:
- }
- if makeCostStats {
- realCost <<= 4
- l := 0
- for l < 9 && realCost > avgTimeCost {
- l++
- realCost >>= 1
- }
- atomic.AddUint64(&ct.stats[code][l], 1)
- }
- }
- // realCost calculates the final cost of a request based on actual serving time,
- // incoming and outgoing message size
- //
- // Note: message size is only taken into account if bandwidth limitation is applied
- // and the cost based on either message size is greater than the cost based on
- // serving time. A maximum of the three costs is applied instead of their sum
- // because the three limited resources (serving thread time and i/o bandwidth) can
- // also be maxed out simultaneously.
- func (ct *costTracker) realCost(servingTime uint64, inSize, outSize uint32) uint64 {
- cost := float64(servingTime)
- inSizeCost := float64(inSize) * ct.inSizeFactor
- if inSizeCost > cost {
- cost = inSizeCost
- }
- outSizeCost := float64(outSize) * ct.outSizeFactor
- if outSizeCost > cost {
- cost = outSizeCost
- }
- return uint64(cost * ct.globalFactor())
- }
- // printStats prints the distribution of real request cost relative to the average estimates
- func (ct *costTracker) printStats() {
- if ct.stats == nil {
- return
- }
- for code, arr := range ct.stats {
- log.Info("Request cost statistics", "code", code, "1/16", arr[0], "1/8", arr[1], "1/4", arr[2], "1/2", arr[3], "1", arr[4], "2", arr[5], "4", arr[6], "8", arr[7], "16", arr[8], ">16", arr[9])
- }
- }
- type (
- // requestCostTable assigns a cost estimate function to each request type
- // which is a linear function of the requested amount
- // (cost = baseCost + reqCost * amount)
- requestCostTable map[uint64]*requestCosts
- requestCosts struct {
- baseCost, reqCost uint64
- }
- // RequestCostList is a list representation of request costs which is used for
- // database storage and communication through the network
- RequestCostList []requestCostListItem
- requestCostListItem struct {
- MsgCode, BaseCost, ReqCost 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
- }
- // decode converts a cost list to a cost table
- func (list RequestCostList) decode(protocolLength uint64) requestCostTable {
- table := make(requestCostTable)
- for _, e := range list {
- if e.MsgCode < protocolLength {
- table[e.MsgCode] = &requestCosts{
- baseCost: e.BaseCost,
- reqCost: e.ReqCost,
- }
- }
- }
- return table
- }
- // testCostList returns a dummy request cost list used by tests
- func testCostList(testCost uint64) RequestCostList {
- cl := make(RequestCostList, len(reqAvgTimeCost))
- var max uint64
- for code := range reqAvgTimeCost {
- if code > max {
- max = code
- }
- }
- i := 0
- for code := uint64(0); code <= max; code++ {
- if _, ok := reqAvgTimeCost[code]; ok {
- cl[i].MsgCode = code
- cl[i].BaseCost = testCost
- cl[i].ReqCost = 0
- i++
- }
- }
- return cl
- }
|