From 163f03f319734901c959938a99e7a32c7cd06726 Mon Sep 17 00:00:00 2001 From: Zsolt Felfoldi Date: Sun, 20 Aug 2017 01:02:55 +0200 Subject: [PATCH] les: protocol versions and topics for clients and servers --- ethstats/ethstats.go | 2 +- les/api_backend.go | 3 +++ les/backend.go | 37 +++++++++++++++++++++++++++---------- les/handler.go | 9 +++++---- les/helper_test.go | 8 +++++++- les/protocol.go | 9 ++++++--- les/server.go | 26 +++++++++++++++++--------- 7 files changed, 66 insertions(+), 28 deletions(-) diff --git a/ethstats/ethstats.go b/ethstats/ethstats.go index bb03dc72b0..77784ff4ac 100644 --- a/ethstats/ethstats.go +++ b/ethstats/ethstats.go @@ -379,7 +379,7 @@ func (s *Service) login(conn *websocket.Conn) error { protocol = fmt.Sprintf("eth/%d", eth.ProtocolVersions[0]) } else { network = fmt.Sprintf("%d", infos.Protocols["les"].(*eth.EthNodeInfo).Network) - protocol = fmt.Sprintf("les/%d", les.ProtocolVersions[0]) + protocol = fmt.Sprintf("les/%d", les.ClientProtocolVersions[0]) } auth := &authMsg{ Id: s.node, diff --git a/les/api_backend.go b/les/api_backend.go index 90188894d0..ef58465042 100644 --- a/les/api_backend.go +++ b/les/api_backend.go @@ -174,6 +174,9 @@ func (b *LesApiBackend) AccountManager() *accounts.Manager { } func (b *LesApiBackend) BloomStatus() (uint64, uint64) { + if b.eth.bbIndexer == nil { + return 0, 0 + } sections, _, _ := b.eth.bbIndexer.Sections() return light.BloomTrieFrequency, sections } diff --git a/les/backend.go b/les/backend.go index 2f29a4229a..4d8bcd3b05 100644 --- a/les/backend.go +++ b/les/backend.go @@ -99,7 +99,8 @@ func New(ctx *node.ServiceContext, config *eth.Config) (*LightEthereum, error) { engine: eth.CreateConsensusEngine(ctx, config, chainConfig, chainDb), shutdownChan: make(chan bool), networkId: config.NetworkId, - bbIndexer: eth.NewBloomBitsProcessor(chainDb, light.BloomTrieFrequency), + bloomRequests: make(chan chan *bloombits.Retrieval), + bbIndexer: eth.NewBloomIndexer(chainDb, light.BloomTrieFrequency), chtIndexer: light.NewChtIndexer(chainDb, true), bltIndexer: light.NewBloomTrieIndexer(chainDb, true), } @@ -118,10 +119,8 @@ func New(ctx *node.ServiceContext, config *eth.Config) (*LightEthereum, error) { core.WriteChainConfig(chainDb, genesisHash, chainConfig) } - //leth.bbIndexer.Start(leth.blockchain) - leth.txPool = light.NewTxPool(leth.chainConfig, leth.blockchain, leth.relay) - if leth.protocolManager, err = NewProtocolManager(leth.chainConfig, true, config.NetworkId, leth.eventMux, leth.engine, leth.peers, leth.blockchain, nil, chainDb, leth.odr, leth.relay, quitSync, &leth.wg); err != nil { + if leth.protocolManager, err = NewProtocolManager(leth.chainConfig, true, ClientProtocolVersions, config.NetworkId, leth.eventMux, leth.engine, leth.peers, leth.blockchain, nil, chainDb, leth.odr, leth.relay, quitSync, &leth.wg); err != nil { return nil, err } leth.ApiBackend = &LesApiBackend{leth, nil} @@ -133,8 +132,17 @@ func New(ctx *node.ServiceContext, config *eth.Config) (*LightEthereum, error) { return leth, nil } -func lesTopic(genesisHash common.Hash) discv5.Topic { - return discv5.Topic("LES@" + common.Bytes2Hex(genesisHash.Bytes()[0:8])) +func lesTopic(genesisHash common.Hash, protocolVersion uint) discv5.Topic { + var name string + switch protocolVersion { + case lpv1: + name = "LES" + case lpv2: + name = "LES2" + default: + panic(nil) + } + return discv5.Topic(name + common.Bytes2Hex(genesisHash.Bytes()[0:8])) } type LightDummyAPI struct{} @@ -209,7 +217,10 @@ func (s *LightEthereum) Protocols() []p2p.Protocol { func (s *LightEthereum) Start(srvr *p2p.Server) error { log.Warn("Light client mode is an experimental feature") s.netRPCService = ethapi.NewPublicNetAPI(srvr, s.networkId) - s.serverPool.start(srvr, lesTopic(s.blockchain.Genesis().Hash())) + // search the topic belonging to the oldest supported protocol because + // servers always advertise all supported protocols + protocolVersion := ClientProtocolVersions[len(ClientProtocolVersions)-1] + s.serverPool.start(srvr, lesTopic(s.blockchain.Genesis().Hash(), protocolVersion)) s.protocolManager.Start() return nil } @@ -218,9 +229,15 @@ func (s *LightEthereum) Start(srvr *p2p.Server) error { // Ethereum protocol. func (s *LightEthereum) Stop() error { s.odr.Stop() - s.bbIndexer.Close() - s.chtIndexer.Close() - s.bltIndexer.Close() + if s.bbIndexer != nil { + s.bbIndexer.Close() + } + if s.chtIndexer != nil { + s.chtIndexer.Close() + } + if s.bltIndexer != nil { + s.bltIndexer.Close() + } s.blockchain.Stop() s.protocolManager.Stop() s.txPool.Stop() diff --git a/les/handler.go b/les/handler.go index 4c6dfd323f..fc0a21d288 100644 --- a/les/handler.go +++ b/les/handler.go @@ -127,7 +127,7 @@ type ProtocolManager struct { // NewProtocolManager returns a new ethereum sub protocol manager. The Ethereum sub protocol manages peers capable // with the ethereum network. -func NewProtocolManager(chainConfig *params.ChainConfig, lightSync bool, networkId uint64, mux *event.TypeMux, engine consensus.Engine, peers *peerSet, blockchain BlockChain, txpool txPool, chainDb ethdb.Database, odr *LesOdr, txrelay *LesTxRelay, quitSync chan struct{}, wg *sync.WaitGroup) (*ProtocolManager, error) { +func NewProtocolManager(chainConfig *params.ChainConfig, lightSync bool, protocolVersions []uint, networkId uint64, mux *event.TypeMux, engine consensus.Engine, peers *peerSet, blockchain BlockChain, txpool txPool, chainDb ethdb.Database, odr *LesOdr, txrelay *LesTxRelay, quitSync chan struct{}, wg *sync.WaitGroup) (*ProtocolManager, error) { // Create the protocol manager with the base fields manager := &ProtocolManager{ lightSync: lightSync, @@ -149,15 +149,16 @@ func NewProtocolManager(chainConfig *params.ChainConfig, lightSync bool, network manager.retriever = odr.retriever manager.reqDist = odr.retriever.dist } + // Initiate a sub-protocol for every implemented version we can handle - manager.SubProtocols = make([]p2p.Protocol, 0, len(ProtocolVersions)) - for i, version := range ProtocolVersions { + manager.SubProtocols = make([]p2p.Protocol, 0, len(protocolVersions)) + for _, version := range protocolVersions { // Compatible, initialize the sub-protocol version := version // Closure for the run manager.SubProtocols = append(manager.SubProtocols, p2p.Protocol{ Name: "les", Version: version, - Length: ProtocolLengths[i], + Length: ProtocolLengths[version], Run: func(p *p2p.Peer, rw p2p.MsgReadWriter) error { var entry *poolEntry peer := manager.newPeer(int(version), networkId, p, rw) diff --git a/les/helper_test.go b/les/helper_test.go index b33454e1d0..f7677b735b 100644 --- a/les/helper_test.go +++ b/les/helper_test.go @@ -156,7 +156,13 @@ func newTestProtocolManager(lightSync bool, blocks int, generator func(int, *cor chain = blockchain } - pm, err := NewProtocolManager(gspec.Config, lightSync, NetworkId, evmux, engine, peers, chain, nil, db, odr, nil, make(chan struct{}), new(sync.WaitGroup)) + var protocolVersions []uint + if lightSync { + protocolVersions = ClientProtocolVersions + } else { + protocolVersions = ServerProtocolVersions + } + pm, err := NewProtocolManager(gspec.Config, lightSync, protocolVersions, NetworkId, evmux, engine, peers, chain, nil, db, odr, nil, make(chan struct{}), new(sync.WaitGroup)) if err != nil { return nil, err } diff --git a/les/protocol.go b/les/protocol.go index deffeee666..80654712e8 100644 --- a/les/protocol.go +++ b/les/protocol.go @@ -32,11 +32,14 @@ const ( lpv2 = 2 ) -// Supported versions of the les protocol (first is primary). -var ProtocolVersions = []uint{lpv1, lpv2} +// Supported versions of the les protocol (first is primary) +var ( + ClientProtocolVersions = []uint{lpv2, lpv1} + ServerProtocolVersions = []uint{lpv2, lpv1} +) // Number of implemented message corresponding to different protocol versions. -var ProtocolLengths = []uint64{15, 19} +var ProtocolLengths = map[uint]uint64{lpv1: 15, lpv2: 19} const ( NetworkId = 1 diff --git a/les/server.go b/les/server.go index a6ceab9d35..fa832ab7ff 100644 --- a/les/server.go +++ b/les/server.go @@ -40,7 +40,7 @@ type LesServer struct { fcManager *flowcontrol.ClientManager // nil if our node is client only fcCostStats *requestCostStats defParams *flowcontrol.ServerParams - lesTopic discv5.Topic + lesTopics []discv5.Topic quitSync chan struct{} chtIndexer, bltIndexer *core.ChainIndexer @@ -48,16 +48,21 @@ type LesServer struct { func NewLesServer(eth *eth.Ethereum, config *eth.Config) (*LesServer, error) { quitSync := make(chan struct{}) - pm, err := NewProtocolManager(eth.BlockChain().Config(), false, config.NetworkId, eth.EventMux(), eth.Engine(), newPeerSet(), eth.BlockChain(), eth.TxPool(), eth.ChainDb(), nil, nil, quitSync, new(sync.WaitGroup)) + pm, err := NewProtocolManager(eth.BlockChain().Config(), false, ServerProtocolVersions, config.NetworkId, eth.EventMux(), eth.Engine(), newPeerSet(), eth.BlockChain(), eth.TxPool(), eth.ChainDb(), nil, nil, quitSync, new(sync.WaitGroup)) if err != nil { return nil, err } pm.blockLoop() + lesTopics := make([]discv5.Topic, len(ServerProtocolVersions)) + for i, pv := range ServerProtocolVersions { + lesTopics[i] = lesTopic(eth.BlockChain().Genesis().Hash(), pv) + } + srv := &LesServer{ protocolManager: pm, quitSync: quitSync, - lesTopic: lesTopic(eth.BlockChain().Genesis().Hash()), + lesTopics: lesTopics, chtIndexer: light.NewChtIndexer(eth.ChainDb(), false), bltIndexer: light.NewBloomTrieIndexer(eth.ChainDb(), false), } @@ -80,13 +85,16 @@ func (s *LesServer) Protocols() []p2p.Protocol { // Start starts the LES server func (s *LesServer) Start(srvr *p2p.Server) { s.protocolManager.Start() - go func() { - logger := log.New("topic", s.lesTopic) - logger.Info("Starting topic registration") - defer logger.Info("Terminated topic registration") + for _, topic := range s.lesTopics { + topic := topic + go func() { + logger := log.New("topic", topic) + logger.Info("Starting topic registration") + defer logger.Info("Terminated topic registration") - srvr.DiscV5.RegisterTopic(s.lesTopic, s.quitSync) - }() + srvr.DiscV5.RegisterTopic(topic, s.quitSync) + }() + } } func (s *LesServer) SetBloomBitsIndexer(bbIndexer *core.ChainIndexer) {