Finish and Test the simultaneous connect problem in libp2p peers (#1041)

* Implement support for SimOpen.

config/config.go

@@ -143,7 +143,7 @@ func (cfg *Config) addTransports(ctx context.Context, h host.Host) (err error) {
 	if cfg.Insecure {
 		upgrader.Secure = makeInsecureTransport(h.ID(), cfg.PeerKey)
 	} else {
-		upgrader.Secure, err = makeSecurityTransport(h, cfg.SecurityTransports)
+		upgrader.Secure, err = makeSecurityMuxer(h, cfg.SecurityTransports)
 		if err != nil {
 			return err
 		}

config/security.go

@@ -50,13 +50,13 @@ func SecurityConstructor(security interface{}) (SecC, error) {
 	}, nil
 }
 
-func makeInsecureTransport(id peer.ID, privKey crypto.PrivKey) sec.SecureTransport {
+func makeInsecureTransport(id peer.ID, privKey crypto.PrivKey) sec.SecureMuxer {
 	secMuxer := new(csms.SSMuxer)
 	secMuxer.AddTransport(insecure.ID, insecure.NewWithIdentity(id, privKey))
 	return secMuxer
 }
 
-func makeSecurityTransport(h host.Host, tpts []MsSecC) (sec.SecureTransport, error) {
+func makeSecurityMuxer(h host.Host, tpts []MsSecC) (sec.SecureMuxer, error) {
 	secMuxer := new(csms.SSMuxer)
 	transportSet := make(map[string]struct{}, len(tpts))
 	for _, tptC := range tpts {

go.mod

@@ -3,21 +3,21 @@ module github.com/libp2p/go-libp2p
 go 1.12
 
 require (
-	github.com/gogo/protobuf v1.3.1
+	github.com/gogo/protobuf v1.3.2
 	github.com/ipfs/go-cid v0.0.7
-	github.com/ipfs/go-datastore v0.4.4
+	github.com/ipfs/go-datastore v0.4.5
 	github.com/ipfs/go-detect-race v0.0.1
 	github.com/ipfs/go-ipfs-util v0.0.2
 	github.com/ipfs/go-log v1.0.4
 	github.com/jbenet/go-cienv v0.1.0
 	github.com/jbenet/goprocess v0.1.4
 	github.com/libp2p/go-addr-util v0.0.2
-	github.com/libp2p/go-conn-security-multistream v0.2.0
+	github.com/libp2p/go-conn-security-multistream v0.2.1
 	github.com/libp2p/go-eventbus v0.2.1
 	github.com/libp2p/go-libp2p-autonat v0.4.0
 	github.com/libp2p/go-libp2p-blankhost v0.2.0
 	github.com/libp2p/go-libp2p-circuit v0.4.0
-	github.com/libp2p/go-libp2p-core v0.8.0
+	github.com/libp2p/go-libp2p-core v0.8.1
 	github.com/libp2p/go-libp2p-discovery v0.5.0
 	github.com/libp2p/go-libp2p-loggables v0.1.0
 	github.com/libp2p/go-libp2p-mplex v0.4.1
@@ -25,10 +25,10 @@ require (
 	github.com/libp2p/go-libp2p-netutil v0.1.0
 	github.com/libp2p/go-libp2p-noise v0.1.1
 	github.com/libp2p/go-libp2p-peerstore v0.2.6
-	github.com/libp2p/go-libp2p-swarm v0.4.0
+	github.com/libp2p/go-libp2p-swarm v0.4.1-0.20210119090647-fca9031125ef
 	github.com/libp2p/go-libp2p-testing v0.4.0
 	github.com/libp2p/go-libp2p-tls v0.1.3
-	github.com/libp2p/go-libp2p-transport-upgrader v0.4.0
+	github.com/libp2p/go-libp2p-transport-upgrader v0.4.1
 	github.com/libp2p/go-libp2p-yamux v0.5.1
 	github.com/libp2p/go-msgio v0.0.6
 	github.com/libp2p/go-netroute v0.1.3
@@ -38,9 +38,7 @@ require (
 	github.com/miekg/dns v1.1.31 // indirect
 	github.com/multiformats/go-multiaddr v0.3.1
 	github.com/multiformats/go-multiaddr-dns v0.2.0
-	github.com/multiformats/go-multistream v0.2.0
-	github.com/onsi/ginkgo v1.12.1 // indirect
+	github.com/multiformats/go-multistream v0.2.1
 	github.com/stretchr/testify v1.6.1
 	github.com/whyrusleeping/mdns v0.0.0-20190826153040-b9b60ed33aa9
-	google.golang.org/protobuf v1.23.0 // indirect
 )

libp2p_test.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"regexp"
 	"strings"
+	"sync"
 	"testing"
 
 	"github.com/libp2p/go-libp2p-core/connmgr"
@@ -12,7 +13,10 @@ import (
 	"github.com/libp2p/go-libp2p-core/host"
 	"github.com/libp2p/go-libp2p-core/peer"
 	"github.com/libp2p/go-libp2p-core/transport"
+	noise "github.com/libp2p/go-libp2p-noise"
 	"github.com/libp2p/go-tcp-transport"
+	ma "github.com/multiformats/go-multiaddr"
+	"github.com/stretchr/testify/require"
 
 	tptu "github.com/libp2p/go-libp2p-transport-upgrader"
 )
@@ -192,3 +196,52 @@ func TestChainOptions(t *testing.T) {
 		}
 	}
 }
+
+func TestTcpSimultaneousConnect(t *testing.T) {
+	ctx := context.Background()
+
+	// Host1
+	h1, err := New(ctx, Transport(tcp.NewTCPTransport), Security(noise.ID, noise.New),
+		ListenAddrs(ma.StringCast("/ip4/0.0.0.0/tcp/0")))
+	require.NoError(t, err)
+	defer h1.Close()
+
+	// Host2
+	h2, err := New(ctx, Transport(tcp.NewTCPTransport), Security(noise.ID, noise.New),
+		ListenAddrs(ma.StringCast("/ip4/0.0.0.0/tcp/0")))
+	require.NoError(t, err)
+	defer h2.Close()
+
+	h1Info := peer.AddrInfo{
+		ID:    h1.ID(),
+		Addrs: h1.Addrs(),
+	}
+
+	h2Info := peer.AddrInfo{
+		ID:    h2.ID(),
+		Addrs: h2.Addrs(),
+	}
+
+	wg := new(sync.WaitGroup)
+
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		for i := 0; i < 20; i++ {
+			require.NoError(t, h1.Connect(context.Background(), h2Info))
+			require.NoError(t, h1.Network().ClosePeer(h2.ID()))
+		}
+	}()
+
+	// use another peer to constantly connect/disconnect with first peer.
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		for i := 0; i < 20; i++ {
+			require.NoError(t, h2.Connect(context.Background(), h1Info))
+			require.NoError(t, h2.Network().ClosePeer(h1.ID()))
+		}
+	}()
+
+	wg.Wait()
+}