implement basic utp dialing and listening support

License: MIT
Signed-off-by: Jeromy <jeromyj@gmail.com>

p2p/net/conn/dial.go

@@ -4,12 +4,10 @@ import (
 	"fmt"
 	"math/rand"
 	"net"
-	"strings"
 	"syscall"
 
 	ma "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr"
 	manet "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr-net"
-	reuseport "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-reuseport"
 	context "github.com/ipfs/go-ipfs/Godeps/_workspace/src/golang.org/x/net/context"
 	lgbl "github.com/ipfs/go-ipfs/util/eventlog/loggables"
 
@@ -95,84 +93,28 @@ func (d *Dialer) Dial(ctx context.Context, raddr ma.Multiaddr, remote peer.ID) (
 	return connOut, nil
 }
 
-// rawConnDial dials the underlying net.Conn + manet.Conns
-func (d *Dialer) rawConnDial(ctx context.Context, raddr ma.Multiaddr, remote peer.ID) (manet.Conn, error) {
-
-	// before doing anything, check we're going to be able to dial.
-	// we may not support the given address.
-	if _, _, err := manet.DialArgs(raddr); err != nil {
-		return nil, err
-	}
-
-	if strings.HasPrefix(raddr.String(), "/ip4/0.0.0.0") {
-		log.Event(ctx, "connDialZeroAddr", lgbl.Dial("conn", d.LocalPeer, remote, nil, raddr))
-		return nil, fmt.Errorf("Attempted to connect to zero address: %s", raddr)
-	}
+func (d *Dialer) AddDialer(pd ProtoDialer) {
+	d.Dialers = append(d.Dialers, pd)
+}
 
-	// get local addr to use.
-	laddr := pickLocalAddr(d.LocalAddrs, raddr)
-	logdial := lgbl.Dial("conn", d.LocalPeer, remote, laddr, raddr)
-	defer log.EventBegin(ctx, "connDialRawConn", logdial).Done()
-
-	// make a copy of the manet.Dialer, we may need to change its timeout.
-	madialer := d.Dialer
-
-	if laddr != nil && reuseportIsAvailable() {
-		// we're perhaps going to dial twice. half the timeout, so we can afford to.
-		// otherwise our context would expire right after the first dial.
-		madialer.Dialer.Timeout = (madialer.Dialer.Timeout / 2)
-
-		// dial using reuseport.Dialer, because we're probably reusing addrs.
-		// this is optimistic, as the reuseDial may fail to bind the port.
-		rpev := log.EventBegin(ctx, "connDialReusePort", logdial)
-		if nconn, retry, reuseErr := reuseDial(madialer.Dialer, laddr, raddr); reuseErr == nil {
-			// if it worked, wrap the raw net.Conn with our manet.Conn
-			logdial["reuseport"] = "success"
-			rpev.Done()
-			return manet.WrapNetConn(nconn)
-		} else if !retry {
-			// reuseDial is sure this is a legitimate dial failure, not a reuseport failure.
-			logdial["reuseport"] = "failure"
-			logdial["error"] = reuseErr
-			rpev.Done()
-			return nil, reuseErr
-		} else {
-			// this is a failure to reuse port. log it.
-			logdial["reuseport"] = "retry"
-			logdial["error"] = reuseErr
-			rpev.Done()
+// returns dialer that can dial the given address
+func (d *Dialer) subDialerForAddr(raddr ma.Multiaddr) ProtoDialer {
+	for _, pd := range d.Dialers {
+		if pd.Matches(raddr) {
+			return pd
 		}
 	}
-
-	defer log.EventBegin(ctx, "connDialManet", logdial).Done()
-	return madialer.Dial(raddr)
+	return nil
 }
 
-func reuseDial(dialer net.Dialer, laddr, raddr ma.Multiaddr) (conn net.Conn, retry bool, err error) {
-	if laddr == nil {
-		// if we're given no local address no sense in using reuseport to dial, dial out as usual.
-		return nil, true, reuseport.ErrReuseFailed
-	}
-
-	// give reuse.Dialer the manet.Dialer's Dialer.
-	// (wow, Dialer should've so been an interface...)
-	rd := reuseport.Dialer{dialer}
-
-	// get the local net.Addr manually
-	rd.D.LocalAddr, err = manet.ToNetAddr(laddr)
-	if err != nil {
-		return nil, true, err // something wrong with laddr. retry without.
-	}
-
-	// get the raddr dial args for rd.dial
-	network, netraddr, err := manet.DialArgs(raddr)
-	if err != nil {
-		return nil, true, err // something wrong with laddr. retry without.
+// rawConnDial dials the underlying net.Conn + manet.Conns
+func (d *Dialer) rawConnDial(ctx context.Context, raddr ma.Multiaddr, remote peer.ID) (manet.Conn, error) {
+	sd := d.subDialerForAddr(raddr)
+	if sd == nil {
+		return nil, fmt.Errorf("no dialer for %s", raddr)
 	}
 
-	// rd.Dial gets us a net.Conn with SO_REUSEPORT and SO_REUSEADDR set.
-	conn, err = rd.Dial(network, netraddr)
-	return conn, reuseErrShouldRetry(err), err // hey! it worked!
+	return sd.Dial(raddr)
 }
 
 // reuseErrShouldRetry diagnoses whether to retry after a reuse error.

p2p/net/conn/dial_test.go

@@ -70,6 +70,7 @@ func setupConn(t *testing.T, ctx context.Context, secure bool) (a, b Conn, p1, p
 		LocalPeer:  p2.ID,
 		PrivateKey: key2,
 	}
+	d2.AddDialer(new(BasicMaDialer))
 
 	var c2 Conn
 
@@ -152,6 +153,7 @@ func testDialer(t *testing.T, secure bool) {
 		LocalPeer:  p2.ID,
 		PrivateKey: key2,
 	}
+	d2.AddDialer(new(BasicMaDialer))
 
 	go echoListen(ctx, l1)
 
@@ -227,6 +229,7 @@ func testDialerCloseEarly(t *testing.T, secure bool) {
 		LocalPeer: p2.ID,
 		// PrivateKey: key2, -- dont give it key. we'll just close the conn.
 	}
+	d2.AddDialer(new(BasicMaDialer))
 
 	errs := make(chan error, 100)
 	done := make(chan struct{}, 1)
@@ -253,7 +256,7 @@ func testDialerCloseEarly(t *testing.T, secure bool) {
 
 	c, err := d2.Dial(ctx, p1.Addr, p1.ID)
 	if err != nil {
-		errs <- err
+		t.Fatal(err)
 	}
 	c.Close() // close it early.
 

p2p/net/conn/interface.go

@@ -54,16 +54,16 @@ type Conn interface {
 // Dial function as before, but it would have many arguments, as dialing is
 // no longer simple (need a peerstore, a local peer, a context, a network, etc)
 type Dialer struct {
-
-	// Dialer is an optional manet.Dialer to use.
-	Dialer manet.Dialer
-
 	// LocalPeer is the identity of the local Peer.
 	LocalPeer peer.ID
 
 	// LocalAddrs is a set of local addresses to use.
 	LocalAddrs []ma.Multiaddr
 
+	// Dialers are the sub-dialers usable by this dialer
+	// selected in order based on the address being dialed
+	Dialers []ProtoDialer
+
 	// PrivateKey used to initialize a secure connection.
 	// Warning: if PrivateKey is nil, connection will not be secured.
 	PrivateKey ic.PrivKey

p2p/net/conn/listen.go

@@ -154,6 +154,10 @@ func Listen(ctx context.Context, addr ma.Multiaddr, local peer.ID, sk ic.PrivKey
 		return nil, err
 	}
 
+	return WrapManetListener(ctx, ml, local, sk)
+}
+
+func WrapManetListener(ctx context.Context, ml manet.Listener, local peer.ID, sk ic.PrivKey) (Listener, error) {
 	l := &listener{
 		Listener: ml,
 		local:    local,

p2p/net/conn/transport.go

@@ -0,0 +1,191 @@
+package conn
+
+import (
+	"net"
+
+	ma "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr"
+	manet "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr-net"
+	mautp "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr-net/utp"
+	reuseport "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-reuseport"
+)
+
+type Transport interface {
+	manet.Listener
+	ProtoDialer
+}
+
+type ProtoDialer interface {
+	Dial(raddr ma.Multiaddr) (manet.Conn, error)
+	Matches(ma.Multiaddr) bool
+}
+
+type TcpReuseTransport struct {
+	list  manet.Listener
+	laddr ma.Multiaddr
+
+	rd       reuseport.Dialer
+	madialer manet.Dialer
+}
+
+var _ Transport = (*TcpReuseTransport)(nil)
+
+func NewTcpReuseTransport(base manet.Dialer, laddr ma.Multiaddr) (*TcpReuseTransport, error) {
+	rd := reuseport.Dialer{base.Dialer}
+
+	list, err := manet.Listen(laddr)
+	if err != nil {
+		return nil, err
+	}
+
+	// get the local net.Addr manually
+	la, err := manet.ToNetAddr(laddr)
+	if err != nil {
+		return nil, err // something wrong with laddr.
+	}
+
+	rd.D.LocalAddr = la
+
+	return &TcpReuseTransport{
+		list:     list,
+		laddr:    laddr,
+		rd:       rd,
+		madialer: base,
+	}, nil
+}
+
+func (d *TcpReuseTransport) Dial(raddr ma.Multiaddr) (manet.Conn, error) {
+	network, netraddr, err := manet.DialArgs(raddr)
+	if err != nil {
+		return nil, err
+	}
+
+	conn, err := d.rd.Dial(network, netraddr)
+	if err == nil {
+		return manet.WrapNetConn(conn)
+	}
+	if !reuseErrShouldRetry(err) {
+		return nil, err
+	}
+
+	return d.madialer.Dial(raddr)
+}
+
+func (d *TcpReuseTransport) Matches(a ma.Multiaddr) bool {
+	return IsTcpMultiaddr(a)
+}
+
+func (d *TcpReuseTransport) Accept() (manet.Conn, error) {
+	c, err := d.list.Accept()
+	if err != nil {
+		return nil, err
+	}
+
+	return manet.WrapNetConn(c)
+}
+
+func (d *TcpReuseTransport) Addr() net.Addr {
+	return d.rd.D.LocalAddr
+}
+
+func (t *TcpReuseTransport) Multiaddr() ma.Multiaddr {
+	return t.list.Multiaddr()
+}
+
+func (t *TcpReuseTransport) NetListener() net.Listener {
+	return t.list.NetListener()
+}
+
+func (d *TcpReuseTransport) Close() error {
+	return d.list.Close()
+}
+
+func IsTcpMultiaddr(a ma.Multiaddr) bool {
+	p := a.Protocols()
+	return len(p) == 2 && (p[0].Name == "ip4" || p[0].Name == "ip6") && p[1].Name == "tcp"
+}
+
+func IsUtpMultiaddr(a ma.Multiaddr) bool {
+	p := a.Protocols()
+	return len(p) == 3 && p[2].Name == "utp"
+}
+
+type UtpReuseTransport struct {
+	s     *mautp.Socket
+	laddr ma.Multiaddr
+}
+
+func NewUtpReuseTransport(laddr ma.Multiaddr) (*UtpReuseTransport, error) {
+	network, addr, err := manet.DialArgs(laddr)
+	if err != nil {
+		return nil, err
+	}
+
+	us, err := mautp.NewSocket(network, addr)
+	if err != nil {
+		return nil, err
+	}
+
+	mmm, err := manet.FromNetAddr(us.Addr())
+	if err != nil {
+		return nil, err
+	}
+
+	return &UtpReuseTransport{
+		s:     us,
+		laddr: mmm,
+	}, nil
+}
+
+func (d *UtpReuseTransport) Matches(a ma.Multiaddr) bool {
+	p := a.Protocols()
+	return len(p) == 3 && p[2].Name == "utp"
+}
+
+func (d *UtpReuseTransport) Dial(raddr ma.Multiaddr) (manet.Conn, error) {
+	network, netraddr, err := manet.DialArgs(raddr)
+	if err != nil {
+		return nil, err
+	}
+
+	c, err := d.s.Dial(network, netraddr)
+	if err != nil {
+		return nil, err
+	}
+
+	return manet.WrapNetConn(c)
+}
+
+func (d *UtpReuseTransport) Accept() (manet.Conn, error) {
+	c, err := d.s.Accept()
+	if err != nil {
+		return nil, err
+	}
+
+	return manet.WrapNetConn(c)
+}
+
+func (t *UtpReuseTransport) Close() error {
+	return t.s.Close()
+}
+
+func (t *UtpReuseTransport) Addr() net.Addr {
+	return t.s.Addr()
+}
+
+func (t *UtpReuseTransport) Multiaddr() ma.Multiaddr {
+	return t.laddr
+}
+
+func (t *UtpReuseTransport) NetListener() net.Listener {
+	return t.s
+}
+
+type BasicMaDialer struct{}
+
+func (d *BasicMaDialer) Dial(raddr ma.Multiaddr) (manet.Conn, error) {
+	return manet.Dial(raddr)
+}
+
+func (d *BasicMaDialer) Matches(a ma.Multiaddr) bool {
+	return true
+}

p2p/net/swarm/addr/addr.go

@@ -18,8 +18,8 @@ var log = logging.Logger("p2p/net/swarm/addr")
 var SupportedTransportStrings = []string{
 	"/ip4/tcp",
 	"/ip6/tcp",
-	// "/ip4/udp/utp", disabled because the lib is broken
-	// "/ip6/udp/utp", disabled because the lib is broken
+	"/ip4/udp/utp",
+	"/ip6/udp/utp",
 	// "/ip4/udp/udt", disabled because the lib doesnt work on arm
 	// "/ip6/udp/udt", disabled because the lib doesnt work on arm
 }
@@ -104,6 +104,7 @@ func AddrUsable(a ma.Multiaddr, partial bool) bool {
 				return false
 			}
 		}
+
 		return true
 	}