add holepunching support

go.mod

@@ -6,7 +6,7 @@ require (
 	github.com/golang/mock v1.6.0
 	github.com/ipfs/go-log v1.0.4
 	github.com/klauspost/compress v1.11.7
-	github.com/libp2p/go-libp2p-core v0.8.0
+	github.com/libp2p/go-libp2p-core v0.8.5
 	github.com/libp2p/go-libp2p-tls v0.1.3
 	github.com/libp2p/go-netroute v0.1.3
 	github.com/lucas-clemente/quic-go v0.21.1

go.sum

@@ -226,8 +226,8 @@ github.com/libp2p/go-buffer-pool v0.0.2/go.mod h1:MvaB6xw5vOrDl8rYZGLFdKAuk/hRoR
 github.com/libp2p/go-flow-metrics v0.0.1/go.mod h1:Iv1GH0sG8DtYN3SVJ2eG221wMiNpZxBdp967ls1g+k8=
 github.com/libp2p/go-flow-metrics v0.0.3/go.mod h1:HeoSNUrOJVK1jEpDqVEiUOIXqhbnS27omG0uWU5slZs=
 github.com/libp2p/go-libp2p-core v0.0.1/go.mod h1:g/VxnTZ/1ygHxH3dKok7Vno1VfpvGcGip57wjTU4fco=
-github.com/libp2p/go-libp2p-core v0.8.0 h1:5K3mT+64qDTKbV3yTdbMCzJ7O6wbNsavAEb8iqBvBcI=
-github.com/libp2p/go-libp2p-core v0.8.0/go.mod h1:FfewUH/YpvWbEB+ZY9AQRQ4TAD8sJBt/G1rVvhz5XT8=
+github.com/libp2p/go-libp2p-core v0.8.5 h1:aEgbIcPGsKy6zYcC+5AJivYFedhYa4sW7mIpWpUaLKw=
+github.com/libp2p/go-libp2p-core v0.8.5/go.mod h1:FfewUH/YpvWbEB+ZY9AQRQ4TAD8sJBt/G1rVvhz5XT8=
 github.com/libp2p/go-libp2p-tls v0.1.3 h1:twKMhMu44jQO+HgQK9X8NHO5HkeJu2QbhLzLJpa8oNM=
 github.com/libp2p/go-libp2p-tls v0.1.3/go.mod h1:wZfuewxOndz5RTnCAxFliGjvYSDA40sKitV4c50uI1M=
 github.com/libp2p/go-maddr-filter v0.1.0/go.mod h1:VzZhTXkMucEGGEOSKddrwGiOv0tUhgnKqNEmIAz/bPU=

listener.go

@@ -12,7 +12,7 @@ import (
 
 	p2ptls "github.com/libp2p/go-libp2p-tls"
 
-	quic "github.com/lucas-clemente/quic-go"
+	"github.com/lucas-clemente/quic-go"
 	ma "github.com/multiformats/go-multiaddr"
 )
 
@@ -74,6 +74,20 @@ func (l *listener) Accept() (tpt.CapableConn, error) {
 			sess.CloseWithError(errorCodeConnectionGating, "connection gated")
 			continue
 		}
+
+		// return through active hole punching if any
+		key := sess.RemoteAddr().String()
+		l.transport.holePunchingMx.Lock()
+		holePunch, ok := l.transport.holePunching[key]
+		l.transport.holePunchingMx.Unlock()
+		if ok {
+			select {
+			case holePunch.connCh <- conn:
+				continue
+			case <-holePunch.ctx.Done():
+			}
+		}
+
 		return conn, nil
 	}
 }
@@ -92,6 +106,7 @@ func (l *listener) setupConn(sess quic.Session) (*conn, error) {
 	if err != nil {
 		return nil, err
 	}
+
 	remoteMultiaddr, err := toQuicMultiaddr(sess.RemoteAddr())
 	if err != nil {
 		return nil, err

transport.go

@@ -1,11 +1,15 @@
 package libp2pquic
 
 import (
+	"bytes"
 	"context"
 	"errors"
 	"fmt"
 	"io"
+	"math/rand"
 	"net"
+	"sync"
+	"time"
 
 	"github.com/libp2p/go-libp2p-core/connmgr"
 	n "github.com/libp2p/go-libp2p-core/network"
@@ -27,6 +31,8 @@ import (
 
 var log = logging.Logger("quic-transport")
 
+var ErrHolePunching = errors.New("hole punching attempted; no active dial")
+
 var quicDialContext = quic.DialContext // so we can mock it in tests
 
 var quicConfig = &quic.Config{
@@ -96,10 +102,18 @@ type transport struct {
 	serverConfig *quic.Config
 	clientConfig *quic.Config
 	gater        connmgr.ConnectionGater
+
+	holePunchingMx sync.Mutex
+	holePunching   map[string]activeHolePunch
 }
 
 var _ tpt.Transport = &transport{}
 
+type activeHolePunch struct {
+	connCh chan tpt.CapableConn
+	ctx    context.Context
+}
+
 // NewTransport creates a new QUIC transport
 func NewTransport(key ic.PrivKey, psk pnet.PSK, gater connmgr.ConnectionGater) (tpt.Transport, error) {
 	if len(psk) > 0 {
@@ -138,6 +152,7 @@ func NewTransport(key ic.PrivKey, psk pnet.PSK, gater connmgr.ConnectionGater) (
 		serverConfig: config,
 		clientConfig: config.Clone(),
 		gater:        gater,
+		holePunching: make(map[string]activeHolePunch),
 	}, nil
 }
 
@@ -156,6 +171,13 @@ func (t *transport) Dial(ctx context.Context, raddr ma.Multiaddr, p peer.ID) (tp
 		return nil, err
 	}
 	tlsConf, keyCh := t.identity.ConfigForPeer(p)
+
+	if simConnect, _ := n.GetSimultaneousConnect(ctx); simConnect {
+		if bytes.Compare([]byte(t.localPeer), []byte(p)) < 0 {
+			return t.holePunch(ctx, network, addr)
+		}
+	}
+
 	pconn, err := t.connManager.Dial(network, addr)
 	if err != nil {
 		return nil, err
@@ -202,6 +224,53 @@ func (t *transport) Dial(ctx context.Context, raddr ma.Multiaddr, p peer.ID) (tp
 	return conn, nil
 }
 
+func (t *transport) holePunch(ctx context.Context, network string, addr *net.UDPAddr) (tpt.CapableConn, error) {
+	pconn, err := t.connManager.Dial(network, addr)
+	if err != nil {
+		return nil, err
+	}
+	defer pconn.DecreaseCount()
+
+	ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
+	defer cancel()
+
+	connCh := make(chan tpt.CapableConn, 1)
+
+	key := addr.String()
+	t.holePunchingMx.Lock()
+	t.holePunching[key] = activeHolePunch{connCh: connCh, ctx: ctx}
+	t.holePunchingMx.Unlock()
+
+	defer func() {
+		t.holePunchingMx.Lock()
+		delete(t.holePunching, key)
+		t.holePunchingMx.Unlock()
+	}()
+
+	payload := make([]byte, 64)
+	for i := 0; ; i++ {
+		if _, err := rand.Read(payload); err != nil {
+			return nil, err
+		}
+		if _, err := pconn.UDPConn.WriteToUDP(payload, addr); err != nil {
+			return nil, err
+		}
+
+		maxSleep := 10 * (i + 1) * (i + 1) // in ms
+		if maxSleep > 200 {
+			maxSleep = 200
+		}
+		sleep := 10*time.Millisecond + time.Duration(rand.Intn(maxSleep))*time.Millisecond
+		select {
+		case c := <-connCh:
+			return c, nil
+		case <-time.After(sleep):
+		case <-ctx.Done():
+			return nil, ErrHolePunching
+		}
+	}
+}
+
 // Don't use mafmt.QUIC as we don't want to dial DNS addresses. Just /ip{4,6}/udp/quic
 var dialMatcher = mafmt.And(mafmt.IP, mafmt.Base(ma.P_UDP), mafmt.Base(ma.P_QUIC))