forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-20 10:52:25 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

les: client pool improvements

Browse source
This commit is contained in:
Zsolt Felfoldi 2020-01-03 20:45:53 +01:00
parent 9907044bf5
commit 58f9dc5604
6 changed files with 960 additions and 426 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

173
les/balance.go
View file

@ -17,24 +17,101 @@
package les
import (
"math"
"sync"
"time"
"github.com/ethereum/go-ethereum/common/mclock"
)
const maxBalance = math.MaxInt64
const (
balanceCallbackQueue = iota
balanceCallbackZero
balanceCallbackCount
)
// expirationController controls the exponential expiration of positive and negative
// balances
type expirationController interface {
posExpiration(mclock.AbsTime) uint64
negExpiration(mclock.AbsTime) uint64
}
// expiredValue is a scalar value that is continuously expired (decreased exponentially)
// based on the logarithmic expiration offset value provided by the expirationController.
// The actual value can be calculated as base*2^(exp-logOffset/log2Multiplier).
//
// Note: this representation is basically a floating point value capable of handling
// 64 bit exponents. The operations are not general purpose floating point operations
// though because a monotonously increasing exponent is assumed and therefore if an
// operation is performed with two different exponents then simply the higher one is used.
type expiredValue struct {
base, exp uint64
}
// value calculates the value at the given moment
func (e expiredValue) value(logOffset uint64) uint64 {
return uint64(math.Exp(float64(int64(e.exp*log2Multiplier-logOffset))/logMultiplier) * float64(e.base))
}
// add adds a signed value at the given moment
func (e *expiredValue) add(amount int64, logOffset uint64) int64 {
addExp := logOffset / log2Multiplier
baseMul := math.Exp(float64(logOffset%log2Multiplier) / logMultiplier)
if addExp < e.exp {
baseMul /= math.Pow(2, float64(e.exp-addExp))
}
if addExp > e.exp {
e.base >>= (addExp - e.exp)
e.exp = addExp
}
add := int64(float64(amount) * baseMul)
if add >= 0 || uint64(-add) <= e.base {
e.base += uint64(add)
return amount
} else {
e.base = 0
return int64(-float64(e.base) / baseMul)
}
}
// addExp adds another expiredValue
func (e *expiredValue) addExp(a expiredValue) {
if e.exp > a.exp {
a.base >>= (e.exp - a.exp)
}
if e.exp < a.exp {
e.base >>= (a.exp - e.exp)
e.exp = a.exp
}
e.base += a.base
}
// subExp subtracts another expiredValue
func (e *expiredValue) subExp(a expiredValue) {
if e.exp > a.exp {
a.base >>= (e.exp - a.exp)
}
if e.exp < a.exp {
e.base >>= (a.exp - e.exp)
e.exp = a.exp
}
if e.base > a.base {
e.base -= a.base
} else {
e.base = 0
}
}
// balanceTracker keeps track of the positive and negative balances of a connected
// client and calculates actual and projected future priority values required by
// prque.LazyQueue.
type balanceTracker struct {
lock sync.Mutex
clock mclock.Clock
exp expirationController
stopped bool
capacity uint64
balance balance
@ -53,7 +130,7 @@ type balanceTracker struct {
// balance represents a pair of positive and negative balances
type balance struct {
pos, neg uint64
pos, neg expiredValue
}
// balanceCallback represents a single callback that is activated when client priority
@ -65,6 +142,7 @@ type balanceCallback struct {
}
// init initializes balanceTracker
// Note: capacity should never be zero
func (bt *balanceTracker) init(clock mclock.Clock, capacity uint64) {
bt.clock = clock
bt.initTime, bt.lastUpdate = clock.Now(), clock.Now() // Init timestamps
@ -95,10 +173,41 @@ func (bt *balanceTracker) stop(now mclock.AbsTime) {
// first to disconnect. Positive balance translates to negative priority. If positive
// balance is zero then negative balance translates to a positive priority.
func (bt *balanceTracker) balanceToPriority(b balance) int64 {
if b.pos > 0 {
return ^int64(b.pos / bt.capacity)
if b.pos.base > 0 {
return -int64(b.pos.value(bt.exp.posExpiration(bt.clock.Now())) / bt.capacity)
}
return int64(b.neg)
return int64(b.neg.value(bt.exp.negExpiration(bt.clock.Now())))
}
// posBalanceMissing calculates the missing amount of positive balance in order to
// connect at targetCapacity, stay connected for the given amount of time and then
// still have a priority of targetPriority
func (bt *balanceTracker) posBalanceMissing(targetPriority int64, targetCapacity uint64, after time.Duration) uint64 {
now := bt.clock.Now()
if targetPriority > 0 {
negPrice := uint64(float64(after) * bt.negTimeFactor)
negBalance := bt.balance.neg.value(bt.exp.negExpiration(now))
if negPrice+negBalance < uint64(targetPriority) {
return 0
}
if uint64(targetPriority) > negBalance && bt.negTimeFactor > 1e-100 {
if negTime := time.Duration(float64(uint64(targetPriority)-negBalance) / bt.negTimeFactor); negTime < after {
after -= negTime
} else {
after = 0
}
}
targetPriority = 0
}
posRequired := uint64(float64(-targetPriority)*float64(targetCapacity)+float64(after)*bt.timeFactor) + 1
if posRequired >= maxBalance {
return math.MaxUint64 // target not reachable
}
posBalance := bt.balance.pos.value(bt.exp.posExpiration(now))
if posRequired > posBalance {
return posRequired - posBalance
}
return 0
}
// reducedBalance estimates the reduced balance at a given time in the fututre based
@ -106,20 +215,19 @@ func (bt *balanceTracker) balanceToPriority(b balance) int64 {
func (bt *balanceTracker) reducedBalance(at mclock.AbsTime, avgReqCost float64) balance {
dt := float64(at - bt.lastUpdate)
b := bt.balance
if b.pos != 0 {
if b.pos.base != 0 {
factor := bt.timeFactor + bt.requestFactor*avgReqCost
diff := uint64(dt * factor)
if diff <= b.pos {
b.pos -= diff
diff := -int64(dt * factor)
dd := b.pos.add(diff, bt.exp.posExpiration(at))
if dd == diff {
dt = 0
} else {
dt -= float64(b.pos) / factor
b.pos = 0
dt += float64(dd) / factor
}
}
if dt != 0 {
if dt > 0 {
factor := bt.negTimeFactor + bt.negRequestFactor*avgReqCost
b.neg += uint64(dt * factor)
b.neg.add(int64(dt*factor), bt.exp.negExpiration(at))
}
return b
}
@ -130,20 +238,22 @@ func (bt *balanceTracker) reducedBalance(at mclock.AbsTime, avgReqCost float64)
// Note: the function assumes that the balance has been recently updated and
// calculates the time starting from the last update.
func (bt *balanceTracker) timeUntil(priority int64) (time.Duration, bool) {
now := bt.clock.Now()
var dt float64
if bt.balance.pos != 0 {
if bt.balance.pos.base != 0 {
posBalance := bt.balance.pos.value(bt.exp.posExpiration(now))
if bt.timeFactor < 1e-100 {
return 0, false
}
if priority < 0 {
newBalance := uint64(^priority) * bt.capacity
if newBalance > bt.balance.pos {
newBalance := uint64(-priority) * bt.capacity
if newBalance > posBalance {
return 0, false
}
dt = float64(bt.balance.pos-newBalance) / bt.timeFactor
dt = float64(posBalance-newBalance) / bt.timeFactor
return time.Duration(dt), true
} else {
dt = float64(bt.balance.pos) / bt.timeFactor
dt = float64(posBalance) / bt.timeFactor
}
} else {
if priority < 0 {
@ -151,16 +261,18 @@ func (bt *balanceTracker) timeUntil(priority int64) (time.Duration, bool) {
}
}
// if we have a positive balance then dt equals the time needed to get it to zero
if uint64(priority) > bt.balance.neg {
negBalance := bt.balance.neg.value(bt.exp.negExpiration(now))
if uint64(priority) > negBalance {
if bt.negTimeFactor < 1e-100 {
return 0, false
}
dt += float64(uint64(priority)-bt.balance.neg) / bt.negTimeFactor
dt += float64(uint64(priority)-negBalance) / bt.negTimeFactor
}
return time.Duration(dt), true
}
// setCapacity updates the capacity value used for priority calculation
// Note: capacity should never be zero
func (bt *balanceTracker) setCapacity(capacity uint64) {
bt.lock.Lock()
defer bt.lock.Unlock()
@ -262,26 +374,26 @@ func (bt *balanceTracker) updateAfter(dt time.Duration) {
}
// requestCost should be called after serving a request for the given peer
func (bt *balanceTracker) requestCost(cost uint64) {
func (bt *balanceTracker) requestCost(cost uint64) uint64 {
bt.lock.Lock()
defer bt.lock.Unlock()
if bt.stopped {
return
return 0
}
now := bt.clock.Now()
bt.addBalance(now)
fcost := float64(cost)
if bt.balance.pos != 0 {
posExp := bt.exp.posExpiration(now)
if bt.balance.pos.base != 0 {
if bt.requestFactor != 0 {
c := uint64(fcost * bt.requestFactor)
if bt.balance.pos >= c {
bt.balance.pos -= c
c := -int64(fcost * bt.requestFactor)
cc := bt.balance.pos.add(c, posExp)
if c == cc {
fcost = 0
} else {
fcost *= 1 - float64(bt.balance.pos)/float64(c)
bt.balance.pos = 0
fcost *= 1 - float64(cc)/float64(c)
}
bt.checkCallbacks(now)
} else {
@ -290,15 +402,16 @@ func (bt *balanceTracker) requestCost(cost uint64) {
}
if fcost > 0 {
if bt.negRequestFactor != 0 {
bt.balance.neg += uint64(fcost * bt.negRequestFactor)
bt.balance.neg.add(int64(fcost*bt.negRequestFactor), bt.exp.negExpiration(now))
bt.checkCallbacks(now)
}
}
bt.sumReqCost += cost
return bt.balance.pos.value(posExp)
}
// getBalance returns the current positive and negative balance
func (bt *balanceTracker) getBalance(now mclock.AbsTime) (uint64, uint64) {
func (bt *balanceTracker) getBalance(now mclock.AbsTime) (expiredValue, expiredValue) {
bt.lock.Lock()
defer bt.lock.Unlock()
@ -307,7 +420,7 @@ func (bt *balanceTracker) getBalance(now mclock.AbsTime) (uint64, uint64) {
}
// setBalance sets the positive and negative balance to the given values
func (bt *balanceTracker) setBalance(pos, neg uint64) error {
func (bt *balanceTracker) setBalance(pos, neg expiredValue) error {
bt.lock.Lock()
defer bt.lock.Unlock()

79
les/balance_test.go
View file

@ -23,18 +23,31 @@ import (
"github.com/ethereum/go-ethereum/common/mclock"
)
type zeroExpCtrl struct{}
func (z zeroExpCtrl) posExpiration(mclock.AbsTime) uint64 {
return 0
}
func (z zeroExpCtrl) negExpiration(mclock.AbsTime) uint64 {
return 0
}
func expval(v uint64) expiredValue {
return expiredValue{base: v}
}
func TestSetBalance(t *testing.T) {
var clock = &mclock.Simulated{}
var inputs = []struct {
pos uint64
neg uint64
pos, neg expiredValue
}{
{1000, 0},
{0, 1000},
{1000, 1000},
{expval(1000), expval(0)},
{expval(0), expval(1000)},
{expval(1000), expval(1000)},
}
tracker := balanceTracker{}
tracker := balanceTracker{exp: zeroExpCtrl{}}
tracker.init(clock, 1000)
defer tracker.stop(clock.Now())
@ -53,14 +66,14 @@ func TestSetBalance(t *testing.T) {
func TestBalanceTimeCost(t *testing.T) {
var (
clock = &mclock.Simulated{}
tracker = balanceTracker{}
tracker = balanceTracker{exp: zeroExpCtrl{}}
)
tracker.init(clock, 1000)
defer tracker.stop(clock.Now())
tracker.setFactors(false, 1, 1)
tracker.setFactors(true, 1, 1)
tracker.setBalance(uint64(time.Minute), 0) // 1 minute time allowance
tracker.setBalance(expval(uint64(time.Minute)), expval(0)) // 1 minute time allowance
var inputs = []struct {
runTime time.Duration
@ -74,21 +87,21 @@ func TestBalanceTimeCost(t *testing.T) {
}
for _, i := range inputs {
clock.Run(i.runTime)
if pos, _ := tracker.getBalance(clock.Now()); pos != i.expPos {
if pos, _ := tracker.getBalance(clock.Now()); pos != expval(i.expPos) {
t.Fatalf("Positive balance mismatch, want %v, got %v", i.expPos, pos)
}
if _, neg := tracker.getBalance(clock.Now()); neg != i.expNeg {
if _, neg := tracker.getBalance(clock.Now()); neg != expval(i.expNeg) {
t.Fatalf("Negative balance mismatch, want %v, got %v", i.expNeg, neg)
}
}
tracker.setBalance(uint64(time.Minute), 0) // Refill 1 minute time allowance
tracker.setBalance(expval(uint64(time.Minute)), expval(0)) // Refill 1 minute time allowance
for _, i := range inputs {
clock.Run(i.runTime)
if pos, _ := tracker.getBalance(clock.Now()); pos != i.expPos {
if pos, _ := tracker.getBalance(clock.Now()); pos != expval(i.expPos) {
t.Fatalf("Positive balance mismatch, want %v, got %v", i.expPos, pos)
}
if _, neg := tracker.getBalance(clock.Now()); neg != i.expNeg {
if _, neg := tracker.getBalance(clock.Now()); neg != expval(i.expNeg) {
t.Fatalf("Negative balance mismatch, want %v, got %v", i.expNeg, neg)
}
}
@ -97,14 +110,14 @@ func TestBalanceTimeCost(t *testing.T) {
func TestBalanceReqCost(t *testing.T) {
var (
clock = &mclock.Simulated{}
tracker = balanceTracker{}
tracker = balanceTracker{exp: zeroExpCtrl{}}
)
tracker.init(clock, 1000)
defer tracker.stop(clock.Now())
tracker.setFactors(false, 1, 1)
tracker.setFactors(true, 1, 1)
tracker.setBalance(uint64(time.Minute), 0) // 1 minute time serving time allowance
tracker.setBalance(expval(uint64(time.Minute)), expval(0)) // 1 minute time serving time allowance
var inputs = []struct {
reqCost uint64
expPos uint64
@ -117,10 +130,10 @@ func TestBalanceReqCost(t *testing.T) {
}
for _, i := range inputs {
tracker.requestCost(i.reqCost)
if pos, _ := tracker.getBalance(clock.Now()); pos != i.expPos {
if pos, _ := tracker.getBalance(clock.Now()); pos != expval(i.expPos) {
t.Fatalf("Positive balance mismatch, want %v, got %v", i.expPos, pos)
}
if _, neg := tracker.getBalance(clock.Now()); neg != i.expNeg {
if _, neg := tracker.getBalance(clock.Now()); neg != expval(i.expNeg) {
t.Fatalf("Negative balance mismatch, want %v, got %v", i.expNeg, neg)
}
}
@ -129,7 +142,7 @@ func TestBalanceReqCost(t *testing.T) {
func TestBalanceToPriority(t *testing.T) {
var (
clock = &mclock.Simulated{}
tracker = balanceTracker{}
tracker = balanceTracker{exp: zeroExpCtrl{}}
)
tracker.init(clock, 1000) // cap = 1000
defer tracker.stop(clock.Now())
@ -141,13 +154,13 @@ func TestBalanceToPriority(t *testing.T) {
neg uint64
priority int64
}{
{1000, 0, ^int64(1)},
{2000, 0, ^int64(2)}, // Higher balance, lower priority value
{1000, 0, -1},
{2000, 0, -2}, // Higher balance, lower priority value
{0, 0, 0},
{0, 1000, 1000},
}
for _, i := range inputs {
tracker.setBalance(i.pos, i.neg)
tracker.setBalance(expval(i.pos), expval(i.neg))
priority := tracker.getPriority(clock.Now())
if priority != i.priority {
t.Fatalf("Priority mismatch, want %v, got %v", i.priority, priority)
@ -158,30 +171,30 @@ func TestBalanceToPriority(t *testing.T) {
func TestEstimatedPriority(t *testing.T) {
var (
clock = &mclock.Simulated{}
tracker = balanceTracker{}
tracker = balanceTracker{exp: zeroExpCtrl{}}
)
tracker.init(clock, 1000000000) // cap = 1000,000,000
defer tracker.stop(clock.Now())
tracker.setFactors(false, 1, 1)
tracker.setFactors(true, 1, 1)
tracker.setBalance(uint64(time.Minute), 0)
tracker.setBalance(expval(uint64(time.Minute)), expval(0))
var inputs = []struct {
runTime time.Duration // time cost
futureTime time.Duration // diff of future time
reqCost uint64 // single request cost
priority int64 // expected estimated priority
}{
{time.Second, time.Second, 0, ^int64(58)},
{0, time.Second, 0, ^int64(58)},
{time.Second, time.Second, 0, -58},
{0, time.Second, 0, -58},
// 2 seconds time cost, 1 second estimated time cost, 10^9 request cost,
// 10^9 estimated request cost per second.
{time.Second, time.Second, 1000000000, ^int64(55)},
{time.Second, time.Second, 1000000000, -55},
// 3 seconds time cost, 3 second estimated time cost, 10^9*2 request cost,
// 4*10^9 estimated request cost.
{time.Second, 3 * time.Second, 1000000000, ^int64(48)},
{time.Second, 3 * time.Second, 1000000000, -48},
// All positive balance is used up
{time.Second * 55, 0, 0, 0},
@ -202,7 +215,7 @@ func TestEstimatedPriority(t *testing.T) {
func TestCallbackChecking(t *testing.T) {
var (
clock = &mclock.Simulated{}
tracker = balanceTracker{}
tracker = balanceTracker{exp: zeroExpCtrl{}}
)
tracker.init(clock, 1000000) // cap = 1000,000
defer tracker.stop(clock.Now())
@ -213,11 +226,11 @@ func TestCallbackChecking(t *testing.T) {
priority int64
expDiff time.Duration
}{
{^int64(500), time.Millisecond * 500},
{-500, time.Millisecond * 500},
{0, time.Second},
{int64(time.Second), 2 * time.Second},
}
tracker.setBalance(uint64(time.Second), 0)
tracker.setBalance(expval(uint64(time.Second)), expval(0))
for _, i := range inputs {
diff, _ := tracker.timeUntil(i.priority)
if diff != i.expDiff {
@ -229,7 +242,7 @@ func TestCallbackChecking(t *testing.T) {
func TestCallback(t *testing.T) {
var (
clock = &mclock.Simulated{}
tracker = balanceTracker{}
tracker = balanceTracker{exp: zeroExpCtrl{}}
)
tracker.init(clock, 1000) // cap = 1000
defer tracker.stop(clock.Now())
@ -237,7 +250,7 @@ func TestCallback(t *testing.T) {
tracker.setFactors(true, 1, 1)
callCh := make(chan struct{}, 1)
tracker.setBalance(uint64(time.Minute), 0)
tracker.setBalance(expval(uint64(time.Minute)), expval(0))
tracker.addCallback(balanceCallbackZero, 0, func() { callCh <- struct{}{} })
clock.Run(time.Minute)
@ -247,7 +260,7 @@ func TestCallback(t *testing.T) {
t.Fatalf("Callback hasn't been called yet")
}
tracker.setBalance(uint64(time.Minute), 0)
tracker.setBalance(expval(uint64(time.Minute)), expval(0))
tracker.addCallback(balanceCallbackZero, 0, func() { callCh <- struct{}{} })
tracker.removeCallback(balanceCallbackZero)

779
les/clientpool.go
View file

File diff suppressed because it is too large Load diff

269
les/clientpool_test.go
View file

@ -56,29 +56,34 @@ func TestClientPoolL100C300P20(t *testing.T) {
const testClientPoolTicks = 100000
type poolTestPeer int
func (i poolTestPeer) ID() enode.ID {
return enode.ID{byte(i % 256), byte(i >> 8)}
}
func (i poolTestPeer) freeClientId() string {
return fmt.Sprintf("addr #%d", i)
}
func (i poolTestPeer) updateCapacity(uint64) {}
type poolTestPeerWithCap struct {
poolTestPeer
type poolTestPeer struct {
index int
disconnCh chan int
cap uint64
}
func (i *poolTestPeerWithCap) updateCapacity(cap uint64) { i.cap = cap }
func newPoolTestPeer(i int, disconnCh chan int) *poolTestPeer {
return &poolTestPeer{index: i, disconnCh: disconnCh}
}
func (i poolTestPeer) freezeClient() {}
func (i *poolTestPeer) ID() enode.ID {
return enode.ID{byte(i.index % 256), byte(i.index >> 8)}
}
func testClientPool(t *testing.T, connLimit, clientCount, paidCount int, randomDisconnect bool) {
func (i *poolTestPeer) freeClientId() string {
return fmt.Sprintf("addr #%d", i)
}
func (i *poolTestPeer) updateCapacity(cap uint64) {
i.cap = cap
if cap == 0 && i.disconnCh != nil {
i.disconnCh <- i.index
}
}
func (i *poolTestPeer) freezeClient() {}
func testClientPool(t *testing.T, activeLimit, clientCount, paidCount int, randomDisconnect bool) {
rand.Seed(time.Now().UnixNano())
var (
clock mclock.Simulated
@ -89,15 +94,16 @@ func testClientPool(t *testing.T, connLimit, clientCount, paidCount int, randomD
disconnFn = func(id enode.ID) {
disconnCh <- int(id[0]) + int(id[1])<<8
}
pool = newClientPool(db, 1, &clock, disconnFn)
pool = newClientPool(db, 1, 1, &clock, disconnFn)
)
pool.disableBias = true
pool.setLimits(connLimit, uint64(connLimit))
pool.setLimits(activeLimit, uint64(activeLimit))
pool.setDefaultFactors(priceFactors{1, 0, 1}, priceFactors{1, 0, 1})
// pool should accept new peers up to its connected limit
for i := 0; i < connLimit; i++ {
if pool.connect(poolTestPeer(i), 0) {
for i := 0; i < activeLimit; i++ {
if cap, _ := pool.connect(newPoolTestPeer(i, disconnCh), 0); cap != 0 {
connected[i] = true
} else {
t.Fatalf("Test peer #%d rejected", i)
@ -111,28 +117,30 @@ func testClientPool(t *testing.T, connLimit, clientCount, paidCount int, randomD
// give a positive balance to some of the peers
amount := testClientPoolTicks / 2 * int64(time.Second) // enough for half of the simulation period
for i := 0; i < paidCount; i++ {
pool.addBalance(poolTestPeer(i).ID(), amount, "")
pool.addBalance(newPoolTestPeer(i, disconnCh).ID(), amount, "")
}
}
i := rand.Intn(clientCount)
if connected[i] {
if randomDisconnect {
pool.disconnect(poolTestPeer(i))
pool.disconnect(newPoolTestPeer(i, disconnCh))
connected[i] = false
connTicks[i] += tickCounter
}
} else {
if pool.connect(poolTestPeer(i), 0) {
if cap, _ := pool.connect(newPoolTestPeer(i, disconnCh), 0); cap != 0 {
connected[i] = true
connTicks[i] -= tickCounter
} else {
pool.disconnect(newPoolTestPeer(i, disconnCh))
}
}
pollDisconnects:
for {
select {
case i := <-disconnCh:
pool.disconnect(poolTestPeer(i))
pool.disconnect(newPoolTestPeer(i, disconnCh))
if connected[i] {
connTicks[i] += tickCounter
connected[i] = false
@ -143,10 +151,10 @@ func testClientPool(t *testing.T, connLimit, clientCount, paidCount int, randomD
}
}
expTicks := testClientPoolTicks/2*connLimit/clientCount + testClientPoolTicks/2*(connLimit-paidCount)/(clientCount-paidCount)
expTicks := testClientPoolTicks/2*activeLimit/clientCount + testClientPoolTicks/2*(activeLimit-paidCount)/(clientCount-paidCount)
expMin := expTicks - expTicks/5
expMax := expTicks + expTicks/5
paidTicks := testClientPoolTicks/2*connLimit/clientCount + testClientPoolTicks/2
paidTicks := testClientPoolTicks/2*activeLimit/clientCount + testClientPoolTicks/2
paidMin := paidTicks - paidTicks/5
paidMax := paidTicks + paidTicks/5
@ -172,15 +180,15 @@ func TestConnectPaidClient(t *testing.T) {
clock mclock.Simulated
db = rawdb.NewMemoryDatabase()
)
pool := newClientPool(db, 1, &clock, nil)
pool := newClientPool(db, 1, 1, &clock, nil)
defer pool.stop()
pool.setLimits(10, uint64(10))
pool.setDefaultFactors(priceFactors{1, 0, 1}, priceFactors{1, 0, 1})
// Add balance for an external client and mark it as paid client
pool.addBalance(poolTestPeer(0).ID(), 1000, "")
pool.addBalance(newPoolTestPeer(0, nil).ID(), 1000, "")
if !pool.connect(poolTestPeer(0), 10) {
if cap, _ := pool.connect(newPoolTestPeer(0, nil), 10); cap == 0 {
t.Fatalf("Failed to connect paid client")
}
}
@ -190,16 +198,16 @@ func TestConnectPaidClientToSmallPool(t *testing.T) {
clock mclock.Simulated
db = rawdb.NewMemoryDatabase()
)
pool := newClientPool(db, 1, &clock, nil)
pool := newClientPool(db, 1, 1, &clock, nil)
defer pool.stop()
pool.setLimits(10, uint64(10)) // Total capacity limit is 10
pool.setDefaultFactors(priceFactors{1, 0, 1}, priceFactors{1, 0, 1})
// Add balance for an external client and mark it as paid client
pool.addBalance(poolTestPeer(0).ID(), 1000, "")
pool.addBalance(newPoolTestPeer(0, nil).ID(), 1000, "")
// Connect a fat paid client to pool, should reject it.
if pool.connect(poolTestPeer(0), 100) {
if cap, _ := pool.connect(newPoolTestPeer(0, nil), 100); cap != 0 {
t.Fatalf("Connected fat paid client, should reject it")
}
}
@ -210,23 +218,23 @@ func TestConnectPaidClientToFullPool(t *testing.T) {
db = rawdb.NewMemoryDatabase()
)
removeFn := func(enode.ID) {} // Noop
pool := newClientPool(db, 1, &clock, removeFn)
pool := newClientPool(db, 1, 1, &clock, removeFn)
defer pool.stop()
pool.setLimits(10, uint64(10)) // Total capacity limit is 10
pool.setDefaultFactors(priceFactors{1, 0, 1}, priceFactors{1, 0, 1})
for i := 0; i < 10; i++ {
pool.addBalance(poolTestPeer(i).ID(), 1000000000, "")
pool.connect(poolTestPeer(i), 1)
pool.addBalance(newPoolTestPeer(i, nil).ID(), 1000000000, "")
pool.connect(newPoolTestPeer(i, nil), 1)
}
pool.addBalance(poolTestPeer(11).ID(), 1000, "") // Add low balance to new paid client
if pool.connect(poolTestPeer(11), 1) {
pool.addBalance(newPoolTestPeer(11, nil).ID(), 1000, "") // Add low balance to new paid client
if cap, _ := pool.connect(newPoolTestPeer(11, nil), 1); cap != 0 {
t.Fatalf("Low balance paid client should be rejected")
}
clock.Run(time.Second)
pool.addBalance(poolTestPeer(12).ID(), 1000000000*60*3, "") // Add high balance to new paid client
if !pool.connect(poolTestPeer(12), 1) {
t.Fatalf("High balance paid client should be accpected")
pool.addBalance(newPoolTestPeer(12, nil).ID(), 1000000000*60*3+1, "") // 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")
}
}
@ -237,19 +245,19 @@ func TestPaidClientKickedOut(t *testing.T) {
kickedCh = make(chan int, 1)
)
removeFn := func(id enode.ID) { kickedCh <- int(id[0]) }
pool := newClientPool(db, 1, &clock, removeFn)
pool := newClientPool(db, 1, 1, &clock, removeFn)
defer pool.stop()
pool.setLimits(10, uint64(10)) // Total capacity limit is 10
pool.setDefaultFactors(priceFactors{1, 0, 1}, priceFactors{1, 0, 1})
for i := 0; i < 10; i++ {
pool.addBalance(poolTestPeer(i).ID(), 1000000000, "") // 1 second allowance
pool.connect(poolTestPeer(i), 1)
pool.addBalance(newPoolTestPeer(i, kickedCh).ID(), 1000000000, "") // 1 second allowance
pool.connect(newPoolTestPeer(i, kickedCh), 1)
clock.Run(time.Millisecond)
}
clock.Run(time.Second)
clock.Run(connectedBias)
if !pool.connect(poolTestPeer(11), 0) {
clock.Run(activeBias)
if cap, _ := pool.connect(newPoolTestPeer(11, kickedCh), 0); cap == 0 {
t.Fatalf("Free client should be accectped")
}
select {
@ -267,11 +275,11 @@ func TestConnectFreeClient(t *testing.T) {
clock mclock.Simulated
db = rawdb.NewMemoryDatabase()
)
pool := newClientPool(db, 1, &clock, nil)
pool := newClientPool(db, 1, 1, &clock, nil)
defer pool.stop()
pool.setLimits(10, uint64(10))
pool.setDefaultFactors(priceFactors{1, 0, 1}, priceFactors{1, 0, 1})
if !pool.connect(poolTestPeer(0), 10) {
if cap, _ := pool.connect(newPoolTestPeer(0, nil), 10); cap == 0 {
t.Fatalf("Failed to connect free client")
}
}
@ -282,24 +290,24 @@ func TestConnectFreeClientToFullPool(t *testing.T) {
db = rawdb.NewMemoryDatabase()
)
removeFn := func(enode.ID) {} // Noop
pool := newClientPool(db, 1, &clock, removeFn)
pool := newClientPool(db, 1, 1, &clock, removeFn)
defer pool.stop()
pool.setLimits(10, uint64(10)) // Total capacity limit is 10
pool.setDefaultFactors(priceFactors{1, 0, 1}, priceFactors{1, 0, 1})
for i := 0; i < 10; i++ {
pool.connect(poolTestPeer(i), 1)
pool.connect(newPoolTestPeer(i, nil), 1)
}
if pool.connect(poolTestPeer(11), 1) {
if cap, _ := pool.connect(newPoolTestPeer(11, nil), 1); cap != 0 {
t.Fatalf("New free client should be rejected")
}
clock.Run(time.Minute)
if pool.connect(poolTestPeer(12), 1) {
if cap, _ := pool.connect(newPoolTestPeer(12, nil), 1); cap != 0 {
t.Fatalf("New free client should be rejected")
}
clock.Run(time.Millisecond)
clock.Run(4 * time.Minute)
if !pool.connect(poolTestPeer(13), 1) {
if cap, _ := pool.connect(newPoolTestPeer(13, nil), 1); cap == 0 {
t.Fatalf("Old client connects more than 5min should be kicked")
}
}
@ -311,21 +319,22 @@ func TestFreeClientKickedOut(t *testing.T) {
kicked = make(chan int, 10)
)
removeFn := func(id enode.ID) { kicked <- int(id[0]) }
pool := newClientPool(db, 1, &clock, removeFn)
pool := newClientPool(db, 1, 1, &clock, removeFn)
defer pool.stop()
pool.setLimits(10, uint64(10)) // Total capacity limit is 10
pool.setDefaultFactors(priceFactors{1, 0, 1}, priceFactors{1, 0, 1})
for i := 0; i < 10; i++ {
pool.connect(poolTestPeer(i), 1)
pool.connect(newPoolTestPeer(i, kicked), 1)
clock.Run(time.Millisecond)
}
if pool.connect(poolTestPeer(10), 1) {
if cap, _ := pool.connect(newPoolTestPeer(10, kicked), 1); cap != 0 {
t.Fatalf("New free client should be rejected")
}
pool.disconnect(newPoolTestPeer(10, kicked))
clock.Run(5 * time.Minute)
for i := 0; i < 10; i++ {
pool.connect(poolTestPeer(i+10), 1)
pool.connect(newPoolTestPeer(i+10, kicked), 1)
}
for i := 0; i < 10; i++ {
select {
@ -346,18 +355,18 @@ func TestPositiveBalanceCalculation(t *testing.T) {
kicked = make(chan int, 10)
)
removeFn := func(id enode.ID) { kicked <- int(id[0]) } // Noop
pool := newClientPool(db, 1, &clock, removeFn)
pool := newClientPool(db, 1, 1, &clock, removeFn)
defer pool.stop()
pool.setLimits(10, uint64(10)) // Total capacity limit is 10
pool.setDefaultFactors(priceFactors{1, 0, 1}, priceFactors{1, 0, 1})
pool.addBalance(poolTestPeer(0).ID(), int64(time.Minute*3), "")
pool.connect(poolTestPeer(0), 10)
pool.addBalance(newPoolTestPeer(0, kicked).ID(), int64(time.Minute*3), "")
pool.connect(newPoolTestPeer(0, kicked), 10)
clock.Run(time.Minute)
pool.disconnect(poolTestPeer(0))
pb := pool.ndb.getOrNewPB(poolTestPeer(0).ID())
if pb.value != uint64(time.Minute*2) {
pool.disconnect(newPoolTestPeer(0, kicked))
pb := pool.ndb.getOrNewPB(newPoolTestPeer(0, kicked).ID())
if pb.value != expval(uint64(time.Minute*2)) {
t.Fatalf("Positive balance mismatch, want %v, got %v", uint64(time.Minute*2), pb.value)
}
}
@ -369,16 +378,14 @@ func TestDowngradePriorityClient(t *testing.T) {
kicked = make(chan int, 10)
)
removeFn := func(id enode.ID) { kicked <- int(id[0]) } // Noop
pool := newClientPool(db, 1, &clock, removeFn)
pool := newClientPool(db, 1, 1, &clock, removeFn)
defer pool.stop()
pool.setLimits(10, uint64(10)) // Total capacity limit is 10
pool.setDefaultFactors(priceFactors{1, 0, 1}, priceFactors{1, 0, 1})
p := &poolTestPeerWithCap{
poolTestPeer: poolTestPeer(0),
}
p := newPoolTestPeer(0, kicked)
pool.addBalance(p.ID(), int64(time.Minute), "")
pool.connect(p, 10)
p.cap, _ = pool.connect(p, 10)
if p.cap != 10 {
t.Fatalf("The capcacity of priority peer hasn't been updated, got: %d", p.cap)
}
@ -388,14 +395,14 @@ func TestDowngradePriorityClient(t *testing.T) {
if p.cap != 1 {
t.Fatalf("The capcacity of peer should be downgraded, got: %d", p.cap)
}
pb := pool.ndb.getOrNewPB(poolTestPeer(0).ID())
if pb.value != 0 {
pb := pool.ndb.getOrNewPB(newPoolTestPeer(0, kicked).ID())
if pb.value.base != 0 {
t.Fatalf("Positive balance mismatch, want %v, got %v", 0, pb.value)
}
pool.addBalance(poolTestPeer(0).ID(), int64(time.Minute), "")
pb = pool.ndb.getOrNewPB(poolTestPeer(0).ID())
if pb.value != uint64(time.Minute) {
pool.addBalance(newPoolTestPeer(0, kicked).ID(), int64(time.Minute), "")
pb = pool.ndb.getOrNewPB(newPoolTestPeer(0, kicked).ID())
if pb.value != expval(uint64(time.Minute)) {
t.Fatalf("Positive balance mismatch, want %v, got %v", uint64(time.Minute), pb.value)
}
}
@ -404,37 +411,35 @@ func TestNegativeBalanceCalculation(t *testing.T) {
var (
clock mclock.Simulated
db = rawdb.NewMemoryDatabase()
kicked = make(chan int, 10)
)
removeFn := func(id enode.ID) { kicked <- int(id[0]) } // Noop
pool := newClientPool(db, 1, &clock, removeFn)
pool := newClientPool(db, 1, 1, &clock, nil)
defer pool.stop()
pool.setLimits(10, uint64(10)) // Total capacity limit is 10
pool.setDefaultFactors(priceFactors{1, 0, 1}, priceFactors{1, 0, 1})
for i := 0; i < 10; i++ {
pool.connect(poolTestPeer(i), 1)
pool.connect(newPoolTestPeer(i, nil), 1)
}
clock.Run(time.Second)
clock.Run(time.Millisecond * 999)
for i := 0; i < 10; i++ {
pool.disconnect(poolTestPeer(i))
nb := pool.ndb.getOrNewNB(poolTestPeer(i).freeClientId())
pool.disconnect(newPoolTestPeer(i, nil))
nb := pool.ndb.getOrNewNB(newPoolTestPeer(i, nil).freeClientId())
if nb.logValue != 0 {
t.Fatalf("Short connection shouldn't be recorded")
}
}
for i := 0; i < 10; i++ {
pool.connect(poolTestPeer(i), 1)
pool.connect(newPoolTestPeer(i, nil), 1)
}
clock.Run(time.Minute)
for i := 0; i < 10; i++ {
pool.disconnect(poolTestPeer(i))
nb := pool.ndb.getOrNewNB(poolTestPeer(i).freeClientId())
nb.logValue -= pool.logOffset(clock.Now())
nb.logValue /= fixedPointMultiplier
if nb.logValue != int64(math.Log(float64(time.Minute/time.Second))) {
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)
}
}
@ -453,8 +458,8 @@ func TestNodeDB(t *testing.T) {
balance interface{}
positive bool
}{
{enode.ID{0x00, 0x01, 0x02}, "", posBalance{value: 100}, true},
{enode.ID{0x00, 0x01, 0x02}, "", posBalance{value: 200}, true},
{enode.ID{0x00, 0x01, 0x02}, "", posBalance{value: expval(100)}, true},
{enode.ID{0x00, 0x01, 0x02}, "", posBalance{value: expval(200)}, true},
{enode.ID{}, "127.0.0.1", negBalance{logValue: 10}, false},
{enode.ID{}, "127.0.0.1", negBalance{logValue: 20}, false},
}
@ -484,9 +489,9 @@ func TestNodeDB(t *testing.T) {
}
}
}
ndb.setCumulativeTime(100)
if ndb.getCumulativeTime() != 100 {
t.Fatalf("Cumulative time mismatch, want %v, got %v", 100, ndb.getCumulativeTime())
ndb.setExpiration(100, 200)
if pos, neg := ndb.getExpiration(); pos != 100 || neg != 200 {
t.Fatalf("Expiration mismatch, want %v / %v, got %v / %v", 100, 200, pos, neg)
}
}
@ -541,3 +546,83 @@ func TestNodeDBExpiration(t *testing.T) {
t.Fatalf("Failed to evict useless negative balances, want %v, got %d", 4, iterated)
}
}
func TestInactiveClient(t *testing.T) {
var (
clock mclock.Simulated
db = rawdb.NewMemoryDatabase()
)
pool := newClientPool(db, 1, 1, &clock, nil)
defer pool.stop()
pool.setLimits(2, uint64(2)) // Total capacity limit is 10
p1 := newPoolTestPeer(1, nil)
p2 := newPoolTestPeer(2, nil)
p3 := newPoolTestPeer(3, nil)
pool.addBalance(p1.ID(), 1000, "")
pool.addBalance(p3.ID(), 2000, "")
// p1: 1000 p2: 0 p3: 2000
p1.cap, _ = pool.connect(p1, 1)
if p1.cap != 1 {
t.Fatalf("Failed to connect peer #1")
}
p2.cap, _ = pool.connect(p2, 1)
if p2.cap != 1 {
t.Fatalf("Failed to connect peer #2")
}
p3.cap, _ = pool.connect(p3, 1)
if p3.cap != 1 {
t.Fatalf("Failed to connect peer #3")
}
if p2.cap != 0 {
t.Fatalf("Failed to deactivate peer #2")
}
pool.addBalance(p2.ID(), 3000, "")
// p1: 1000 p2: 3000 p3: 2000
if p2.cap != 1 {
t.Fatalf("Failed to activate peer #2")
}
if p1.cap != 0 {
t.Fatalf("Failed to deactivate peer #1")
}
pool.addBalance(p2.ID(), -2500, "")
// p1: 1000 p2: 500 p3: 2000
if p1.cap != 1 {
t.Fatalf("Failed to activate peer #1")
}
if p2.cap != 0 {
t.Fatalf("Failed to deactivate peer #2")
}
pool.setDefaultFactors(priceFactors{1e-9, 0, 0}, priceFactors{1e-9, 0, 0})
p4 := newPoolTestPeer(4, nil)
pool.addBalance(p4.ID(), 1500, "")
// p1: 1000 p2: 500 p3: 2000 p4: 1500
p4.cap, _ = pool.connect(p4, 1)
if p4.cap != 1 {
t.Fatalf("Failed to activate peer #4")
}
if p1.cap != 0 {
t.Fatalf("Failed to deactivate peer #1")
}
clock.Run(time.Second * 600)
// manually trigger a check to avoid a long real-time wait
pool.lock.Lock()
pool.tryActivateClients()
pool.lock.Unlock()
// p1: 1000 p2: 500 p3: 2000 p4: 900
if p1.cap != 1 {
t.Fatalf("Failed to activate peer #1")
}
if p4.cap != 0 {
t.Fatalf("Failed to deactivate peer #4")
}
pool.disconnect(p2)
pool.disconnect(p4)
pool.addBalance(p1.ID(), -1000, "")
if p1.cap != 1 {
t.Fatalf("Should not deactivate peer #1")
}
if p2.cap != 0 {
t.Fatalf("Should not activate peer #2")
}
}

28
les/costtracker.go
View file

@ -159,15 +159,9 @@ func newCostTracker(db ethdb.Database, config *eth.Config) (*costTracker, uint64
}
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
var minRecharge uint64
ct.minBufLimit, minRecharge = costList.decode(ProtocolLengths[ServerProtocolVersions[len(ServerProtocolVersions)-1]]).reqParams()
return ct, minRecharge
}
// stop stops the cost tracker and saves the cost factor statistics to the database
@ -480,6 +474,22 @@ func (table requestCostTable) getMaxCost(code, amount uint64) uint64 {
return costs.baseCost + amount*costs.reqCost
}
func (table requestCostTable) reqParams() (minRecharge, minBufLimit uint64) {
for code, c := range table {
amount := minBufferReqAmount[code]
cost := c.baseCost + amount*c.reqCost
if cost > minBufLimit {
minBufLimit = cost
}
}
minBufLimit *= uint64(minBufferMultiplier)
if minBufLimit < 1 {
minBufLimit = 1
}
minRecharge = (minBufLimit-1)/bufLimitRatio + 1
return
}
// decode converts a cost list to a cost table
func (list RequestCostList) decode(protocolLength uint64) requestCostTable {
table := make(requestCostTable)

8
les/flowcontrol/control.go
View file

@ -185,6 +185,14 @@ func (node *ClientNode) UpdateParams(params ServerParams) {
}
}
// Params returns the current server parameters
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)