[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, run `echo \"[]\" >> nodes.json` to get started.",
 	Args:  cobra.MinimumNArgs(1),
 	PreRunE: func(cmd *cobra.Command, args []string) (err error) {
 		inputCrawlParams.NodesFile = args[0]
@@ -54,7 +55,7 @@ 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
 		}
@@ -84,13 +85,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)
 	},
 }
 

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,14 +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()
+		if input, err := p2p.ReadNodeSet(args[0]); err == nil {
+			nodes = input
 			inputPingParams.Listen = false
 		} else if node, err := p2p.ParseNode(args[0]); err == nil {
 			nodes = append(nodes, node)
@@ -98,14 +98,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)
 		}