les: docs and minor fixes

This commit is contained in:
Zsolt Felfoldi 2019-01-26 20:38:32 +01:00
parent 1b1552b0ef
commit 927b3dfb15
4 changed files with 24 additions and 13 deletions

View file

@ -33,7 +33,7 @@ var (
ErrTotalCap = errors.New("total capacity exceeded") ErrTotalCap = errors.New("total capacity exceeded")
ErrUnknownBenchmarkType = errors.New("unknown benchmark type") 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. // 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) return hexutil.Uint64(api.server.priorityClientPool.clients[id].cap)
} }
// clientPool is implemented by both the free and priority client pools
type clientPool interface { type clientPool interface {
peerSetNotify peerSetNotify
setLimits(count int, totalCap uint64) setLimits(count int, totalCap uint64)
@ -155,7 +156,7 @@ type scheduledUpdate struct {
totalCap, id uint64 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 { type priorityClientInfo struct {
cap uint64 // zero for non-priority clients cap uint64 // zero for non-priority clients
connected bool 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 // registerPeer is called when a new client is connected. If the client has no
// is called when the assigned capacity is changed while the client is connected. // priority assigned then it is passed to the child pool which may either keep it
// It returns the priority capacity or zero if the client is not prioritized. // or disconnect it.
// It also returns whether the client can be accepted.
// //
// Note: priorityClientPool also stores a record about free clients while they are // 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 // connected in order to be able to assign priority to them later.
// function if necessary.
func (v *priorityClientPool) registerPeer(p *peer) { func (v *priorityClientPool) registerPeer(p *peer) {
v.lock.Lock() v.lock.Lock()
defer v.lock.Unlock() defer v.lock.Unlock()
@ -210,8 +209,8 @@ func (v *priorityClientPool) registerPeer(p *peer) {
} }
} }
// disconnect should be called when a client is disconnected. // unregisterPeer is called when a client is disconnected. If the client has no
// It should be called for all clients accepted by connect even if not prioritized. // priority assigned then it is also removed from the child pool.
func (v *priorityClientPool) unregisterPeer(p *peer) { func (v *priorityClientPool) unregisterPeer(p *peer) {
v.lock.Lock() v.lock.Lock()
defer v.lock.Unlock() defer v.lock.Unlock()
@ -311,7 +310,6 @@ func (v *priorityClientPool) setLimitsNow(count int, totalCap uint64) {
} }
} }
} }
v.maxPeers = count v.maxPeers = count
v.totalCap = totalCap v.totalCap = totalCap
if v.child != nil { if v.child != nil {

View file

@ -44,6 +44,7 @@ type requestBenchmark interface {
request(peer *peer, index int) error request(peer *peer, index int) error
} }
// benchmarkBlockHeaders implements requestBenchmark
type benchmarkBlockHeaders struct { type benchmarkBlockHeaders struct {
amount, skip int amount, skip int
reverse, byHash bool reverse, byHash bool
@ -78,6 +79,7 @@ func (b *benchmarkBlockHeaders) request(peer *peer, index int) error {
} }
} }
// benchmarkBodiesOrReceipts implements requestBenchmark
type benchmarkBodiesOrReceipts struct { type benchmarkBodiesOrReceipts struct {
receipts bool receipts bool
hashes []common.Hash hashes []common.Hash
@ -100,6 +102,7 @@ func (b *benchmarkBodiesOrReceipts) request(peer *peer, index int) error {
} }
} }
// benchmarkProofsOrCode implements requestBenchmark
type benchmarkProofsOrCode struct { type benchmarkProofsOrCode struct {
code bool code bool
headHash common.Hash headHash common.Hash
@ -120,6 +123,7 @@ func (b *benchmarkProofsOrCode) request(peer *peer, index int) error {
} }
} }
// benchmarkHelperTrie implements requestBenchmark
type benchmarkHelperTrie struct { type benchmarkHelperTrie struct {
bloom bool bloom bool
reqCount int reqCount int
@ -162,6 +166,7 @@ func (b *benchmarkHelperTrie) request(peer *peer, index int) error {
return peer.RequestHelperTrieProofs(0, 0, reqs) return peer.RequestHelperTrieProofs(0, 0, reqs)
} }
// benchmarkTxSend implements requestBenchmark
type benchmarkTxSend struct { type benchmarkTxSend struct {
txs types.Transactions txs types.Transactions
} }
@ -189,6 +194,7 @@ func (b *benchmarkTxSend) request(peer *peer, index int) error {
return peer.SendTxs(0, 0, enc) return peer.SendTxs(0, 0, enc)
} }
// benchmarkTxStatus implements requestBenchmark
type benchmarkTxStatus struct{} type benchmarkTxStatus struct{}
func (b *benchmarkTxStatus) init(pm *ProtocolManager, count int) error { func (b *benchmarkTxStatus) init(pm *ProtocolManager, count int) error {

View file

@ -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() { func (s *LesServer) startEventLoop() {
s.protocolManager.wg.Add(1) s.protocolManager.wg.Add(1)
@ -154,7 +157,7 @@ func (s *LesServer) startEventLoop() {
totalCapacity := s.fcManager.SubscribeTotalCapacity(totalCapacityCh) totalCapacity := s.fcManager.SubscribeTotalCapacity(totalCapacityCh)
go func() { go func() {
for { updateRecharge := func() {
if processing { if processing {
s.protocolManager.servingQueue.setThreads(s.thcBlockProcessing) s.protocolManager.servingQueue.setThreads(s.thcBlockProcessing)
s.fcManager.SetRechargeCurve(flowcontrol.PieceWiseLinear{{0, 0}, {totalRecharge, totalRecharge}}) s.fcManager.SetRechargeCurve(flowcontrol.PieceWiseLinear{{0, 0}, {totalRecharge, totalRecharge}})
@ -162,9 +165,13 @@ func (s *LesServer) startEventLoop() {
s.protocolManager.servingQueue.setThreads(s.thcNormal) s.protocolManager.servingQueue.setThreads(s.thcNormal)
s.fcManager.SetRechargeCurve(flowcontrol.PieceWiseLinear{{0, 0}, {totalRecharge / 10, totalRecharge}, {totalRecharge, totalRecharge}}) s.fcManager.SetRechargeCurve(flowcontrol.PieceWiseLinear{{0, 0}, {totalRecharge / 10, totalRecharge}, {totalRecharge, totalRecharge}})
} }
}
for {
select { select {
case processing = <-blockProcFeed: case processing = <-blockProcFeed:
updateRecharge()
case totalRecharge = <-totalRechargeCh: case totalRecharge = <-totalRechargeCh:
updateRecharge()
case totalCapacity = <-totalCapacityCh: case totalCapacity = <-totalCapacityCh:
s.priorityClientPool.setLimits(s.maxPeers, totalCapacity) s.priorityClientPool.setLimits(s.maxPeers, totalCapacity)
case <-s.protocolManager.quitSync: case <-s.protocolManager.quitSync:

View file

@ -181,7 +181,7 @@ func (n *ExecNode) Start(snapshots map[string][]byte) (err error) {
} }
// start the one-shot server that waits for startup information // 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() defer cancel()
statusURL, statusC := n.waitForStartupJSON(ctx) statusURL, statusC := n.waitForStartupJSON(ctx)