gaspersic/go-ethereum
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

les: rework clientpool (#20077)

Browse Source 
* les: rework clientpool
This commit is contained in:
gary rong
2019-11-02 20:02:35 +08:00
committed by Felföldi Zsolt
parent 44b74cfc40
commit 0ce5e113be
9 changed files with 976 additions and 268 deletions
Download Patch File Download Diff File Expand all files Collapse all files

408
les/clientpool_test.go
View File

@ -17,8 +17,11 @@
package les
import (
"bytes"
"fmt"
"math"
"math/rand"
"reflect"
"testing"
"time"
@ -51,7 +54,7 @@ func TestClientPoolL100C300P20(t *testing.T) {
testClientPool(t, 100, 300, 20, false)
}
const testClientPoolTicks = 500000
const testClientPoolTicks = 100000
type poolTestPeer int
@ -76,8 +79,9 @@ 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, 10000, &clock, disconnFn)
pool = newClientPool(db, 1, &clock, disconnFn)
)
pool.disableBias = true
pool.setLimits(connLimit, uint64(connLimit))
pool.setPriceFactors(priceFactors{1, 0, 1}, priceFactors{1, 0, 1})
@ -89,16 +93,9 @@ func testClientPool(t *testing.T, connLimit, clientCount, paidCount int, randomD
t.Fatalf("Test peer #%d rejected", i)
}
}
// since all accepted peers are new and should not be kicked out, the next one should be rejected
if pool.connect(poolTestPeer(connLimit), 0) {
connected[connLimit] = true
t.Fatalf("Peer accepted over connected limit")
}
// randomly connect and disconnect peers, expect to have a similar total connection time at the end
for tickCounter := 0; tickCounter < testClientPoolTicks; tickCounter++ {
clock.Run(1 * time.Second)
//time.Sleep(time.Microsecond * 100)
if tickCounter == testClientPoolTicks/4 {
// give a positive balance to some of the peers
@ -137,11 +134,11 @@ func testClientPool(t *testing.T, connLimit, clientCount, paidCount int, randomD
}
expTicks := testClientPoolTicks/2*connLimit/clientCount + testClientPoolTicks/2*(connLimit-paidCount)/(clientCount-paidCount)
expMin := expTicks - expTicks/10
expMax := expTicks + expTicks/10
expMin := expTicks - expTicks/5
expMax := expTicks + expTicks/5
paidTicks := testClientPoolTicks/2*connLimit/clientCount + testClientPoolTicks/2
paidMin := paidTicks - paidTicks/10
paidMax := paidTicks + paidTicks/10
paidMin := paidTicks - paidTicks/5
paidMax := paidTicks + paidTicks/5
// check if the total connected time of peers are all in the expected range
for i, c := range connected {
@ -157,24 +154,371 @@ func testClientPool(t *testing.T, connLimit, clientCount, paidCount int, randomD
t.Errorf("Total connected time of test node #%d (%d) outside expected range (%d to %d)", i, connTicks[i], min, max)
}
}
// a previously unknown peer should be accepted now
if !pool.connect(poolTestPeer(54321), 0) {
t.Fatalf("Previously unknown peer rejected")
}
// close and restart pool
pool.stop()
pool = newClientPool(db, 1, 10000, &clock, func(id enode.ID) {})
pool.setLimits(connLimit, uint64(connLimit))
// try connecting all known peers (connLimit should be filled up)
for i := 0; i < clientCount; i++ {
pool.connect(poolTestPeer(i), 0)
}
// expect pool to remember known nodes and kick out one of them to accept a new one
if !pool.connect(poolTestPeer(54322), 0) {
t.Errorf("Previously unknown peer rejected after restarting pool")
}
pool.stop()
}
func TestConnectPaidClient(t *testing.T) {
var (
clock mclock.Simulated
db = rawdb.NewMemoryDatabase()
)
pool := newClientPool(db, 1, &clock, nil)
defer pool.stop()
pool.setLimits(10, uint64(10))
pool.setPriceFactors(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, false)
if !pool.connect(poolTestPeer(0), 10) {
t.Fatalf("Failed to connect paid client")
}
}
func TestConnectPaidClientToSmallPool(t *testing.T) {
var (
clock mclock.Simulated
db = rawdb.NewMemoryDatabase()
)
pool := newClientPool(db, 1, &clock, nil)
defer pool.stop()
pool.setLimits(10, uint64(10)) // Total capacity limit is 10
pool.setPriceFactors(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, false)
// Connect a fat paid client to pool, should reject it.
if pool.connect(poolTestPeer(0), 100) {
t.Fatalf("Connected fat paid client, should reject it")
}
}
func TestConnectPaidClientToFullPool(t *testing.T) {
var (
clock mclock.Simulated
db = rawdb.NewMemoryDatabase()
)
removeFn := func(enode.ID) {} // Noop
pool := newClientPool(db, 1, &clock, removeFn)
defer pool.stop()
pool.setLimits(10, uint64(10)) // Total capacity limit is 10
pool.setPriceFactors(priceFactors{1, 0, 1}, priceFactors{1, 0, 1})
for i := 0; i < 10; i++ {
pool.addBalance(poolTestPeer(i).ID(), 1000000000, false)
pool.connect(poolTestPeer(i), 1)
}
pool.addBalance(poolTestPeer(11).ID(), 1000, false) // Add low balance to new paid client
if pool.connect(poolTestPeer(11), 1) {
t.Fatalf("Low balance paid client should be rejected")
}
clock.Run(time.Second)
pool.addBalance(poolTestPeer(12).ID(), 1000000000*60*3, false) // Add high balance to new paid client
if !pool.connect(poolTestPeer(12), 1) {
t.Fatalf("High balance paid client should be accpected")
}
}
func TestPaidClientKickedOut(t *testing.T) {
var (
clock mclock.Simulated
db = rawdb.NewMemoryDatabase()
kickedCh = make(chan int, 1)
)
removeFn := func(id enode.ID) { kickedCh <- int(id[0]) }
pool := newClientPool(db, 1, &clock, removeFn)
defer pool.stop()
pool.setLimits(10, uint64(10)) // Total capacity limit is 10
pool.setPriceFactors(priceFactors{1, 0, 1}, priceFactors{1, 0, 1})
for i := 0; i < 10; i++ {
pool.addBalance(poolTestPeer(i).ID(), 1000000000, false) // 1 second allowance
pool.connect(poolTestPeer(i), 1)
clock.Run(time.Millisecond)
}
clock.Run(time.Second)
clock.Run(connectedBias)
if !pool.connect(poolTestPeer(11), 0) {
t.Fatalf("Free client should be accectped")
}
select {
case id := <-kickedCh:
if id != 0 {
t.Fatalf("Kicked client mismatch, want %v, got %v", 0, id)
}
case <-time.NewTimer(time.Second).C:
t.Fatalf("timeout")
}
}
func TestConnectFreeClient(t *testing.T) {
var (
clock mclock.Simulated
db = rawdb.NewMemoryDatabase()
)
pool := newClientPool(db, 1, &clock, nil)
defer pool.stop()
pool.setLimits(10, uint64(10))
pool.setPriceFactors(priceFactors{1, 0, 1}, priceFactors{1, 0, 1})
if !pool.connect(poolTestPeer(0), 10) {
t.Fatalf("Failed to connect free client")
}
}
func TestConnectFreeClientToFullPool(t *testing.T) {
var (
clock mclock.Simulated
db = rawdb.NewMemoryDatabase()
)
removeFn := func(enode.ID) {} // Noop
pool := newClientPool(db, 1, &clock, removeFn)
defer pool.stop()
pool.setLimits(10, uint64(10)) // Total capacity limit is 10
pool.setPriceFactors(priceFactors{1, 0, 1}, priceFactors{1, 0, 1})
for i := 0; i < 10; i++ {
pool.connect(poolTestPeer(i), 1)
}
if pool.connect(poolTestPeer(11), 1) {
t.Fatalf("New free client should be rejected")
}
clock.Run(time.Minute)
if pool.connect(poolTestPeer(12), 1) {
t.Fatalf("New free client should be rejected")
}
clock.Run(time.Millisecond)
clock.Run(4 * time.Minute)
if !pool.connect(poolTestPeer(13), 1) {
t.Fatalf("Old client connects more than 5min should be kicked")
}
}
func TestFreeClientKickedOut(t *testing.T) {
var (
clock mclock.Simulated
db = rawdb.NewMemoryDatabase()
kicked = make(chan int, 10)
)
removeFn := func(id enode.ID) { kicked <- int(id[0]) }
pool := newClientPool(db, 1, &clock, removeFn)
defer pool.stop()
pool.setLimits(10, uint64(10)) // Total capacity limit is 10
pool.setPriceFactors(priceFactors{1, 0, 1}, priceFactors{1, 0, 1})
for i := 0; i < 10; i++ {
pool.connect(poolTestPeer(i), 1)
clock.Run(100 * time.Millisecond)
}
if pool.connect(poolTestPeer(11), 1) {
t.Fatalf("New free client should be rejected")
}
clock.Run(5 * time.Minute)
for i := 0; i < 10; i++ {
pool.connect(poolTestPeer(i+10), 1)
}
for i := 0; i < 10; i++ {
select {
case id := <-kicked:
if id != i {
t.Fatalf("Kicked client mismatch, want %v, got %v", i, id)
}
case <-time.NewTimer(time.Second).C:
t.Fatalf("timeout")
}
}
}
func TestPositiveBalanceCalculation(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)
defer pool.stop()
pool.setLimits(10, uint64(10)) // Total capacity limit is 10
pool.setPriceFactors(priceFactors{1, 0, 1}, priceFactors{1, 0, 1})
pool.addBalance(poolTestPeer(0).ID(), uint64(time.Minute*3), false)
pool.connect(poolTestPeer(0), 10)
clock.Run(time.Minute)
pool.disconnect(poolTestPeer(0))
pb := pool.ndb.getOrNewPB(poolTestPeer(0).ID())
if pb.value != uint64(time.Minute*2) {
t.Fatalf("Positive balance mismatch, want %v, got %v", uint64(time.Minute*2), pb.value)
}
}
func TestDowngradePriorityClient(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)
defer pool.stop()
pool.setLimits(10, uint64(10)) // Total capacity limit is 10
pool.setPriceFactors(priceFactors{1, 0, 1}, priceFactors{1, 0, 1})
pool.addBalance(poolTestPeer(0).ID(), uint64(time.Minute), false)
pool.connect(poolTestPeer(0), 10)
clock.Run(time.Minute) // All positive balance should be used up.
time.Sleep(300 * time.Millisecond) // Ensure the callback is called
pb := pool.ndb.getOrNewPB(poolTestPeer(0).ID())
if pb.value != 0 {
t.Fatalf("Positive balance mismatch, want %v, got %v", 0, pb.value)
}
pool.addBalance(poolTestPeer(0).ID(), uint64(time.Minute), false)
pb = pool.ndb.getOrNewPB(poolTestPeer(0).ID())
if pb.value != uint64(time.Minute) {
t.Fatalf("Positive balance mismatch, want %v, got %v", uint64(time.Minute), pb.value)
}
}
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)
defer pool.stop()
pool.setLimits(10, uint64(10)) // Total capacity limit is 10
pool.setPriceFactors(priceFactors{1, 0, 1}, priceFactors{1, 0, 1})
for i := 0; i < 10; i++ {
pool.connect(poolTestPeer(i), 1)
}
clock.Run(time.Second)
for i := 0; i < 10; i++ {
pool.disconnect(poolTestPeer(i))
nb := pool.ndb.getOrNewNB(poolTestPeer(i).freeClientId())
if nb.logValue != 0 {
t.Fatalf("Short connection shouldn't be recorded")
}
}
for i := 0; i < 10; i++ {
pool.connect(poolTestPeer(i), 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))) {
t.Fatalf("Negative balance mismatch, want %v, got %v", int64(math.Log(float64(time.Minute/time.Second))), nb.logValue)
}
}
}
func TestNodeDB(t *testing.T) {
ndb := newNodeDB(rawdb.NewMemoryDatabase(), mclock.System{})
defer ndb.close()
if !bytes.Equal(ndb.verbuf[:], []byte{0x00, 0x00}) {
t.Fatalf("version buffer mismatch, want %v, got %v", []byte{0x00, 0x00}, ndb.verbuf)
}
var cases = []struct {
id enode.ID
ip string
balance interface{}
positive bool
}{
{enode.ID{0x00, 0x01, 0x02}, "", posBalance{value: 100, lastTotal: 200}, true},
{enode.ID{0x00, 0x01, 0x02}, "", posBalance{value: 200, lastTotal: 300}, true},
{enode.ID{}, "127.0.0.1", negBalance{logValue: 10}, false},
{enode.ID{}, "127.0.0.1", negBalance{logValue: 20}, false},
}
for _, c := range cases {
if c.positive {
ndb.setPB(c.id, c.balance.(posBalance))
if pb := ndb.getOrNewPB(c.id); !reflect.DeepEqual(pb, c.balance.(posBalance)) {
t.Fatalf("Positive balance mismatch, want %v, got %v", c.balance.(posBalance), pb)
}
} else {
ndb.setNB(c.ip, c.balance.(negBalance))
if nb := ndb.getOrNewNB(c.ip); !reflect.DeepEqual(nb, c.balance.(negBalance)) {
t.Fatalf("Negative balance mismatch, want %v, got %v", c.balance.(negBalance), nb)
}
}
}
for _, c := range cases {
if c.positive {
ndb.delPB(c.id)
if pb := ndb.getOrNewPB(c.id); !reflect.DeepEqual(pb, posBalance{}) {
t.Fatalf("Positive balance mismatch, want %v, got %v", posBalance{}, pb)
}
} else {
ndb.delNB(c.ip)
if nb := ndb.getOrNewNB(c.ip); !reflect.DeepEqual(nb, negBalance{}) {
t.Fatalf("Negative balance mismatch, want %v, got %v", negBalance{}, nb)
}
}
}
ndb.setCumulativeTime(100)
if ndb.getCumulativeTime() != 100 {
t.Fatalf("Cumulative time mismatch, want %v, got %v", 100, ndb.getCumulativeTime())
}
}
func TestNodeDBExpiration(t *testing.T) {
var (
iterated int
done = make(chan struct{}, 1)
)
callback := func(now mclock.AbsTime, b negBalance) bool {
iterated += 1
return true
}
clock := &mclock.Simulated{}
ndb := newNodeDB(rawdb.NewMemoryDatabase(), clock)
defer ndb.close()
ndb.nbEvictCallBack = callback
ndb.cleanupHook = func() { done <- struct{}{} }
var cases = []struct {
ip string
balance negBalance
}{
{"127.0.0.1", negBalance{logValue: 1}},
{"127.0.0.2", negBalance{logValue: 1}},
{"127.0.0.3", negBalance{logValue: 1}},
{"127.0.0.4", negBalance{logValue: 1}},
}
for _, c := range cases {
ndb.setNB(c.ip, c.balance)
}
time.Sleep(100 * time.Millisecond) // Ensure the db expirer is registered.
clock.Run(time.Hour + time.Minute)
select {
case <-done:
case <-time.NewTimer(time.Second).C:
t.Fatalf("timeout")
}
if iterated != 4 {
t.Fatalf("Failed to evict useless negative balances, want %v, got %d", 4, iterated)
}
for _, c := range cases {
ndb.setNB(c.ip, c.balance)
}
clock.Run(time.Hour + time.Minute)
select {
case <-done:
case <-time.NewTimer(time.Second).C:
t.Fatalf("timeout")
}
if iterated != 8 {
t.Fatalf("Failed to evict useless negative balances, want %v, got %d", 4, iterated)
}
}