Merge pull request #1740 from filecoin-project/feat/bootstrapper-profile

add profile for bootstrappers

cmd/lotus/daemon.go

@@ -33,6 +33,7 @@ import (
 	"github.com/filecoin-project/lotus/metrics"
 	"github.com/filecoin-project/lotus/node"
 	"github.com/filecoin-project/lotus/node/modules"
+	"github.com/filecoin-project/lotus/node/modules/dtypes"
 	"github.com/filecoin-project/lotus/node/modules/testing"
 	"github.com/filecoin-project/lotus/node/repo"
 	"github.com/filecoin-project/sector-storage/ffiwrapper"
@@ -86,6 +87,10 @@ var DaemonCmd = &cli.Command{
 			Name:  "pprof",
 			Usage: "specify name of file for writing cpu profile to",
 		},
+		&cli.StringFlag{
+			Name:  "profile",
+			Usage: "specify type of node",
+		},
 	},
 	Action: func(cctx *cli.Context) error {
 		if prof := cctx.String("pprof"); prof != "" {
@@ -100,6 +105,16 @@ var DaemonCmd = &cli.Command{
 			defer pprof.StopCPUProfile()
 		}
 
+		var isBootstrapper dtypes.Bootstrapper
+		switch profile := cctx.String("profile"); profile {
+		case "bootstrapper":
+			isBootstrapper = true
+		case "":
+			// do nothing
+		default:
+			return fmt.Errorf("unrecognized profile type: %q", profile)
+		}
+
 		ctx, _ := tag.New(context.Background(), tag.Insert(metrics.Version, build.BuildVersion), tag.Insert(metrics.Commit, build.CurrentCommit))
 		{
 			dir, err := homedir.Expand(cctx.String("repo"))
@@ -160,6 +175,7 @@ var DaemonCmd = &cli.Command{
 		stop, err := node.New(ctx,
 			node.FullAPI(&api),
 
+			node.Override(new(dtypes.Bootstrapper), isBootstrapper),
 			node.Online(),
 			node.Repo(r),
 

node/builder.go

@@ -142,12 +142,14 @@ type Settings struct {
 
 	Online bool // Online option applied
 	Config bool // Config option applied
+
 }
 
 func defaults() []Option {
 	return []Option{
 		Override(new(helpers.MetricsCtx), context.Background),
 		Override(new(record.Validator), modules.RecordValidator),
+		Override(new(dtypes.Bootstrapper), dtypes.Bootstrapper(false)),
 
 		// Filecoin modules
 
@@ -178,7 +180,12 @@ func libp2p() Option {
 		Override(ConnectionManagerKey, lp2p.ConnectionManager(50, 200, 20*time.Second, nil)),
 		Override(AutoNATSvcKey, lp2p.AutoNATService),
 
-		Override(new(*pubsub.PubSub), lp2p.GossipSub(&config.Pubsub{})),
+		Override(new(*pubsub.PubSub), lp2p.GossipSub),
+		Override(new(*config.Pubsub), func(bs dtypes.Bootstrapper) *config.Pubsub {
+			return &config.Pubsub{
+				Bootstrapper: bool(bs),
+			}
+		}),
 
 		Override(PstoreAddSelfKeysKey, lp2p.PstoreAddSelfKeys),
 		Override(StartListeningKey, lp2p.StartListening(config.DefaultFullNode().Libp2p.ListenAddresses)),
@@ -354,7 +361,8 @@ func ConfigCommon(cfg *config.Common) Option {
 				cfg.Libp2p.ConnMgrHigh,
 				time.Duration(cfg.Libp2p.ConnMgrGrace),
 				cfg.Libp2p.ProtectedPeers)),
-			Override(new(*pubsub.PubSub), lp2p.GossipSub(&cfg.Pubsub)),
+			Override(new(*pubsub.PubSub), lp2p.GossipSub),
+			Override(new(*config.Pubsub), &cfg.Pubsub),
 
 			ApplyIf(func(s *Settings) bool { return len(cfg.Libp2p.BootstrapPeers) > 0 },
 				Override(new(dtypes.BootstrapPeers), modules.ConfigBootstrap(cfg.Libp2p.BootstrapPeers)),

node/modules/dtypes/bootstrap.go

@@ -3,3 +3,5 @@ package dtypes
 import "github.com/libp2p/go-libp2p-core/peer"
 
 type BootstrapPeers []peer.AddrInfo
+
+type Bootstrapper bool

node/modules/lp2p/host.go

@@ -71,19 +71,23 @@ func MockHost(mn mocknet.Mocknet, id peer.ID, ps peerstore.Peerstore) (RawHost,
 }
 
 func DHTRouting(mode dht.ModeOpt) interface{} {
-	return func(mctx helpers.MetricsCtx, lc fx.Lifecycle, host RawHost, dstore dtypes.MetadataDS, validator record.Validator, nn dtypes.NetworkName) (BaseIpfsRouting, error) {
+	return func(mctx helpers.MetricsCtx, lc fx.Lifecycle, host RawHost, dstore dtypes.MetadataDS, validator record.Validator, nn dtypes.NetworkName, bs dtypes.Bootstrapper) (BaseIpfsRouting, error) {
 		ctx := helpers.LifecycleCtx(mctx, lc)
 
-		d, err := dht.New(
+		if bs {
-			ctx, host,
+			mode = dht.ModeServer
-			dht.Mode(mode),
+		}
+
+		opts := []dht.Option{dht.Mode(mode),
 			dht.Datastore(dstore),
 			dht.Validator(validator),
 			dht.ProtocolPrefix(build.DhtProtocolName(nn)),
 			dht.QueryFilter(dht.PublicQueryFilter),
 			dht.RoutingTableFilter(dht.PublicRoutingTableFilter),
 			dht.DisableProviders(),
-			dht.DisableValues(),
+			dht.DisableValues()}
+		d, err := dht.New(
+			ctx, host, opts...,
 		)
 
 		if err != nil {

node/modules/lp2p/pubsub.go

@@ -28,160 +28,141 @@ func init() {
 	pubsub.GossipSubDirectConnectInitialDelay = 30 * time.Second
 }
 
-func GossipSub(cfg *config.Pubsub) interface{} {
+func GossipSub(mctx helpers.MetricsCtx, lc fx.Lifecycle, host host.Host, nn dtypes.NetworkName, bp dtypes.BootstrapPeers, cfg *config.Pubsub) (service *pubsub.PubSub, err error) {
-	return func(mctx helpers.MetricsCtx, lc fx.Lifecycle, host host.Host, nn dtypes.NetworkName, bp dtypes.BootstrapPeers) (service *pubsub.PubSub, err error) {
+	bootstrappers := make(map[peer.ID]struct{})
-		bootstrappers := make(map[peer.ID]struct{})
+	for _, pi := range bp {
-		for _, pi := range bp {
+		bootstrappers[pi.ID] = struct{}{}
-			bootstrappers[pi.ID] = struct{}{}
+	}
-		}
+	isBootstrapNode := cfg.Bootstrapper
-		isBootstrapNode := cfg.Bootstrapper
 
-		options := []pubsub.Option{
+	options := []pubsub.Option{
-			// Gossipsubv1.1 configuration
+		// Gossipsubv1.1 configuration
-			pubsub.WithFloodPublish(true),
+		pubsub.WithFloodPublish(true),
-			pubsub.WithPeerScore(
+		pubsub.WithPeerScore(
-				&pubsub.PeerScoreParams{
+			&pubsub.PeerScoreParams{
-					AppSpecificScore: func(p peer.ID) float64 {
+				AppSpecificScore: func(p peer.ID) float64 {
-						// return a heavy positive score for bootstrappers so that we don't unilaterally prune
+					// return a heavy positive score for bootstrappers so that we don't unilaterally prune
-						// them and accept PX from them.
+					// them and accept PX from them.
-						// we don't do that in the bootstrappers themselves to avoid creating a closed mesh
+					// we don't do that in the bootstrappers themselves to avoid creating a closed mesh
-						// between them (however we might want to consider doing just that)
+					// between them (however we might want to consider doing just that)
-						_, ok := bootstrappers[p]
+					_, ok := bootstrappers[p]
-						if ok && !isBootstrapNode {
+					if ok && !isBootstrapNode {
-							return 2500
+						return 2500
-						}
+					}
 
-						// TODO: we want to  plug the application specific score to the node itself in order
+					// TODO: we want to  plug the application specific score to the node itself in order
-						//       to provide feedback to the pubsub system based on observed behaviour
+					//       to provide feedback to the pubsub system based on observed behaviour
-						return 0
+					return 0
+				},
+				AppSpecificWeight: 1,
+
+				// This sets the IP colocation threshold to 1 peer per
+				IPColocationFactorThreshold: 1,
+				IPColocationFactorWeight:    -100,
+				// TODO we want to whitelist IPv6 /64s that belong to datacenters etc
+				// IPColocationFactorWhitelist: map[string]struct{}{},
+
+				DecayInterval: pubsub.DefaultDecayInterval,
+				DecayToZero:   pubsub.DefaultDecayToZero,
+
+				// this retains non-positive scores for 6 hours
+				RetainScore: 6 * time.Hour,
+
+				// topic parameters
+				Topics: map[string]*pubsub.TopicScoreParams{
+					build.BlocksTopic(nn): {
+						// expected 10 blocks/min
+						TopicWeight: 0.1, // max is 50, max mesh penalty is -10, single invalid message is -100
+
+						// 1 tick per second, maxes at 1 after 1 hour
+						TimeInMeshWeight:  0.00027, // ~1/3600
+						TimeInMeshQuantum: time.Second,
+						TimeInMeshCap:     1,
+
+						// deliveries decay after 1 hour, cap at 100 blocks
+						FirstMessageDeliveriesWeight: 5, // max value is 500
+						FirstMessageDeliveriesDecay:  pubsub.ScoreParameterDecay(time.Hour),
+						FirstMessageDeliveriesCap:    100, // 100 blocks in an hour
+
+						// tracks deliveries in the last minute
+						// penalty activates at 1 minute and expects ~0.4 blocks
+						MeshMessageDeliveriesWeight:     -576, // max penalty is -100
+						MeshMessageDeliveriesDecay:      pubsub.ScoreParameterDecay(time.Minute),
+						MeshMessageDeliveriesCap:        10,      // 10 blocks in a minute
+						MeshMessageDeliveriesThreshold:  0.41666, // 10/12/2 blocks/min
+						MeshMessageDeliveriesWindow:     10 * time.Millisecond,
+						MeshMessageDeliveriesActivation: time.Minute,
+
+						// decays after 15 min
+						MeshFailurePenaltyWeight: -576,
+						MeshFailurePenaltyDecay:  pubsub.ScoreParameterDecay(15 * time.Minute),
+
+						// invalid messages decay after 1 hour
+						InvalidMessageDeliveriesWeight: -1000,
+						InvalidMessageDeliveriesDecay:  pubsub.ScoreParameterDecay(time.Hour),
 					},
-					AppSpecificWeight: 1,
+					build.MessagesTopic(nn): {
+						// expected > 1 tx/second
+						TopicWeight: 0.05, // max is 25, max mesh penalty is -5, single invalid message is -100
 
-					// This sets the IP colocation threshold to 1 peer per
+						// 1 tick per second, maxes at 1 hour
-					IPColocationFactorThreshold: 1,
+						TimeInMeshWeight:  0.0002778, // ~1/3600
-					IPColocationFactorWeight:    -100,
+						TimeInMeshQuantum: time.Second,
-					// TODO we want to whitelist IPv6 /64s that belong to datacenters etc
+						TimeInMeshCap:     1,
-					// IPColocationFactorWhitelist: map[string]struct{}{},
 
-					DecayInterval: pubsub.DefaultDecayInterval,
+						// deliveries decay after 10min, cap at 1000 tx
-					DecayToZero:   pubsub.DefaultDecayToZero,
+						FirstMessageDeliveriesWeight: 0.5, // max value is 500
+						FirstMessageDeliveriesDecay:  pubsub.ScoreParameterDecay(10 * time.Minute),
+						FirstMessageDeliveriesCap:    1000,
 
-					// this retains non-positive scores for 6 hours
+						// tracks deliveries in the last minute
-					RetainScore: 6 * time.Hour,
+						// penalty activates at 1 min and expects 2.5 txs
+						MeshMessageDeliveriesWeight:     -16, // max penalty is -100
+						MeshMessageDeliveriesDecay:      pubsub.ScoreParameterDecay(time.Minute),
+						MeshMessageDeliveriesCap:        100, // 100 txs in a minute
+						MeshMessageDeliveriesThreshold:  2.5, // 60/12/2 txs/minute
+						MeshMessageDeliveriesWindow:     10 * time.Millisecond,
+						MeshMessageDeliveriesActivation: time.Minute,
 
-					// topic parameters
+						// decays after 5min
-					Topics: map[string]*pubsub.TopicScoreParams{
+						MeshFailurePenaltyWeight: -16,
-						build.BlocksTopic(nn): {
+						MeshFailurePenaltyDecay:  pubsub.ScoreParameterDecay(5 * time.Minute),
-							// expected 10 blocks/min
-							TopicWeight: 0.1, // max is 50, max mesh penalty is -10, single invalid message is -100
 
-							// 1 tick per second, maxes at 1 after 1 hour
+						// invalid messages decay after 1 hour
-							TimeInMeshWeight:  0.00027, // ~1/3600
+						InvalidMessageDeliveriesWeight: -2000,
-							TimeInMeshQuantum: time.Second,
+						InvalidMessageDeliveriesDecay:  pubsub.ScoreParameterDecay(time.Hour),
-							TimeInMeshCap:     1,
-
-							// deliveries decay after 1 hour, cap at 100 blocks
-							FirstMessageDeliveriesWeight: 5, // max value is 500
-							FirstMessageDeliveriesDecay:  pubsub.ScoreParameterDecay(time.Hour),
-							FirstMessageDeliveriesCap:    100, // 100 blocks in an hour
-
-							// tracks deliveries in the last minute
-							// penalty activates at 1 minute and expects ~0.4 blocks
-							MeshMessageDeliveriesWeight:     -576, // max penalty is -100
-							MeshMessageDeliveriesDecay:      pubsub.ScoreParameterDecay(time.Minute),
-							MeshMessageDeliveriesCap:        10,      // 10 blocks in a minute
-							MeshMessageDeliveriesThreshold:  0.41666, // 10/12/2 blocks/min
-							MeshMessageDeliveriesWindow:     10 * time.Millisecond,
-							MeshMessageDeliveriesActivation: time.Minute,
-
-							// decays after 15 min
-							MeshFailurePenaltyWeight: -576,
-							MeshFailurePenaltyDecay:  pubsub.ScoreParameterDecay(15 * time.Minute),
-
-							// invalid messages decay after 1 hour
-							InvalidMessageDeliveriesWeight: -1000,
-							InvalidMessageDeliveriesDecay:  pubsub.ScoreParameterDecay(time.Hour),
-						},
-						build.MessagesTopic(nn): {
-							// expected > 1 tx/second
-							TopicWeight: 0.05, // max is 25, max mesh penalty is -5, single invalid message is -100
-
-							// 1 tick per second, maxes at 1 hour
-							TimeInMeshWeight:  0.0002778, // ~1/3600
-							TimeInMeshQuantum: time.Second,
-							TimeInMeshCap:     1,
-
-							// deliveries decay after 10min, cap at 1000 tx
-							FirstMessageDeliveriesWeight: 0.5, // max value is 500
-							FirstMessageDeliveriesDecay:  pubsub.ScoreParameterDecay(10 * time.Minute),
-							FirstMessageDeliveriesCap:    1000,
-
-							// tracks deliveries in the last minute
-							// penalty activates at 1 min and expects 2.5 txs
-							MeshMessageDeliveriesWeight:     -16, // max penalty is -100
-							MeshMessageDeliveriesDecay:      pubsub.ScoreParameterDecay(time.Minute),
-							MeshMessageDeliveriesCap:        100, // 100 txs in a minute
-							MeshMessageDeliveriesThreshold:  2.5, // 60/12/2 txs/minute
-							MeshMessageDeliveriesWindow:     10 * time.Millisecond,
-							MeshMessageDeliveriesActivation: time.Minute,
-
-							// decays after 5min
-							MeshFailurePenaltyWeight: -16,
-							MeshFailurePenaltyDecay:  pubsub.ScoreParameterDecay(5 * time.Minute),
-
-							// invalid messages decay after 1 hour
-							InvalidMessageDeliveriesWeight: -2000,
-							InvalidMessageDeliveriesDecay:  pubsub.ScoreParameterDecay(time.Hour),
-						},
 					},
 				},
-				&pubsub.PeerScoreThresholds{
+			},
-					GossipThreshold:             -500,
+			&pubsub.PeerScoreThresholds{
-					PublishThreshold:            -1000,
+				GossipThreshold:             -500,
-					GraylistThreshold:           -2500,
+				PublishThreshold:            -1000,
-					AcceptPXThreshold:           1000,
+				GraylistThreshold:           -2500,
-					OpportunisticGraftThreshold: 2.5,
+				AcceptPXThreshold:           1000,
-				},
+				OpportunisticGraftThreshold: 2.5,
-			),
+			},
-		}
+		),
+	}
 
-		// enable Peer eXchange on bootstrappers
+	// enable Peer eXchange on bootstrappers
-		if isBootstrapNode {
+	if isBootstrapNode {
-			// turn off the mesh in bootstrappers -- only do gossip and PX
+		// turn off the mesh in bootstrappers -- only do gossip and PX
-			pubsub.GossipSubD = 0
+		pubsub.GossipSubD = 0
-			pubsub.GossipSubDscore = 0
+		pubsub.GossipSubDscore = 0
-			pubsub.GossipSubDlo = 0
+		pubsub.GossipSubDlo = 0
-			pubsub.GossipSubDhi = 0
+		pubsub.GossipSubDhi = 0
-			pubsub.GossipSubDlazy = 1024
+		pubsub.GossipSubDlazy = 1024
-			pubsub.GossipSubGossipFactor = 0.5
+		pubsub.GossipSubGossipFactor = 0.5
-			// turn on PX
+		// turn on PX
-			options = append(options, pubsub.WithPeerExchange(true))
+		options = append(options, pubsub.WithPeerExchange(true))
-		}
+	}
 
-		// direct peers
+	// direct peers
-		if cfg.DirectPeers != nil {
+	if cfg.DirectPeers != nil {
-			var directPeerInfo []peer.AddrInfo
+		var directPeerInfo []peer.AddrInfo
 
-			for _, addr := range cfg.DirectPeers {
+		for _, addr := range cfg.DirectPeers {
-				a, err := ma.NewMultiaddr(addr)
+			a, err := ma.NewMultiaddr(addr)
-				if err != nil {
-					return nil, err
-				}
-
-				pi, err := peer.AddrInfoFromP2pAddr(a)
-				if err != nil {
-					return nil, err
-				}
-
-				directPeerInfo = append(directPeerInfo, *pi)
-			}
-
-			options = append(options, pubsub.WithDirectPeers(directPeerInfo))
-		}
-
-		// tracer
-		if cfg.RemoteTracer != "" {
-			a, err := ma.NewMultiaddr(cfg.RemoteTracer)
 			if err != nil {
 				return nil, err
 			}
@@ -191,21 +172,38 @@ func GossipSub(cfg *config.Pubsub) interface{} {
 				return nil, err
 			}
 
-			tr, err := pubsub.NewRemoteTracer(context.TODO(), host, *pi)
+			directPeerInfo = append(directPeerInfo, *pi)
-			if err != nil {
-				return nil, err
-			}
-
-			trw := newTracerWrapper(tr)
-			options = append(options, pubsub.WithEventTracer(trw))
 		}
 
-		// TODO: we want to hook the peer score inspector so that we can gain visibility
+		options = append(options, pubsub.WithDirectPeers(directPeerInfo))
-		//       in peer scores for debugging purposes -- this might be trigged by metrics collection
-		// options = append(options, pubsub.WithPeerScoreInspect(XXX, time.Second))
-
-		return pubsub.NewGossipSub(helpers.LifecycleCtx(mctx, lc), host, options...)
 	}
+
+	// tracer
+	if cfg.RemoteTracer != "" {
+		a, err := ma.NewMultiaddr(cfg.RemoteTracer)
+		if err != nil {
+			return nil, err
+		}
+
+		pi, err := peer.AddrInfoFromP2pAddr(a)
+		if err != nil {
+			return nil, err
+		}
+
+		tr, err := pubsub.NewRemoteTracer(context.TODO(), host, *pi)
+		if err != nil {
+			return nil, err
+		}
+
+		trw := newTracerWrapper(tr)
+		options = append(options, pubsub.WithEventTracer(trw))
+	}
+
+	// TODO: we want to hook the peer score inspector so that we can gain visibility
+	//       in peer scores for debugging purposes -- this might be trigged by metrics collection
+	// options = append(options, pubsub.WithPeerScoreInspect(XXX, time.Second))
+
+	return pubsub.NewGossipSub(helpers.LifecycleCtx(mctx, lc), host, options...)
 }
 
 func HashMsgId(m *pubsub_pb.Message) string {