diff --git a/les/api.go b/les/api.go index a933cbd068..dbeb4a6317 100644 --- a/les/api.go +++ b/les/api.go @@ -19,11 +19,13 @@ package les import ( "context" "errors" + "fmt" "sync" "time" "github.com/ethereum/go-ethereum/common/hexutil" "github.com/ethereum/go-ethereum/common/mclock" + "github.com/ethereum/go-ethereum/les/csvlogger" "github.com/ethereum/go-ethereum/p2p/enode" "github.com/ethereum/go-ethereum/rpc" ) @@ -144,6 +146,8 @@ type priorityClientPool struct { totalCap, totalCapAnnounced uint64 totalConnectedCap, freeClientCap uint64 maxPeers, priorityCount int + logger *csvlogger.Logger + logTotalPriConn *csvlogger.Channel subs tcSubs updateSchedule []scheduledUpdate @@ -164,12 +168,14 @@ type priorityClientInfo struct { } // newPriorityClientPool creates a new priority client pool -func newPriorityClientPool(freeClientCap uint64, ps *peerSet, child clientPool) *priorityClientPool { +func newPriorityClientPool(freeClientCap uint64, ps *peerSet, child clientPool, logger *csvlogger.Logger) *priorityClientPool { return &priorityClientPool{ - clients: make(map[enode.ID]priorityClientInfo), - freeClientCap: freeClientCap, - ps: ps, - child: child, + clients: make(map[enode.ID]priorityClientInfo), + freeClientCap: freeClientCap, + ps: ps, + child: child, + logger: logger, + logTotalPriConn: logger.NewChannel("totalPriConn", 0), } } @@ -185,6 +191,7 @@ func (v *priorityClientPool) registerPeer(p *peer) { id := p.ID() c := v.clients[id] + v.logger.Event(fmt.Sprintf("priorityClientPool: registerPeer cap=%d connected=%v, %x", c.cap, c.connected, id.Bytes())) if c.connected { return } @@ -192,6 +199,7 @@ func (v *priorityClientPool) registerPeer(p *peer) { v.child.registerPeer(p) } if c.cap != 0 && v.totalConnectedCap+c.cap > v.totalCap { + v.logger.Event(fmt.Sprintf("priorityClientPool: rejected, %x", id.Bytes())) go v.ps.Unregister(p.id) return } @@ -202,6 +210,8 @@ func (v *priorityClientPool) registerPeer(p *peer) { if c.cap != 0 { v.priorityCount++ v.totalConnectedCap += c.cap + v.logger.Event(fmt.Sprintf("priorityClientPool: accepted with %d capacity, %x", c.cap, id.Bytes())) + v.logTotalPriConn.Update(float64(v.totalConnectedCap)) if v.child != nil { v.child.setLimits(v.maxPeers-v.priorityCount, v.totalCap-v.totalConnectedCap) } @@ -217,6 +227,7 @@ func (v *priorityClientPool) unregisterPeer(p *peer) { id := p.ID() c := v.clients[id] + v.logger.Event(fmt.Sprintf("priorityClientPool: unregisterPeer cap=%d connected=%v, %x", c.cap, c.connected, id.Bytes())) if !c.connected { return } @@ -225,6 +236,7 @@ func (v *priorityClientPool) unregisterPeer(p *peer) { v.clients[id] = c v.priorityCount-- v.totalConnectedCap -= c.cap + v.logTotalPriConn.Update(float64(v.totalConnectedCap)) if v.child != nil { v.child.setLimits(v.maxPeers-v.priorityCount, v.totalCap-v.totalConnectedCap) } @@ -299,8 +311,10 @@ func (v *priorityClientPool) setLimitsNow(count int, totalCap uint64) { if v.priorityCount > count || v.totalConnectedCap > totalCap { for id, c := range v.clients { if c.connected { + v.logger.Event(fmt.Sprintf("priorityClientPool: setLimitsNow kicked out, %x", id.Bytes())) c.connected = false v.totalConnectedCap -= c.cap + v.logTotalPriConn.Update(float64(v.totalConnectedCap)) v.priorityCount-- v.clients[id] = c go v.ps.Unregister(c.peer.id) @@ -356,6 +370,7 @@ func (v *priorityClientPool) setClientCapacity(id enode.ID, cap uint64) error { v.priorityCount-- } v.totalConnectedCap += cap - c.cap + v.logTotalPriConn.Update(float64(v.totalConnectedCap)) if v.child != nil { v.child.setLimits(v.maxPeers-v.priorityCount, v.totalCap-v.totalConnectedCap) } @@ -374,6 +389,9 @@ func (v *priorityClientPool) setClientCapacity(id enode.ID, cap uint64) error { } else { delete(v.clients, id) } + if c.connected { + v.logger.Event(fmt.Sprintf("priorityClientPool: changed capacity to %d, %x", cap, id.Bytes())) + } return nil } diff --git a/les/costtracker.go b/les/costtracker.go index 332d733431..3b1327a2f4 100644 --- a/les/costtracker.go +++ b/les/costtracker.go @@ -18,6 +18,7 @@ package les import ( "encoding/binary" + "fmt" "math" "sync" "sync/atomic" @@ -26,6 +27,7 @@ import ( "github.com/ethereum/go-ethereum/common/mclock" "github.com/ethereum/go-ethereum/eth" "github.com/ethereum/go-ethereum/ethdb" + "github.com/ethereum/go-ethereum/les/csvlogger" "github.com/ethereum/go-ethereum/les/flowcontrol" "github.com/ethereum/go-ethereum/log" ) @@ -99,16 +101,22 @@ type costTracker struct { gfLock sync.RWMutex totalRechargeCh chan uint64 - stats map[uint64][]uint64 + stats map[uint64][]uint64 + logger *csvlogger.Logger + logRecentUsage, logTotalRecharge, logRelCost *csvlogger.Channel } // newCostTracker creates a cost tracker and loads the cost factor statistics from the database -func newCostTracker(db ethdb.Database, config *eth.Config) *costTracker { +func newCostTracker(db ethdb.Database, config *eth.Config, logger *csvlogger.Logger) *costTracker { utilTarget := float64(config.LightServ) * flowcontrol.FixedPointMultiplier / 100 ct := &costTracker{ - db: db, - stopCh: make(chan chan struct{}), - utilTarget: utilTarget, + db: db, + stopCh: make(chan chan struct{}), + utilTarget: utilTarget, + logger: logger, + logRelCost: logger.NewMinMaxChannel("relativeCost", true), + logRecentUsage: logger.NewMinMaxChannel("recentUsage", true), + logTotalRecharge: logger.NewChannel("totalRecharge", 0.01), } if config.LightBandwidthIn > 0 { ct.inSizeFactor = utilTarget / float64(config.LightBandwidthIn) @@ -201,14 +209,23 @@ func (ct *costTracker) gfLoop() { case r := <-ct.gfUpdateCh: now := mclock.Now() max := r.servingTime * gf + if ct.logRelCost != nil && r.avgTime > 1e-20 { + ct.logRelCost.Update(max / r.avgTime) + } if r.avgTime > max { max = r.avgTime } dt := float64(now - expUpdate) expUpdate = now gfUsage = gfUsage*math.Exp(-dt/float64(gfUsageTC)) + max*1000000/float64(gfUsageTC) + totalRecharge := ct.utilTarget * gf + ct.logRecentUsage.Update(gfUsage) + ct.logTotalRecharge.Update(totalRecharge) + if r.servingTime > 1000000000 { + ct.logger.Event(fmt.Sprintf("Very long servingTime = %f avgTime = %f costFactor = %f", r.servingTime, r.avgTime, gf)) + } - if gfUsage >= gfUsageThreshold*ct.utilTarget*gf { + if gfUsage >= gfUsageThreshold*totalRecharge { gfSum += r.avgTime gfWeight += r.servingTime if time.Duration(now-lastUpdate) > time.Second { @@ -224,7 +241,7 @@ func (ct *costTracker) gfLoop() { ct.gfLock.Unlock() if ch != nil { select { - case ct.totalRechargeCh <- uint64(ct.utilTarget * gf): + case ct.totalRechargeCh <- uint64(totalRecharge): default: } } diff --git a/les/csvlogger/csvlogger.go b/les/csvlogger/csvlogger.go new file mode 100644 index 0000000000..b82710aaf7 --- /dev/null +++ b/les/csvlogger/csvlogger.go @@ -0,0 +1,210 @@ +// Copyright 2019 The go-ethereum Authors +// This file is part of the go-ethereum library. +// +// The go-ethereum library is free software: you can redistribute it and/or modify +// it under the terms of the GNU Lesser General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. +// +// The go-ethereum library is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU Lesser General Public License for more details. +// +// You should have received a copy of the GNU Lesser General Public License +// along with the go-ethereum library. If not, see . + +package csvlogger + +import ( + "fmt" + "os" + "sync" + "time" + + "github.com/ethereum/go-ethereum/common/mclock" + "github.com/ethereum/go-ethereum/log" +) + +type Logger struct { + file *os.File + started mclock.AbsTime + channels []*Channel + period time.Duration + stopCh, stopped chan struct{} + storeCh chan string + eventHeader string +} + +func NewLogger(fileName string, period time.Duration, eventHeader string) *Logger { + f, err := os.Create(fileName) + if err != nil { + log.Error("Error creating log file", "name", fileName, "error", err) + return nil + } + return &Logger{ + file: f, + period: period, + stopCh: make(chan struct{}), + storeCh: make(chan string, 1), + eventHeader: eventHeader, + } +} + +func (l *Logger) NewChannel(name string, threshold float64) *Channel { + if l == nil { + return nil + } + c := &Channel{ + logger: l, + name: name, + threshold: threshold, + } + l.channels = append(l.channels, c) + return c +} + +func (l *Logger) NewMinMaxChannel(name string, zeroDefault bool) *Channel { + if l == nil { + return nil + } + c := &Channel{ + logger: l, + name: name, + minmax: true, + mmZeroDefault: zeroDefault, + } + l.channels = append(l.channels, c) + return c +} + +func (l *Logger) store(event string) { + s := fmt.Sprintf("%g", float64(mclock.Now()-l.started)/1000000000) + for _, ch := range l.channels { + s += ", " + ch.store() + } + if event != "" { + s += ", " + event + } + l.file.WriteString(s + "\n") +} + +func (l *Logger) Start() { + if l == nil { + return + } + l.started = mclock.Now() + s := "Time" + for _, ch := range l.channels { + s += ", " + ch.header() + } + if l.eventHeader != "" { + s += ", " + l.eventHeader + } + l.file.WriteString(s + "\n") + fmt.Println(s) + go func() { + timer := time.NewTimer(l.period) + for { + select { + case <-timer.C: + l.store("") + timer.Reset(l.period) + case event := <-l.storeCh: + l.store(event) + if !timer.Stop() { + <-timer.C + } + timer.Reset(l.period) + case <-l.stopCh: + close(l.stopped) + return + } + } + }() +} + +func (l *Logger) Stop() { + if l == nil { + return + } + l.stopped = make(chan struct{}) + close(l.stopCh) + <-l.stopped + l.file.Close() +} + +func (l *Logger) Event(event string) { + if l == nil { + return + } + select { + case l.storeCh <- event: + case <-l.stopCh: + } +} + +type Channel struct { + logger *Logger + lock sync.Mutex + name string + threshold, storeMin, storeMax, lastValue, min, max float64 + minmax, mmSet, mmZeroDefault bool +} + +func (lc *Channel) Update(value float64) { + if lc == nil { + return + } + lc.lock.Lock() + defer lc.lock.Unlock() + + lc.lastValue = value + if lc.minmax { + if value > lc.max || !lc.mmSet { + lc.max = value + } + if value < lc.min || !lc.mmSet { + lc.min = value + } + lc.mmSet = true + } else { + if value < lc.storeMin || value > lc.storeMax { + select { + case lc.logger.storeCh <- "": + default: + } + } + } +} + +func (lc *Channel) store() (s string) { + lc.lock.Lock() + defer lc.lock.Unlock() + + if lc.minmax { + s = fmt.Sprintf("%g, %g", lc.min, lc.max) + lc.mmSet = false + if lc.mmZeroDefault { + lc.min = 0 + } else { + lc.min = lc.lastValue + } + lc.max = lc.min + } else { + s = fmt.Sprintf("%g", lc.lastValue) + lc.storeMin = lc.lastValue * (1 - lc.threshold) + lc.storeMax = lc.lastValue * (1 + lc.threshold) + if lc.lastValue < 0 { + lc.storeMin, lc.storeMax = lc.storeMax, lc.storeMin + } + } + return +} + +func (lc *Channel) header() string { + if lc.minmax { + return lc.name + " (min), " + lc.name + " (max)" + } + return lc.name +} diff --git a/les/flowcontrol/manager.go b/les/flowcontrol/manager.go index 532e6a4050..b80bf666d5 100644 --- a/les/flowcontrol/manager.go +++ b/les/flowcontrol/manager.go @@ -24,6 +24,7 @@ import ( "github.com/ethereum/go-ethereum/common/mclock" "github.com/ethereum/go-ethereum/common/prque" + "github.com/ethereum/go-ethereum/les/csvlogger" ) // cmNodeFields are ClientNode fields used by the client manager @@ -68,6 +69,8 @@ type ClientManager struct { capLastUpdate mclock.AbsTime totalCapacityCh chan uint64 + logTotalCap *csvlogger.Channel + // recharge integrator is increasing in each moment with a rate of // (totalRecharge / sumRecharge)*FixedPointMultiplier or 0 if sumRecharge==0 rcLastUpdate mclock.AbsTime // last time the recharge integrator was updated @@ -101,11 +104,12 @@ type ClientManager struct { // starting from zero in order to not let a single low-priority client use up // the entire server capacity and thus ensure quick availability for others at // any moment. -func NewClientManager(curve PieceWiseLinear, clock mclock.Clock) *ClientManager { +func NewClientManager(curve PieceWiseLinear, clock mclock.Clock, logger *csvlogger.Logger) *ClientManager { cm := &ClientManager{ clock: clock, rcQueue: prque.New(func(a interface{}, i int) { a.(*ClientNode).queueIndex = i }), capLastUpdate: clock.Now(), + logTotalCap: logger.NewChannel("totalCapacity", 0.01), } if curve != nil { cm.SetRechargeCurve(curve) @@ -337,6 +341,7 @@ func (cm *ClientManager) refreshCapacity() { if totalCapacity >= cm.totalCapacity*0.999 && totalCapacity <= cm.totalCapacity*1.001 { return } + cm.logTotalCap.Update(totalCapacity) cm.totalCapacity = totalCapacity if cm.totalCapacityCh != nil { select { diff --git a/les/freeclient.go b/les/freeclient.go index d859337c26..27e6abf2a4 100644 --- a/les/freeclient.go +++ b/les/freeclient.go @@ -26,6 +26,7 @@ import ( "github.com/ethereum/go-ethereum/common/mclock" "github.com/ethereum/go-ethereum/common/prque" "github.com/ethereum/go-ethereum/ethdb" + "github.com/ethereum/go-ethereum/les/csvlogger" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/rlp" ) @@ -52,6 +53,8 @@ type freeClientPool struct { connectedLimit, totalLimit int freeClientCap uint64 + logger *csvlogger.Logger + logTotalFreeConn *csvlogger.Channel addressMap map[string]*freeClientPoolEntry connPool, disconnPool *prque.Prque @@ -66,16 +69,18 @@ const ( ) // newFreeClientPool creates a new free client pool -func newFreeClientPool(db ethdb.Database, freeClientCap uint64, totalLimit int, clock mclock.Clock, removePeer func(string)) *freeClientPool { +func newFreeClientPool(db ethdb.Database, freeClientCap uint64, totalLimit int, clock mclock.Clock, removePeer func(string), logger *csvlogger.Logger) *freeClientPool { pool := &freeClientPool{ - db: db, - clock: clock, - addressMap: make(map[string]*freeClientPoolEntry), - connPool: prque.New(poolSetIndex), - disconnPool: prque.New(poolSetIndex), - freeClientCap: freeClientCap, - totalLimit: totalLimit, - removePeer: removePeer, + db: db, + clock: clock, + addressMap: make(map[string]*freeClientPoolEntry), + connPool: prque.New(poolSetIndex), + disconnPool: prque.New(poolSetIndex), + freeClientCap: freeClientCap, + totalLimit: totalLimit, + logger: logger, + logTotalFreeConn: logger.NewChannel("totalFreeConn", 0), + removePeer: removePeer, } pool.loadFromDb() return pool @@ -107,7 +112,9 @@ func (f *freeClientPool) connect(address, id string) bool { return false } + f.logger.Event("freeClientPool: connecting from " + address + ", " + id) if f.connectedLimit == 0 { + f.logger.Event("freeClientPool: rejected, " + id) log.Debug("Client rejected", "address", address) return false } @@ -119,6 +126,7 @@ func (f *freeClientPool) connect(address, id string) bool { f.addressMap[address] = e } else { if e.connected { + f.logger.Event("freeClientPool: already connected, " + id) log.Debug("Client already connected", "address", address) return false } @@ -131,9 +139,11 @@ func (f *freeClientPool) connect(address, id string) bool { if e.linUsage+int64(connectedBias)-i.linUsage < 0 { // kick it out and accept the new client f.dropClient(i, now) + f.logger.Event("freeClientPool: kicked out, " + i.id) } else { // keep the old client and reject the new one f.connPool.Push(i, i.linUsage) + f.logger.Event("freeClientPool: rejected, " + id) log.Debug("Client rejected", "address", address) return false } @@ -142,9 +152,11 @@ func (f *freeClientPool) connect(address, id string) bool { e.connected = true e.id = id f.connPool.Push(e, e.linUsage) + f.logTotalFreeConn.Update(float64(uint64(f.connPool.Size()) * f.freeClientCap)) if f.connPool.Size()+f.disconnPool.Size() > f.totalLimit { f.disconnPool.Pop() } + f.logger.Event("freeClientPool: accepted, " + id) log.Debug("Client accepted", "address", address) return true } @@ -174,9 +186,11 @@ func (f *freeClientPool) disconnect(address string) { } f.connPool.Remove(e.index) + f.logTotalFreeConn.Update(float64(uint64(f.connPool.Size()) * f.freeClientCap)) f.calcLogUsage(e, now) e.connected = false f.disconnPool.Push(e, -e.logUsage) + f.logger.Event("freeClientPool: disconnected, " + e.id) log.Debug("Client disconnected", "address", address) } @@ -194,6 +208,7 @@ func (f *freeClientPool) setLimits(count int, totalCap uint64) { for f.connPool.Size() > f.connectedLimit { i := f.connPool.PopItem().(*freeClientPoolEntry) f.dropClient(i, now) + f.logger.Event("freeClientPool: setLimits kicked out, " + i.id) } } @@ -201,6 +216,7 @@ func (f *freeClientPool) setLimits(count int, totalCap uint64) { // disconnected pool func (f *freeClientPool) dropClient(i *freeClientPoolEntry, now mclock.AbsTime) { f.connPool.Remove(i.index) + f.logTotalFreeConn.Update(float64(uint64(f.connPool.Size()) * f.freeClientCap)) f.calcLogUsage(i, now) i.connected = false f.disconnPool.Push(i, -i.logUsage) diff --git a/les/freeclient_test.go b/les/freeclient_test.go index 1918222641..e5f264604c 100644 --- a/les/freeclient_test.go +++ b/les/freeclient_test.go @@ -61,7 +61,7 @@ func testFreeClientPool(t *testing.T, connLimit, clientCount int) { } disconnCh <- i } - pool = newFreeClientPool(db, 1, 10000, &clock, disconnFn) + pool = newFreeClientPool(db, 1, 10000, &clock, disconnFn, nil) ) pool.setLimits(connLimit, uint64(connLimit)) @@ -130,7 +130,7 @@ func testFreeClientPool(t *testing.T, connLimit, clientCount int) { // close and restart pool pool.stop() - pool = newFreeClientPool(db, 1, 10000, &clock, disconnFn) + pool = newFreeClientPool(db, 1, 10000, &clock, disconnFn, nil) pool.setLimits(connLimit, uint64(connLimit)) // try connecting all known peers (connLimit should be filled up) diff --git a/les/handler.go b/les/handler.go index f53a4722f1..d810987e44 100644 --- a/les/handler.go +++ b/les/handler.go @@ -34,6 +34,7 @@ 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/csvlogger" "github.com/ethereum/go-ethereum/light" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/p2p" @@ -118,6 +119,7 @@ type ProtocolManager struct { wg *sync.WaitGroup eventMux *event.TypeMux + logger *csvlogger.Logger // Callbacks synced func() bool @@ -272,6 +274,7 @@ func (pm *ProtocolManager) handle(p *peer) error { // Ignore maxPeers if this is a trusted peer // In server mode we try to check into the client pool after handshake if pm.client && pm.peers.Len() >= pm.maxPeers && !p.Peer.Info().Network.Trusted { + pm.logger.Event("Rejected (too many peers), " + p.id) return p2p.DiscTooManyPeers } // Reject light clients if server is not synced. @@ -290,6 +293,7 @@ func (pm *ProtocolManager) handle(p *peer) error { ) if err := p.Handshake(td, hash, number, genesis.Hash(), pm.server); err != nil { p.Log().Debug("Light Ethereum handshake failed", "err", err) + pm.logger.Event("Handshake error: " + err.Error() + ", " + p.id) return err } if p.fcClient != nil { @@ -303,9 +307,12 @@ func (pm *ProtocolManager) handle(p *peer) error { // Register the peer locally if err := pm.peers.Register(p); err != nil { p.Log().Error("Light Ethereum peer registration failed", "err", err) + pm.logger.Event("Peer registration error: " + err.Error() + ", " + p.id) return err } + pm.logger.Event("Connection established, " + p.id) defer func() { + pm.logger.Event("Closed connection, " + p.id) pm.removePeer(p.id) }() @@ -326,6 +333,7 @@ func (pm *ProtocolManager) handle(p *peer) error { // main loop. handle incoming messages. for { if err := pm.handleMsg(p); err != nil { + pm.logger.Event("Message handling error: " + err.Error() + ", " + p.id) p.Log().Debug("Light Ethereum message handling failed", "err", err) if p.fcServer != nil { p.fcServer.DumpLogs() diff --git a/les/server.go b/les/server.go index 6b93b846db..f04e49db54 100644 --- a/les/server.go +++ b/les/server.go @@ -19,6 +19,7 @@ package les import ( "crypto/ecdsa" "sync" + "time" "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common/mclock" @@ -26,6 +27,7 @@ import ( "github.com/ethereum/go-ethereum/core/rawdb" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/eth" + "github.com/ethereum/go-ethereum/les/csvlogger" "github.com/ethereum/go-ethereum/les/flowcontrol" "github.com/ethereum/go-ethereum/light" "github.com/ethereum/go-ethereum/log" @@ -47,6 +49,7 @@ type LesServer struct { privateKey *ecdsa.PrivateKey quitSync chan struct{} onlyAnnounce bool + csvLogger *csvlogger.Logger thcNormal, thcBlockProcessing int // serving thread count for normal operation and block processing mode @@ -83,6 +86,8 @@ func NewLesServer(eth *eth.Ethereum, config *eth.Config) (*LesServer, error) { for i, pv := range AdvertiseProtocolVersions { lesTopics[i] = lesTopic(eth.BlockChain().Genesis().Hash(), pv) } + var csvLogger *csvlogger.Logger + csvLogger = csvlogger.NewLogger("/tmp/server.csv", time.Second*10, "event, peerId") srv := &LesServer{ lesCommons: lesCommons{ @@ -93,20 +98,22 @@ func NewLesServer(eth *eth.Ethereum, config *eth.Config) (*LesServer, error) { bloomTrieIndexer: light.NewBloomTrieIndexer(eth.ChainDb(), nil, params.BloomBitsBlocks, params.BloomTrieFrequency), protocolManager: pm, }, - costTracker: newCostTracker(eth.ChainDb(), config), + costTracker: newCostTracker(eth.ChainDb(), config, csvLogger), quitSync: quitSync, lesTopics: lesTopics, onlyAnnounce: config.OnlyAnnounce, + csvLogger: csvLogger, } logger := log.New() pm.server = srv + pm.logger = csvLogger srv.thcNormal = config.LightServ * 4 / 100 if srv.thcNormal < 4 { srv.thcNormal = 4 } srv.thcBlockProcessing = config.LightServ/100 + 1 - srv.fcManager = flowcontrol.NewClientManager(nil, &mclock.System{}) + srv.fcManager = flowcontrol.NewClientManager(nil, &mclock.System{}, csvLogger) chtSectionCount, _, _ := srv.chtIndexer.Sections() if chtSectionCount != 0 { @@ -205,10 +212,11 @@ func (s *LesServer) Start(srvr *p2p.Server) { log.Warn("Light peer count limited", "specified", s.maxPeers, "allowed", freePeers) } - s.freeClientPool = newFreeClientPool(s.chainDb, s.freeClientCap, 10000, mclock.System{}, func(id string) { go s.protocolManager.removePeer(id) }) - s.priorityClientPool = newPriorityClientPool(s.freeClientCap, s.protocolManager.peers, s.freeClientPool) + s.freeClientPool = newFreeClientPool(s.chainDb, s.freeClientCap, 10000, mclock.System{}, func(id string) { go s.protocolManager.removePeer(id) }, s.csvLogger) + s.priorityClientPool = newPriorityClientPool(s.freeClientCap, s.protocolManager.peers, s.freeClientPool, s.csvLogger) s.protocolManager.peers.notify(s.priorityClientPool) + s.csvLogger.Start() s.startEventLoop() s.protocolManager.Start(s.config.LightPeers) if srvr.DiscV5 != nil { @@ -241,6 +249,7 @@ func (s *LesServer) Stop() { s.freeClientPool.stop() s.costTracker.stop() s.protocolManager.Stop() + s.csvLogger.Stop() } // todo(rjl493456442) separate client and server implementation.