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: implement inactive peer state

Browse source
This commit is contained in:
Zsolt Felfoldi 2019-12-20 18:47:46 +01:00
parent 21f760b68e
commit 0fc06ea8af
14 changed files with 600 additions and 306 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
les/balance.go
View file

@ -68,6 +68,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
@ -99,7 +100,7 @@ func (bt *balanceTracker) stop(now mclock.AbsTime) {
// 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)
return -int64(b.pos / bt.capacity)
}
return int64(b.neg)
}
@ -107,7 +108,7 @@ func (bt *balanceTracker) balanceToPriority(b balance) int64 {
func (bt *balanceTracker) posBalanceMissing(targetPriority int64, targetCapacity uint64, after time.Duration) uint64 {
if targetPriority > 0 {
negPrice := uint64(float64(after) * bt.negTimeFactor)
if negPrice+bt.balance.neg <= uint64(targetPriority) {
if negPrice+bt.balance.neg < uint64(targetPriority) {
return 0
}
if uint64(targetPriority) > bt.balance.neg && bt.negTimeFactor > 1e-100 {
@ -119,7 +120,7 @@ func (bt *balanceTracker) posBalanceMissing(targetPriority int64, targetCapacity
}
targetPriority = 0
}
posRequired := uint64(float64(^targetPriority)*float64(targetCapacity) + float64(after)*bt.timeFactor)
posRequired := uint64(float64(-targetPriority)*float64(targetCapacity)+float64(after)*bt.timeFactor) + 1
if posRequired >= maxBalance {
return math.MaxUint64 // target not reachable
}
@ -164,7 +165,7 @@ func (bt *balanceTracker) timeUntil(priority int64) (time.Duration, bool) {
return 0, false
}
if priority < 0 {
newBalance := uint64(^priority) * bt.capacity
newBalance := uint64(-priority) * bt.capacity
if newBalance > bt.balance.pos {
return 0, false
}
@ -189,6 +190,7 @@ func (bt *balanceTracker) timeUntil(priority int64) (time.Duration, bool) {
}
// 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()

14
les/balance_test.go
View file

@ -141,8 +141,8 @@ 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},
}
@ -172,16 +172,16 @@ func TestEstimatedPriority(t *testing.T) {
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},
@ -213,7 +213,7 @@ 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},
}

44
les/client_handler.go
View file

@ -116,28 +116,48 @@ func (h *clientHandler) handle(p *peer) error {
p.Log().Debug("Light Ethereum handshake failed", "err", err)
return err
}
// Register the peer locally
if err := h.backend.peers.Register(p); err != nil {
p.Log().Error("Light Ethereum peer registration failed", "err", err)
return err
}
serverConnectionGauge.Update(int64(h.backend.peers.Len()))
connectedAt := mclock.Now()
defer func() {
h.backend.peers.Unregister(p.id)
var (
connectedAt mclock.AbsTime
lastActive bool
)
activate := func() {
// Register the peer locally
if err := h.backend.peers.Register(p); err != nil {
p.Log().Error("Light Ethereum peer registration failed", "err", err)
return
}
serverConnectionGauge.Update(int64(h.backend.peers.Len()))
connectedAt = mclock.Now()
h.fetcher.announce(p, p.headInfo)
lastActive = true
}
deactivate := func() {
h.backend.peers.Unregister(p)
connectionTimer.Update(time.Duration(mclock.Now() - connectedAt))
serverConnectionGauge.Update(int64(h.backend.peers.Len()))
lastActive = false
}
defer func() {
if lastActive {
deactivate()
}
h.backend.peers.Disconnect(p.id)
}()
h.fetcher.announce(p, p.headInfo)
// pool entry can be nil during the unit test.
if p.poolEntry != nil {
h.backend.serverPool.registered(p.poolEntry)
}
// Spawn a main loop to handle all incoming messages.
for {
if p.active && !lastActive {
activate()
}
if !p.active && lastActive {
deactivate()
}
if err := h.handleMsg(p); err != nil {
p.Log().Debug("Light Ethereum message handling failed", "err", err)
p.fcServer.DumpLogs()
@ -400,7 +420,7 @@ func (h *clientHandler) makeLespayCall(p *peer, cmd []byte, handler func([]byte)
}
func (h *clientHandler) removePeer(id string) {
h.backend.peers.Unregister(id)
h.backend.peers.Disconnect(id)
}
type peerConnection struct {

322
les/clientpool.go
View file

@ -39,19 +39,20 @@ const (
negBalanceExpTC = time.Hour // time constant for exponentially reducing negative balance
fixedPointMultiplier = 0x1000000 // constant to convert logarithms to fixed point format
lazyQueueRefresh = time.Second * 10 // refresh period of the connected queue
tryActivatePeriod = time.Second * 5 // periodically check whether inactive clients can be activated
persistCumulativeTimeRefresh = time.Minute * 5 // refresh period of the cumulative running time persistence
posBalanceCacheLimit = 8192 // the maximum number of cached items in positive balance queue
negBalanceCacheLimit = 8192 // the maximum number of cached items in negative balance queue
fullRatioTC = time.Hour
// connectedBias is applied to already connected clients So that
// activeBias is applied to already connected clients So that
// already connected client won't be kicked out very soon and we
// can ensure all connected clients can have enough time to request
// or sync some data.
//
// todo(rjl493456442) make it configurable. It can be the option of
// free trial time!
connectedBias = time.Minute * 3
activeBias = time.Minute * 3
)
// clientPool implements a client database that assigns a priority to each client
@ -61,7 +62,7 @@ const (
// then negative balance is accumulated.
//
// Balance tracking and priority calculation for connected clients is done by
// balanceTracker. connectedQueue ensures that clients with the lowest positive or
// balanceTracker. activeQueue ensures that clients with the lowest positive or
// highest negative balance get evicted when the total capacity allowance is full
// and new clients with a better balance want to connect.
//
@ -83,24 +84,25 @@ type clientPool struct {
closed bool
removePeer func(enode.ID)
connectedMap map[enode.ID]*clientInfo
connectedQueue *prque.LazyQueue
connectedMap map[enode.ID]*clientInfo
activeQueue *prque.LazyQueue
inactiveQueue *prque.Prque
connectedBalances, disconnectedBalances uint64
activeBalances, inactiveBalances uint64
lastConnectedBalanceUpdate, fullRatioLastUpdate mclock.AbsTime
fullRatio float64
defaultPosFactors, defaultNegFactors priceFactors
connLimit int // The maximum number of connections that clientpool can support
capLimit uint64 // The maximum cumulative capacity that clientpool can support
connectedCap uint64 // The sum of the capacity of the current clientpool connected
priorityConnected uint64 // The sum of the capacity of currently connected priority clients
minCap uint64 // The minimal capacity value allowed for any client
freeClientCap uint64 // The capacity value of each free client
startTime mclock.AbsTime // The timestamp at which the clientpool started running
cumulativeTime int64 // The cumulative running time of clientpool at the start point.
disableBias bool // Disable connection bias(used in testing)
activeLimit int // The maximum number of connections that clientpool can support
capLimit uint64 // The maximum cumulative capacity that clientpool can support
activeCap uint64 // The sum of the capacity of the current clientpool connected
priorityActive uint64 // The sum of the capacity of currently connected priority clients
minCap uint64 // The minimal capacity value allowed for any client
freeClientCap uint64 // The capacity value of each free client
startTime mclock.AbsTime // The timestamp at which the clientpool started running
cumulativeTime int64 // The cumulative running time of clientpool at the start point.
disableBias bool // Disable connection bias(used in testing)
}
// clientPeer represents a client in the pool.
@ -120,36 +122,37 @@ type clientInfo struct {
address string
id enode.ID
freeID string
active bool
connectedAt mclock.AbsTime
capacity uint64
priority bool
pool *clientPool
peer clientPeer
queueIndex int // position in connectedQueue
queueIndex int // position in activeQueue
balanceTracker balanceTracker
posFactors, negFactors priceFactors
balanceMetaInfo string
}
// connSetIndex callback updates clientInfo item index in connectedQueue
// connSetIndex callback updates clientInfo item index in activeQueue
func connSetIndex(a interface{}, index int) {
a.(*clientInfo).queueIndex = index
}
// connPriority callback returns actual priority of clientInfo item in connectedQueue
// connPriority callback returns actual priority of clientInfo item in activeQueue
func connPriority(a interface{}, now mclock.AbsTime) int64 {
c := a.(*clientInfo)
return c.balanceTracker.getPriority(now)
}
// connMaxPriority callback returns estimated maximum priority of clientInfo item in connectedQueue
// connMaxPriority callback returns estimated maximum priority of clientInfo item in activeQueue
func connMaxPriority(a interface{}, until mclock.AbsTime) int64 {
c := a.(*clientInfo)
pri := c.balanceTracker.estimatedPriority(until, true)
c.balanceTracker.addCallback(balanceCallbackQueue, pri+1, func() {
c.pool.lock.Lock()
if c.queueIndex != -1 {
c.pool.connectedQueue.Update(c.queueIndex)
if c.active && c.queueIndex != -1 {
c.pool.activeQueue.Update(c.queueIndex)
}
c.pool.lock.Unlock()
})
@ -172,7 +175,8 @@ func newClientPool(db ethdb.Database, minCap, freeClientCap uint64, clock mclock
ndb: ndb,
clock: clock,
connectedMap: make(map[enode.ID]*clientInfo),
connectedQueue: prque.NewLazyQueue(connSetIndex, connPriority, connMaxPriority, clock, lazyQueueRefresh),
activeQueue: prque.NewLazyQueue(connSetIndex, connPriority, connMaxPriority, clock, lazyQueueRefresh),
inactiveQueue: prque.New(connSetIndex),
minCap: minCap,
freeClientCap: freeClientCap,
removePeer: removePeer,
@ -193,7 +197,7 @@ func newClientPool(db ethdb.Database, minCap, freeClientCap uint64, clock mclock
stop = true
}
for i := 0; i < l; i++ {
pool.disconnectedBalances += pool.ndb.getOrNewPB(ids[i]).value
pool.inactiveBalances += pool.ndb.getOrNewPB(ids[i]).value
}
if stop {
break
@ -210,8 +214,16 @@ func newClientPool(db ethdb.Database, minCap, freeClientCap uint64, clock mclock
select {
case <-clock.After(lazyQueueRefresh):
pool.lock.Lock()
pool.connectedQueue.Refresh()
pool.activeQueue.Refresh()
pool.lock.Unlock()
case <-pool.stopCh:
return
}
}
}()
go func() {
for {
select {
case <-clock.After(persistCumulativeTimeRefresh):
pool.ndb.setCumulativeTime(pool.logOffset(clock.Now()))
case <-pool.stopCh:
@ -219,6 +231,18 @@ func newClientPool(db ethdb.Database, minCap, freeClientCap uint64, clock mclock
}
}
}()
go func() {
for {
select {
case <-clock.After(tryActivatePeriod):
pool.lock.Lock()
pool.tryActivateClients()
pool.lock.Unlock()
case <-pool.stopCh:
return
}
}
}()
return pool
}
@ -234,8 +258,8 @@ func (f *clientPool) stop() {
func (f *clientPool) updateFullRatio() {
full := float64(1)
if f.priorityConnected < f.capLimit {
freeCap := f.capLimit - f.priorityConnected
if f.priorityActive < f.capLimit {
freeCap := f.capLimit - f.priorityActive
if freeCap > f.freeClientCap {
freeCapThreshold := f.capLimit / 4
if freeCap > freeCapThreshold {
@ -275,43 +299,37 @@ func (f *clientPool) totalTokenAmount() uint64 {
now := f.clock.Now()
if now > f.lastConnectedBalanceUpdate+mclock.AbsTime(time.Second) {
f.connectedBalances = 0
f.activeBalances = 0
for _, c := range f.connectedMap {
pos, _ := c.balanceTracker.getBalance(now)
f.connectedBalances += pos
f.activeBalances += pos
}
f.lastConnectedBalanceUpdate = now
}
return f.connectedBalances + f.disconnectedBalances
return f.activeBalances + f.inactiveBalances
}
// connect should be called after a successful handshake. If the connection was
// rejected, there is no need to call disconnect.
func (f *clientPool) connect(peer clientPeer, capacity uint64) bool {
func (f *clientPool) connect(peer clientPeer, reqCapacity uint64) (uint64, error) {
f.lock.Lock()
defer f.lock.Unlock()
// Short circuit if clientPool is already closed.
if f.closed {
return false
return 0, fmt.Errorf("Client pool is already closed")
}
// Dedup connected peers.
id, freeID := peer.ID(), peer.freeClientId()
if _, ok := f.connectedMap[id]; ok {
clientRejectedMeter.Mark(1)
log.Debug("Client already connected", "address", freeID, "id", peerIdToString(id))
return false
return 0, fmt.Errorf("Client already connected address = %s id = %s", freeID, peerIdToString(id))
}
var missing uint64
missing, capacity = f.capAvailable(id, freeID, capacity, 0, true)
if missing != 0 {
return false
}
// capacity is available, create and add client
pb := f.ndb.getOrNewPB(id)
nb := f.ndb.getOrNewNB(freeID)
e := &clientInfo{
capacity: capacity,
capacity: reqCapacity,
pool: f,
peer: peer,
address: freeID,
@ -324,33 +342,38 @@ func (f *clientPool) connect(peer clientPeer, capacity uint64) bool {
negFactors: f.defaultNegFactors,
balanceMetaInfo: pb.meta,
}
f.initBalanceTracker(&e.balanceTracker, pb, nb, capacity)
// Register new client to connection queue.
f.disconnectedBalances -= pb.value
f.connectedBalances += pb.value
missing, capacity := f.capAvailable(id, freeID, reqCapacity, 0, true)
f.connectedMap[id] = e
f.connectedQueue.Push(e)
f.connectedCap += e.capacity
if missing != 0 {
// capacity is not available, add client to inactive queue
f.initBalanceTracker(&e.balanceTracker, pb, nb, capacity, false)
f.inactiveQueue.Push(e, -connPriority(e, f.clock.Now()))
return 0, nil
}
// capacity is available, add client
e.active = true
e.capacity = capacity
f.initBalanceTracker(&e.balanceTracker, pb, nb, capacity, true)
// Register new client to connection queue.
f.inactiveBalances -= pb.value
f.activeBalances += pb.value
f.activeQueue.Push(e)
f.activeCap += e.capacity
// If the current client is a paid client, monitor the status of client,
// downgrade it to normal client if positive balance is used up.
if e.priority {
f.updateFullRatio()
f.priorityConnected += capacity
f.priorityActive += capacity
e.balanceTracker.addCallback(balanceCallbackZero, 0, func() { f.balanceExhausted(id) })
}
// If the capacity of client is not the default value(free capacity), notify
// it to update capacity.
if e.capacity != f.freeClientCap {
e.peer.updateCapacity(e.capacity)
}
totalConnectedGauge.Update(int64(f.connectedCap))
totalConnectedGauge.Update(int64(f.activeCap))
clientConnectedMeter.Mark(1)
log.Debug("Client accepted", "address", freeID)
return true
return e.capacity, nil
}
func (f *clientPool) initBalanceTracker(bt *balanceTracker, pb posBalance, nb negBalance, capacity uint64) {
func (f *clientPool) initBalanceTracker(bt *balanceTracker, pb posBalance, nb negBalance, capacity uint64, active bool) {
posBalance := pb.value
var negBalance uint64
if nb.logValue != 0 {
@ -358,7 +381,11 @@ func (f *clientPool) initBalanceTracker(bt *balanceTracker, pb posBalance, nb ne
}
bt.init(f.clock, capacity)
bt.setBalance(posBalance, negBalance)
updatePriceFactors(bt, f.defaultPosFactors, f.defaultNegFactors, capacity)
if active {
updatePriceFactors(bt, f.defaultPosFactors, f.defaultNegFactors, capacity)
} else {
zeroPriceFactors(bt)
}
}
// disconnect should be called when a connection is terminated. If the disconnection
@ -372,13 +399,20 @@ func (f *clientPool) disconnect(p clientPeer) {
if f.closed {
return
}
// Short circuit if the peer hasn't been registered.
e := f.connectedMap[p.ID()]
if e == nil {
e, ok := f.connectedMap[p.ID()]
if !ok {
log.Debug("Client not connected", "address", p.freeClientId(), "id", peerIdToString(p.ID()))
return
}
f.dropClient(e, f.clock.Now(), false)
if e.active {
f.deactivateClient(e)
}
f.finalizeBalance(e, f.clock.Now())
f.inactiveQueue.Remove(e.queueIndex)
delete(f.connectedMap, e.id)
clientDisconnectedMeter.Mark(1)
log.Debug("Client disconnected", "address", e.address)
f.tryActivateClients()
}
// capAvailable checks whether the current priority level of the given client is enough to
@ -392,19 +426,19 @@ func (f *clientPool) capAvailable(id enode.ID, freeID string, capacity uint64, m
if capacity < f.minCap {
capacity = f.minCap
}
newCapacity := f.connectedCap + capacity
newCount := f.connectedQueue.Size() + 1
newCapacity := f.activeCap + capacity
newCount := f.activeQueue.Size() + 1
client := f.connectedMap[id]
if client != nil {
if client != nil && client.active {
newCapacity -= client.capacity
newCount--
}
if newCapacity > f.capLimit || newCount > f.connLimit {
if newCapacity > f.capLimit || newCount > f.activeLimit {
var (
popList []*clientInfo
targetPriority int64
)
f.connectedQueue.MultiPop(func(data interface{}, priority int64) bool {
f.activeQueue.MultiPop(func(data interface{}, priority int64) bool {
c := data.(*clientInfo)
popList = append(popList, c)
if c != client {
@ -412,9 +446,9 @@ func (f *clientPool) capAvailable(id enode.ID, freeID string, capacity uint64, m
newCapacity -= c.capacity
newCount--
}
return newCapacity > f.capLimit || newCount > f.connLimit
return newCapacity > f.capLimit || newCount > f.activeLimit
})
if newCapacity > f.capLimit || newCount > f.connLimit {
if newCapacity > f.capLimit || newCount > f.activeLimit {
missing = math.MaxUint64
} else {
var bt *balanceTracker
@ -422,12 +456,12 @@ func (f *clientPool) capAvailable(id enode.ID, freeID string, capacity uint64, m
bt = &client.balanceTracker
} else {
bt = &balanceTracker{}
f.initBalanceTracker(bt, f.ndb.getOrNewPB(id), f.ndb.getOrNewNB(freeID), capacity)
f.initBalanceTracker(bt, f.ndb.getOrNewPB(id), f.ndb.getOrNewNB(freeID), capacity, true)
}
if capacity != f.freeClientCap && targetPriority >= -1 {
targetPriority = -2
if capacity != f.freeClientCap && targetPriority >= 0 {
targetPriority = -1
}
bias := connectedBias
bias := activeBias
if f.disableBias {
bias = 0
}
@ -441,9 +475,9 @@ func (f *clientPool) capAvailable(id enode.ID, freeID string, capacity uint64, m
}
for _, c := range popList {
if kick && c != client {
f.dropClient(c, f.clock.Now(), true)
f.deactivateClient(c)
} else {
f.connectedQueue.Push(c)
f.activeQueue.Push(c)
}
}
}
@ -482,28 +516,56 @@ func (f *clientPool) setDefaultFactors(posFactors, negFactors priceFactors) {
f.defaultNegFactors = negFactors
}
// dropClient removes a client from the connected queue and finalizes its balance.
// If kick is true then it also initiates the disconnection.
func (f *clientPool) dropClient(e *clientInfo, now mclock.AbsTime, kick bool) {
if _, ok := f.connectedMap[e.id]; !ok {
func (f *clientPool) deactivateClient(e *clientInfo) {
if _, ok := f.connectedMap[e.id]; !ok || !e.active {
return
}
f.finalizeBalance(e, now)
f.connectedQueue.Remove(e.queueIndex)
delete(f.connectedMap, e.id)
f.connectedCap -= e.capacity
f.activeQueue.Remove(e.queueIndex)
f.activeCap -= e.capacity
if e.priority {
f.updateFullRatio()
f.priorityConnected -= e.capacity
f.priorityActive -= e.capacity
}
totalConnectedGauge.Update(int64(f.connectedCap))
if kick {
clientKickedMeter.Mark(1)
log.Debug("Client kicked out", "address", e.address)
f.removePeer(e.id)
} else {
clientDisconnectedMeter.Mark(1)
log.Debug("Client disconnected", "address", e.address)
e.active = false
e.peer.updateCapacity(0)
totalConnectedGauge.Update(int64(f.activeCap))
f.inactiveQueue.Push(e, -connPriority(e, f.clock.Now()))
//TODO start timer
}
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)
if missing != 0 {
f.inactiveQueue.Push(e, -connPriority(e, now))
return
}
// capacity is available, activate client
e.active = true
e.capacity = capacity
e.peer.updateCapacity(capacity)
balance, _ := e.balanceTracker.getBalance(now)
e.balanceTracker.setCapacity(capacity)
updatePriceFactors(&e.balanceTracker, f.defaultPosFactors, f.defaultNegFactors, capacity)
// Register activated client to connection queue.
f.inactiveBalances -= balance
f.activeBalances += balance
f.activeQueue.Push(e)
f.activeCap += e.capacity
// If the current client is a paid client, monitor the status of client,
// downgrade it to normal client if positive balance is used up.
if e.priority {
f.updateFullRatio()
f.priorityActive += capacity
e.balanceTracker.addCallback(balanceCallbackZero, 0, func() { f.balanceExhausted(e.id) })
}
e.peer.updateCapacity(e.capacity)
totalConnectedGauge.Update(int64(f.activeCap))
clientConnectedMeter.Mark(1)
log.Debug("Client activated", "address", e.freeID)
}
}
@ -513,7 +575,7 @@ func (f *clientPool) capacityInfo() (uint64, uint64, uint64) {
f.lock.Lock()
defer f.lock.Unlock()
return f.capLimit, f.connectedCap, f.priorityConnected
return f.capLimit, f.activeCap, f.priorityActive
}
// finalizeBalance stops the balance tracker, retrieves the final balances and
@ -521,8 +583,8 @@ func (f *clientPool) capacityInfo() (uint64, uint64, uint64) {
func (f *clientPool) finalizeBalance(c *clientInfo, now mclock.AbsTime) {
c.balanceTracker.stop(now)
pos, neg := c.balanceTracker.getBalance(now)
f.disconnectedBalances += pos
f.connectedBalances -= pos
f.inactiveBalances += pos
f.activeBalances -= pos
pb, nb := f.ndb.getOrNewPB(c.id), f.ndb.getOrNewNB(c.address)
pb.value = pos
@ -549,12 +611,12 @@ func (f *clientPool) balanceExhausted(id enode.ID) {
}
if c.priority {
f.updateFullRatio()
f.priorityConnected -= c.capacity
f.priorityActive -= c.capacity
}
c.priority = false
if c.capacity != f.freeClientCap {
f.connectedCap += f.freeClientCap - c.capacity
totalConnectedGauge.Update(int64(f.connectedCap))
f.activeCap += f.freeClientCap - c.capacity
totalConnectedGauge.Update(int64(f.activeCap))
c.capacity = f.freeClientCap
c.balanceTracker.setCapacity(c.capacity)
c.peer.updateCapacity(c.capacity)
@ -564,20 +626,22 @@ func (f *clientPool) balanceExhausted(id enode.ID) {
f.ndb.setPB(id, pb)
}
// setConnLimit sets the maximum number and total capacity of connected clients,
// setactiveLimit sets the maximum number and total capacity of connected clients,
// dropping some of them if necessary.
func (f *clientPool) setLimits(totalConn int, totalCap uint64) {
f.lock.Lock()
defer f.lock.Unlock()
f.updateFullRatio()
f.connLimit = totalConn
f.activeLimit = totalConn
f.capLimit = totalCap
if f.connectedCap > f.capLimit || f.connectedQueue.Size() > f.connLimit {
f.connectedQueue.MultiPop(func(data interface{}, priority int64) bool {
f.dropClient(data.(*clientInfo), mclock.Now(), true)
return f.connectedCap > f.capLimit || f.connectedQueue.Size() > f.connLimit
if f.activeCap > f.capLimit || f.activeQueue.Size() > f.activeLimit {
f.activeQueue.MultiPop(func(data interface{}, priority int64) bool {
f.deactivateClient(data.(*clientInfo))
return f.activeCap > f.capLimit || f.activeQueue.Size() > f.activeLimit
})
} else {
f.tryActivateClients()
}
}
@ -599,15 +663,16 @@ func (f *clientPool) setCapacity(id enode.ID, freeID string, capacity uint64, mi
if c == nil {
return 0, capacity, fmt.Errorf("client %064x is not connected", c.id[:])
}
f.connectedCap += capacity - c.capacity
f.activeCap += capacity - c.capacity
f.updateFullRatio()
f.priorityConnected += capacity - c.capacity
f.priorityActive += capacity - c.capacity
c.capacity = capacity
c.balanceTracker.setCapacity(capacity)
f.connectedQueue.Update(c.queueIndex)
totalConnectedGauge.Update(int64(f.connectedCap))
f.activeQueue.Update(c.queueIndex)
totalConnectedGauge.Update(int64(f.activeCap))
updatePriceFactors(&c.balanceTracker, c.posFactors, c.negFactors, c.capacity)
c.peer.updateCapacity(c.capacity)
f.tryActivateClients()
}
return 0, capacity, nil
}
@ -625,11 +690,11 @@ func (f *clientPool) requestCost(p *peer, cost uint64) uint64 {
f.lock.Lock()
defer f.lock.Unlock()
info, exist := f.connectedMap[p.ID()]
if !exist || f.closed {
c := f.connectedMap[p.ID()]
if c == nil || f.closed {
return 0
}
return info.balanceTracker.requestCost(cost)
return c.balanceTracker.requestCost(cost)
}
// logOffset calculates the time-dependent offset for the logarithmic
@ -650,6 +715,11 @@ func updatePriceFactors(bt *balanceTracker, posFactors, negFactors priceFactors,
bt.setFactors(false, posFactors.timeFactor+float64(capacity)*posFactors.capacityFactor/1000000, posFactors.requestFactor)
}
func zeroPriceFactors(bt *balanceTracker) {
bt.setFactors(true, 0, 0)
bt.setFactors(false, 0, 0)
}
// getPosBalance retrieves a single positive balance entry from cache or the database
func (f *clientPool) getPosBalance(id enode.ID) posBalance {
f.lock.Lock()
@ -692,22 +762,32 @@ func (f *clientPool) addBalance(id enode.ID, amount int64, meta string) (uint64,
f.ndb.setPB(id, pb)
if c != nil {
c.balanceTracker.setBalance(pb.value, negBalance)
if !c.priority && pb.value > 0 {
// The capacity should be adjusted based on the requirement,
// but we have no idea about the new capacity, need a second
// call to udpate it.
f.updateFullRatio()
c.priority = true
f.priorityConnected += c.capacity
c.balanceTracker.addCallback(balanceCallbackZero, 0, func() { f.balanceExhausted(id) })
if c.active {
f.activeQueue.Update(c.queueIndex)
if !c.priority && pb.value > 0 {
// The capacity should be adjusted based on the requirement,
// but we have no idea about the new capacity, need a second
// call to udpate it.
f.updateFullRatio()
f.priorityActive += c.capacity
c.balanceTracker.addCallback(balanceCallbackZero, 0, func() { f.balanceExhausted(id) })
}
c.balanceMetaInfo = meta
f.activeBalances += pb.value - oldBalance
} else {
f.inactiveQueue.Remove(c.queueIndex)
f.inactiveQueue.Push(c, -connPriority(c, f.clock.Now()))
f.inactiveBalances += pb.value - oldBalance
}
if pb.value > 0 {
c.priority = true
// if balance is set to zero then reverting to non-priority status
// is handled by the balanceExhausted callback
}
// if balance is set to zero then reverting to non-priority status
// is handled by the balanceExhausted callback
c.balanceMetaInfo = meta
f.connectedBalances += pb.value - oldBalance
} else {
f.disconnectedBalances += pb.value - oldBalance
f.inactiveBalances += pb.value - oldBalance
}
f.tryActivateClients()
return oldBalance, pb.value, nil
}

223
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)}
type poolTestPeer struct {
index int
disconnCh chan int
cap uint64
}
func (i poolTestPeer) freeClientId() string {
func newPoolTestPeer(i int, disconnCh chan int) *poolTestPeer {
return &poolTestPeer{index: i, disconnCh: disconnCh}
}
func (i *poolTestPeer) ID() enode.ID {
return enode.ID{byte(i.index % 256), byte(i.index >> 8)}
}
func (i *poolTestPeer) freeClientId() string {
return fmt.Sprintf("addr #%d", i)
}
func (i poolTestPeer) updateCapacity(uint64) {}
type poolTestPeerWithCap struct {
poolTestPeer
cap uint64
func (i *poolTestPeer) updateCapacity(cap uint64) {
i.cap = cap
if cap == 0 && i.disconnCh != nil {
i.disconnCh <- i.index
}
}
func (i *poolTestPeerWithCap) updateCapacity(cap uint64) { i.cap = cap }
func (i *poolTestPeer) freezeClient() {}
func (i poolTestPeer) freezeClient() {}
func testClientPool(t *testing.T, connLimit, clientCount, paidCount int, randomDisconnect bool) {
func testClientPool(t *testing.T, activeLimit, clientCount, paidCount int, randomDisconnect bool) {
rand.Seed(time.Now().UnixNano())
var (
clock mclock.Simulated
@ -91,13 +96,14 @@ func testClientPool(t *testing.T, connLimit, clientCount, paidCount int, randomD
}
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
@ -178,9 +186,9 @@ func TestConnectPaidClient(t *testing.T) {
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")
}
}
@ -196,10 +204,10 @@ func TestConnectPaidClientToSmallPool(t *testing.T) {
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")
}
}
@ -216,17 +224,17 @@ func TestConnectPaidClientToFullPool(t *testing.T) {
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")
}
}
@ -243,13 +251,13 @@ func TestPaidClientKickedOut(t *testing.T) {
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 {
@ -271,7 +279,7 @@ func TestConnectFreeClient(t *testing.T) {
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")
}
}
@ -288,18 +296,18 @@ func TestConnectFreeClientToFullPool(t *testing.T) {
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")
}
}
@ -317,15 +325,16 @@ func TestFreeClientKickedOut(t *testing.T) {
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 {
@ -351,12 +360,12 @@ func TestPositiveBalanceCalculation(t *testing.T) {
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())
pool.disconnect(newPoolTestPeer(0, kicked))
pb := pool.ndb.getOrNewPB(newPoolTestPeer(0, kicked).ID())
if pb.value != uint64(time.Minute*2) {
t.Fatalf("Positive balance mismatch, want %v, got %v", uint64(time.Minute*2), pb.value)
}
@ -374,11 +383,9 @@ func TestDowngradePriorityClient(t *testing.T) {
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,13 +395,13 @@ 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())
pb := pool.ndb.getOrNewPB(newPoolTestPeer(0, kicked).ID())
if pb.value != 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())
pool.addBalance(newPoolTestPeer(0, kicked).ID(), int64(time.Minute), "")
pb = pool.ndb.getOrNewPB(newPoolTestPeer(0, kicked).ID())
if pb.value != uint64(time.Minute) {
t.Fatalf("Positive balance mismatch, want %v, got %v", uint64(time.Minute), pb.value)
}
@ -402,36 +409,34 @@ func TestDowngradePriorityClient(t *testing.T) {
func TestNegativeBalanceCalculation(t *testing.T) {
var (
clock mclock.Simulated
db = rawdb.NewMemoryDatabase()
kicked = make(chan int, 10)
clock mclock.Simulated
db = rawdb.NewMemoryDatabase()
)
removeFn := func(id enode.ID) { kicked <- int(id[0]) } // Noop
pool := newClientPool(db, 1, 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)
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())
pool.disconnect(newPoolTestPeer(i, nil))
nb := pool.ndb.getOrNewNB(newPoolTestPeer(i, nil).freeClientId())
nb.logValue -= pool.logOffset(clock.Now())
nb.logValue /= fixedPointMultiplier
if nb.logValue != int64(math.Log(float64(time.Minute/time.Second))) {
@ -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)

22
les/handler_test.go
View file

@ -60,7 +60,7 @@ func TestGetBlockHeadersLes3(t *testing.T) { testGetBlockHeaders(t, 3) }
func TestGetBlockHeadersLes4(t *testing.T) { testGetBlockHeaders(t, 4) }
func testGetBlockHeaders(t *testing.T, protocol int) {
server, tearDown := newServerEnv(t, downloader.MaxHashFetch+15, protocol, nil, false, true, 0)
server, tearDown := newServerEnv(t, downloader.MaxHashFetch+15, protocol, nil, false, true, 0, true)
defer tearDown()
bc := server.handler.blockchain
@ -190,7 +190,7 @@ func TestGetBlockBodiesLes3(t *testing.T) { testGetBlockBodies(t, 3) }
func TestGetBlockBodiesLes4(t *testing.T) { testGetBlockBodies(t, 4) }
func testGetBlockBodies(t *testing.T, protocol int) {
server, tearDown := newServerEnv(t, downloader.MaxBlockFetch+15, protocol, nil, false, true, 0)
server, tearDown := newServerEnv(t, downloader.MaxBlockFetch+15, protocol, nil, false, true, 0, true)
defer tearDown()
bc := server.handler.blockchain
@ -269,7 +269,7 @@ func TestGetCodeLes4(t *testing.T) { testGetCode(t, 4) }
func testGetCode(t *testing.T, protocol int) {
// Assemble the test environment
server, tearDown := newServerEnv(t, 4, protocol, nil, false, true, 0)
server, tearDown := newServerEnv(t, 4, protocol, nil, false, true, 0, true)
defer tearDown()
bc := server.handler.blockchain
@ -299,7 +299,7 @@ func TestGetStaleCodeLes3(t *testing.T) { testGetStaleCode(t, 3) }
func TestGetStaleCodeLes4(t *testing.T) { testGetStaleCode(t, 4) }
func testGetStaleCode(t *testing.T, protocol int) {
server, tearDown := newServerEnv(t, core.TriesInMemory+4, protocol, nil, false, true, 0)
server, tearDown := newServerEnv(t, core.TriesInMemory+4, protocol, nil, false, true, 0, true)
defer tearDown()
bc := server.handler.blockchain
@ -325,7 +325,7 @@ func TestGetReceiptLes4(t *testing.T) { testGetReceipt(t, 4) }
func testGetReceipt(t *testing.T, protocol int) {
// Assemble the test environment
server, tearDown := newServerEnv(t, 4, protocol, nil, false, true, 0)
server, tearDown := newServerEnv(t, 4, protocol, nil, false, true, 0, true)
defer tearDown()
bc := server.handler.blockchain
@ -353,7 +353,7 @@ func TestGetProofsLes4(t *testing.T) { testGetProofs(t, 4) }
func testGetProofs(t *testing.T, protocol int) {
// Assemble the test environment
server, tearDown := newServerEnv(t, 4, protocol, nil, false, true, 0)
server, tearDown := newServerEnv(t, 4, protocol, nil, false, true, 0, true)
defer tearDown()
bc := server.handler.blockchain
@ -388,7 +388,7 @@ func TestGetStaleProofLes3(t *testing.T) { testGetStaleProof(t, 3) }
func TestGetStaleProofLes4(t *testing.T) { testGetStaleProof(t, 4) }
func testGetStaleProof(t *testing.T, protocol int) {
server, tearDown := newServerEnv(t, core.TriesInMemory+4, protocol, nil, false, true, 0)
server, tearDown := newServerEnv(t, core.TriesInMemory+4, protocol, nil, false, true, 0, true)
defer tearDown()
bc := server.handler.blockchain
@ -436,7 +436,7 @@ func testGetCHTProofs(t *testing.T, protocol int) {
time.Sleep(10 * time.Millisecond)
}
}
server, tearDown := newServerEnv(t, int(config.ChtSize+config.ChtConfirms), protocol, waitIndexers, false, true, 0)
server, tearDown := newServerEnv(t, int(config.ChtSize+config.ChtConfirms), protocol, waitIndexers, false, true, 0, true)
defer tearDown()
bc := server.handler.blockchain
@ -485,7 +485,7 @@ func testGetBloombitsProofs(t *testing.T, protocol int) {
time.Sleep(10 * time.Millisecond)
}
}
server, tearDown := newServerEnv(t, int(config.BloomTrieSize+config.BloomTrieConfirms), protocol, waitIndexers, false, true, 0)
server, tearDown := newServerEnv(t, int(config.BloomTrieSize+config.BloomTrieConfirms), protocol, waitIndexers, false, true, 0, true)
defer tearDown()
bc := server.handler.blockchain
@ -523,7 +523,7 @@ func TestTransactionStatusLes3(t *testing.T) { testTransactionStatus(t, 3) }
func TestTransactionStatusLes4(t *testing.T) { testTransactionStatus(t, 4) }
func testTransactionStatus(t *testing.T, protocol int) {
server, tearDown := newServerEnv(t, 0, protocol, nil, false, true, 0)
server, tearDown := newServerEnv(t, 0, protocol, nil, false, true, 0, true)
defer tearDown()
server.handler.addTxsSync = true
@ -620,7 +620,7 @@ func TestStopResumeLes3(t *testing.T) { testStopResume(t, 3) }
func TestStopResumeLes4(t *testing.T) { testStopResume(t, 4) }
func testStopResume(t *testing.T, protocol int) {
server, tearDown := newServerEnv(t, 0, protocol, nil, true, true, testBufLimit/10)
server, tearDown := newServerEnv(t, 0, protocol, nil, true, true, testBufLimit/10, true)
defer tearDown()
server.handler.server.costTracker.testing = true

2
les/odr_test.go
View file

@ -236,7 +236,7 @@ func testOdr(t *testing.T, protocol int, expFail uint64, checkCached bool, fn od
// still expect all retrievals to pass, now data should be cached locally
if checkCached {
client.handler.backend.peers.Unregister(client.peer.peer.id)
client.handler.backend.peers.Disconnect(client.peer.peer.id)
time.Sleep(time.Millisecond * 10) // ensure that all peerSetNotify callbacks are executed
test(5)
}

129
les/peer.go
View file

@ -101,6 +101,9 @@ type peer struct {
responseCount uint64
invalidCount uint32
active bool
activate, deactivate func()
poolEntry *poolEntry
hasBlock func(common.Hash, uint64, bool) bool
responseErrors int
@ -297,12 +300,20 @@ func (p *peer) updateCapacity(cap uint64) {
p.responseLock.Lock()
defer p.responseLock.Unlock()
p.fcParams = flowcontrol.ServerParams{MinRecharge: cap, BufLimit: cap * bufLimitRatio}
p.fcClient.UpdateParams(p.fcParams)
var kvList keyValueList
kvList = kvList.add("flowControl/MRR", cap)
kvList = kvList.add("flowControl/BL", cap*bufLimitRatio)
p.queueSend(func() { p.SendAnnounce(announceData{Update: kvList}) })
if !p.active && cap != 0 && p.activate != nil {
p.activate()
}
if cap != 0 || p.version >= lpv4 {
p.fcParams = flowcontrol.ServerParams{MinRecharge: cap, BufLimit: cap * bufLimitRatio}
p.fcClient.UpdateParams(p.fcParams)
var kvList keyValueList
kvList = kvList.add("flowControl/BL", cap*bufLimitRatio)
kvList = kvList.add("flowControl/MRR", cap)
p.queueSend(func() { p.SendAnnounce(announceData{Update: kvList}) })
}
if p.active && cap == 0 && p.deactivate != nil {
p.deactivate()
}
}
func (p *peer) responseID() uint64 {
@ -628,8 +639,15 @@ func (p *peer) Handshake(td *big.Int, head common.Hash, headNum uint64, genesis
send = send.add("serveRecentState", stateRecent)
send = send.add("txRelay", nil)
}
send = send.add("flowControl/BL", server.defParams.BufLimit)
send = send.add("flowControl/MRR", server.defParams.MinRecharge)
p.active = p.version < lpv4
if p.active {
p.fcParams = server.defParams
} else {
p.fcParams = flowcontrol.ServerParams{}
}
send = send.add("flowControl/BL", p.fcParams.BufLimit)
send = send.add("flowControl/MRR", p.fcParams.MinRecharge)
var costList RequestCostList
if server.costTracker.testCostList != nil {
@ -639,7 +657,6 @@ func (p *peer) Handshake(td *big.Int, head common.Hash, headNum uint64, genesis
}
send = send.add("flowControl/MRC", costList)
p.fcCosts = costList.decode(ProtocolLengths[uint(p.version)])
p.fcParams = server.defParams
// Add advertised checkpoint and register block height which
// client can verify the checkpoint validity.
@ -710,7 +727,7 @@ func (p *peer) Handshake(td *big.Int, head common.Hash, headNum uint64, genesis
// set default announceType on server side
p.announceType = announceTypeSimple
}
p.fcClient = flowcontrol.NewClientNode(server.fcManager, server.defParams)
p.fcClient = flowcontrol.NewClientNode(server.fcManager, p.fcParams)
}
} else {
if recv.get("serveChainSince", &p.chainSince) != nil {
@ -747,6 +764,7 @@ func (p *peer) Handshake(td *big.Int, head common.Hash, headNum uint64, genesis
p.fcParams = sParams
p.fcServer = flowcontrol.NewServerNode(sParams, &mclock.System{})
p.fcCosts = MRC.decode(ProtocolLengths[uint(p.version)])
p.active = p.paramsUseful()
recv.get("checkpoint/value", &p.checkpoint)
recv.get("checkpoint/registerHeight", &p.checkpointNumber)
@ -772,10 +790,12 @@ func (p *peer) updateFlowControl(update keyValueMap) {
}
// If any of the flow control params is nil, refuse to update.
var params flowcontrol.ServerParams
updated := false
if update.get("flowControl/BL", &params.BufLimit) == nil && update.get("flowControl/MRR", &params.MinRecharge) == nil {
// todo can light client set a minimal acceptable flow control params?
p.fcParams = params
p.fcServer.UpdateParams(params)
updated = true
}
var MRC RequestCostList
if update.get("flowControl/MRC", &MRC) == nil {
@ -783,7 +803,16 @@ func (p *peer) updateFlowControl(update keyValueMap) {
for code, cost := range costUpdate {
p.fcCosts[code] = cost
}
updated = true
}
if updated {
p.active = p.paramsUseful()
}
}
func (p *peer) paramsUseful() bool {
reqRecharge, reqBufLimit := p.fcCosts.reqParams()
return p.fcParams.MinRecharge >= reqRecharge && p.fcParams.BufLimit >= reqBufLimit
}
// String implements fmt.Stringer.
@ -803,16 +832,17 @@ type peerSetNotify interface {
// peerSet represents the collection of active peers currently participating in
// the Light Ethereum sub-protocol.
type peerSet struct {
peers map[string]*peer
lock sync.RWMutex
notifyList []peerSetNotify
closed bool
active, inactive map[string]*peer
lock sync.RWMutex
notifyList []peerSetNotify
closed bool
}
// newPeerSet creates a new peer set to track the active participants.
func newPeerSet() *peerSet {
return &peerSet{
peers: make(map[string]*peer),
active: make(map[string]*peer),
inactive: make(map[string]*peer),
}
}
@ -820,8 +850,8 @@ func newPeerSet() *peerSet {
func (ps *peerSet) notify(n peerSetNotify) {
ps.lock.Lock()
ps.notifyList = append(ps.notifyList, n)
peers := make([]*peer, 0, len(ps.peers))
for _, p := range ps.peers {
peers := make([]*peer, 0, len(ps.active))
for _, p := range ps.active {
peers = append(peers, p)
}
ps.lock.Unlock()
@ -839,11 +869,13 @@ func (ps *peerSet) Register(p *peer) error {
ps.lock.Unlock()
return errClosed
}
if _, ok := ps.peers[p.id]; ok {
if _, ok := ps.active[p.id]; ok {
ps.lock.Unlock()
return errAlreadyRegistered
}
ps.peers[p.id] = p
ps.active[p.id] = p
delete(ps.inactive, p.id)
p.sendQueue = newExecQueue(100)
peers := make([]peerSetNotify, len(ps.notifyList))
copy(peers, ps.notifyList)
@ -856,14 +888,15 @@ func (ps *peerSet) Register(p *peer) error {
}
// Unregister removes a remote peer from the active set, disabling any further
// actions to/from that particular entity. It also initiates disconnection at the networking layer.
func (ps *peerSet) Unregister(id string) error {
// actions to/from that particular entity.
func (ps *peerSet) Unregister(p *peer) error {
ps.lock.Lock()
if p, ok := ps.peers[id]; !ok {
if _, ok := ps.active[p.id]; !ok {
ps.lock.Unlock()
return errNotRegistered
} else {
delete(ps.peers, id)
delete(ps.active, p.id)
ps.inactive[p.id] = p
peers := make([]peerSetNotify, len(ps.notifyList))
copy(peers, ps.notifyList)
ps.lock.Unlock()
@ -871,22 +904,38 @@ func (ps *peerSet) Unregister(id string) error {
for _, n := range peers {
n.unregisterPeer(p)
}
p.sendQueue.quit()
p.Peer.Disconnect(p2p.DiscUselessPeer)
return nil
}
}
// AllPeerIDs returns a list of all registered peer IDs
// Disconnect removes a remote peer from either the active or inactive set and
// initiates disconnection at the networking layer.
func (ps *peerSet) Disconnect(id string) error {
ps.lock.Lock()
p, ok := ps.active[id]
if ok {
delete(ps.active, id)
} else {
if p, ok = ps.inactive[id]; !ok {
ps.lock.Unlock()
return errNotRegistered
}
delete(ps.inactive, id)
}
ps.lock.Unlock()
p.Peer.Disconnect(p2p.DiscUselessPeer)
return nil
}
// AllPeerIDs returns a list of all active peer IDs
func (ps *peerSet) AllPeerIDs() []string {
ps.lock.RLock()
defer ps.lock.RUnlock()
res := make([]string, len(ps.peers))
res := make([]string, len(ps.active))
idx := 0
for id := range ps.peers {
for id := range ps.active {
res[idx] = id
idx++
}
@ -898,15 +947,18 @@ func (ps *peerSet) Peer(id string) *peer {
ps.lock.RLock()
defer ps.lock.RUnlock()
return ps.peers[id]
if p, ok := ps.active[id]; ok {
return p
}
return ps.inactive[id]
}
// Len returns if the current number of peers in the set.
// Len returns if the current number of peers in the active set.
func (ps *peerSet) Len() int {
ps.lock.RLock()
defer ps.lock.RUnlock()
return len(ps.peers)
return len(ps.active)
}
// BestPeer retrieves the known peer with the currently highest total difficulty.
@ -918,7 +970,7 @@ func (ps *peerSet) BestPeer() *peer {
bestPeer *peer
bestTd *big.Int
)
for _, p := range ps.peers {
for _, p := range ps.active {
if td := p.Td(); bestPeer == nil || td.Cmp(bestTd) > 0 {
bestPeer, bestTd = p, td
}
@ -926,14 +978,14 @@ func (ps *peerSet) BestPeer() *peer {
return bestPeer
}
// AllPeers returns all peers in a list
// AllPeers returns all active peers in a list
func (ps *peerSet) AllPeers() []*peer {
ps.lock.RLock()
defer ps.lock.RUnlock()
list := make([]*peer, len(ps.peers))
list := make([]*peer, len(ps.active))
i := 0
for _, peer := range ps.peers {
for _, peer := range ps.active {
list[i] = peer
i++
}
@ -946,7 +998,10 @@ func (ps *peerSet) Close() {
ps.lock.Lock()
defer ps.lock.Unlock()
for _, p := range ps.peers {
for _, p := range ps.active {
p.Disconnect(p2p.DiscQuitting)
}
for _, p := range ps.inactive {
p.Disconnect(p2p.DiscQuitting)
}
ps.closed = true

2
les/retrieve.go
View file

@ -345,7 +345,7 @@ func (r *sentReq) tryRequest() {
if hrto {
pp.Log().Debug("Request timed out hard")
if r.rm.peers != nil {
r.rm.peers.Unregister(pp.id)
r.rm.peers.Disconnect(pp.id)
}
}

2
les/server.go
View file

@ -116,7 +116,7 @@ func NewLesServer(e *eth.Ethereum, config *eth.Config) (*LesServer, error) {
srv.maxCapacity = totalRecharge
}
srv.fcManager.SetCapacityLimits(srv.freeCapacity, srv.maxCapacity, srv.freeCapacity*2)
srv.clientPool = newClientPool(srv.chainDb, srv.minCapacity, srv.freeCapacity, mclock.System{}, func(id enode.ID) { go srv.peers.Unregister(peerIdToString(id)) })
srv.clientPool = newClientPool(srv.chainDb, srv.minCapacity, srv.freeCapacity, mclock.System{}, func(id enode.ID) { go srv.peers.Disconnect(peerIdToString(id)) })
srv.clientPool.setDefaultFactors(priceFactors{0, 1, 1}, priceFactors{0, 1, 1})
srv.tokenSale = newTokenSale(srv.clientPool, 0.1)

75
les/server_handler.go
View file

@ -58,10 +58,7 @@ const (
MaxTxStatus = 256 // Amount of transactions to queried per request
)
var (
errTooManyInvalidRequest = errors.New("too many invalid requests made")
errFullClientPool = errors.New("client pool is full")
)
var errTooManyInvalidRequest = errors.New("too many invalid requests made")
// serverHandler is responsible for serving light client and process
// all incoming light requests.
@ -139,28 +136,58 @@ func (h *serverHandler) handle(p *peer) error {
}
defer p.fcClient.Disconnect()
// Disconnect the inbound peer if it's rejected by clientPool
if !h.server.clientPool.connect(p, 0) {
p.Log().Debug("Light Ethereum peer registration failed", "err", errFullClientPool)
return errFullClientPool
var (
connectedAt mclock.AbsTime
wg *sync.WaitGroup // Wait group used to track all in-flight task routines.
)
p.activate = func() {
// Register the peer locally
if err := h.server.peers.Register(p); err != nil {
h.server.clientPool.disconnect(p)
p.Log().Error("Light Ethereum peer registration failed", "err", err)
return
}
clientConnectionGauge.Update(int64(h.server.peers.Len()))
connectedAt = mclock.Now()
wg = new(sync.WaitGroup)
p.active = true
}
// Register the peer locally
if err := h.server.peers.Register(p); err != nil {
h.server.clientPool.disconnect(p)
p.Log().Error("Light Ethereum peer registration failed", "err", err)
return err
}
clientConnectionGauge.Update(int64(h.server.peers.Len()))
var wg sync.WaitGroup // Wait group used to track all in-flight task routines.
connectedAt := mclock.Now()
defer func() {
wg.Wait() // Ensure all background task routines have exited.
h.server.peers.Unregister(p.id)
h.server.clientPool.disconnect(p)
p.deactivate = func() {
h.server.peers.Unregister(p)
if p.version < lpv4 {
h.server.peers.Disconnect(p.id)
}
clientConnectionGauge.Update(int64(h.server.peers.Len()))
connectionTimer.Update(time.Duration(mclock.Now() - connectedAt))
p.active = false
}
if p.active {
p.activate()
}
if capacity, err := h.server.clientPool.connect(p, 0); err != nil {
// Disconnect the inbound peer if it's rejected by clientPool
p.Log().Debug("Light Ethereum peer registration failed", "err", err)
return err
} else if capacity != p.fcParams.MinRecharge {
if p.version < lpv4 {
h.server.peers.Disconnect(p.id)
} else {
p.updateCapacity(capacity)
}
}
defer func() {
wg.Wait() // Ensure all background task routines have exited.
h.server.clientPool.disconnect(p)
p.responseLock.Lock()
if p.active {
p.deactivate()
}
p.activate = nil
p.deactivate = nil
p.responseLock.Unlock()
h.server.peers.Disconnect(p.id)
}()
// Spawn a main loop to handle all incoming messages.
@ -171,7 +198,7 @@ func (h *serverHandler) handle(p *peer) error {
return err
default:
}
if err := h.handleMsg(p, &wg); err != nil {
if err := h.handleMsg(p, wg); err != nil {
p.Log().Debug("Light Ethereum message handling failed", "err", err)
return err
}

31
les/test_helper.go
View file

@ -77,10 +77,10 @@ var (
processConfirms = big.NewInt(1)
// The token bucket buffer limit for testing purpose.
testBufLimit = uint64(1000000)
testBufLimit = uint64(6000)
// The buffer recharging speed for testing purpose.
testBufRecharge = uint64(1000)
testBufRecharge = uint64(1)
)
/*
@ -393,8 +393,13 @@ func (p *testPeer) handshake(t *testing.T, td *big.Int, head common.Hash, headNu
expList = expList.add("serveStateSince", uint64(0))
expList = expList.add("serveRecentState", uint64(core.TriesInMemory-4))
expList = expList.add("txRelay", nil)
expList = expList.add("flowControl/BL", testBufLimit)
expList = expList.add("flowControl/MRR", testBufRecharge)
if p.peer.version >= lpv4 {
expList = expList.add("flowControl/BL", uint64(0))
expList = expList.add("flowControl/MRR", uint64(0))
} else {
expList = expList.add("flowControl/BL", testBufLimit)
expList = expList.add("flowControl/MRR", testBufRecharge)
}
expList = expList.add("flowControl/MRC", costList)
if err := p2p.ExpectMsg(p.app, StatusMsg, expList); err != nil {
@ -403,9 +408,16 @@ func (p *testPeer) handshake(t *testing.T, td *big.Int, head common.Hash, headNu
if err := p2p.Send(p.app, StatusMsg, sendList); err != nil {
t.Fatalf("status send: %v", err)
}
p.peer.fcParams = flowcontrol.ServerParams{
BufLimit: testBufLimit,
MinRecharge: testBufRecharge,
}
func (p *testPeer) expectCapUpdate(t *testing.T) {
if p.peer.version >= lpv4 {
var expList keyValueList
expList = expList.add("flowControl/BL", testBufLimit)
expList = expList.add("flowControl/MRR", testBufRecharge)
if err := p2p.ExpectMsg(p.app, AnnounceMsg, announceData{Update: expList}); err != nil {
t.Fatalf("status recv: %v", err)
}
}
}
@ -436,7 +448,7 @@ type testServer struct {
bloomTrieIndexer *core.ChainIndexer
}
func newServerEnv(t *testing.T, blocks int, protocol int, callback indexerCallback, simClock bool, newPeer bool, testCost uint64) (*testServer, func()) {
func newServerEnv(t *testing.T, blocks int, protocol int, callback indexerCallback, simClock bool, newPeer bool, testCost uint64, expectCapUpdate bool) (*testServer, func()) {
db := rawdb.NewMemoryDatabase()
indexers := testIndexers(db, nil, light.TestServerIndexerConfig)
@ -477,6 +489,9 @@ func newServerEnv(t *testing.T, blocks int, protocol int, callback indexerCallba
cIndexer.Close()
bIndexer.Close()
}
if expectCapUpdate {
server.peer.expectCapUpdate(t)
}
return server, teardown
}

2
les/ulc_test.go
View file

@ -126,7 +126,7 @@ func connect(server *serverHandler, serverId enode.ID, client *clientHandler, pr
// newServerPeer creates server peer.
func newServerPeer(t *testing.T, blocks int, protocol int) (*testServer, *enode.Node, func()) {
s, teardown := newServerEnv(t, blocks, protocol, nil, false, false, 0)
s, teardown := newServerEnv(t, blocks, protocol, nil, false, false, 0, false)
key, err := crypto.GenerateKey()
if err != nil {
t.Fatal("generate key err:", err)