[DVT-819][DVT-903] Use p2p.Server and refactor

cmd/p2p/crawl/crawl.go

@@ -23,6 +23,7 @@ type (
 		NodesFile            string
 		Database             string
 		RevalidationInterval string
+
 		revalidationInterval time.Duration
 	}
 )
@@ -36,7 +37,7 @@ var (
 var CrawlCmd = &cobra.Command{
 	Use:   "crawl [nodes file]",
 	Short: "Crawl a network on the devp2p layer and generate a nodes JSON file.",
-	Long:  "If no nodes.json file exists, run `echo \"{}\" >> nodes.json` to get started.",
+	Long:  "If no nodes.json file exists, it will be created.",
 	Args:  cobra.MinimumNArgs(1),
 	PreRunE: func(cmd *cobra.Command, args []string) (err error) {
 		inputCrawlParams.NodesFile = args[0]
@@ -54,18 +55,17 @@ var CrawlCmd = &cobra.Command{
 		return nil
 	},
 	RunE: func(cmd *cobra.Command, args []string) error {
-		inputSet, err := p2p.LoadNodesJSON(inputCrawlParams.NodesFile)
+		nodes, err := p2p.ReadNodeSet(inputCrawlParams.NodesFile)
 		if err != nil {
-			return err
+			log.Warn().Err(err).Msgf("Creating nodes file %v because it does not exist", inputCrawlParams.NodesFile)
 		}
 
 		var cfg discover.Config
 		cfg.PrivateKey, _ = crypto.GenerateKey()
-		bn, err := p2p.ParseBootnodes(inputCrawlParams.Bootnodes)
+		cfg.Bootnodes, err = p2p.ParseBootnodes(inputCrawlParams.Bootnodes)
 		if err != nil {
 			return fmt.Errorf("unable to parse bootnodes: %w", err)
 		}
-		cfg.Bootnodes = bn
 
 		db, err := enode.OpenDB(inputCrawlParams.Database)
 		if err != nil {
@@ -84,13 +84,13 @@ var CrawlCmd = &cobra.Command{
 		}
 		defer disc.Close()
 
-		c := newCrawler(inputSet, disc, disc.RandomNodes())
+		c := newCrawler(nodes, disc, disc.RandomNodes())
 		c.revalidateInterval = inputCrawlParams.revalidationInterval
 
 		log.Info().Msg("Starting crawl")
 
 		output := c.run(inputCrawlParams.timeout, inputCrawlParams.Threads)
-		return p2p.WriteNodesJSON(inputCrawlParams.NodesFile, output)
+		return p2p.WriteNodeSet(inputCrawlParams.NodesFile, output)
 	},
 }
 
@@ -101,9 +101,9 @@ required, so other nodes in the network can discover each other.`)
 	if err := CrawlCmd.MarkPersistentFlagRequired("bootnodes"); err != nil {
 		log.Error().Err(err).Msg("Failed to mark bootnodes as required persistent flag")
 	}
-	CrawlCmd.PersistentFlags().StringVarP(&inputCrawlParams.Timeout, "timeout", "t", "30m0s", "Time limit for the crawl.")
-	CrawlCmd.PersistentFlags().IntVarP(&inputCrawlParams.Threads, "parallel", "p", 16, "How many parallel discoveries to attempt.")
-	CrawlCmd.PersistentFlags().Uint64VarP(&inputCrawlParams.NetworkID, "network-id", "n", 0, "Filter discovered nodes by this network id.")
-	CrawlCmd.PersistentFlags().StringVarP(&inputCrawlParams.Database, "database", "d", "", "Node database for updating and storing client information.")
-	CrawlCmd.PersistentFlags().StringVarP(&inputCrawlParams.RevalidationInterval, "revalidation-interval", "r", "10m", "The amount of time it takes to retry connecting to a failed peer.")
+	CrawlCmd.PersistentFlags().StringVarP(&inputCrawlParams.Timeout, "timeout", "t", "30m0s", "Time limit for the crawl")
+	CrawlCmd.PersistentFlags().IntVarP(&inputCrawlParams.Threads, "parallel", "p", 16, "How many parallel discoveries to attempt")
+	CrawlCmd.PersistentFlags().Uint64VarP(&inputCrawlParams.NetworkID, "network-id", "n", 0, "Filter discovered nodes by this network id")
+	CrawlCmd.PersistentFlags().StringVarP(&inputCrawlParams.Database, "database", "d", "", "Node database for updating and storing client information")
+	CrawlCmd.PersistentFlags().StringVarP(&inputCrawlParams.RevalidationInterval, "revalidation-interval", "r", "10m", "Time before retrying to connect to a failed peer")
 }

cmd/p2p/crawl/crawl_util.go

@@ -12,7 +12,7 @@ import (
 )
 
 type crawler struct {
-	input     p2p.NodeSet
+	input     []*enode.Node
 	output    p2p.NodeSet
 	disc      resolver
 	iters     []enode.Iterator
@@ -37,21 +37,21 @@ type resolver interface {
 	RequestENR(*enode.Node) (*enode.Node, error)
 }
 
-func newCrawler(input p2p.NodeSet, disc resolver, iters ...enode.Iterator) *crawler {
+func newCrawler(input []*enode.Node, disc resolver, iters ...enode.Iterator) *crawler {
 	c := &crawler{
 		input:     input,
 		output:    make(p2p.NodeSet, len(input)),
 		disc:      disc,
 		iters:     iters,
-		inputIter: enode.IterNodes(input.Nodes()),
+		inputIter: enode.IterNodes(input),
 		ch:        make(chan *enode.Node),
 		closed:    make(chan struct{}),
 	}
 	c.iters = append(c.iters, c.inputIter)
 	// Copy input to output initially. Any nodes that fail validation
 	// will be dropped from output during the run.
-	for id, n := range input {
-		c.output[id] = n
+	for _, n := range input {
+		c.output[n.ID()] = n.URLv4()
 	}
 	return c
 }
@@ -74,10 +74,8 @@ func (c *crawler) run(timeout time.Duration, nthreads int) p2p.NodeSet {
 	}
 	var (
 		added   uint64
-		updated uint64
 		skipped uint64
 		recent  uint64
-		removed uint64
 		wg      sync.WaitGroup
 	)
 	wg.Add(nthreads)
@@ -92,12 +90,8 @@ func (c *crawler) run(timeout time.Duration, nthreads int) p2p.NodeSet {
 						atomic.AddUint64(&skipped, 1)
 					case nodeSkipRecent:
 						atomic.AddUint64(&recent, 1)
-					case nodeRemoved:
-						atomic.AddUint64(&removed, 1)
 					case nodeAdded:
 						atomic.AddUint64(&added, 1)
-					default:
-						atomic.AddUint64(&updated, 1)
 					}
 				case <-c.closed:
 					return
@@ -125,9 +119,7 @@ loop:
 		case <-statusTicker.C:
 			log.Info().
 				Uint64("added", atomic.LoadUint64(&added)).
-				Uint64("updated", atomic.LoadUint64(&updated)).
-				Uint64("removed", atomic.LoadUint64(&removed)).
-				Uint64("ignored(recent)", atomic.LoadUint64(&removed)).
+				Uint64("ignored(recent)", atomic.LoadUint64(&recent)).
 				Uint64("ignored(incompatible)", atomic.LoadUint64(&skipped)).
 				Msg("Crawling in progress")
 		}
@@ -184,12 +176,10 @@ func shouldSkipNode(n *enode.Node) bool {
 // what changed.
 func (c *crawler) updateNode(n *enode.Node) int {
 	c.mu.RLock()
-	node, ok := c.output[n.ID()]
+	_, ok := c.output[n.ID()]
 	c.mu.RUnlock()
 
-	// Skip validation of recently-seen nodes.
-	if ok && time.Since(node.LastCheck) < c.revalidateInterval {
-		log.Debug().Str("id", n.ID().String()).Msg("Skipping node")
+	if ok {
 		return nodeSkipRecent
 	}
 
@@ -198,43 +188,15 @@ func (c *crawler) updateNode(n *enode.Node) int {
 		return nodeSkipIncompat
 	}
 
-	// Request the node record.
-	status := nodeUpdated
-	node.LastCheck = truncNow()
-
-	if nn, err := c.disc.RequestENR(n); err != nil {
-		if node.Score == 0 {
-			// Node doesn't implement EIP-868.
-			log.Debug().Str("id", n.ID().String()).Msg("Skipping node")
-			return nodeSkipIncompat
-		}
-		node.Score /= 2
-	} else {
-		node.N = nn
-		node.Seq = nn.Seq()
-		node.Score++
-		if node.FirstResponse.IsZero() {
-			node.FirstResponse = node.LastCheck
-			status = nodeAdded
-		}
-		node.LastResponse = node.LastCheck
+	nn, err := c.disc.RequestENR(n)
+	if err != nil {
+		return nodeSkipIncompat
 	}
 
 	// Store/update node in output set.
 	c.mu.Lock()
-	defer c.mu.Unlock()
-
-	if node.Score <= 0 {
-		log.Debug().Str("id", n.ID().String()).Msg("Removing node")
-		delete(c.output, n.ID())
-		return nodeRemoved
-	}
-
-	log.Debug().Str("id", n.ID().String()).Uint64("seq", n.Seq()).Int("score", node.Score).Msg("Updating node")
-	c.output[n.ID()] = node
-	return status
-}
+	c.output[nn.ID()] = nn.URLv4()
+	c.mu.Unlock()
 
-func truncNow() time.Time {
-	return time.Now().UTC().Truncate(1 * time.Second)
+	return nodeAdded
 }

cmd/p2p/ping/ping.go

@@ -39,15 +39,14 @@ var PingCmd = &cobra.Command{
 	Long: `Ping nodes by either giving a single enode/enr or an entire nodes file.
 
 This command will establish a handshake and status exchange to get the Hello and
-Status messages and output JSON. If providing a enode/enr rather than a node file,
-then the connection will remain open by default (--listen=true), and you can see
-other messages the peer sends (e.g. blocks, transactions, etc.).`,
+Status messages and output JSON. If providing a enode/enr rather than a nodes
+file, then the connection will remain open by default (--listen=true), and you
+can see other messages the peer sends (e.g. blocks, transactions, etc.).`,
 	Args: cobra.MinimumNArgs(1),
 	RunE: func(cmd *cobra.Command, args []string) error {
 		nodes := []*enode.Node{}
-		if inputSet, err := p2p.LoadNodesJSON(args[0]); err == nil {
-			nodes = inputSet.Nodes()
-			inputPingParams.Listen = false
+		if input, err := p2p.ReadNodeSet(args[0]); err == nil {
+			nodes = input
 		} else if node, err := p2p.ParseNode(args[0]); err == nil {
 			nodes = append(nodes, node)
 		} else {
@@ -65,7 +64,17 @@ other messages the peer sends (e.g. blocks, transactions, etc.).`,
 		sem := make(chan bool, inputPingParams.Threads)
 
 		count := &p2p.MessageCount{}
-		go p2p.LogMessageCount(count, time.NewTicker(time.Second))
+		go func() {
+			ticker := time.NewTicker(2 * time.Second)
+			for {
+				<-ticker.C
+				c := count.Load()
+				if !c.IsEmpty() {
+					log.Info().Interface("counts", c).Send()
+					count.Clear()
+				}
+			}
+		}()
 
 		// Ping each node in the slice.
 		for _, n := range nodes {
@@ -98,14 +107,19 @@ other messages the peer sends (e.g. blocks, transactions, etc.).`,
 					errStr = err.Error()
 				} else if inputPingParams.Listen {
 					// If the dial and peering were successful, listen to the peer for messages.
-					if err := conn.ReadAndServe(nil, count); err != nil {
+					if err := conn.ReadAndServe(count); err != nil {
 						log.Error().Err(err).Msg("Received error")
 					}
 				}
 
 				// Save the results to the output map.
 				mutex.Lock()
-				output[node.ID()] = pingNodeJSON{node, hello, status, errStr}
+				output[node.ID()] = pingNodeJSON{
+					Record: node,
+					Hello:  hello,
+					Status: status,
+					Error:  errStr,
+				}
 				mutex.Unlock()
 			}(n)
 		}
@@ -128,8 +142,8 @@ other messages the peer sends (e.g. blocks, transactions, etc.).`,
 }
 
 func init() {
-	PingCmd.PersistentFlags().StringVarP(&inputPingParams.OutputFile, "output", "o", "", "Write ping results to output file. (default stdout)")
-	PingCmd.PersistentFlags().IntVarP(&inputPingParams.Threads, "parallel", "p", 16, "How many parallel pings to attempt.")
+	PingCmd.PersistentFlags().StringVarP(&inputPingParams.OutputFile, "output", "o", "", "Write ping results to output file (default stdout)")
+	PingCmd.PersistentFlags().IntVarP(&inputPingParams.Threads, "parallel", "p", 16, "How many parallel pings to attempt")
 	PingCmd.PersistentFlags().BoolVarP(&inputPingParams.Listen, "listen", "l", true,
 		`Keep the connection open and listen to the peer. This only works if the first
 argument is an enode/enr, not a nodes file.`)