diff --git a/les/api.go b/les/api.go index 5b5a8dea2e..d4f36a8251 100644 --- a/les/api.go +++ b/les/api.go @@ -33,7 +33,7 @@ var ( ErrTotalCap = errors.New("total capacity exceeded") ErrUnknownBenchmarkType = errors.New("unknown benchmark type") - dropCapacityDelay = time.Second + dropCapacityDelay = time.Second // delay applied to decreasing capacity changes ) // PrivateLightServerAPI provides an API to access the LES light server. @@ -129,6 +129,7 @@ func (api *PrivateLightServerAPI) GetClientCapacity(id enode.ID) hexutil.Uint64 return hexutil.Uint64(api.server.priorityClientPool.clients[id].cap) } +// clientPool is implemented by both the free and priority client pools type clientPool interface { peerSetNotify setLimits(count int, totalCap uint64) @@ -155,7 +156,7 @@ type scheduledUpdate struct { totalCap, id uint64 } -// priorityClientInfo entries exist for all prioritized clients and currently connected free clients +// priorityClientInfo entries exist for all prioritized clients and currently connected non-priority clients type priorityClientInfo struct { cap uint64 // zero for non-priority clients connected bool @@ -172,14 +173,12 @@ func newPriorityClientPool(freeClientCap uint64, ps *peerSet, child clientPool) } } -// connect should be called when a new client is connected. The callback function -// is called when the assigned capacity is changed while the client is connected. -// It returns the priority capacity or zero if the client is not prioritized. -// It also returns whether the client can be accepted. +// registerPeer is called when a new client is connected. If the client has no +// priority assigned then it is passed to the child pool which may either keep it +// or disconnect it. // // 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. +// connected in order to be able to assign priority to them later. func (v *priorityClientPool) registerPeer(p *peer) { v.lock.Lock() defer v.lock.Unlock() @@ -210,8 +209,8 @@ func (v *priorityClientPool) registerPeer(p *peer) { } } -// disconnect should be called when a client is disconnected. -// It should be called for all clients accepted by connect even if not prioritized. +// unregisterPeer is called when a client is disconnected. If the client has no +// priority assigned then it is also removed from the child pool. func (v *priorityClientPool) unregisterPeer(p *peer) { v.lock.Lock() defer v.lock.Unlock() @@ -311,7 +310,6 @@ func (v *priorityClientPool) setLimitsNow(count int, totalCap uint64) { } } } - v.maxPeers = count v.totalCap = totalCap if v.child != nil { diff --git a/les/benchmark.go b/les/benchmark.go index 4d310a1306..abf3f13a85 100644 --- a/les/benchmark.go +++ b/les/benchmark.go @@ -44,6 +44,7 @@ type requestBenchmark interface { request(peer *peer, index int) error } +// benchmarkBlockHeaders implements requestBenchmark type benchmarkBlockHeaders struct { amount, skip int reverse, byHash bool @@ -78,6 +79,7 @@ func (b *benchmarkBlockHeaders) request(peer *peer, index int) error { } } +// benchmarkBodiesOrReceipts implements requestBenchmark type benchmarkBodiesOrReceipts struct { receipts bool hashes []common.Hash @@ -100,6 +102,7 @@ func (b *benchmarkBodiesOrReceipts) request(peer *peer, index int) error { } } +// benchmarkProofsOrCode implements requestBenchmark type benchmarkProofsOrCode struct { code bool headHash common.Hash @@ -120,6 +123,7 @@ func (b *benchmarkProofsOrCode) request(peer *peer, index int) error { } } +// benchmarkHelperTrie implements requestBenchmark type benchmarkHelperTrie struct { bloom bool reqCount int @@ -162,6 +166,7 @@ func (b *benchmarkHelperTrie) request(peer *peer, index int) error { return peer.RequestHelperTrieProofs(0, 0, reqs) } +// benchmarkTxSend implements requestBenchmark type benchmarkTxSend struct { txs types.Transactions } @@ -189,6 +194,7 @@ func (b *benchmarkTxSend) request(peer *peer, index int) error { return peer.SendTxs(0, 0, enc) } +// benchmarkTxStatus implements requestBenchmark type benchmarkTxStatus struct{} func (b *benchmarkTxStatus) init(pm *ProtocolManager, count int) error { diff --git a/les/server.go b/les/server.go index db6586c963..87126c779a 100644 --- a/les/server.go +++ b/les/server.go @@ -142,6 +142,9 @@ func (s *LesServer) APIs() []rpc.API { } } +// startEventLoop starts an event handler loop that updates the recharge curve of +// the client manager and adjusts the client pool's size according to the total +// capacity updates coming from the client manager func (s *LesServer) startEventLoop() { s.protocolManager.wg.Add(1) @@ -154,7 +157,7 @@ func (s *LesServer) startEventLoop() { totalCapacity := s.fcManager.SubscribeTotalCapacity(totalCapacityCh) go func() { - for { + updateRecharge := func() { if processing { s.protocolManager.servingQueue.setThreads(s.thcBlockProcessing) s.fcManager.SetRechargeCurve(flowcontrol.PieceWiseLinear{{0, 0}, {totalRecharge, totalRecharge}}) @@ -162,9 +165,13 @@ func (s *LesServer) startEventLoop() { s.protocolManager.servingQueue.setThreads(s.thcNormal) s.fcManager.SetRechargeCurve(flowcontrol.PieceWiseLinear{{0, 0}, {totalRecharge / 10, totalRecharge}, {totalRecharge, totalRecharge}}) } + } + for { select { case processing = <-blockProcFeed: + updateRecharge() case totalRecharge = <-totalRechargeCh: + updateRecharge() case totalCapacity = <-totalCapacityCh: s.priorityClientPool.setLimits(s.maxPeers, totalCapacity) case <-s.protocolManager.quitSync: diff --git a/p2p/simulations/adapters/exec.go b/p2p/simulations/adapters/exec.go index d61d165f41..bd8bcbc856 100644 --- a/p2p/simulations/adapters/exec.go +++ b/p2p/simulations/adapters/exec.go @@ -181,7 +181,7 @@ func (n *ExecNode) Start(snapshots map[string][]byte) (err error) { } // start the one-shot server that waits for startup information - ctx, cancel := context.WithTimeout(context.Background(), 40*time.Second) + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() statusURL, statusC := n.waitForStartupJSON(ctx)