Skip to content

Commit

Permalink
implement basic utp dialing and listening support
Browse files Browse the repository at this point in the history
License: MIT
Signed-off-by: Jeromy <[email protected]>
  • Loading branch information
whyrusleeping committed Oct 20, 2015
1 parent d7fce58 commit 4be92ef
Show file tree
Hide file tree
Showing 13 changed files with 296 additions and 150 deletions.
2 changes: 1 addition & 1 deletion Godeps/Godeps.json

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

88 changes: 15 additions & 73 deletions p2p/net/conn/dial.go
Original file line number Diff line number Diff line change
Expand Up @@ -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"

Expand Down Expand Up @@ -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.
Expand Down
5 changes: 4 additions & 1 deletion p2p/net/conn/dial_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down Expand Up @@ -152,6 +153,7 @@ func testDialer(t *testing.T, secure bool) {
LocalPeer: p2.ID,
PrivateKey: key2,
}
d2.AddDialer(new(BasicMaDialer))

go echoListen(ctx, l1)

Expand Down Expand Up @@ -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)
Expand All @@ -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.

Expand Down
8 changes: 4 additions & 4 deletions p2p/net/conn/interface.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
4 changes: 4 additions & 0 deletions p2p/net/conn/listen.go
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down
191 changes: 191 additions & 0 deletions p2p/net/conn/transport.go
Original file line number Diff line number Diff line change
@@ -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
}
5 changes: 3 additions & 2 deletions p2p/net/swarm/addr/addr.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
}
Expand Down Expand Up @@ -104,6 +104,7 @@ func AddrUsable(a ma.Multiaddr, partial bool) bool {
return false
}
}

return true
}

Expand Down
Loading

0 comments on commit 4be92ef

Please sign in to comment.