diff --git a/les/balance.go b/les/balance.go index 1eae56071c..1d3cbd143e 100644 --- a/les/balance.go +++ b/les/balance.go @@ -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() diff --git a/les/balance_test.go b/les/balance_test.go index b571c2cc5c..2cf4fcc730 100644 --- a/les/balance_test.go +++ b/les/balance_test.go @@ -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}, } diff --git a/les/client_handler.go b/les/client_handler.go index 7d167c33f7..485a1395b5 100644 --- a/les/client_handler.go +++ b/les/client_handler.go @@ -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 { diff --git a/les/clientpool.go b/les/clientpool.go index 89393d8e30..86d3f988f4 100644 --- a/les/clientpool.go +++ b/les/clientpool.go @@ -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 } diff --git a/les/clientpool_test.go b/les/clientpool_test.go index 23564d4052..5b4170ec00 100644 --- a/les/clientpool_test.go +++ b/les/clientpool_test.go @@ -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") + } +} diff --git a/les/costtracker.go b/les/costtracker.go index 81da045660..abf8618ffd 100644 --- a/les/costtracker.go +++ b/les/costtracker.go @@ -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) diff --git a/les/handler_test.go b/les/handler_test.go index 7293493d2a..08796b49e8 100644 --- a/les/handler_test.go +++ b/les/handler_test.go @@ -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 diff --git a/les/odr_test.go b/les/odr_test.go index b47a4d080b..b6836a057f 100644 --- a/les/odr_test.go +++ b/les/odr_test.go @@ -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) } diff --git a/les/peer.go b/les/peer.go index cc5ee95df0..0e5e4d196c 100644 --- a/les/peer.go +++ b/les/peer.go @@ -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", ¶ms.BufLimit) == nil && update.get("flowControl/MRR", ¶ms.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 diff --git a/les/retrieve.go b/les/retrieve.go index c806117902..a0d494b6c3 100644 --- a/les/retrieve.go +++ b/les/retrieve.go @@ -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) } } diff --git a/les/server.go b/les/server.go index 706c7506ac..9d9fba3876 100644 --- a/les/server.go +++ b/les/server.go @@ -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) diff --git a/les/server_handler.go b/les/server_handler.go index 6014ceeb98..f1a798eb67 100644 --- a/les/server_handler.go +++ b/les/server_handler.go @@ -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 } diff --git a/les/test_helper.go b/les/test_helper.go index 544c3d9e59..3fd0844cba 100644 --- a/les/test_helper.go +++ b/les/test_helper.go @@ -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 } diff --git a/les/ulc_test.go b/les/ulc_test.go index 861cf7cd82..9353d115b9 100644 --- a/les/ulc_test.go +++ b/les/ulc_test.go @@ -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)