diff --git a/les/api.go b/les/api.go index 5998ca7462..169464e76a 100644 --- a/les/api.go +++ b/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 - pm *ProtocolManager - priority *priorityClientPool + server *LesServer } // 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, - pm: server.protocolManager, - priority: priority, + server: server, } } @@ -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 - pm *ProtocolManager - clients map[enode.ID]priorityClientInfo - totalCap, totalpriorityCap, totalConnectedCap uint64 - priorityCount int + lock sync.Mutex + child clientPool + ps *peerSet + clients map[enode.ID]priorityClientInfo + 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 } - - 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 + return api.server.priorityClientPool.setClientCapacity(id, cap) } // GetClientCapacity returns the capacity assigned to a given client func (api *PrivateLesServerAPI) GetClientCapacity(id enode.ID) hexutil.Uint64 { - api.priority.lock.Lock() - defer api.priority.lock.Unlock() + api.server.priorityClientPool.lock.Lock() + 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 != 0 { + if c.cap == cap { + 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 - v.pm.clientPool.setConnLimit(v.pm.maxFreePeers(v.priorityCount, v.totalConnectedCap)) + return nil } diff --git a/les/freeclient.go b/les/freeclient.go index 9fbc43443c..cc7c6a8412 100644 --- a/les/freeclient.go +++ b/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 - lock sync.Mutex - clock mclock.Clock - closed bool + db ethdb.Database + lock sync.Mutex + clock mclock.Clock + 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, - clock: clock, - addressMap: make(map[string]*freeClientPoolEntry), - connPool: prque.New(poolSetIndex), - disconnPool: prque.New(poolSetIndex), - connectedLimit: connectedLimit, - totalLimit: totalLimit, + db: db, + clock: clock, + addressMap: make(map[string]*freeClientPoolEntry), + connPool: prque.New(poolSetIndex), + disconnPool: prque.New(poolSetIndex), + freeClientCap: freeClientCap, + 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. -// -// Note: the disconnectFn callback should not block. -func (f *freeClientPool) connect(address string, disconnectFn func()) bool { +func (f *freeClientPool) connect(address, id string) 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 diff --git a/les/freeclient_test.go b/les/freeclient_test.go index e95abc7aad..d93f6c653a 100644 --- a/les/freeclient_test.go +++ b/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 - db = ethdb.NewMemDatabase() - pool = newFreeClientPool(db, connLimit, 10000, &clock) - connected = make([]bool, clientCount) - connTicks = make([]int, clientCount) - disconnCh = make(chan int, clientCount) - ) - peerId := func(i int) string { - return fmt.Sprintf("test peer #%d", i) - } - disconnFn := func(i int) func() { - return func() { + clock mclock.Simulated + db = ethdb.NewMemDatabase() + connected = make([]bool, clientCount) + connTicks = make([]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("id #%d", i) + } + 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() diff --git a/les/handler.go b/les/handler.go index 944ea488bd..5fb8904ba4 100644 --- a/les/handler.go +++ b/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 - txpool txPool - txrelay *LesTxRelay - networkId uint64 - chainConfig *params.ChainConfig - iConfig *light.IndexerConfig - blockchain BlockChain - chainDb ethdb.Database - odr *LesOdr - server *LesServer - serverPool *serverPool - clientPool *freeClientPool - freeClientCap uint64 - priorityClientPool *priorityClientPool - lesTopic discv5.Topic - reqDist *requestDistributor - retriever *retrieveManager - servingQueue *servingQueue + lightSync bool + txpool txPool + txrelay *LesTxRelay + networkId uint64 + chainConfig *params.ChainConfig + iConfig *light.IndexerConfig + blockchain BlockChain + chainDb ethdb.Database + odr *LesOdr + server *LesServer + serverPool *serverPool + lesTopic discv5.Topic + 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() diff --git a/les/helper_test.go b/les/helper_test.go index 7434b38e90..053f752de1 100644 --- a/les/helper_test.go +++ b/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 diff --git a/les/peer.go b/les/peer.go index c369231d9a..3ac1e08e5b 100644 --- a/les/peer.go +++ b/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, } } diff --git a/les/server.go b/les/server.go index f40174e1f3..06d0dd0931 100644 --- a/les/server.go +++ b/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() }