les: protocol versions and topics for clients and servers

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,

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
 }

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 {
+func lesTopic(genesisHash common.Hash, protocolVersion uint) discv5.Topic {
-	return discv5.Topic("LES@" + common.Bytes2Hex(genesisHash.Bytes()[0:8]))
+	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()
+	if s.bbIndexer != nil {
-	s.chtIndexer.Close()
+		s.bbIndexer.Close()
-	s.bltIndexer.Close()
+	}
+	if s.chtIndexer != nil {
+		s.chtIndexer.Close()
+	}
+	if s.bltIndexer != nil {
+		s.bltIndexer.Close()
+	}
 	s.blockchain.Stop()
 	s.protocolManager.Stop()
 	s.txPool.Stop()

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))
+	manager.SubProtocols = make([]p2p.Protocol, 0, len(protocolVersions))
-	for i, version := range 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)

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
 	}

les/protocol.go

@@ -32,11 +32,14 @@ const (
 	lpv2 = 2
 )
 
-// Supported versions of the les protocol (first is primary).
+// Supported versions of the les protocol (first is primary)
-var ProtocolVersions = []uint{lpv1, lpv2}
+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

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() {
+	for _, topic := range s.lesTopics {
-		logger := log.New("topic", s.lesTopic)
+		topic := topic
-		logger.Info("Starting topic registration")
+		go func() {
-		defer logger.Info("Terminated topic registration")
+			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) {