Feat/lastconn boot

core/core.go

@@ -82,7 +82,7 @@ type IpfsNode struct {
 	Discovery            discovery.Service         `optional:"true"`
 	FilesRoot            *mfs.Root
 	RecordValidator      record.Validator
-	//Statestore      storage.StateStorer
+	// Statestore      storage.StateStorer
 
 	// Online
 	PeerHost      p2phost.Host               `optional:"true"` // the network host (server+client)

core/node/groups.go

@@ -214,7 +214,8 @@ func Identity(cfg *config.Config) fx.Option {
 	if cfg.Identity.PrivKey == "" {
 		return fx.Options( // No PK (usually in tests)
 			fx.Provide(PeerID(id)),
-			fx.Provide(libp2p.Peerstore),
+			// fx.Provide(libp2p.Peerstore),
+			fx.Provide(libp2p.PeerstoreDs),
 		)
 	}
 
@@ -236,7 +237,8 @@ func Identity(cfg *config.Config) fx.Option {
 	return fx.Options( // Full identity
 		fx.Provide(PeerID(id)),
 		fx.Provide(PrivateKey(sk)),
-		fx.Provide(libp2p.Peerstore),
+		// fx.Provide(libp2p.Peerstore),
+		fx.Provide(libp2p.PeerstoreDs),
 
 		fx.Invoke(libp2p.PstoreAddSelfKeys),
 	)
@@ -296,6 +298,7 @@ func Online(bcfg *BuildCfg, cfg *config.Config) fx.Option {
 		fx.Provide(Namesys(ipnsCacheSize)),
 		fx.Provide(Peering),
 		PeerWith(cfg.Peering.Peers...),
+		PeerWithLastConn(),
 
 		fx.Invoke(IpnsRepublisher(repubPeriod, recordLifetime)),
 

core/node/libp2p/peerstore.go

@@ -2,6 +2,8 @@ package libp2p
 
 import (
 	"context"
+	"github.com/bittorrent/go-btfs/repo"
+	"github.com/libp2p/go-libp2p/p2p/host/peerstore/pstoreds"
 
 	"github.com/libp2p/go-libp2p/core/peerstore"
 	"github.com/libp2p/go-libp2p/p2p/host/peerstore/pstoremem"
@@ -22,3 +24,17 @@ func Peerstore(lc fx.Lifecycle) peerstore.Peerstore {
 
 	return pstore
 }
+
+func PeerstoreDs(lc fx.Lifecycle, repo repo.Repo) peerstore.Peerstore {
+	pstore, err := pstoreds.NewPeerstore(context.Background(), repo.Datastore(), pstoreds.DefaultOpts())
+	if err != nil {
+		log.Errorln(err)
+		return nil
+	}
+	lc.Append(fx.Hook{
+		OnStop: func(ctx context.Context) error {
+			return pstore.Close()
+		},
+	})
+	return pstore
+}

core/node/peering.go

@@ -2,12 +2,16 @@ package node
 
 import (
 	"context"
-
+	config "github.com/bittorrent/go-btfs-config"
 	"github.com/bittorrent/go-btfs/peering"
-
 	"github.com/libp2p/go-libp2p/core/host"
+	"github.com/libp2p/go-libp2p/core/network"
 	"github.com/libp2p/go-libp2p/core/peer"
 	"go.uber.org/fx"
+	"golang.org/x/sync/errgroup"
+	"math/rand"
+	"sync/atomic"
+	"time"
 )
 
 // Peering constructs the peering service and hooks it into fx's lifetime
@@ -33,3 +37,167 @@ func PeerWith(peers ...peer.AddrInfo) fx.Option {
 		}
 	})
 }
+
+const (
+	maxNLastConn    = 10
+	maxTryLimit     = 100
+	maxTimeDuration = 20 * time.Second
+	connTimeout     = 3 * time.Second
+)
+
+func loadConnPeers(host host.Host, cfg *config.Config) map[peer.ID]bool {
+	peerIds := host.Peerstore().PeersWithAddrs()
+
+	bootstrap, err := cfg.BootstrapPeers()
+	if err != nil {
+		logger.Warn("failed to parse bootstrap peers from config")
+	}
+
+	filter := make(map[peer.ID]bool, len(bootstrap))
+	for _, id := range bootstrap {
+		filter[id.ID] = true
+	}
+
+	canConnect := make(map[peer.ID]bool)
+	for _, id := range peerIds {
+		if host.Network().Connectedness(id) == network.Connected || id == host.ID() || filter[id] {
+			continue
+		}
+		canConnect[id] = true
+	}
+	return canConnect
+}
+
+func randomSubsetOfPeers(in map[peer.ID]bool, max int) map[peer.ID]bool {
+	c := 0
+	for _, v := range in {
+		if v {
+			c++
+		}
+	}
+
+	if max > c {
+		max = c
+	}
+
+	out := make(map[peer.ID]bool, max)
+
+	tem := make([]peer.ID, 0)
+	for k, v := range in {
+		if v {
+			tem = append(tem, k)
+		}
+	}
+
+	for _, val := range rand.Perm(max) {
+		out[tem[val]] = true
+	}
+
+	return out
+}
+
+func clearTriedPeer(peers map[peer.ID]bool, triedPeer map[peer.ID]bool) map[peer.ID]bool {
+	for k := range triedPeer {
+		peers[k] = false
+	}
+	return peers
+}
+
+func doConcurrentConn(ctx context.Context, host host.Host, peers map[peer.ID]bool) int32 {
+	if len(peers) < 1 {
+		return 0
+	}
+
+	g := errgroup.Group{}
+	g.SetLimit(maxNLastConn)
+
+	connected := int32(0)
+
+	for id := range peers {
+		peerId := id
+		g.Go(func() error {
+			if err := host.Connect(ctx, host.Peerstore().PeerInfo(peerId)); err != nil {
+				logger.Debugf("connect to last connection peer %s, error %v", peerId, err)
+				return nil
+			}
+			atomic.AddInt32(&connected, 1)
+			return nil
+		})
+	}
+	_ = g.Wait()
+
+	return connected
+}
+
+func tryConn(host host.Host, peers map[peer.ID]bool) {
+
+	success := make(chan struct{})
+	useOut := make(chan struct{})
+	maxTry := make(chan struct{})
+	timeout := make(chan struct{})
+
+	timer := time.NewTimer(maxTimeDuration)
+
+	needPeerCount := maxNLastConn
+	canTryPeerCount := maxTryLimit
+
+	ctx := context.Background()
+	ctx, cancel := context.WithTimeout(ctx, connTimeout)
+	defer cancel()
+
+	go func() {
+		for {
+			select {
+			case <-timer.C:
+				timeout <- struct{}{}
+				return
+			default:
+				conns := randomSubsetOfPeers(peers, needPeerCount)
+				peers = clearTriedPeer(peers, conns)
+
+				connCount := doConcurrentConn(ctx, host, conns)
+
+				needPeerCount -= int(connCount)
+				canTryPeerCount -= len(conns)
+
+				if len(conns) <= 0 {
+					useOut <- struct{}{}
+					return
+				}
+
+				if needPeerCount <= 0 {
+					success <- struct{}{}
+					return
+				}
+
+				if canTryPeerCount <= 0 {
+					maxTry <- struct{}{}
+					return
+				}
+			}
+		}
+	}()
+
+	select {
+	case <-timeout:
+		logger.Debugf("connect to last connection timeout")
+		return
+	case <-success:
+		logger.Debugf("connect to last connection success")
+		return
+	case <-useOut:
+		logger.Debugf("connect to last connection use out")
+		return
+	case <-maxTry:
+		logger.Debugf("connect to last connection try limited")
+		return
+	}
+}
+
+// PeerWithLastConn tryConn to connect to last peers
+func PeerWithLastConn() fx.Option {
+	return fx.Invoke(func(host host.Host, cfg *config.Config) {
+		peers := loadConnPeers(host, cfg)
+		tryConn(host, peers)
+	})
+}