From 77ae572cfe6186ad13bff2ea4cdba4339e9e1826 Mon Sep 17 00:00:00 2001 From: rjl493456442 Date: Tue, 20 Aug 2019 09:52:17 +0800 Subject: [PATCH] les: address comments --- les/client_handler.go | 5 +++-- les/metrics.go | 2 +- les/odr.go | 5 +++-- les/server.go | 6 +++--- les/server_handler.go | 22 ++++++++++++---------- 5 files changed, 22 insertions(+), 18 deletions(-) diff --git a/les/client_handler.go b/les/client_handler.go index 3d5d90cbc5..aff05ddbc9 100644 --- a/les/client_handler.go +++ b/les/client_handler.go @@ -22,6 +22,7 @@ import ( "time" "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/common/mclock" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/eth/downloader" "github.com/ethereum/go-ethereum/light" @@ -115,10 +116,10 @@ func (h *clientHandler) handle(p *peer) error { } serverConnectionGauge.Update(int64(h.backend.peers.Len())) - connectedAt := time.Now() + connectedAt := mclock.Now() defer func() { h.backend.peers.Unregister(p.id) - connectionTimer.UpdateSince(connectedAt) + connectionTimer.Update(time.Duration(mclock.Now() - connectedAt)) serverConnectionGauge.Update(int64(h.backend.peers.Len())) }() diff --git a/les/metrics.go b/les/metrics.go index 96bf4eb16f..b286e2ac6e 100644 --- a/les/metrics.go +++ b/les/metrics.go @@ -72,7 +72,7 @@ var ( requestServedMeter = metrics.NewRegisteredMeter("les/server/req/avgServedTime", nil) requestServedTimer = metrics.NewRegisteredTimer("les/server/req/servedTime", nil) - requestEstimatedMeter = metrics.NewRegisteredMeter("les/server/req/argEstimatedTime", nil) + requestEstimatedMeter = metrics.NewRegisteredMeter("les/server/req/avgEstimatedTime", nil) requestEstimatedTimer = metrics.NewRegisteredTimer("les/server/req/estimatedTime", nil) relativeCostHistogram = metrics.NewRegisteredHistogram("les/server/req/relative", nil, metrics.NewExpDecaySample(1028, 0.015)) diff --git a/les/odr.go b/les/odr.go index 54d07ca937..136ecf4df4 100644 --- a/les/odr.go +++ b/les/odr.go @@ -20,6 +20,7 @@ import ( "context" "time" + "github.com/ethereum/go-ethereum/common/mclock" "github.com/ethereum/go-ethereum/core" "github.com/ethereum/go-ethereum/ethdb" "github.com/ethereum/go-ethereum/light" @@ -121,11 +122,11 @@ func (odr *LesOdr) Retrieve(ctx context.Context, req light.OdrRequest) (err erro return func() { lreq.Request(reqID, p) } }, } - sent := time.Now() + sent := mclock.Now() if err = odr.retriever.retrieve(ctx, reqID, rq, func(p distPeer, msg *Msg) error { return lreq.Validate(odr.db, msg) }, odr.stop); err == nil { // retrieved from network, store in db req.StoreResult(odr.db) - requestRTT.UpdateSince(sent) + requestRTT.Update(time.Duration(mclock.Now() - sent)) } else { log.Debug("Failed to retrieve data from network", "err", err) } diff --git a/les/server.go b/les/server.go index 28db632322..416cabd136 100644 --- a/les/server.go +++ b/les/server.go @@ -221,7 +221,7 @@ func (s *LesServer) capacityManagement() { var ( busy bool freePeers uint64 - blockProcess time.Time + blockProcess mclock.AbsTime ) updateRecharge := func() { if busy { @@ -238,9 +238,9 @@ func (s *LesServer) capacityManagement() { select { case busy = <-processCh: if busy { - blockProcess = time.Now() + blockProcess = mclock.Now() } else { - blockProcessingTimer.UpdateSince(blockProcess) + blockProcessingTimer.Update(time.Duration(mclock.Now() - blockProcess)) } updateRecharge() case totalRecharge = <-totalRechargeCh: diff --git a/les/server_handler.go b/les/server_handler.go index 4ef4d822d4..af9c077bc1 100644 --- a/les/server_handler.go +++ b/les/server_handler.go @@ -137,12 +137,12 @@ func (h *serverHandler) handle(p *peer) error { p.balanceTracker.init(&mclock.System{}, 1) } - connectedAt := time.Now() + connectedAt := mclock.Now() defer func() { p.balanceTracker = nil h.server.peers.Unregister(p.id) clientConnectionGauge.Update(int64(h.server.peers.Len())) - connectionTimer.UpdateSince(connectedAt) + connectionTimer.Update(time.Duration(mclock.Now() - connectedAt)) }() // Spawn a main loop to handle all incoming messages. @@ -180,8 +180,9 @@ func (h *serverHandler) handleMsg(p *peer) error { var ( maxCost uint64 task *servingTask - respId = p.responseID() ) + p.responseCount++ + responseCount := p.responseCount // accept returns an indicator whether the request can be served. // If so, deduct the max cost from the flow control buffer. accept := func(reqID, reqCnt, maxCnt uint64) bool { @@ -193,7 +194,7 @@ func (h *serverHandler) handleMsg(p *peer) error { } // Prepaid max cost units before request been serving. maxCost = p.fcCosts.getMaxCost(msg.Code, reqCnt) - accepted, bufShort, priority := p.fcClient.AcceptRequest(reqID, respId, maxCost) + accepted, bufShort, priority := p.fcClient.AcceptRequest(reqID, responseCount, maxCost) if !accepted { p.freezeClient() p.Log().Error("Request came too early", "remaining", common.PrettyDuration(time.Duration(bufShort*1000000/p.fcParams.MinRecharge))) @@ -212,7 +213,7 @@ func (h *serverHandler) handleMsg(p *peer) error { if task.start() { return true } - p.fcClient.RequestProcessed(reqID, respId, maxCost, inSizeCost) + p.fcClient.RequestProcessed(reqID, responseCount, maxCost, inSizeCost) return false } // sendResponse sends back the response and updates the flow control statistic. @@ -223,7 +224,7 @@ func (h *serverHandler) handleMsg(p *peer) error { // Short circuit if the client is already frozen. if p.isFrozen() { realCost := h.server.costTracker.realCost(servingTime, msg.Size, 0) - p.fcClient.RequestProcessed(reqID, respId, maxCost, realCost) + p.fcClient.RequestProcessed(reqID, responseCount, maxCost, realCost) return } // Positive correction buffer value with real cost. @@ -231,12 +232,13 @@ func (h *serverHandler) handleMsg(p *peer) error { if reply != nil { replySize = reply.size() } - realCost := h.server.costTracker.realCost(servingTime, msg.Size, replySize) - // Assign a fake cost for testing purpose. + var realCost uint64 if h.server.costTracker.testing { - realCost = maxCost + realCost = maxCost // Assign a fake cost for testing purpose + } else { + realCost = h.server.costTracker.realCost(servingTime, msg.Size, replySize) } - bv := p.fcClient.RequestProcessed(reqID, respId, maxCost, realCost) + bv := p.fcClient.RequestProcessed(reqID, responseCount, maxCost, realCost) if amount != 0 { // Feed cost tracker request serving statistic. h.server.costTracker.updateStats(msg.Code, amount, servingTime, realCost)