From 45bc1f2a35c491e5187772ceba0836236f2902fa Mon Sep 17 00:00:00 2001 From: rjl493456442 Date: Mon, 24 Feb 2020 09:47:28 +0800 Subject: [PATCH] les: stash --- les/api.go | 2 +- les/clientpool.go | 62 ++++++++++++++++++++++++++---------------- les/clientpool_test.go | 17 ++++++------ 3 files changed, 47 insertions(+), 34 deletions(-) diff --git a/les/api.go b/les/api.go index 47009624df..ff2487a42a 100644 --- a/les/api.go +++ b/les/api.go @@ -154,7 +154,7 @@ func (api *PrivateLightServerAPI) setParams(params map[string]interface{}, clien setFactor(&negFactors.requestFactor) case !defParams && name == "capacity": if capacity, ok := value.(float64); ok && uint64(capacity) >= api.server.minCapacity { - _, _, err = api.server.clientPool.setCapacity(client.id, client.freeID, uint64(capacity), 0, true) + _, _, err = api.server.clientPool.setCapacity(client.id, client.address, uint64(capacity), 0, true) // Don't have to call factor update explicitly. It's already done // in setCapacity function. } else { diff --git a/les/clientpool.go b/les/clientpool.go index b2ba18e5c9..c55a14827f 100644 --- a/les/clientpool.go +++ b/les/clientpool.go @@ -113,20 +113,20 @@ type clientPoolPeer interface { freeze() } -// clientInfo represents a connected client +// clientInfo defines all information required by clientpool. type clientInfo struct { - address string - id enode.ID - freeID string - active bool - connectedAt mclock.AbsTime - capacity uint64 - priority bool - pool *clientPool - peer clientPoolPeer - queueIndex int // position in activeQueue - balanceTracker balanceTracker - posFactors, negFactors priceFactors + id enode.ID + address string + active bool + capacity uint64 + priority bool + pool *clientPool + peer clientPoolPeer + connectedAt mclock.AbsTime + queueIndex int + balanceTracker balanceTracker + posFactors priceFactors + negFactors priceFactors } // connSetIndex callback updates clientInfo item index in activeQueue @@ -206,8 +206,9 @@ func newClientPool(db ethdb.Database, minCap, freeClientCap uint64, clock mclock break } } - // If the negative balance of free client is low enough, - // delete this entry. + // The positive and negative balances of clients are stored in database + // and both of these decay exponentially over time. Delete them if the + // value is small enough. ndb.evictCallBack = func(now mclock.AbsTime, neg bool, b tokenBalance) bool { var expiration float64 if neg { @@ -215,7 +216,7 @@ func newClientPool(db ethdb.Database, minCap, freeClientCap uint64, clock mclock } else { expiration = pool.posExpiration(now) } - return b.value.value(expiration) < uint64(time.Second) + return b.value.value(expiration) <= uint64(time.Second) } go func() { for { @@ -299,8 +300,14 @@ func (f *clientPool) updateFreeRatio() { } f.averageFreeRatio -= (f.freeRatio - f.averageFreeRatio) * math.Expm1(-float64(dt)/float64(freeRatioTC)) f.freeRatioLastUpdate = now - f.posExp += float64(dt) / float64(f.posExpTC) * f.freeRatio - f.negExp += float64(dt) / float64(f.negExpTC) * f.freeRatio + + dt /= mclock.AbsTime(time.Second) + if f.posExpTC != 0 { + f.posExp += float64(dt) / float64(f.posExpTC) * f.freeRatio + } + if f.negExpTC != 0 { + f.negExp += float64(dt) / float64(f.negExpTC) * f.freeRatio + } f.expLock.Unlock() } @@ -331,10 +338,14 @@ func (f *clientPool) posExpiration(now mclock.AbsTime) float64 { f.expLock.RLock() defer f.expLock.RUnlock() + if f.posExpTC == 0 { + return 0 + } dt := now - f.freeRatioLastUpdate if dt < 0 { dt = 0 } + dt /= mclock.AbsTime(time.Second) return f.posExp + float64(dt)/float64(f.posExpTC)*f.freeRatio } @@ -344,10 +355,14 @@ func (f *clientPool) negExpiration(now mclock.AbsTime) float64 { f.expLock.RLock() defer f.expLock.RUnlock() + if f.negExpTC == 0 { + return 0 + } dt := now - f.freeRatioLastUpdate if dt < 0 { dt = 0 } + dt /= mclock.AbsTime(time.Second) return f.negExp + float64(dt)/float64(f.negExpTC)*f.freeRatio } @@ -402,18 +417,17 @@ func (f *clientPool) connect(peer clientPoolPeer, reqCapacity uint64) (uint64, e if _, ok := f.connectedMap[id]; ok { clientRejectedMeter.Mark(1) log.Debug("Client already connected", "address", freeID, "id", peerIdToString(id)) - return 0, fmt.Errorf("Client already connected address = %s id = %s", freeID, peerIdToString(id)) + return 0, fmt.Errorf("Client already connected address=%s id=%s", freeID, peerIdToString(id)) } pb := f.ndb.getOrNewBalance(id.Bytes(), false) nb := f.ndb.getOrNewBalance([]byte(freeID), true) e := &clientInfo{ + id: id, + address: freeID, capacity: reqCapacity, pool: f, peer: peer, - address: freeID, queueIndex: -1, - id: id, - freeID: freeID, connectedAt: f.clock.Now(), priority: pb.value.base != 0, posFactors: f.defaultPosFactors, @@ -629,7 +643,7 @@ func (f *clientPool) tryActivateClients() { now := f.clock.Now() for f.inactiveQueue.Size() != 0 { e := f.inactiveQueue.PopItem().(*clientInfo) - missing, capacity := f.capAvailable(e.id, e.freeID, e.capacity, 0, true) + missing, capacity := f.capAvailable(e.id, e.address, e.capacity, 0, true) if missing != 0 { f.inactiveQueue.Push(e, -connPriority(e, now)) return @@ -657,7 +671,7 @@ func (f *clientPool) tryActivateClients() { e.peer.updateCapacity(e.capacity) totalConnectedGauge.Update(int64(f.activeCap)) clientConnectedMeter.Mark(1) - log.Debug("Client activated", "address", e.freeID) + log.Debug("Client activated", "address", e.address) } } diff --git a/les/clientpool_test.go b/les/clientpool_test.go index bbe185ce53..a34deff1b8 100644 --- a/les/clientpool_test.go +++ b/les/clientpool_test.go @@ -221,7 +221,7 @@ func TestConnectPaidClientToFullPool(t *testing.T) { pool.setDefaultFactors(priceFactors{1, 0, 1}, priceFactors{1, 0, 1}) for i := 0; i < 10; i++ { - pool.addBalance(newPoolTestPeer(i, nil).ID(), 1000000000) + pool.addBalance(newPoolTestPeer(i, nil).ID(), int64(time.Second)) pool.connect(newPoolTestPeer(i, nil), 1) } pool.addBalance(newPoolTestPeer(11, nil).ID(), 1000) // Add low balance to new paid client @@ -229,7 +229,7 @@ func TestConnectPaidClientToFullPool(t *testing.T) { t.Fatalf("Low balance paid client should be rejected") } clock.Run(time.Second) - pool.addBalance(newPoolTestPeer(12, nil).ID(), 1000000000*60*3+1) // Add high balance to new paid client + pool.addBalance(newPoolTestPeer(12, nil).ID(), int64(time.Minute*5)) // Add high balance to new paid client if cap, _ := pool.connect(newPoolTestPeer(12, nil), 1); cap == 0 { t.Fatalf("High balance paid client should be accepted") } @@ -417,7 +417,7 @@ func TestNegativeBalanceCalculation(t *testing.T) { for i := 0; i < 10; i++ { pool.connect(newPoolTestPeer(i, nil), 1) } - clock.Run(time.Millisecond * 999) + clock.Run(time.Second) for i := 0; i < 10; i++ { pool.disconnect(newPoolTestPeer(i, nil)) @@ -432,12 +432,11 @@ func TestNegativeBalanceCalculation(t *testing.T) { clock.Run(time.Minute) for i := 0; i < 10; i++ { pool.disconnect(newPoolTestPeer(i, nil)) - //nb := pool.ndb.getOrNewNB(newPoolTestPeer(i, nil).freeClientId()) - //// nb.logValue -= pool.negExpiration(clock.Now()) - //nb.logValue = uint64(float64(nb.logValue) / logMultiplier) - //if nb.logValue != uint64(math.Log(float64(time.Minute/time.Second))) { - // t.Fatalf("Negative balance mismatch, want %v, got %v", int64(math.Log(float64(time.Minute/time.Second))), nb.logValue) - //} + nb := pool.ndb.getOrNewBalance([]byte(newPoolTestPeer(i, nil).freeClientId()), true) + value := nb.value.value(pool.negExpiration(clock.Now())) + if value != uint64(time.Minute) { + t.Fatalf("Negative balance mismatch, want %v, got %v", time.Minute, value) + } } }