vdn: support init network, add event loop;

cmd/geth/chaincmd.go

@@ -31,6 +31,8 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/libp2p/go-libp2p/core/peer"
+
 	"github.com/olekukonko/tablewriter"
 	"github.com/urfave/cli/v2"
 
@@ -83,6 +85,7 @@ It expects the genesis file as argument.`,
 		Flags: []cli.Flag{
 			utils.InitNetworkDir,
 			utils.InitNetworkPort,
+			utils.InitVDNPort,
 			utils.InitNetworkSize,
 			utils.InitNetworkIps,
 			configFileFlag,
@@ -362,9 +365,10 @@ func createNodeConfig(baseConfig gethConfig, enodes []*enode.Node, ip string, po
 }
 
 // Create configs for nodes in the cluster
-func createNodeConfigs(baseConfig gethConfig, initDir string, ips []string, ports []int, size int) ([]gethConfig, error) {
+func createNodeConfigs(baseConfig gethConfig, initDir string, ips []string, ports []int, vdnPorts []int, size int) ([]gethConfig, error) {
 	// Create the nodes
 	enodes := make([]*enode.Node, size)
+	peerIDs := make([]peer.ID, size)
 	for i := 0; i < size; i++ {
 		nodeConfig := baseConfig.Node
 		nodeConfig.DataDir = path.Join(initDir, fmt.Sprintf("node%d", i))
@@ -374,13 +378,25 @@ func createNodeConfigs(baseConfig gethConfig, initDir string, ips []string, port
 		}
 		pk := stack.Config().NodeKey()
 		enodes[i] = enode.NewV4(&pk.PublicKey, net.ParseIP(ips[i]), ports[i], ports[i])
+		stack.Config().VDN.NodeKeyPath = stack.Config().VDNNodeKeyPath()
+		peerID, _, err := stack.Config().VDN.LoadPrivateKey()
+		if err != nil {
+			return nil, err
+		}
+		peerIDs[i] = peerID
 	}
 
 	// Create the configs
 	configs := make([]gethConfig, size)
 	for i := 0; i < size; i++ {
 		configs[i] = createNodeConfig(baseConfig, enodes, ips[i], ports[i], size, i)
 	}
+
+	// set VDN bootnode, using the first node as bootnode default.
+	// TODO(galaio): may using enode url than multiaddr url in future?
+	for i := 1; i < size; i++ {
+		configs[i].Node.VDN.BootstrapPeers = []string{fmt.Sprintf("/ip4/%v/tcp/%d/p2p/%v", ips[0], vdnPorts[0], peerIDs[0])}
+	}
 	return configs, nil
 }
 
@@ -398,6 +414,10 @@ func initNetwork(ctx *cli.Context) error {
 	if port <= 0 {
 		utils.Fatalf("port should be greater than 0")
 	}
+	vdnPort := ctx.Int(utils.InitVDNPort.Name)
+	if vdnPort <= 0 {
+		utils.Fatalf("VDN port should be greater than 0")
+	}
 	ipStr := ctx.String(utils.InitNetworkIps.Name)
 	cfgFile := ctx.String(configFileFlag.Name)
 
@@ -411,6 +431,7 @@ func initNetwork(ctx *cli.Context) error {
 	}
 
 	ports := createPorts(ipStr, port, size)
+	vdnPorts := createPorts(ipStr, vdnPort, size)
 
 	// Make sure we have a valid genesis JSON
 	genesisPath := ctx.Args().First()
@@ -435,7 +456,7 @@ func initNetwork(ctx *cli.Context) error {
 		return err
 	}
 
-	configs, err := createNodeConfigs(config, initDir, ips, ports, size)
+	configs, err := createNodeConfigs(config, initDir, ips, ports, vdnPorts, size)
 	if err != nil {
 		utils.Fatalf("Failed to create node configs: %v", err)
 	}

cmd/utils/flags.go

@@ -24,7 +24,6 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"github.com/ethereum/go-ethereum/vdn"
 	"math"
 	"math/big"
 	"net"
@@ -37,6 +36,8 @@ import (
 	"strings"
 	"time"
 
+	"github.com/ethereum/go-ethereum/vdn"
+
 	"github.com/ethereum/go-ethereum/accounts"
 	"github.com/ethereum/go-ethereum/accounts/keystore"
 	"github.com/ethereum/go-ethereum/beacon/fakebeacon"
@@ -1156,6 +1157,11 @@ Please note that --` + MetricsHTTPFlag.Name + ` must be set to start the server.
 		Usage: "the p2p port of the nodes in the network",
 		Value: 30311,
 	}
+	InitVDNPort = &cli.IntFlag{
+		Name:  "init.vdn-port",
+		Usage: "the vdn p2p port of the nodes in the network",
+		Value: 13000,
+	}
 	MetricsInfluxDBOrganizationFlag = &cli.StringFlag{
 		Name:     "metrics.influxdb.organization",
 		Usage:    "InfluxDB organization name (v2 only)",
@@ -1694,10 +1700,10 @@ func SetValidatorP2PConfig(ctx *cli.Context, cfg *vdn.Config) {
 		cfg.QUICPort = ctx.Int(VDNQuicPortFlag.Name)
 	}
 	if ctx.IsSet(VDNBootnodesFlag.Name) {
-		cfg.BootStrapAddrs = SplitAndTrim(ctx.String(VDNBootnodesFlag.Name))
+		cfg.BootstrapPeers = SplitAndTrim(ctx.String(VDNBootnodesFlag.Name))
 	}
 	if ctx.IsSet(VDNNodeKeyFileFlag.Name) {
-		cfg.PrivateKeyPath = ctx.String(VDNNodeKeyFileFlag.Name)
+		cfg.NodeKeyPath = ctx.String(VDNNodeKeyFileFlag.Name)
 	}
 }
 

node/config.go

@@ -40,7 +40,7 @@ const (
 	datadirStaticNodes     = "static-nodes.json"  // Path within the datadir to the static node list
 	datadirTrustedNodes    = "trusted-nodes.json" // Path within the datadir to the trusted node list
 	datadirNodeDatabase    = "nodes"              // Path within the datadir to store the node infos
-	datadirVDNPrivateKey   = "nodekey_vdn"
+	datadirVDNNodeKey      = "nodekey_vdn"
 )
 
 // Config represents a small collection of configuration values to fine tune the
@@ -359,6 +359,10 @@ func (c *Config) name() string {
 	return c.Name
 }
 
+func (c *Config) VDNNodeKeyPath() string {
+	return c.ResolvePath(datadirVDNNodeKey)
+}
+
 // These resources are resolved differently for "geth" instances.
 var isOldGethResource = map[string]bool{
 	"chaindata":          true,

node/defaults.go

@@ -80,7 +80,7 @@ var DefaultConfig = Config{
 	Instance: 1,
 	VDN: vdn.Config{
 		HostAddress:      "127.0.0.1",
-		PrivateKeyPath:   "", // by default, it will be auto set PrivateKeyPath/nodekey_vdn
+		NodeKeyPath:      "", // by default, it will be auto set DataDir/geth/nodekey_vdn
 		QUICPort:         13000,
 		TCPPort:          13000,
 		MaxPeers:         50,

node/node.go

@@ -208,8 +208,8 @@ func New(conf *Config) (*Node, error) {
 
 	// Configure validator p2p server.
 	if conf.EnableMining {
-		if conf.VDN.PrivateKeyPath == "" {
-			conf.VDN.PrivateKeyPath = conf.ResolvePath(datadirVDNPrivateKey)
+		if conf.VDN.NodeKeyPath == "" {
+			conf.VDN.NodeKeyPath = conf.VDNNodeKeyPath()
 		}
 		node.valServer, err = vdn.NewServer(&conf.VDN)
 		if err != nil {

vdn/config.go

@@ -1,7 +1,11 @@
 package vdn
 
 import (
+	"crypto/ecdsa"
+
 	"github.com/ethereum/go-ethereum/log"
+	"github.com/libp2p/go-libp2p/core/peer"
+	"github.com/pkg/errors"
 )
 
 const (
@@ -11,9 +15,9 @@ const (
 
 type Config struct {
 	StaticPeers      []string // using multi address format
-	BootStrapAddrs   []string
+	BootstrapPeers   []string
 	HostAddress      string // it indicates listen IP addr, it better a external IP for public service
-	PrivateKeyPath   string // the path saves the key and cached nodes
+	NodeKeyPath      string // the path saves the key and cached nodes
 	QUICPort         int
 	TCPPort          int
 	MaxPeers         int
@@ -25,16 +29,32 @@ type Config struct {
 
 func (cfg *Config) SanityCheck() error {
 	if cfg.QueueSize == 0 {
-		log.Warn("Invalid pubsub queue size of %d, set %d", cfg.QueueSize, defaultPubsubQueueSize)
+		log.Warn("Invalid pubsub queue size", "input", cfg.QueueSize, "default", defaultPubsubQueueSize)
 		cfg.QueueSize = defaultPubsubQueueSize
 	}
 	if cfg.MaxPeers == 0 {
-		log.Warn("Invalid MaxPeers size of %d, set %d", cfg.QueueSize, defaultMaxPeers)
+		log.Warn("Invalid MaxPeers size", "input", cfg.MaxPeers, "default", defaultMaxPeers)
 		cfg.MaxPeers = defaultMaxPeers
 	}
-	if len(cfg.BootStrapAddrs) == 0 && cfg.EnableDiscovery {
-		log.Warn("Invalid BootStrapAddrs, disable discovery")
+	if len(cfg.BootstrapPeers) == 0 && cfg.EnableDiscovery {
+		log.Warn("Invalid BootstrapPeers, disable discovery")
 		cfg.EnableDiscovery = false
 	}
 	return nil
 }
+
+func (cfg *Config) LoadPrivateKey() (peer.ID, *ecdsa.PrivateKey, error) {
+	priv, err := LoadPrivateKey(cfg.NodeKeyPath)
+	if err != nil {
+		return "", nil, errors.Wrap(err, "failed to load node key")
+	}
+	ipriv, err := ConvertToInterfacePrivkey(priv)
+	if err != nil {
+		return "", nil, errors.Wrap(err, "failed to convert node key")
+	}
+	peerId, err := peer.IDFromPrivateKey(ipriv)
+	if err != nil {
+		return "", nil, errors.Wrap(err, "failed to parse peer id")
+	}
+	return peerId, priv, nil
+}

vdn/connection_gater.go

@@ -33,10 +33,7 @@ func (*Server) InterceptPeerDial(_ peer.ID) (allow bool) {
 // multiaddr for the given peer.
 func (s *Server) InterceptAddrDial(pid peer.ID, m multiaddr.Multiaddr) (allow bool) {
 	// Disallow bad peers from dialing in.
-	if s.watcher.IsBad(pid) != nil {
-		return false
-	}
-	return true
+	return s.watcher.IsBad(pid) == nil
 }
 
 // InterceptAccept checks whether the incidental inbound connection is allowed.

vdn/crypto.go

@@ -18,10 +18,10 @@ func LoadPrivateKey(keyPath string) (*ecdsa.PrivateKey, error) {
 	}
 	priv, err := crypto.GenerateKey()
 	if err != nil {
-		return nil, errors.Wrapf(err, "GenerateKey err")
+		return nil, errors.Wrap(err, "GenerateKey err")
 	}
 	if err := crypto.SaveECDSA(keyPath, priv); err != nil {
-		return nil, errors.Wrapf(err, "SaveKey err")
+		return nil, errors.Wrap(err, "SaveKey err")
 	}
 	return priv, nil
 }

vdn/gossip_scoring_params.go

@@ -82,7 +82,8 @@ func (s *Server) topicScoreParams(topic string) (*pubsub.TopicScoreParams, error
 // Based on the lighthouse beacon block parameters.
 // https://gist.github.com/blacktemplar/5c1862cb3f0e32a1a7fb0b25e79e6e2c
 func defaultBlockTopicParams() *pubsub.TopicScoreParams {
-	decayEpoch := time.Duration(5)
+	decayEpoch := 5
+	decayEpochDuration := 5 * oneEpochDuration()
 	blocksInEpoch := blocksPerEpoch()
 	meshWeight := -0.717
 	invalidDecayPeriod := 50 * oneEpochDuration()
@@ -100,13 +101,13 @@ func defaultBlockTopicParams() *pubsub.TopicScoreParams {
 		FirstMessageDeliveriesDecay:     scoreDecay(20 * oneEpochDuration()),
 		FirstMessageDeliveriesCap:       23,
 		MeshMessageDeliveriesWeight:     meshWeight,
-		MeshMessageDeliveriesDecay:      scoreDecay(decayEpoch * oneEpochDuration()),
+		MeshMessageDeliveriesDecay:      scoreDecay(decayEpochDuration),
 		MeshMessageDeliveriesCap:        float64(blocksInEpoch * uint64(decayEpoch)),
 		MeshMessageDeliveriesThreshold:  float64(blocksInEpoch*uint64(decayEpoch)) / 10,
 		MeshMessageDeliveriesWindow:     2 * time.Second,
 		MeshMessageDeliveriesActivation: 4 * oneEpochDuration(),
 		MeshFailurePenaltyWeight:        meshWeight,
-		MeshFailurePenaltyDecay:         scoreDecay(decayEpoch * oneEpochDuration()),
+		MeshFailurePenaltyDecay:         scoreDecay(decayEpochDuration),
 		InvalidMessageDeliveriesWeight:  -140.4475,
 		InvalidMessageDeliveriesDecay:   scoreDecay(invalidDecayPeriod),
 	}

vdn/pubsub.go

@@ -17,18 +17,11 @@ const (
 	gossipSubDhi = 12 // topic stable mesh high watermark
 
 	// gossip parameters
-	gossipSubMcacheLen    = 6   // number of windows to retain full messages in cache for `IWANT` responses
-	gossipSubMcacheGossip = 3   // number of windows to gossip about
-	gossipSubSeenTTL      = 768 // number of seconds to retain message IDs ( 2 epochs)
-
-	// fanout ttl
-	gossipSubFanoutTTL = 60000000000 // TTL for fanout maps for topics we are not subscribed to but have published to, in nano seconds
+	gossipSubMcacheLen    = 6 // number of windows to retain full messages in cache for `IWANT` responses
+	gossipSubMcacheGossip = 3 // number of windows to gossip about
 
 	// heartbeat interval
 	gossipSubHeartbeatInterval = 700 * time.Millisecond // frequency of heartbeat, milliseconds
-
-	// misc
-	rSubD = 8 // random gossip target
 )
 
 // JoinTopic will join PubSub topic, if not already joined.
@@ -109,9 +102,9 @@ func (s *Server) pubsubOptions() []pubsub.Option {
 		// Notice: Eth 2.0 uses no author, no sign & verify.
 		pubsub.WithMessageAuthor(""),
 		pubsub.WithMessageSignaturePolicy(pubsub.StrictSign),
-		pubsub.WithPeerOutboundQueueSize(int(s.cfg.QueueSize)),
+		pubsub.WithPeerOutboundQueueSize(s.cfg.QueueSize),
 		pubsub.WithMaxMessageSize(GossipMaxSize),
-		pubsub.WithValidateQueueSize(int(s.cfg.QueueSize)),
+		pubsub.WithValidateQueueSize(s.cfg.QueueSize),
 		pubsub.WithPeerScore(peerScoringParams()),
 		pubsub.WithPeerScoreInspect(s.peerInspector, time.Minute),
 		pubsub.WithGossipSubParams(pubsubGossipParam()),