les: stash

This commit is contained in:
rjl493456442 2020-02-24 09:47:28 +08:00
parent 55d5cf8d97
commit 45bc1f2a35
3 changed files with 47 additions and 34 deletions

View file

@ -154,7 +154,7 @@ func (api *PrivateLightServerAPI) setParams(params map[string]interface{}, clien
setFactor(&negFactors.requestFactor) setFactor(&negFactors.requestFactor)
case !defParams && name == "capacity": case !defParams && name == "capacity":
if capacity, ok := value.(float64); ok && uint64(capacity) >= api.server.minCapacity { 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 // Don't have to call factor update explicitly. It's already done
// in setCapacity function. // in setCapacity function.
} else { } else {

View file

@ -113,20 +113,20 @@ type clientPoolPeer interface {
freeze() freeze()
} }
// clientInfo represents a connected client // clientInfo defines all information required by clientpool.
type clientInfo struct { type clientInfo struct {
address string id enode.ID
id enode.ID address string
freeID string active bool
active bool capacity uint64
connectedAt mclock.AbsTime priority bool
capacity uint64 pool *clientPool
priority bool peer clientPoolPeer
pool *clientPool connectedAt mclock.AbsTime
peer clientPoolPeer queueIndex int
queueIndex int // position in activeQueue balanceTracker balanceTracker
balanceTracker balanceTracker posFactors priceFactors
posFactors, negFactors priceFactors negFactors priceFactors
} }
// connSetIndex callback updates clientInfo item index in activeQueue // connSetIndex callback updates clientInfo item index in activeQueue
@ -206,8 +206,9 @@ func newClientPool(db ethdb.Database, minCap, freeClientCap uint64, clock mclock
break break
} }
} }
// If the negative balance of free client is low enough, // The positive and negative balances of clients are stored in database
// delete this entry. // 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 { ndb.evictCallBack = func(now mclock.AbsTime, neg bool, b tokenBalance) bool {
var expiration float64 var expiration float64
if neg { if neg {
@ -215,7 +216,7 @@ func newClientPool(db ethdb.Database, minCap, freeClientCap uint64, clock mclock
} else { } else {
expiration = pool.posExpiration(now) expiration = pool.posExpiration(now)
} }
return b.value.value(expiration) < uint64(time.Second) return b.value.value(expiration) <= uint64(time.Second)
} }
go func() { go func() {
for { for {
@ -299,8 +300,14 @@ func (f *clientPool) updateFreeRatio() {
} }
f.averageFreeRatio -= (f.freeRatio - f.averageFreeRatio) * math.Expm1(-float64(dt)/float64(freeRatioTC)) f.averageFreeRatio -= (f.freeRatio - f.averageFreeRatio) * math.Expm1(-float64(dt)/float64(freeRatioTC))
f.freeRatioLastUpdate = now 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() f.expLock.Unlock()
} }
@ -331,10 +338,14 @@ func (f *clientPool) posExpiration(now mclock.AbsTime) float64 {
f.expLock.RLock() f.expLock.RLock()
defer f.expLock.RUnlock() defer f.expLock.RUnlock()
if f.posExpTC == 0 {
return 0
}
dt := now - f.freeRatioLastUpdate dt := now - f.freeRatioLastUpdate
if dt < 0 { if dt < 0 {
dt = 0 dt = 0
} }
dt /= mclock.AbsTime(time.Second)
return f.posExp + float64(dt)/float64(f.posExpTC)*f.freeRatio 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() f.expLock.RLock()
defer f.expLock.RUnlock() defer f.expLock.RUnlock()
if f.negExpTC == 0 {
return 0
}
dt := now - f.freeRatioLastUpdate dt := now - f.freeRatioLastUpdate
if dt < 0 { if dt < 0 {
dt = 0 dt = 0
} }
dt /= mclock.AbsTime(time.Second)
return f.negExp + float64(dt)/float64(f.negExpTC)*f.freeRatio 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 { if _, ok := f.connectedMap[id]; ok {
clientRejectedMeter.Mark(1) clientRejectedMeter.Mark(1)
log.Debug("Client already connected", "address", freeID, "id", peerIdToString(id)) 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) pb := f.ndb.getOrNewBalance(id.Bytes(), false)
nb := f.ndb.getOrNewBalance([]byte(freeID), true) nb := f.ndb.getOrNewBalance([]byte(freeID), true)
e := &clientInfo{ e := &clientInfo{
id: id,
address: freeID,
capacity: reqCapacity, capacity: reqCapacity,
pool: f, pool: f,
peer: peer, peer: peer,
address: freeID,
queueIndex: -1, queueIndex: -1,
id: id,
freeID: freeID,
connectedAt: f.clock.Now(), connectedAt: f.clock.Now(),
priority: pb.value.base != 0, priority: pb.value.base != 0,
posFactors: f.defaultPosFactors, posFactors: f.defaultPosFactors,
@ -629,7 +643,7 @@ func (f *clientPool) tryActivateClients() {
now := f.clock.Now() now := f.clock.Now()
for f.inactiveQueue.Size() != 0 { for f.inactiveQueue.Size() != 0 {
e := f.inactiveQueue.PopItem().(*clientInfo) 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 { if missing != 0 {
f.inactiveQueue.Push(e, -connPriority(e, now)) f.inactiveQueue.Push(e, -connPriority(e, now))
return return
@ -657,7 +671,7 @@ func (f *clientPool) tryActivateClients() {
e.peer.updateCapacity(e.capacity) e.peer.updateCapacity(e.capacity)
totalConnectedGauge.Update(int64(f.activeCap)) totalConnectedGauge.Update(int64(f.activeCap))
clientConnectedMeter.Mark(1) clientConnectedMeter.Mark(1)
log.Debug("Client activated", "address", e.freeID) log.Debug("Client activated", "address", e.address)
} }
} }

View file

@ -221,7 +221,7 @@ func TestConnectPaidClientToFullPool(t *testing.T) {
pool.setDefaultFactors(priceFactors{1, 0, 1}, priceFactors{1, 0, 1}) pool.setDefaultFactors(priceFactors{1, 0, 1}, priceFactors{1, 0, 1})
for i := 0; i < 10; i++ { 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.connect(newPoolTestPeer(i, nil), 1)
} }
pool.addBalance(newPoolTestPeer(11, nil).ID(), 1000) // Add low balance to new paid client 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") t.Fatalf("Low balance paid client should be rejected")
} }
clock.Run(time.Second) 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 { if cap, _ := pool.connect(newPoolTestPeer(12, nil), 1); cap == 0 {
t.Fatalf("High balance paid client should be accepted") t.Fatalf("High balance paid client should be accepted")
} }
@ -417,7 +417,7 @@ func TestNegativeBalanceCalculation(t *testing.T) {
for i := 0; i < 10; i++ { for i := 0; i < 10; i++ {
pool.connect(newPoolTestPeer(i, nil), 1) pool.connect(newPoolTestPeer(i, nil), 1)
} }
clock.Run(time.Millisecond * 999) clock.Run(time.Second)
for i := 0; i < 10; i++ { for i := 0; i < 10; i++ {
pool.disconnect(newPoolTestPeer(i, nil)) pool.disconnect(newPoolTestPeer(i, nil))
@ -432,12 +432,11 @@ func TestNegativeBalanceCalculation(t *testing.T) {
clock.Run(time.Minute) clock.Run(time.Minute)
for i := 0; i < 10; i++ { for i := 0; i < 10; i++ {
pool.disconnect(newPoolTestPeer(i, nil)) pool.disconnect(newPoolTestPeer(i, nil))
//nb := pool.ndb.getOrNewNB(newPoolTestPeer(i, nil).freeClientId()) nb := pool.ndb.getOrNewBalance([]byte(newPoolTestPeer(i, nil).freeClientId()), true)
//// nb.logValue -= pool.negExpiration(clock.Now()) value := nb.value.value(pool.negExpiration(clock.Now()))
//nb.logValue = uint64(float64(nb.logValue) / logMultiplier) if value != uint64(time.Minute) {
//if nb.logValue != uint64(math.Log(float64(time.Minute/time.Second))) { t.Fatalf("Negative balance mismatch, want %v, got %v", time.Minute, value)
// t.Fatalf("Negative balance mismatch, want %v, got %v", int64(math.Log(float64(time.Minute/time.Second))), nb.logValue) }
//}
} }
} }