les: factored out client pool logic from handler

les/api.go

@@ -31,23 +31,13 @@ var (
 // PublicLesServerAPI  provides an API to access the les server.
 // It offers only methods that operate on public data that is freely available to anyone.
 type PrivateLesServerAPI struct {
-	server   *LesServer
+	server *LesServer
-	pm       *ProtocolManager
-	priority *priorityClientPool
 }
 
 // NewPublicLesServerAPI creates a new les server API.
 func NewPrivateLesServerAPI(server *LesServer) *PrivateLesServerAPI {
-	priority := &priorityClientPool{
-		clients:  make(map[enode.ID]priorityClientInfo),
-		totalCap: server.totalCapacity,
-		pm:       server.protocolManager,
-	}
-	server.protocolManager.priorityClientPool = priority
 	return &PrivateLesServerAPI{
-		server:   server,
+		server: server,
-		pm:       server.protocolManager,
-		priority: priority,
 	}
 }
 
@@ -58,23 +48,29 @@ func (api *PrivateLesServerAPI) TotalCapacity() hexutil.Uint64 {
 
 // MinimumCapacity queries minimum assignable capacity for a single client
 func (api *PrivateLesServerAPI) MinimumCapacity() hexutil.Uint64 {
-	return hexutil.Uint64(api.server.minCapacity)
+	return hexutil.Uint64(minCapacity)
+}
+
+type clientPool interface {
+	peerSetNotify
+	setLimits(count int, totalCap uint64)
 }
 
 // priorityClientPool stores information about prioritized clients
 type priorityClientPool struct {
-	lock                                          sync.Mutex
+	lock                                       sync.Mutex
-	pm                                            *ProtocolManager
+	child                                      clientPool
-	clients                                       map[enode.ID]priorityClientInfo
+	ps                                         *peerSet
-	totalCap, totalpriorityCap, totalConnectedCap uint64
+	clients                                    map[enode.ID]priorityClientInfo
-	priorityCount                                 int
+	totalCap, totalConnectedCap, freeClientCap uint64
+	maxPeers, priorityCount                    int
 }
 
 // priorityClientInfo entries exist for all prioritized clients and currently connected free clients
 type priorityClientInfo struct {
 	cap       uint64 // zero for non-priority clients
 	connected bool
-	updateCap func(uint64)
+	peer      *peer
 }
 
 // SetClientCapacity sets the priority capacity assigned to a given client.
@@ -85,49 +81,27 @@ type priorityClientInfo struct {
 // Note: assigned capacity can be changed while the client is connected with
 // immediate effect.
 func (api *PrivateLesServerAPI) SetClientCapacity(id enode.ID, cap uint64) error {
-	if cap != 0 && cap < api.server.minCapacity {
+	if cap != 0 && cap < minCapacity {
 		return ErrMinCap
 	}
-
+	return api.server.priorityClientPool.setClientCapacity(id, cap)
-	api.priority.lock.Lock()
-	defer api.priority.lock.Unlock()
-
-	c := api.priority.clients[id]
-	if api.priority.totalpriorityCap+cap > api.priority.totalCap+c.cap {
-		return ErrTotalCap
-	}
-	api.priority.totalpriorityCap += cap - c.cap
-	if c.updateCap != nil && cap != 0 {
-		c.updateCap(cap)
-	}
-	if c.connected {
-		if c.cap != 0 {
-			api.priority.priorityCount--
-		}
-		if cap != 0 {
-			api.priority.priorityCount++
-		}
-		api.priority.totalConnectedCap += cap - c.cap
-		api.pm.clientPool.setConnLimit(api.pm.maxFreePeers(api.priority.priorityCount, api.priority.totalConnectedCap))
-	}
-	if c.updateCap != nil && cap == 0 {
-		c.updateCap(cap)
-	}
-	if cap != 0 || c.connected {
-		c.cap = cap
-		api.priority.clients[id] = c
-	} else {
-		delete(api.priority.clients, id)
-	}
-	return nil
 }
 
 // GetClientCapacity returns the capacity assigned to a given client
 func (api *PrivateLesServerAPI) GetClientCapacity(id enode.ID) hexutil.Uint64 {
-	api.priority.lock.Lock()
+	api.server.priorityClientPool.lock.Lock()
-	defer api.priority.lock.Unlock()
+	defer api.server.priorityClientPool.lock.Unlock()
 
-	return hexutil.Uint64(api.priority.clients[id].cap)
+	return hexutil.Uint64(api.server.priorityClientPool.clients[id].cap)
+}
+
+func newPriorityClientPool(freeClientCap uint64, ps *peerSet, child clientPool) *priorityClientPool {
+	return &priorityClientPool{
+		clients:       make(map[enode.ID]priorityClientInfo),
+		freeClientCap: freeClientCap,
+		ps:            ps,
+		child:         child,
+	}
 }
 
 // connect should be called when a new client is connected. The callback function
@@ -138,39 +112,125 @@ func (api *PrivateLesServerAPI) GetClientCapacity(id enode.ID) hexutil.Uint64 {
 // Note: priorityClientPool also stores a record about free clients while they are
 // connected in order to be able to assign priority to them later with the callback
 // function if necessary.
-func (v *priorityClientPool) connect(id enode.ID, updateCap func(uint64)) (uint64, bool) {
+func (v *priorityClientPool) registerPeer(p *peer) {
 	v.lock.Lock()
 	defer v.lock.Unlock()
 
+	id := p.ID()
 	c := v.clients[id]
 	if c.connected {
-		return 0, false
+		return
 	}
+	if c.cap == 0 && v.child != nil {
+		v.child.registerPeer(p)
+	}
+	if c.cap != 0 && v.totalConnectedCap+c.cap > v.totalCap {
+		v.ps.Unregister(p.id)
+		return
+	}
+
 	c.connected = true
-	c.updateCap = updateCap
+	c.peer = p
 	v.clients[id] = c
 	if c.cap != 0 {
 		v.priorityCount++
+		v.totalConnectedCap += c.cap
+		if v.child != nil {
+			v.child.setLimits(v.maxPeers-v.priorityCount, v.totalCap-v.totalConnectedCap)
+		}
+		p.updateCapacity(c.cap)
 	}
-	v.totalConnectedCap += c.cap
-	v.pm.clientPool.setConnLimit(v.pm.maxFreePeers(v.priorityCount, v.totalConnectedCap))
-	return c.cap, true
 }
 
 // disconnect should be called when a client is disconnected.
 // It should be called for all clients accepted by connect even if not prioritized.
-func (v *priorityClientPool) disconnect(id enode.ID) {
+func (v *priorityClientPool) unregisterPeer(p *peer) {
+	v.lock.Lock()
+	defer v.lock.Unlock()
+
+	id := p.ID()
+	c := v.clients[id]
+	if !c.connected {
+		return
+	}
+	if c.cap != 0 {
+		c.connected = false
+		v.clients[id] = c
+		v.priorityCount--
+		v.totalConnectedCap -= c.cap
+		if v.child != nil {
+			v.child.setLimits(v.maxPeers-v.priorityCount, v.totalCap-v.totalConnectedCap)
+		}
+	} else {
+		if v.child != nil {
+			v.child.unregisterPeer(p)
+		}
+		delete(v.clients, id)
+	}
+}
+
+func (v *priorityClientPool) setLimits(count int, totalCap uint64) {
+	v.lock.Lock()
+	defer v.lock.Unlock()
+
+	if v.priorityCount > count || v.totalConnectedCap > totalCap {
+		for id, c := range v.clients {
+			if c.connected {
+				c.connected = false
+				v.totalConnectedCap -= c.cap
+				v.priorityCount--
+				v.clients[id] = c
+				v.ps.Unregister(c.peer.id)
+				if v.priorityCount <= count && v.totalConnectedCap <= totalCap {
+					break
+				}
+			}
+		}
+	}
+
+	v.maxPeers = count
+	v.totalCap = totalCap
+	if v.child != nil {
+		v.child.setLimits(v.maxPeers-v.priorityCount, v.totalCap-v.totalConnectedCap)
+	}
+}
+
+func (v *priorityClientPool) setClientCapacity(id enode.ID, cap uint64) error {
 	v.lock.Lock()
 	defer v.lock.Unlock()
 
 	c := v.clients[id]
-	c.connected = false
+	if c.cap == cap {
-	if c.cap != 0 {
+		return nil
+	}
+	if c.connected {
+		if v.totalConnectedCap+cap > v.totalCap+c.cap {
+			return ErrTotalCap
+		}
+		if c.cap == 0 {
+			if v.child != nil {
+				v.child.unregisterPeer(c.peer)
+			}
+			v.priorityCount++
+			c.peer.updateCapacity(cap)
+		}
+		v.totalConnectedCap += cap - c.cap
+		if v.child != nil {
+			v.child.setLimits(v.maxPeers-v.priorityCount, v.totalCap-v.totalConnectedCap)
+		}
+		if cap == 0 {
+			if v.child != nil {
+				v.child.registerPeer(c.peer)
+			}
+			v.priorityCount--
+			c.peer.updateCapacity(v.freeClientCap)
+		}
+	}
+	if cap != 0 || c.connected {
+		c.cap = cap
 		v.clients[id] = c
-		v.priorityCount--
 	} else {
 		delete(v.clients, id)
 	}
-	v.totalConnectedCap -= c.cap
+	return nil
-	v.pm.clientPool.setConnLimit(v.pm.maxFreePeers(v.priorityCount, v.totalConnectedCap))
 }

les/freeclient.go

@@ -20,6 +20,7 @@ package les
 import (
 	"io"
 	"math"
+	"net"
 	"sync"
 	"time"
 
@@ -44,12 +45,14 @@ import (
 // value for the client. Currently the LES protocol manager uses IP addresses
 // (without port address) to identify clients.
 type freeClientPool struct {
-	db     ethdb.Database
+	db         ethdb.Database
-	lock   sync.Mutex
+	lock       sync.Mutex
-	clock  mclock.Clock
+	clock      mclock.Clock
-	closed bool
+	closed     bool
+	removePeer func(string)
 
 	connectedLimit, totalLimit int
+	freeClientCap              uint64
 
 	addressMap            map[string]*freeClientPoolEntry
 	connPool, disconnPool *prque.Prque
@@ -64,15 +67,16 @@ const (
 )
 
 // newFreeClientPool creates a new free client pool
-func newFreeClientPool(db ethdb.Database, connectedLimit, totalLimit int, clock mclock.Clock) *freeClientPool {
+func newFreeClientPool(db ethdb.Database, freeClientCap uint64, totalLimit int, clock mclock.Clock, removePeer func(string)) *freeClientPool {
 	pool := &freeClientPool{
-		db:             db,
+		db:            db,
-		clock:          clock,
+		clock:         clock,
-		addressMap:     make(map[string]*freeClientPoolEntry),
+		addressMap:    make(map[string]*freeClientPoolEntry),
-		connPool:       prque.New(poolSetIndex),
+		connPool:      prque.New(poolSetIndex),
-		disconnPool:    prque.New(poolSetIndex),
+		disconnPool:   prque.New(poolSetIndex),
-		connectedLimit: connectedLimit,
+		freeClientCap: freeClientCap,
-		totalLimit:     totalLimit,
+		totalLimit:    totalLimit,
+		removePeer:    removePeer,
 	}
 	pool.loadFromDb()
 	return pool
@@ -85,11 +89,16 @@ func (f *freeClientPool) stop() {
 	f.lock.Unlock()
 }
 
+// registerPeer implements clientPool
+func (f *freeClientPool) registerPeer(p *peer) {
+	if addr, ok := p.RemoteAddr().(*net.TCPAddr); ok {
+		f.connect(addr.IP.String(), p.id)
+	}
+}
+
 // connect should be called after a successful handshake. If the connection was
 // rejected, there is no need to call disconnect.
-//
+func (f *freeClientPool) connect(address, id string) bool {
-// Note: the disconnectFn callback should not block.
-func (f *freeClientPool) connect(address string, disconnectFn func()) bool {
 	f.lock.Lock()
 	defer f.lock.Unlock()
 
@@ -99,17 +108,19 @@ func (f *freeClientPool) connect(address string, disconnectFn func()) bool {
 
 	if f.connectedLimit == 0 {
 		log.Debug("Client rejected", "address", address)
+		f.removePeer(id)
 		return false
 	}
 	e := f.addressMap[address]
 	now := f.clock.Now()
 	var recentUsage int64
 	if e == nil {
-		e = &freeClientPoolEntry{address: address, index: -1}
+		e = &freeClientPoolEntry{address: address, index: -1, id: id}
 		f.addressMap[address] = e
 	} else {
 		if e.connected {
 			log.Debug("Client already connected", "address", address)
+			f.removePeer(id)
 			return false
 		}
 		recentUsage = int64(math.Exp(float64(e.logUsage-f.logOffset(now)) / fixedPointMultiplier))
@@ -125,12 +136,12 @@ func (f *freeClientPool) connect(address string, disconnectFn func()) bool {
 			// keep the old client and reject the new one
 			f.connPool.Push(i, i.linUsage)
 			log.Debug("Client rejected", "address", address)
+			f.removePeer(id)
 			return false
 		}
 	}
 	f.disconnPool.Remove(e.index)
 	e.connected = true
-	e.disconnectFn = disconnectFn
 	f.connPool.Push(e, e.linUsage)
 	if f.connPool.Size()+f.disconnPool.Size() > f.totalLimit {
 		f.disconnPool.Pop()
@@ -139,6 +150,13 @@ func (f *freeClientPool) connect(address string, disconnectFn func()) bool {
 	return true
 }
 
+// unregisterPeer implements clientPool
+func (f *freeClientPool) unregisterPeer(p *peer) {
+	if addr, ok := p.RemoteAddr().(*net.TCPAddr); ok {
+		f.disconnect(addr.IP.String())
+	}
+}
+
 // disconnect should be called when a connection is terminated. If the disconnection
 // was initiated by the pool itself using disconnectFn then calling disconnect is
 // not necessary but permitted.
@@ -165,11 +183,14 @@ func (f *freeClientPool) disconnect(address string) {
 
 // setConnLimit sets the maximum number of free client slots and also drops
 // some peers if necessary
-func (f *freeClientPool) setConnLimit(newLimit int) {
+func (f *freeClientPool) setLimits(count int, totalCap uint64) {
 	f.lock.Lock()
 	defer f.lock.Unlock()
 
-	f.connectedLimit = newLimit
+	f.connectedLimit = int(totalCap / f.freeClientCap)
+	if count < f.connectedLimit {
+		f.connectedLimit = count
+	}
 	now := mclock.Now()
 	for f.connPool.Size() > f.connectedLimit {
 		i := f.connPool.PopItem().(*freeClientPoolEntry)
@@ -185,7 +206,7 @@ func (f *freeClientPool) dropClient(i *freeClientPoolEntry, now mclock.AbsTime)
 	i.connected = false
 	f.disconnPool.Push(i, -i.logUsage)
 	log.Debug("Client kicked out", "address", i.address)
-	i.disconnectFn()
+	f.removePeer(i.id)
 }
 
 // logOffset calculates the time-dependent offset for the logarithmic
@@ -270,7 +291,7 @@ func (f *freeClientPool) saveToDb() {
 // even though they are close to each other at any time they may wrap around int64
 // limits over time. Comparison should be performed accordingly.
 type freeClientPoolEntry struct {
-	address            string
+	address, id        string
 	connected          bool
 	disconnectFn       func()
 	linUsage, logUsage int64

les/freeclient_test.go

@@ -21,6 +21,7 @@ package les
 import (
 	"fmt"
 	"math/rand"
+	"strconv"
 	"testing"
 	"time"
 
@@ -44,32 +45,38 @@ const testFreeClientPoolTicks = 500000
 
 func testFreeClientPool(t *testing.T, connLimit, clientCount int) {
 	var (
-		clock     mclock.Simulated
+		clock       mclock.Simulated
-		db        = ethdb.NewMemDatabase()
+		db          = ethdb.NewMemDatabase()
-		pool      = newFreeClientPool(db, connLimit, 10000, &clock)
+		connected   = make([]bool, clientCount)
-		connected = make([]bool, clientCount)
+		connTicks   = make([]int, clientCount)
-		connTicks = make([]int, clientCount)
+		disconnCh   = make(chan int, clientCount)
-		disconnCh = make(chan int, clientCount)
+		peerAddress = func(i int) string {
-	)
+			return fmt.Sprintf("addr #%d", i)
-	peerId := func(i int) string {
+		}
-		return fmt.Sprintf("test peer #%d", i)
+		peerId = func(i int) string {
-	}
+			return fmt.Sprintf("id #%d", i)
-	disconnFn := func(i int) func() {
+		}
-		return func() {
+		disconnFn = func(id string) {
+			i, err := strconv.Atoi(id[4:])
+			if err != nil {
+				panic(err)
+			}
 			disconnCh <- i
 		}
-	}
+		pool = newFreeClientPool(db, 1, 10000, &clock, disconnFn)
+	)
+	pool.setLimits(connLimit, uint64(connLimit))
 
 	// pool should accept new peers up to its connected limit
 	for i := 0; i < connLimit; i++ {
-		if pool.connect(peerId(i), disconnFn(i)) {
+		if pool.connect(peerAddress(i), peerId(i)) {
 			connected[i] = true
 		} else {
 			t.Fatalf("Test peer #%d rejected", i)
 		}
 	}
 	// since all accepted peers are new and should not be kicked out, the next one should be rejected
-	if pool.connect(peerId(connLimit), disconnFn(connLimit)) {
+	if pool.connect(peerAddress(connLimit), peerId(connLimit)) {
 		connected[connLimit] = true
 		t.Fatalf("Peer accepted over connected limit")
 	}
@@ -80,11 +87,11 @@ func testFreeClientPool(t *testing.T, connLimit, clientCount int) {
 
 		i := rand.Intn(clientCount)
 		if connected[i] {
-			pool.disconnect(peerId(i))
+			pool.disconnect(peerAddress(i))
 			connected[i] = false
 			connTicks[i] += tickCounter
 		} else {
-			if pool.connect(peerId(i), disconnFn(i)) {
+			if pool.connect(peerAddress(i), peerId(i)) {
 				connected[i] = true
 				connTicks[i] -= tickCounter
 			}
@@ -93,7 +100,7 @@ func testFreeClientPool(t *testing.T, connLimit, clientCount int) {
 		for {
 			select {
 			case i := <-disconnCh:
-				pool.disconnect(peerId(i))
+				pool.disconnect(peerAddress(i))
 				if connected[i] {
 					connTicks[i] += tickCounter
 					connected[i] = false
@@ -119,20 +126,21 @@ func testFreeClientPool(t *testing.T, connLimit, clientCount int) {
 	}
 
 	// a previously unknown peer should be accepted now
-	if !pool.connect("newPeer", func() {}) {
+	if !pool.connect("newAddr", "newId") {
 		t.Fatalf("Previously unknown peer rejected")
 	}
 
 	// close and restart pool
 	pool.stop()
-	pool = newFreeClientPool(db, connLimit, 10000, &clock)
+	pool = newFreeClientPool(db, 1, 10000, &clock, disconnFn)
+	pool.setLimits(connLimit, uint64(connLimit))
 
 	// try connecting all known peers (connLimit should be filled up)
 	for i := 0; i < clientCount; i++ {
-		pool.connect(peerId(i), func() {})
+		pool.connect(peerAddress(i), peerId(i))
 	}
 	// expect pool to remember known nodes and kick out one of them to accept a new one
-	if !pool.connect("newPeer2", func() {}) {
+	if !pool.connect("newAddr2", "newId2") {
 		t.Errorf("Previously unknown peer rejected after restarting pool")
 	}
 	pool.stop()

les/handler.go

@@ -22,12 +22,10 @@ import (
 	"encoding/json"
 	"fmt"
 	"math/big"
-	"net"
 	"sync"
 	"time"
 
 	"github.com/ethereum/go-ethereum/common"
-	"github.com/ethereum/go-ethereum/common/mclock"
 	"github.com/ethereum/go-ethereum/consensus"
 	"github.com/ethereum/go-ethereum/core"
 	"github.com/ethereum/go-ethereum/core/rawdb"
@@ -37,7 +35,6 @@ import (
 	"github.com/ethereum/go-ethereum/eth/downloader"
 	"github.com/ethereum/go-ethereum/ethdb"
 	"github.com/ethereum/go-ethereum/event"
-	"github.com/ethereum/go-ethereum/les/flowcontrol"
 	"github.com/ethereum/go-ethereum/light"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/ethereum/go-ethereum/p2p"
@@ -91,24 +88,21 @@ type txPool interface {
 }
 
 type ProtocolManager struct {
-	lightSync          bool
+	lightSync    bool
-	txpool             txPool
+	txpool       txPool
-	txrelay            *LesTxRelay
+	txrelay      *LesTxRelay
-	networkId          uint64
+	networkId    uint64
-	chainConfig        *params.ChainConfig
+	chainConfig  *params.ChainConfig
-	iConfig            *light.IndexerConfig
+	iConfig      *light.IndexerConfig
-	blockchain         BlockChain
+	blockchain   BlockChain
-	chainDb            ethdb.Database
+	chainDb      ethdb.Database
-	odr                *LesOdr
+	odr          *LesOdr
-	server             *LesServer
+	server       *LesServer
-	serverPool         *serverPool
+	serverPool   *serverPool
-	clientPool         *freeClientPool
+	lesTopic     discv5.Topic
-	freeClientCap      uint64
+	reqDist      *requestDistributor
-	priorityClientPool *priorityClientPool
+	retriever    *retrieveManager
-	lesTopic           discv5.Topic
+	servingQueue *servingQueue
-	reqDist            *requestDistributor
-	retriever          *retrieveManager
-	servingQueue       *servingQueue
 
 	downloader *downloader.Downloader
 	fetcher    *lightFetcher
@@ -195,42 +189,11 @@ func (pm *ProtocolManager) removePeer(id string) {
 	pm.peers.Unregister(id)
 }
 
-// maxFreePeers returns the maximum number of free client slots based on the number
-// and total capacity of other clients
-func (pm *ProtocolManager) maxFreePeers(otherPeers int, otherCap uint64) int {
-	if otherPeers >= pm.maxPeers || otherCap >= pm.server.totalCapacity {
-		return 0
-	}
-	maxPeers := int((pm.server.totalCapacity - otherCap) / pm.freeClientCap)
-	if maxPeers <= pm.maxPeers-otherPeers {
-		return maxPeers
-	}
-	return pm.maxPeers - otherPeers
-}
-
 func (pm *ProtocolManager) Start(maxPeers int) {
 	pm.maxPeers = maxPeers
-	if pm.server != nil && maxPeers > 0 {
-		pm.freeClientCap = pm.server.totalCapacity / uint64(maxPeers)
-		if pm.freeClientCap < pm.server.minCapacity {
-			pm.freeClientCap = pm.server.minCapacity
-		}
-		if pm.freeClientCap > 0 {
-			pm.server.defParams = flowcontrol.ServerParams{
-				BufLimit:    pm.freeClientCap * bufLimitRatio,
-				MinRecharge: pm.freeClientCap,
-			}
-		}
-	}
-
 	if pm.lightSync {
 		go pm.syncer()
 	} else {
-		freePeers := pm.maxFreePeers(0, 0)
-		if freePeers < maxPeers {
-			log.Warn("Light peer count limited", "specified", maxPeers, "allowed", freePeers)
-		}
-		pm.clientPool = newFreeClientPool(pm.chainDb, freePeers, 10000, mclock.System{})
 		go func() {
 			for range pm.newPeerCh {
 			}
@@ -248,9 +211,6 @@ func (pm *ProtocolManager) Stop() {
 	pm.noMorePeers <- struct{}{}
 
 	close(pm.quitSync) // quits syncer, fetcher
-	if pm.clientPool != nil {
-		pm.clientPool.stop()
-	}
 
 	// Disconnect existing sessions.
 	// This also closes the gate for any new registrations on the peer set.
@@ -334,86 +294,6 @@ func (pm *ProtocolManager) handle(p *peer) error {
 		pm.removePeer(p.id)
 	}()
 
-	if !pm.lightSync && !p.Peer.Info().Network.Trusted {
-		var freeId string
-		if addr, ok := p.RemoteAddr().(*net.TCPAddr); ok {
-			freeId = addr.IP.String()
-		}
-
-		var (
-			free, priority bool
-			lock           sync.Mutex // lock protects access to the free and priority flags
-		)
-
-		defer func() {
-			lock.Lock()
-			if free {
-				pm.clientPool.disconnect(freeId)
-			}
-			lock.Unlock()
-		}()
-
-		updateCap := func(cap uint64) {
-			lock.Lock()
-			defer lock.Unlock()
-
-			if !priority && cap != 0 {
-				// switch to priority mode
-				if free {
-					pm.clientPool.disconnect(freeId)
-					free = false
-				}
-				priority = true
-				p.updateCapacity(cap)
-			}
-			if priority {
-				if cap == 0 {
-					// priority revoked; switch to free client mode or drop
-					if freeId != "" {
-						if !pm.clientPool.connect(freeId, func() { go pm.removePeer(p.id) }) {
-							pm.removePeer(p.id)
-							return
-						}
-						free = true
-					}
-					priority = false
-					p.updateCapacity(pm.freeClientCap)
-				} else {
-					// just update priority capacity
-					p.updateCapacity(cap)
-				}
-			}
-		}
-
-		lock.Lock()
-		if pm.priorityClientPool != nil {
-			// the priority client pool registers currently connected non-priority clients too
-			// in order to be able to notify them if they get priority while connected
-			priorityCap, ok := pm.priorityClientPool.connect(p.ID(), updateCap)
-			if !ok {
-				lock.Unlock()
-				return p2p.DiscAlreadyConnected
-			}
-			// always unregister
-			defer pm.priorityClientPool.disconnect(p.ID())
-			if priorityCap != 0 {
-				priority = true
-				p.updateCapacity(priorityCap)
-			}
-		}
-
-		if !priority && freeId != "" {
-			// if freeId == "" then we are in test mode and let the client connect
-			// without entering the free client pool
-			if !pm.clientPool.connect(freeId, func() { go pm.removePeer(p.id) }) {
-				lock.Unlock()
-				return p2p.DiscTooManyPeers
-			}
-			free = true
-		}
-		lock.Unlock()
-	}
-
 	// Register the peer in the downloader. If the downloader considers it banned, we disconnect
 	if pm.lightSync {
 		p.lock.Lock()

les/helper_test.go

@@ -192,8 +192,6 @@ func newTestProtocolManager(lightSync bool, blocks int, generator func(int, *cor
 		}
 
 		srv.fcManager = flowcontrol.NewClientManager(nil, &mclock.System{})
-		srv.fcCostList = testRCL()
-		srv.fcCostTable = srv.fcCostList.decode()
 	}
 	pm.Start(1000)
 	return pm, nil

les/peer.go

@@ -106,7 +106,7 @@ func newPeer(version int, network uint64, isTrusted bool, p *p2p.Peer, rw p2p.Ms
 		rw:        rw,
 		version:   version,
 		network:   network,
-		id:        fmt.Sprintf("%x", id[:8]),
+		id:        fmt.Sprintf("%x", id),
 		isTrusted: isTrusted,
 	}
 }

les/server.go

@@ -85,6 +85,7 @@ var (
 		GetTxStatusMsg:         {0, 100},
 	}
 	minBufLimit = 100000000
+	minCapacity = uint64((minBufLimit-1)/bufLimitRatio + 1)
 )
 
 type LesServer struct {
@@ -98,7 +99,7 @@ type LesServer struct {
 	quitSync     chan struct{}
 	onlyAnnounce bool
 
-	totalCapacity, minCapacity    uint64
+	totalCapacity                 uint64
 	rcNormal, rcBlockProcessing   flowcontrol.PieceWiseLinear // buffer recharge curve for normal operation and block processing mode
 	thcNormal, thcBlockProcessing int                         // serving thread count for normal operation and block processing mode
 
@@ -107,6 +108,10 @@ type LesServer struct {
 	globalCostFactor float64
 	gcfUpdateCh      chan gcfUpdate
 	gcfLock          sync.RWMutex
+
+	freeClientCap      uint64
+	freeClientPool     *freeClientPool
+	priorityClientPool *priorityClientPool
 }
 
 func NewLesServer(eth *eth.Ethereum, config *eth.Config) (*LesServer, error) {
@@ -184,8 +189,6 @@ func NewLesServer(eth *eth.Ethereum, config *eth.Config) (*LesServer, error) {
 		srv.fcCostStats = newCostStats(srv.makeCostList().decode())
 	}
 
-	srv.minCapacity = uint64((minBufLimit-1)/bufLimitRatio + 1)
-
 	chtV1SectionCount, _, _ := srv.chtIndexer.Sections() // indexer still uses LES/1 4k section size for backwards server compatibility
 	chtV2SectionCount := chtV1SectionCount / (params.CHTFrequencyClient / params.CHTFrequencyServer)
 	if chtV2SectionCount != 0 {
@@ -251,6 +254,29 @@ func (s *LesServer) Protocols() []p2p.Protocol {
 
 // Start starts the LES server
 func (s *LesServer) Start(srvr *p2p.Server) {
+	maxPeers := s.config.LightPeers
+	if maxPeers > 0 {
+		s.freeClientCap = s.totalCapacity / uint64(maxPeers)
+		if s.freeClientCap < minCapacity {
+			s.freeClientCap = minCapacity
+		}
+		if s.freeClientCap > 0 {
+			s.defParams = flowcontrol.ServerParams{
+				BufLimit:    s.freeClientCap * bufLimitRatio,
+				MinRecharge: s.freeClientCap,
+			}
+		}
+	}
+	freePeers := int(s.totalCapacity / s.freeClientCap)
+	if freePeers < maxPeers {
+		log.Warn("Light peer count limited", "specified", maxPeers, "allowed", freePeers)
+	}
+
+	s.freeClientPool = newFreeClientPool(s.chainDb, s.freeClientCap, 10000, mclock.System{}, s.protocolManager.removePeer)
+	s.priorityClientPool = newPriorityClientPool(s.freeClientCap, s.protocolManager.peers, s.freeClientPool)
+	s.priorityClientPool.setLimits(maxPeers, s.totalCapacity)
+	s.protocolManager.peers.notify(s.priorityClientPool)
+
 	s.protocolManager.Start(s.config.LightPeers)
 	if srvr.DiscV5 != nil {
 		for _, topic := range s.lesTopics {
@@ -282,6 +308,7 @@ func (s *LesServer) Stop() {
 	go func() {
 		<-s.protocolManager.noMorePeers
 	}()
+	s.freeClientPool.stop()
 	s.protocolManager.Stop()
 }