forked from cerc-io/plugeth
eth, les: defer starting LES service until ETH initial sync is finished
This commit is contained in:
parent
c8130df1d9
commit
c57c54ce96
@ -104,6 +104,7 @@ type Config struct {
|
|||||||
|
|
||||||
type LesServer interface {
|
type LesServer interface {
|
||||||
Start(srvr *p2p.Server)
|
Start(srvr *p2p.Server)
|
||||||
|
Synced()
|
||||||
Stop()
|
Stop()
|
||||||
Protocols() []p2p.Protocol
|
Protocols() []p2p.Protocol
|
||||||
}
|
}
|
||||||
@ -145,6 +146,7 @@ type Ethereum struct {
|
|||||||
|
|
||||||
func (s *Ethereum) AddLesServer(ls LesServer) {
|
func (s *Ethereum) AddLesServer(ls LesServer) {
|
||||||
s.lesServer = ls
|
s.lesServer = ls
|
||||||
|
s.protocolManager.lesServer = ls
|
||||||
}
|
}
|
||||||
|
|
||||||
// New creates a new Ethereum object (including the
|
// New creates a new Ethereum object (including the
|
||||||
|
@ -87,6 +87,8 @@ type ProtocolManager struct {
|
|||||||
quitSync chan struct{}
|
quitSync chan struct{}
|
||||||
noMorePeers chan struct{}
|
noMorePeers chan struct{}
|
||||||
|
|
||||||
|
lesServer LesServer
|
||||||
|
|
||||||
// wait group is used for graceful shutdowns during downloading
|
// wait group is used for graceful shutdowns during downloading
|
||||||
// and processing
|
// and processing
|
||||||
wg sync.WaitGroup
|
wg sync.WaitGroup
|
||||||
@ -171,7 +173,7 @@ func NewProtocolManager(config *params.ChainConfig, fastSync bool, networkId int
|
|||||||
return blockchain.CurrentBlock().NumberU64()
|
return blockchain.CurrentBlock().NumberU64()
|
||||||
}
|
}
|
||||||
inserter := func(blocks types.Blocks) (int, error) {
|
inserter := func(blocks types.Blocks) (int, error) {
|
||||||
atomic.StoreUint32(&manager.synced, 1) // Mark initial sync done on any fetcher import
|
manager.setSynced() // Mark initial sync done on any fetcher import
|
||||||
return manager.insertChain(blocks)
|
return manager.insertChain(blocks)
|
||||||
}
|
}
|
||||||
manager.fetcher = fetcher.New(blockchain.GetBlockByHash, validator, manager.BroadcastBlock, heighter, inserter, manager.removePeer)
|
manager.fetcher = fetcher.New(blockchain.GetBlockByHash, validator, manager.BroadcastBlock, heighter, inserter, manager.removePeer)
|
||||||
|
@ -180,7 +180,7 @@ func (pm *ProtocolManager) synchronise(peer *peer) {
|
|||||||
if err := pm.downloader.Synchronise(peer.id, pHead, pTd, mode); err != nil {
|
if err := pm.downloader.Synchronise(peer.id, pHead, pTd, mode); err != nil {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
atomic.StoreUint32(&pm.synced, 1) // Mark initial sync done
|
pm.setSynced() // Mark initial sync done
|
||||||
|
|
||||||
// If fast sync was enabled, and we synced up, disable it
|
// If fast sync was enabled, and we synced up, disable it
|
||||||
if atomic.LoadUint32(&pm.fastSync) == 1 {
|
if atomic.LoadUint32(&pm.fastSync) == 1 {
|
||||||
@ -191,3 +191,10 @@ func (pm *ProtocolManager) synchronise(peer *peer) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// setSynced sets the synced flag and notifies the light server if present
|
||||||
|
func (pm *ProtocolManager) setSynced() {
|
||||||
|
if atomic.SwapUint32(&pm.synced, 1) == 0 && pm.lesServer != nil {
|
||||||
|
pm.lesServer.Synced()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
@ -267,9 +267,9 @@ func (pm *ProtocolManager) Start(srvr *p2p.Server) {
|
|||||||
} else {
|
} else {
|
||||||
if topicDisc != nil {
|
if topicDisc != nil {
|
||||||
go func() {
|
go func() {
|
||||||
glog.V(logger.Debug).Infoln("Starting registering topic", string(lesTopic))
|
glog.V(logger.Info).Infoln("Starting registering topic", string(lesTopic))
|
||||||
topicDisc.RegisterTopic(lesTopic, pm.quitSync)
|
topicDisc.RegisterTopic(lesTopic, pm.quitSync)
|
||||||
glog.V(logger.Debug).Infoln("Stopped registering topic", string(lesTopic))
|
glog.V(logger.Info).Infoln("Stopped registering topic", string(lesTopic))
|
||||||
}()
|
}()
|
||||||
}
|
}
|
||||||
go func() {
|
go func() {
|
||||||
|
@ -42,6 +42,9 @@ type LesServer struct {
|
|||||||
fcManager *flowcontrol.ClientManager // nil if our node is client only
|
fcManager *flowcontrol.ClientManager // nil if our node is client only
|
||||||
fcCostStats *requestCostStats
|
fcCostStats *requestCostStats
|
||||||
defParams *flowcontrol.ServerParams
|
defParams *flowcontrol.ServerParams
|
||||||
|
srvr *p2p.Server
|
||||||
|
synced, stopped bool
|
||||||
|
lock sync.Mutex
|
||||||
}
|
}
|
||||||
|
|
||||||
func NewLesServer(eth *eth.Ethereum, config *eth.Config) (*LesServer, error) {
|
func NewLesServer(eth *eth.Ethereum, config *eth.Config) (*LesServer, error) {
|
||||||
@ -67,12 +70,35 @@ func (s *LesServer) Protocols() []p2p.Protocol {
|
|||||||
return s.protocolManager.SubProtocols
|
return s.protocolManager.SubProtocols
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Start only starts the actual service if the ETH protocol has already been synced,
|
||||||
|
// otherwise it will be started by Synced()
|
||||||
func (s *LesServer) Start(srvr *p2p.Server) {
|
func (s *LesServer) Start(srvr *p2p.Server) {
|
||||||
s.protocolManager.Start(srvr)
|
s.lock.Lock()
|
||||||
|
defer s.lock.Unlock()
|
||||||
|
|
||||||
|
s.srvr = srvr
|
||||||
|
if s.synced {
|
||||||
|
s.protocolManager.Start(s.srvr)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Synced notifies the server that the ETH protocol has been synced and LES service can be started
|
||||||
|
func (s *LesServer) Synced() {
|
||||||
|
s.lock.Lock()
|
||||||
|
defer s.lock.Unlock()
|
||||||
|
|
||||||
|
s.synced = true
|
||||||
|
if s.srvr != nil && !s.stopped {
|
||||||
|
s.protocolManager.Start(s.srvr)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Stop stops the LES service
|
||||||
func (s *LesServer) Stop() {
|
func (s *LesServer) Stop() {
|
||||||
|
s.lock.Lock()
|
||||||
|
defer s.lock.Unlock()
|
||||||
|
|
||||||
|
s.stopped = true
|
||||||
s.fcCostStats.store()
|
s.fcCostStats.store()
|
||||||
s.fcManager.Stop()
|
s.fcManager.Stop()
|
||||||
go func() {
|
go func() {
|
||||||
|
Loading…
Reference in New Issue
Block a user