From 91b69d08dab16d10b337d53f6cb8112b187114ab Mon Sep 17 00:00:00 2001 From: Zsolt Felfoldi Date: Sun, 8 Dec 2019 20:06:21 +0100 Subject: [PATCH] les: lespay rate limiting --- les/flowcontrol/control.go | 7 +++ les/server_handler.go | 51 +++++++++++---- les/servingqueue.go | 4 +- les/tokensale.go | 125 ++++++++++++++++++++++++++++++++++--- 4 files changed, 164 insertions(+), 23 deletions(-) diff --git a/les/flowcontrol/control.go b/les/flowcontrol/control.go index 490013677c..de636231ad 100644 --- a/les/flowcontrol/control.go +++ b/les/flowcontrol/control.go @@ -185,6 +185,13 @@ func (node *ClientNode) UpdateParams(params ServerParams) { } } +func (node *ClientNode) Params() ServerParams { + node.lock.Lock() + defer node.lock.Unlock() + + return node.params +} + // updateParams updates the flow control parameters of the node func (node *ClientNode) updateParams(params ServerParams, now mclock.AbsTime) { diff := int64(params.BufLimit - node.params.BufLimit) diff --git a/les/server_handler.go b/les/server_handler.go index a43bf2aef3..f5d8b41fd1 100644 --- a/les/server_handler.go +++ b/les/server_handler.go @@ -377,7 +377,7 @@ func (h *serverHandler) handleMsg(p *peer, wg *sync.WaitGroup) error { first = false } reply := p.ReplyBlockHeaders(req.ReqID, headers) - sendResponse(req.ReqID, query.Amount, p.ReplyBlockHeaders(req.ReqID, headers), task.done()) + sendResponse(req.ReqID, query.Amount, reply, task.done()) if metrics.EnabledExpensive { miscOutHeaderPacketsMeter.Mark(1) miscOutHeaderTrafficMeter.Mark(int64(reply.size())) @@ -841,11 +841,22 @@ func (h *serverHandler) handleMsg(p *peer, wg *sync.WaitGroup) error { clientErrorMeter.Mark(1) return errResp(ErrDecode, "msg %v: %v", msg, err) } - reply := h.server.tokenSale.runCommand(req.Cmd, p.ID(), p.freeClientId()) - p.ReplyLespay(req.ReqID, reply) - if metrics.EnabledExpensive { - miscOutLespayPacketsMeter.Mark(1) - miscOutLespayTrafficMeter.Mark(int64(len(reply))) + if !h.server.tokenSale.queueCommand(p.id, tokenCmd{ + cmd: req.Cmd, + id: p.ID(), + freeID: p.freeClientId(), + send: func(reply []byte) { + if metrics.EnabledExpensive { + miscOutLespayPacketsMeter.Mark(1) + miscOutLespayTrafficMeter.Mark(int64(len(reply))) + } + p.queueSend(func() { + p.ReplyLespay(req.ReqID, reply) + }) + }, + }, p.fcClient.Params().MinRecharge) { + clientErrorMeter.Mark(1) + return errResp(ErrRequestRejected, "") } default: @@ -985,16 +996,34 @@ func (h *serverHandler) talkRequestHandler(id enode.ID, addr *net.UDPAddr, paylo if !ok { return nil, false } - cmds := make([][]byte, len(c)) - for i, c := range c { - cmds[i], ok = c.([]byte) + resultCh := make(chan []byte, len(c)) + results := make([][]byte, len(c)) + for _, c := range c { + cmd, ok := c.([]byte) if !ok { fmt.Println("type err", reflect.TypeOf(c)) return nil, false } + if !h.server.tokenSale.queueCommand(id.String(), tokenCmd{ + cmd: cmd, + id: id, + freeID: addr.IP.String(), + send: func(reply []byte) { + resultCh <- reply + }, + }, h.server.freeCapacity) { + fmt.Println("failed to queue") + return nil, false + } + } + + for i, _ := range results { + select { + case results[i] = <-resultCh: + case <-h.closeCh: + return nil, false + } } - fmt.Println("ok", ok) - results := h.server.tokenSale.runCommands(cmds, id, addr.IP.String()) fmt.Println("results", results) return results, true } diff --git a/les/servingqueue.go b/les/servingqueue.go index 8842cf9e9d..487e6046a5 100644 --- a/les/servingqueue.go +++ b/les/servingqueue.go @@ -70,7 +70,7 @@ 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() { + if t.peer != nil && t.peer.isFrozen() { return false } t.tokenCh = make(chan runToken, 1) @@ -289,7 +289,7 @@ func (sq *servingQueue) addTask(task *servingTask) { sq.queuedTime += task.expTime sqServedGauge.Update(int64(sq.recentTime)) sqQueuedGauge.Update(int64(sq.queuedTime)) - if sq.recentTime+sq.queuedTime > sq.burstLimit { + if sq.burstLimit != 0 && sq.recentTime+sq.queuedTime > sq.burstLimit { sq.freezePeers() } } diff --git a/les/tokensale.go b/les/tokensale.go index df57e188bf..9179b948f4 100644 --- a/les/tokensale.go +++ b/les/tokensale.go @@ -23,11 +23,15 @@ import ( "sync" "time" + "github.com/ethereum/go-ethereum/common/mclock" "github.com/ethereum/go-ethereum/p2p/enode" "github.com/ethereum/go-ethereum/rlp" ) -const basePriceTC = time.Hour * 10 +const ( + basePriceTC = time.Hour * 10 + tokenQueueTC = time.Hour +) type paymentReceiver interface { info() keyValueList @@ -36,12 +40,15 @@ type paymentReceiver interface { } type tokenSale struct { - lock sync.Mutex + lock, qlock sync.Mutex clientPool *clientPool stopCh chan struct{} receivers map[string]paymentReceiver receiverNames []string basePrice, minBasePrice float64 + + sq *servingQueue + sources map[string]*cmdSource } func newTokenSale(clientPool *clientPool, minBasePrice float64) *tokenSale { @@ -51,8 +58,12 @@ func newTokenSale(clientPool *clientPool, minBasePrice float64) *tokenSale { basePrice: minBasePrice, minBasePrice: minBasePrice, stopCh: make(chan struct{}), + sq: newServingQueue(0, 0), + sources: make(map[string]*cmdSource), } + t.sq.setThreads(1) go func() { + cleanupCounter := 0 for { select { case <-time.After(time.Second * 10): @@ -66,6 +77,12 @@ func newTokenSale(clientPool *clientPool, minBasePrice float64) *tokenSale { t.basePrice = minBasePrice } t.lock.Unlock() + + cleanupCounter++ + if cleanupCounter == 100 { + t.sourceMapCleanup() + cleanupCounter = 0 + } case <-t.stopCh: return } @@ -74,8 +91,104 @@ func newTokenSale(clientPool *clientPool, minBasePrice float64) *tokenSale { return t } +type ( + cmdSource struct { + ch chan tokenCmd + recentTime float64 + lastUpdate mclock.AbsTime + } + + tokenCmd struct { + cmd []byte + id enode.ID + freeID string + send func([]byte) + } +) + +func (c *cmdSource) priority(capacity uint64) int64 { + dt := mclock.Now() - c.lastUpdate + rt := c.recentTime + if dt > 0 { + rt *= math.Exp(-float64(dt) / float64(tokenQueueTC)) + } + return -int64(rt / float64(capacity)) +} + +func (c *cmdSource) addTime(time uint64) { + now := mclock.Now() + dt := now - c.lastUpdate + if dt > 0 { + c.recentTime *= math.Exp(-float64(dt) / float64(tokenQueueTC)) + c.lastUpdate = now + } + c.recentTime += float64(time) +} + +func (t *tokenSale) sourceMapCleanup() { + t.qlock.Lock() + defer t.qlock.Unlock() + + for src, s := range t.sources { + s.addTime(0) + if s.recentTime < float64(time.Millisecond*100) { + delete(t.sources, src) + } + } +} + +func (t *tokenSale) queueCommand(src string, cmd tokenCmd, capacity uint64) bool { + t.qlock.Lock() + defer t.qlock.Unlock() + + s := t.sources[src] + if s == nil { + s = &cmdSource{lastUpdate: mclock.Now()} + t.sources[src] = s + } + if s.ch != nil { + select { + case s.ch <- cmd: + return true + default: + return false + } + } + s.ch = make(chan tokenCmd, 16) + s.ch <- cmd + + go func() { + loop: + for { + select { + case cmd := <-s.ch: + task := t.sq.newTask(nil, 0, s.priority(capacity)) + if !task.start() { + break loop + } + start := mclock.Now() + reply := t.runCommand(cmd.cmd, cmd.id, cmd.freeID) + runTime := mclock.Now() - start + cmd.send(reply) + time.Sleep(time.Duration(runTime) * 9) + task.done() + t.qlock.Lock() + s.addTime(uint64(runTime)) + t.qlock.Unlock() + default: + break loop + } + t.qlock.Lock() + s.ch = nil // TODO map cleanup + t.qlock.Unlock() + } + }() + return true +} + func (t *tokenSale) stop() { close(t.stopCh) + t.sq.stop() } func (t *tokenSale) tokenCost(buyAmount uint64) (float64, bool) { @@ -420,11 +533,3 @@ func (t *tokenSale) runCommand(cmd []byte, id enode.ID, freeID string) []byte { } return res } - -func (t *tokenSale) runCommands(cmds [][]byte, id enode.ID, freeID string) [][]byte { - res := make([][]byte, len(cmds)) - for i, cmd := range cmds { - res[i] = t.runCommand(cmd, id, freeID) - } - return res -}