les: implement server priority API

internal/web3ext/web3ext.go

@@ -445,6 +445,11 @@ web3._extend({
 			params: 2,
 			inputFormatter:[null, null],
 		}),
+		new web3._extend.Method({
+			name: 'freezeClient',
+			call: 'debug_freezeClient',
+			params: 1,
+		}),
 	],
 	properties: []
 });
@@ -798,6 +803,31 @@ web3._extend({
 			call: 'les_getCheckpoint',
 			params: 1
 		}),
+		new web3._extend.Method({
+			name: 'clientInfo',
+			call: 'les_clientInfo',
+			params: 1
+		}),
+		new web3._extend.Method({
+			name: 'priorityClientInfo',
+			call: 'les_priorityClientInfo',
+			params: 3
+		}),
+		new web3._extend.Method({
+			name: 'setClientParams',
+			call: 'les_setClientParams',
+			params: 2
+		}),
+		new web3._extend.Method({
+			name: 'setDefaultParams',
+			call: 'les_setDefaultParams',
+			params: 1
+		}),
+		new web3._extend.Method({
+			name: 'updateBalance',
+			call: 'les_updateBalance',
+			params: 4
+		}),
 	],
 	properties:
 	[
@@ -809,6 +839,10 @@ web3._extend({
 			name: 'checkpointContractAddress',
 			getter: 'les_getCheckpointContractAddress'
 		}),
+		new web3._extend.Property({
+			name: 'serverInfo',
+			getter: 'les_serverInfo'
+		}),
 	]
 });
 `

les/api.go

@@ -17,16 +17,400 @@
 package les
 
 import (
+	"context"
 	"errors"
+	"fmt"
+	"math"
+	"sync"
+	"time"
 
 	"github.com/ethereum/go-ethereum/common/hexutil"
+	"github.com/ethereum/go-ethereum/common/mclock"
+	"github.com/ethereum/go-ethereum/p2p/enode"
+	"github.com/ethereum/go-ethereum/rpc"
 )
 
 var (
 	errNoCheckpoint         = errors.New("no local checkpoint provided")
 	errNotActivated         = errors.New("checkpoint registrar is not activated")
+	errUnknownBenchmarkType = errors.New("unknown benchmark type")
+	errClientNotConnected   = errors.New("client is not connected")
+	errBalanceOverflow      = errors.New("balance overflow")
+	errNoPriority           = errors.New("not enough priority")
 )
 
+const maxBalance = math.MaxInt64
+
+type clientApiFields struct {
+	balanceUpdatePeriod uint64
+}
+
+// PrivateLightServerAPI provides an API to access the LES light server.
+type PrivateLightServerAPI struct {
+	server                               *LesServer
+	defaultPosFactors, defaultNegFactors priceFactors
+	subs                                 map[*eventSub]struct{}
+	lock                                 sync.Mutex
+}
+
+// NewPrivateLightServerAPI creates a new LES light server API.
+func NewPrivateLightServerAPI(server *LesServer) *PrivateLightServerAPI {
+	api := &PrivateLightServerAPI{
+		server:            server,
+		defaultPosFactors: server.clientPool.defaultPosFactors,
+		defaultNegFactors: server.clientPool.defaultNegFactors,
+		subs:              make(map[*eventSub]struct{}),
+	}
+	server.clientPool.eventHook = api.sendEvent
+	return api
+}
+
+// ServerInfo returns global server parameters
+func (api *PrivateLightServerAPI) ServerInfo() map[string]interface{} {
+	res := make(map[string]interface{})
+	res["minimumCapacity"] = api.server.minCapacity
+	res["maximumCapacity"] = api.server.maxCapacity
+	res["freeClientCapacity"] = api.server.freeCapacity
+	res["totalCapacity"], res["totalConnectedCapacity"], res["priorityConnectedCapacity"] = api.server.clientPool.capacityInfo()
+	return res
+}
+
+// ClientInfo returns information about clients listed in the ids list or matching the given tags
+func (api *PrivateLightServerAPI) ClientInfo(ids []enode.ID) map[enode.ID]map[string]interface{} {
+	res := make(map[enode.ID]map[string]interface{})
+	api.server.clientPool.forClients(ids, func(client *clientInfo, id enode.ID) {
+		res[id] = api.clientInfo(client, id)
+	})
+	return res
+}
+
+// PriorityClientInfo returns information about clients with a positive balance
+// in the given ID range (stop excluded). If stop is null then the iterator stops
+// only at the end of the ID space. MaxCount limits the number of results returned.
+// If maxCount limit is applied but there are more potential results then the ID
+// of the next potential result is included in the map with an empty structure
+// assigned to it.
+func (api *PrivateLightServerAPI) PriorityClientInfo(start, stop enode.ID, maxCount int) map[enode.ID]map[string]interface{} {
+	res := make(map[enode.ID]map[string]interface{})
+	ids := api.server.clientPool.ndb.getPosBalanceIDs(start, stop, maxCount+1)
+	if len(ids) > maxCount {
+		res[ids[maxCount]] = make(map[string]interface{})
+		ids = ids[:maxCount]
+	}
+	api.server.clientPool.forClients(ids, func(client *clientInfo, id enode.ID) {
+		res[id] = api.clientInfo(client, id)
+	})
+	return res
+}
+
+// clientInfo creates a client info data structure
+func (api *PrivateLightServerAPI) clientInfo(c *clientInfo, id enode.ID) map[string]interface{} {
+	info := make(map[string]interface{})
+	if c != nil {
+		info["isConnected"] = true
+		info["capacity"] = c.capacity
+		info["pricing/balance"], info["pricing/negBalance"] = c.balanceTracker.getBalance(mclock.Now())
+		info["pricing/balanceMeta"] = c.balanceMetaInfo
+	} else {
+		info["isConnected"] = false
+		pb := api.server.clientPool.getPosBalance(id)
+		info["pricing/balance"], info["pricing/balanceMeta"] = pb.value, pb.meta
+	}
+	return info
+}
+
+// sendEvent sends an event to the subscribers interested in it. For global events client == nil.
+func (api *PrivateLightServerAPI) sendEvent(clientEvent string, client *clientInfo) {
+	if len(api.subs) == 0 {
+		return
+	}
+
+	event := make(map[string]interface{})
+	event["totalCapacity"], event["totalConnectedCapacity"], event["priorityConnectedCapacity"] = api.server.clientPool.capacityInfo()
+	if client != nil {
+		event["clientEvent"] = clientEvent
+		event["clientId"] = client.id
+		event["clientInfo"] = api.clientInfo(client, client.id)
+	}
+
+	for sub := range api.subs {
+		select {
+		case <-sub.rpcSub.Err():
+			delete(api.subs, sub)
+		case <-sub.notifier.Closed():
+			delete(api.subs, sub)
+		default:
+			sub.notifier.Notify(sub.rpcSub.ID, event)
+		}
+	}
+}
+
+// setParams either sets the given parameters for a single client (if ID is specified)
+// or the default parameters applicable to clients connected in the future
+func (api *PrivateLightServerAPI) setParams(params map[string]interface{}, client *clientInfo, id enode.ID, posFactors, negFactors *priceFactors) (updateFactors bool, err error) {
+	if client != nil {
+		posFactors, negFactors = &client.posFactors, &client.negFactors
+	}
+	defParams := id == enode.ID{}
+loop:
+	for name, value := range params {
+		errValue := func() error {
+			return fmt.Errorf("invalid value for parameter '%s'", name)
+		}
+		setFactor := func(v *float64) {
+			if posFactors != nil {
+				if val, ok := value.(float64); ok && val >= 0 {
+					*v = val / float64(time.Second)
+					updateFactors = true
+				} else {
+					err = errValue()
+				}
+			} else {
+				err = errClientNotConnected
+			}
+		}
+
+		processed := true
+		switch name {
+		case "pricing/timeFactor":
+			setFactor(&posFactors.timeFactor)
+		case "pricing/capacityFactor":
+			setFactor(&posFactors.capacityFactor)
+		case "pricing/requestCostFactor":
+			setFactor(&posFactors.requestFactor)
+		case "pricing/negative/timeFactor":
+			setFactor(&negFactors.timeFactor)
+		case "pricing/negative/capacityFactor":
+			setFactor(&negFactors.capacityFactor)
+		case "pricing/negative/requestCostFactor":
+			setFactor(&negFactors.requestFactor)
+		default:
+			processed = false
+			if defParams {
+				err = fmt.Errorf("invalid default parameter '%s'", name)
+				continue loop
+			}
+		}
+		if processed {
+			continue loop
+		}
+		switch name {
+		case "capacity":
+			if client != nil {
+				if capacity, ok := value.(float64); ok && (capacity == 0 || uint64(capacity) >= api.server.minCapacity) {
+					err = api.server.clientPool.setCapacity(client, uint64(capacity))
+					updateFactors = true
+				} else {
+					err = errValue()
+				}
+			} else {
+				err = errClientNotConnected
+			}
+		case "pricing/alert":
+			if client != nil {
+				if val, ok := value.(float64); ok && val >= 0 {
+					api.setBalanceUpdate(client, uint64(val), false)
+				} else {
+					err = errValue()
+				}
+			} else {
+				err = errClientNotConnected
+			}
+		case "pricing/periodicUpdate":
+			if client != nil {
+				if val, ok := value.(float64); ok && val >= 0 {
+					api.setBalanceUpdate(client, uint64(val), true)
+				} else {
+					err = errValue()
+				}
+			} else {
+				err = errClientNotConnected
+			}
+		default:
+			err = fmt.Errorf("invalid client parameter '%s'", name)
+		}
+	}
+	return updateFactors, err
+}
+
+// UpdateBalance updates the balance of a client (either overwrites it or adds to it).
+// It also updates the balance meta info string.
+func (api *PrivateLightServerAPI) UpdateBalance(id enode.ID, value int64, add bool, meta string) error {
+	return api.server.clientPool.updateBalance(id, value, add, meta)
+}
+
+// SetClientParams sets client parameters for all clients listed in the ids list
+// or all connected clients if the list is empty
+func (api *PrivateLightServerAPI) SetClientParams(ids []enode.ID, params map[string]interface{}) error {
+	var finalErr error
+	api.server.clientPool.forClients(ids, func(client *clientInfo, id enode.ID) {
+		update, err := api.setParams(params, client, id, nil, nil)
+		if err != nil {
+			finalErr = err
+		}
+		if update {
+			client.updatePriceFactors()
+		}
+	})
+	return finalErr
+}
+
+// SetDefaultParams sets the default parameters applicable to clients connected in the future
+func (api *PrivateLightServerAPI) SetDefaultParams(params map[string]interface{}) error {
+	update, err := api.setParams(params, nil, enode.ID{}, &api.defaultPosFactors, &api.defaultNegFactors)
+	if update {
+		api.server.clientPool.setDefaultFactors(api.defaultPosFactors, api.defaultNegFactors)
+	}
+	return err
+}
+
+// balanceUpdate sends a price update client event and schedules a new update with the
+// price tracker if necessary.
+func (api *PrivateLightServerAPI) balanceUpdate(client *clientInfo) {
+	api.lock.Lock()
+	defer api.lock.Unlock()
+
+	api.sendEvent("balanceUpdate", client)
+	if client.balanceUpdatePeriod != 0 {
+		api.setBalanceUpdate(client, client.balanceUpdatePeriod, true)
+	}
+}
+
+// setBalanceUpdate schedules a price update when the balance reaches the given limit.
+// If periodic is false then the limit is interpreted as an absolute value while if true
+// it is relative to the current totalAmount value or the its value at the last future update.
+func (api *PrivateLightServerAPI) setBalanceUpdate(client *clientInfo, value uint64, periodic bool) {
+	balance := balance{pos: value}
+	if periodic {
+		client.balanceUpdatePeriod = value
+		balance.pos, _ = client.balanceTracker.getBalance(mclock.Now())
+		if balance.pos > value {
+			balance.pos -= value
+		} else {
+			balance.pos = 0
+		}
+	} else {
+		client.balanceUpdatePeriod = 0
+	}
+	client.balanceTracker.addCallback(balanceCallbackApi, client.balanceTracker.balanceToPriority(balance), func() { api.balanceUpdate(client) })
+}
+
+// eventSub represents an event subscription
+type eventSub struct {
+	notifier *rpc.Notifier
+	rpcSub   *rpc.Subscription
+}
+
+// SubscribeEvent subscribes to global events and client events related to the clients matching the given tags.
+// If totalCapUnderrun is true then totalCapacity updates are only sent when totalCapacity drops under totalConnectedCapacity.
+func (api *PrivateLightServerAPI) SubscribeEvent(ctx context.Context) (*rpc.Subscription, error) {
+	notifier, supported := rpc.NotifierFromContext(ctx)
+	if !supported {
+		return &rpc.Subscription{}, rpc.ErrNotificationsUnsupported
+	}
+	rpcSub := notifier.CreateSubscription()
+	api.subs[&eventSub{notifier, rpcSub}] = struct{}{}
+	return rpcSub, nil
+}
+
+// Benchmark runs a request performance benchmark with a given set of measurement setups
+// in multiple passes specified by passCount. The measurement time for each setup in each
+// pass is specified in milliseconds by length.
+//
+// Note: measurement time is adjusted for each pass depending on the previous ones.
+// Therefore a controlled total measurement time is achievable in multiple passes.
+func (api *PrivateLightServerAPI) Benchmark(setups []map[string]interface{}, passCount, length int) ([]map[string]interface{}, error) {
+	benchmarks := make([]requestBenchmark, len(setups))
+	for i, setup := range setups {
+		if t, ok := setup["type"].(string); ok {
+			getInt := func(field string, def int) int {
+				if value, ok := setup[field].(float64); ok {
+					return int(value)
+				}
+				return def
+			}
+			getBool := func(field string, def bool) bool {
+				if value, ok := setup[field].(bool); ok {
+					return value
+				}
+				return def
+			}
+			switch t {
+			case "header":
+				benchmarks[i] = &benchmarkBlockHeaders{
+					amount:  getInt("amount", 1),
+					skip:    getInt("skip", 1),
+					byHash:  getBool("byHash", false),
+					reverse: getBool("reverse", false),
+				}
+			case "body":
+				benchmarks[i] = &benchmarkBodiesOrReceipts{receipts: false}
+			case "receipts":
+				benchmarks[i] = &benchmarkBodiesOrReceipts{receipts: true}
+			case "proof":
+				benchmarks[i] = &benchmarkProofsOrCode{code: false}
+			case "code":
+				benchmarks[i] = &benchmarkProofsOrCode{code: true}
+			case "cht":
+				benchmarks[i] = &benchmarkHelperTrie{
+					bloom:    false,
+					reqCount: getInt("amount", 1),
+				}
+			case "bloom":
+				benchmarks[i] = &benchmarkHelperTrie{
+					bloom:    true,
+					reqCount: getInt("amount", 1),
+				}
+			case "txSend":
+				benchmarks[i] = &benchmarkTxSend{}
+			case "txStatus":
+				benchmarks[i] = &benchmarkTxStatus{}
+			default:
+				return nil, errUnknownBenchmarkType
+			}
+		} else {
+			return nil, errUnknownBenchmarkType
+		}
+	}
+	rs := api.server.handler.runBenchmark(benchmarks, passCount, time.Millisecond*time.Duration(length))
+	result := make([]map[string]interface{}, len(setups))
+	for i, r := range rs {
+		res := make(map[string]interface{})
+		if r.err == nil {
+			res["totalCount"] = r.totalCount
+			res["avgTime"] = r.avgTime
+			res["maxInSize"] = r.maxInSize
+			res["maxOutSize"] = r.maxOutSize
+		} else {
+			res["error"] = r.err.Error()
+		}
+		result[i] = res
+	}
+	return result, nil
+}
+
+// PrivateDebugAPI provides an API to debug LES light server functionality.
+type PrivateDebugAPI struct {
+	server *LesServer
+}
+
+// NewPrivateDebugAPI creates a new LES light server debug API.
+func NewPrivateDebugAPI(server *LesServer) *PrivateDebugAPI {
+	return &PrivateDebugAPI{
+		server: server,
+	}
+}
+
+// FreezeClient forces a temporary client freeze which normally happens when the server is overloaded
+func (api *PrivateDebugAPI) FreezeClient(id enode.ID) error {
+	err := errClientNotConnected
+	api.server.clientPool.forClients([]enode.ID{id}, func(c *clientInfo, id enode.ID) {
+		c.peer.freezeClient()
+		err = nil
+	})
+	return err
+}
+
 // PrivateLightAPI provides an API to access the LES light server or light client.
 type PrivateLightAPI struct {
 	backend *lesCommons

les/balance.go

@@ -26,6 +26,7 @@ import (
 const (
 	balanceCallbackQueue = iota
 	balanceCallbackZero
+	balanceCallbackApi
 	balanceCallbackCount
 )
 

les/clientpool.go

@@ -17,6 +17,7 @@
 package les
 
 import (
+	"bytes"
 	"encoding/binary"
 	"io"
 	"math"
@@ -79,15 +80,17 @@ type clientPool struct {
 	stopCh     chan struct{}
 	closed     bool
 	removePeer func(enode.ID)
+	eventHook  func(string, *clientInfo)
 
 	connectedMap   map[enode.ID]*clientInfo
 	connectedQueue *prque.LazyQueue
 
-	posFactors, negFactors priceFactors
+	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
 	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.
@@ -103,6 +106,7 @@ type clientPeer interface {
 	ID() enode.ID
 	freeClientId() string
 	updateCapacity(uint64)
+	freezeClient()
 }
 
 // clientInfo represents a connected client
@@ -115,6 +119,9 @@ type clientInfo struct {
 	peer                   clientPeer
 	queueIndex             int // position in connectedQueue
 	balanceTracker         balanceTracker
+	posFactors, negFactors priceFactors
+	balanceMetaInfo        string
+	clientApiFields
 }
 
 // connSetIndex callback updates clientInfo item index in connectedQueue
@@ -223,12 +230,21 @@ func (f *clientPool) connect(peer clientPeer, capacity uint64) bool {
 	)
 	pb := f.ndb.getOrNewPB(id)
 	posBalance = pb.value
-	e := &clientInfo{pool: f, peer: peer, address: freeID, queueIndex: -1, id: id, priority: posBalance != 0}
 
 	nb := f.ndb.getOrNewNB(freeID)
 	if nb.logValue != 0 {
-		negBalance = uint64(math.Exp(float64(nb.logValue-f.logOffset(now)) / fixedPointMultiplier))
-		negBalance *= uint64(time.Second)
+		negBalance = uint64(math.Exp(float64(nb.logValue-f.logOffset(now))/fixedPointMultiplier) * float64(time.Second))
+	}
+	e := &clientInfo{
+		pool:            f,
+		peer:            peer,
+		address:         freeID,
+		queueIndex:      -1,
+		id:              id,
+		priority:        posBalance != 0,
+		posFactors:      f.defaultPosFactors,
+		negFactors:      f.defaultNegFactors,
+		balanceMetaInfo: pb.meta,
 	}
 	// If the client is a free client, assign with a low free capacity,
 	// Otherwise assign with the given value(priority client)
@@ -240,11 +256,14 @@ func (f *clientPool) connect(peer clientPeer, capacity uint64) bool {
 		capacity = f.freeClientCap
 	}
 	e.capacity = capacity
+	if e.priority {
+		f.priorityConnected += capacity
+	}
 
 	// Starts a balance tracker
 	e.balanceTracker.init(f.clock, capacity)
 	e.balanceTracker.setBalance(posBalance, negBalance)
-	f.setClientPriceFactors(e)
+	e.updatePriceFactors()
 
 	// If the number of clients already connected in the clientpool exceeds its
 	// capacity, evict some clients with lowest priority.
@@ -301,6 +320,9 @@ func (f *clientPool) connect(peer clientPeer, capacity uint64) bool {
 	totalConnectedGauge.Update(int64(f.connectedCap))
 	clientConnectedMeter.Mark(1)
 	log.Debug("Client accepted", "address", freeID)
+	if f.eventHook != nil {
+		f.eventHook("connected", e)
+	}
 	return true
 }
 
@@ -324,6 +346,33 @@ func (f *clientPool) disconnect(p clientPeer) {
 	f.dropClient(e, f.clock.Now(), false)
 }
 
+// forClients iterates through a list of clients, calling the callback for each one.
+// If a client is not connected then clientInfo is nil. If the specified list is empty
+// then the callback is called for all connected clients.
+func (f *clientPool) forClients(ids []enode.ID, callback func(*clientInfo, enode.ID)) {
+	f.lock.Lock()
+	defer f.lock.Unlock()
+
+	if len(ids) > 0 {
+		for _, id := range ids {
+			callback(f.connectedMap[id], id)
+		}
+	} else {
+		for _, c := range f.connectedMap {
+			callback(c, c.id)
+		}
+	}
+}
+
+// setDefaultFactors sets the default price factors applied to subsequently connected clients
+func (f *clientPool) setDefaultFactors(posFactors, negFactors priceFactors) {
+	f.lock.Lock()
+	defer f.lock.Unlock()
+
+	f.defaultPosFactors = posFactors
+	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) {
@@ -334,15 +383,33 @@ func (f *clientPool) dropClient(e *clientInfo, now mclock.AbsTime, kick bool) {
 	f.connectedQueue.Remove(e.queueIndex)
 	delete(f.connectedMap, e.id)
 	f.connectedCap -= e.capacity
+	if e.priority {
+		f.priorityConnected -= e.capacity
+	}
 	totalConnectedGauge.Update(int64(f.connectedCap))
 	if kick {
 		clientKickedMeter.Mark(1)
 		log.Debug("Client kicked out", "address", e.address)
+		if f.eventHook != nil {
+			f.eventHook("kicked", e)
+		}
 		f.removePeer(e.id)
 	} else {
 		clientDisconnectedMeter.Mark(1)
 		log.Debug("Client disconnected", "address", e.address)
+		if f.eventHook != nil {
+			f.eventHook("disconnected", e)
 		}
+	}
+}
+
+// capacityInfo returns the total capacity allowance, the total capacity of connected
+// clients and the total capacity of connected and prioritized clients
+func (f *clientPool) capacityInfo() (uint64, uint64, uint64) {
+	f.lock.Lock()
+	defer f.lock.Unlock()
+
+	return f.capLimit, f.connectedCap, f.priorityConnected
 }
 
 // finalizeBalance stops the balance tracker, retrieves the final balances and
@@ -374,6 +441,9 @@ func (f *clientPool) balanceExhausted(id enode.ID) {
 	if c == nil || !c.priority {
 		return
 	}
+	if c.priority {
+		f.priorityConnected -= c.capacity
+	}
 	c.priority = false
 	if c.capacity != f.freeClientCap {
 		f.connectedCap += f.freeClientCap - c.capacity
@@ -382,6 +452,9 @@ func (f *clientPool) balanceExhausted(id enode.ID) {
 		c.peer.updateCapacity(c.capacity)
 	}
 	f.ndb.delPB(id)
+	if f.eventHook != nil {
+		f.eventHook("balanceExhausted", c)
+	}
 }
 
 // setConnLimit sets the maximum number and total capacity of connected clients,
@@ -398,6 +471,57 @@ func (f *clientPool) setLimits(totalConn int, totalCap uint64) {
 			return f.connectedCap > f.capLimit || f.connectedQueue.Size() > f.connLimit
 		})
 	}
+	if f.eventHook != nil {
+		f.eventHook("capacityUpdate", nil)
+	}
+}
+
+// setCapacity sets the assigned capacity of a connected client
+func (f *clientPool) setCapacity(c *clientInfo, capacity uint64) error {
+	if f.connectedMap[c.id] != c {
+		return errClientNotConnected
+	}
+	if c.capacity == capacity {
+		return nil
+	}
+	if !c.priority {
+		return errNoPriority
+	}
+	oldCapacity := c.capacity
+	c.capacity = capacity
+	f.connectedCap += capacity - oldCapacity
+	f.connectedQueue.Remove(c.queueIndex)
+	f.connectedQueue.Push(c)
+	if f.connectedCap > f.capLimit {
+		var kickList []*clientInfo
+		kick := true
+		f.connectedQueue.MultiPop(func(data interface{}, priority int64) bool {
+			client := data.(*clientInfo)
+			kickList = append(kickList, client)
+			f.connectedCap -= client.capacity
+			if client == c {
+				kick = false
+			}
+			return kick && (f.connectedCap > f.capLimit)
+		})
+		if kick {
+			now := mclock.Now()
+			for _, c := range kickList {
+				f.dropClient(c, now, true)
+			}
+		} else {
+			c.capacity = oldCapacity
+			for _, c := range kickList {
+				f.connectedCap += c.capacity
+				f.connectedQueue.Push(c)
+			}
+			return errNoPriority
+		}
+	}
+	totalConnectedGauge.Update(int64(f.connectedCap))
+	f.priorityConnected += capacity - oldCapacity
+	c.peer.updateCapacity(c.capacity)
+	return nil
 }
 
 // requestCost feeds request cost after serving a request from the given peer.
@@ -424,30 +548,23 @@ func (f *clientPool) logOffset(now mclock.AbsTime) int64 {
 	return f.cumulativeTime + cumulativeTime
 }
 
-// setPriceFactors changes pricing factors for both positive and negative balances.
-// Applies to connected clients and also future connections.
-func (f *clientPool) setPriceFactors(posFactors, negFactors priceFactors) {
+// setClientPriceFactors sets the pricing factors for an individual connected client
+func (c *clientInfo) updatePriceFactors() {
+	c.balanceTracker.setFactors(true, c.negFactors.timeFactor+float64(c.capacity)*c.negFactors.capacityFactor/1000000, c.negFactors.requestFactor)
+	c.balanceTracker.setFactors(false, c.posFactors.timeFactor+float64(c.capacity)*c.posFactors.capacityFactor/1000000, c.posFactors.requestFactor)
+}
+
+// getPosBalance retrieves a single positive balance entry from cache or the database
+func (f *clientPool) getPosBalance(id enode.ID) posBalance {
 	f.lock.Lock()
 	defer f.lock.Unlock()
 
-	f.posFactors, f.negFactors = posFactors, negFactors
-	for _, c := range f.connectedMap {
-		f.setClientPriceFactors(c)
-	}
+	return f.ndb.getOrNewPB(id)
 }
 
-// setClientPriceFactors sets the pricing factors for an individual connected client
-func (f *clientPool) setClientPriceFactors(c *clientInfo) {
-	c.balanceTracker.setFactors(true, f.negFactors.timeFactor+float64(c.capacity)*f.negFactors.capacityFactor/1000000, f.negFactors.requestFactor)
-	c.balanceTracker.setFactors(false, f.posFactors.timeFactor+float64(c.capacity)*f.posFactors.capacityFactor/1000000, f.posFactors.requestFactor)
-}
-
-// addBalance updates the positive balance of a client.
-// If setTotal is false then the given amount is added to the balance.
-// If setTotal is true then amount represents the total amount ever added to the
-// given ID and positive balance is increased by (amount-lastTotal) while lastTotal
-// is updated to amount. This method also allows removing positive balance.
-func (f *clientPool) addBalance(id enode.ID, amount uint64, setTotal bool) {
+// updateBalance updates the balance of a client (either overwrites it or adds to it).
+// It also updates the balance meta info string.
+func (f *clientPool) updateBalance(id enode.ID, amount int64, add bool, meta string) error {
 	f.lock.Lock()
 	defer f.lock.Unlock()
 
@@ -463,44 +580,63 @@ func (f *clientPool) addBalance(id enode.ID, amount uint64, setTotal bool) {
 				// but we have no idea about the new capacity, need a second
 				// call to udpate it.
 				c.priority = true
+				if c.priority {
+					f.priorityConnected += c.capacity
+				}
 				c.balanceTracker.addCallback(balanceCallbackZero, 0, func() { f.balanceExhausted(id) })
 			}
+			c.balanceMetaInfo = meta
 		}()
 	}
-	if setTotal {
-		if pb.value+amount > pb.lastTotal {
-			pb.value += amount - pb.lastTotal
+	if add {
+		if amount > 0 {
+			if amount > maxBalance || pb.value > maxBalance-uint64(amount) {
+				return errBalanceOverflow
+			}
+			pb.value += uint64(amount)
 		} else {
+			if uint64(-amount) > pb.value {
 				pb.value = 0
-		}
-		pb.lastTotal = amount
 			} else {
-		pb.value += amount
-		pb.lastTotal += amount
+				pb.value -= uint64(-amount)
 			}
+		}
+	} else {
+		if amount > maxBalance {
+			return errBalanceOverflow
+		}
+		if amount < 0 {
+			amount = 0
+		}
+		pb.value = uint64(amount)
+	}
+	pb.meta = meta
 	f.ndb.setPB(id, pb)
+	return nil
 }
 
 // posBalance represents a recently accessed positive balance entry
 type posBalance struct {
-	value, lastTotal uint64
+	value uint64
+	meta  string
 }
 
 // EncodeRLP implements rlp.Encoder
 func (e *posBalance) EncodeRLP(w io.Writer) error {
-	return rlp.Encode(w, []interface{}{e.value, e.lastTotal})
+	return rlp.Encode(w, []interface{}{e.value, e.meta})
 }
 
 // DecodeRLP implements rlp.Decoder
 func (e *posBalance) DecodeRLP(s *rlp.Stream) error {
 	var entry struct {
-		Value, LastTotal uint64
+		Value uint64
+		Meta  string
 	}
 	if err := s.Decode(&entry); err != nil {
 		return err
 	}
 	e.value = entry.Value
-	e.lastTotal = entry.LastTotal
+	e.meta = entry.Meta
 	return nil
 }
 
@@ -630,6 +766,37 @@ func (db *nodeDB) delPB(id enode.ID) {
 	db.pcache.Remove(string(key))
 }
 
+// getPosBalanceIDs returns a lexicographically ordered list of IDs of accounts
+// with a positive balance
+func (db *nodeDB) getPosBalanceIDs(start, stop enode.ID, maxCount int) (result []enode.ID) {
+	if maxCount <= 0 {
+		return
+	}
+	it := db.db.NewIteratorWithStart(db.key(start.Bytes(), false))
+	defer it.Release()
+	for i := len(stop[:]) - 1; i >= 0; i-- {
+		stop[i]--
+		if stop[i] != 255 {
+			break
+		}
+	}
+	stopKey := db.key(stop.Bytes(), false)
+	keyLen := len(stopKey)
+
+	for it.Next() {
+		var id enode.ID
+		if len(it.Key()) != keyLen || bytes.Compare(it.Key(), stopKey) == 1 {
+			return
+		}
+		copy(id[:], it.Key()[keyLen-len(id):])
+		result = append(result, id)
+		if len(result) == maxCount {
+			return
+		}
+	}
+	return
+}
+
 func (db *nodeDB) getOrNewNB(id string) negBalance {
 	key := db.key([]byte(id), true)
 	item, exist := db.ncache.Get(string(key))

les/clientpool_test.go

@@ -76,6 +76,8 @@ type poolTestPeerWithCap struct {
 
 func (i *poolTestPeerWithCap) updateCapacity(cap uint64) { i.cap = cap }
 
+func (i poolTestPeer) freezeClient() {}
+
 func testClientPool(t *testing.T, connLimit, clientCount, paidCount int, randomDisconnect bool) {
 	rand.Seed(time.Now().UnixNano())
 	var (
@@ -91,7 +93,7 @@ func testClientPool(t *testing.T, connLimit, clientCount, paidCount int, randomD
 	)
 	pool.disableBias = true
 	pool.setLimits(connLimit, uint64(connLimit))
-	pool.setPriceFactors(priceFactors{1, 0, 1}, priceFactors{1, 0, 1})
+	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++ {
@@ -107,9 +109,11 @@ func testClientPool(t *testing.T, connLimit, clientCount, paidCount int, randomD
 
 		if tickCounter == testClientPoolTicks/4 {
 			// give a positive balance to some of the peers
-			amount := uint64(testClientPoolTicks / 2 * 1000000000) // enough for half of the simulation period
+			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, false)
+				pool.forClients([]enode.ID{poolTestPeer(i).ID()}, func(client *clientInfo, id enode.ID) {
+					pool.updateBalance(id, amount, true, "")
+				})
 			}
 		}
 

les/server.go

@@ -50,7 +50,7 @@ type LesServer struct {
 	servingQueue *servingQueue
 	clientPool   *clientPool
 
-	freeCapacity uint64 // The minimal client capacity used for free client.
+	minCapacity, maxCapacity, freeCapacity uint64
 	threadsIdle                            int // Request serving threads count when system is idle.
 	threadsBusy                            int // Request serving threads count when system is busy(block insertion).
 }
@@ -88,7 +88,8 @@ func NewLesServer(e *eth.Ethereum, config *eth.Config) (*LesServer, error) {
 		threadsIdle:  threads,
 	}
 	srv.handler = newServerHandler(srv, e.BlockChain(), e.ChainDb(), e.TxPool(), e.Synced)
-	srv.costTracker, srv.freeCapacity = newCostTracker(e.ChainDb(), config)
+	srv.costTracker, srv.minCapacity = newCostTracker(e.ChainDb(), config)
+	srv.freeCapacity = srv.minCapacity
 
 	// Set up checkpoint oracle.
 	oracle := config.CheckpointOracle
@@ -108,13 +109,13 @@ func NewLesServer(e *eth.Ethereum, config *eth.Config) (*LesServer, error) {
 	// to send requests most of the time. Our goal is to serve as many clients as
 	// possible while the actually used server capacity does not exceed the limits
 	totalRecharge := srv.costTracker.totalRecharge()
-	maxCapacity := srv.freeCapacity * uint64(srv.config.LightPeers)
-	if totalRecharge > maxCapacity {
-		maxCapacity = totalRecharge
+	srv.maxCapacity = srv.freeCapacity * uint64(srv.config.LightPeers)
+	if totalRecharge > srv.maxCapacity {
+		srv.maxCapacity = totalRecharge
 	}
-	srv.fcManager.SetCapacityLimits(srv.freeCapacity, maxCapacity, srv.freeCapacity*2)
+	srv.fcManager.SetCapacityLimits(srv.freeCapacity, srv.maxCapacity, srv.freeCapacity*2)
 	srv.clientPool = newClientPool(srv.chainDb, srv.freeCapacity, mclock.System{}, func(id enode.ID) { go srv.peers.Unregister(peerIdToString(id)) })
-	srv.clientPool.setPriceFactors(priceFactors{0, 1, 1}, priceFactors{0, 1, 1})
+	srv.clientPool.setDefaultFactors(priceFactors{0, 1, 1}, priceFactors{0, 1, 1})
 
 	checkpoint := srv.latestLocalCheckpoint()
 	if !checkpoint.Empty() {
@@ -133,6 +134,18 @@ func (s *LesServer) APIs() []rpc.API {
 			Service:   NewPrivateLightAPI(&s.lesCommons),
 			Public:    false,
 		},
+		{
+			Namespace: "les",
+			Version:   "1.0",
+			Service:   NewPrivateLightServerAPI(s),
+			Public:    false,
+		},
+		{
+			Namespace: "debug",
+			Version:   "1.0",
+			Service:   NewPrivateDebugAPI(s),
+			Public:    false,
+		},
 	}
 }