@cryptotaxi247 / kubo / commits / 334f9d210

p2p/net/conn: use reuseport

Juan Batiz-Benet committed Jan 19, 2015 at 02:54 UTC 334f9d2102f85df3babc3fd3560d82710ee59f01
2 files changed +91 -27
p2p/net/conn/dial.go
+76 -26
@@ -2,11 +2,14 @@ package conn
2
3 import (
4 "fmt"
5 + "math/rand"
6 + "net"
7 "strings"
8
9 context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
10 ma "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr"
11 manet "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr-net"
12 + reuseport "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-reuseport"
13
14 peer "github.com/jbenet/go-ipfs/p2p/peer"
15 debugerror "github.com/jbenet/go-ipfs/util/debugerror"
@@ -22,32 +25,7 @@ func (d *Dialer) String() string {
25 // Example: d.DialAddr(ctx, peer.Addresses()[0], peer)
26 func (d *Dialer) Dial(ctx context.Context, raddr ma.Multiaddr, remote peer.ID) (Conn, error) {
27
25 - _, _, err := manet.DialArgs(raddr)
26 - if err != nil {
27 - return nil, err
28 - }
29 -
30 - if strings.HasPrefix(raddr.String(), "/ip4/0.0.0.0") {
31 - return nil, debugerror.Errorf("Attempted to connect to zero address: %s", raddr)
32 - }
33 -
34 - if len(d.LocalAddrs) > 0 {
35 - laddrs := manet.AddrMatch(raddr, d.LocalAddrs)
36 - if len(laddrs) < 1 {
37 - return nil, debugerror.Errorf("No local address matches %s %s", raddr, d.LocalAddrs)
38 - }
39 -
40 - // TODO pick with a good heuristic
41 - // we use a random one for now to prevent bad addresses from making nodes unreachable
42 - // with a random selection, multiple tries may work.
43 - // laddr := laddrs[rand.Intn(len(laddrs))]
44 -
45 - // TODO: try to get reusing addr/ports to work.
46 - // d.Dialer.LocalAddr = laddr
47 - }
48 -
49 - log.Debugf("%s dialing %s %s", d.LocalPeer, remote, raddr)
50 - maconn, err := d.Dialer.Dial(raddr)
28 + maconn, err := d.rawConnDial(ctx, raddr, remote)
29 if err != nil {
30 return nil, err
31 }
@@ -92,6 +70,78 @@ func (d *Dialer) Dial(ctx context.Context, raddr ma.Multiaddr, remote peer.ID) (
70 return connOut, errOut
71 }
72
73 +// rawConnDial dials the underlying net.Conn + manet.Conns
74 +func (d *Dialer) rawConnDial(ctx context.Context, raddr ma.Multiaddr, remote peer.ID) (manet.Conn, error) {
75 +
76 + // before doing anything, check we're going to be able to dial.
77 + // we may not support the given address.
78 + _, _, err := manet.DialArgs(raddr)
79 + if err != nil {
80 + return nil, err
81 + }
82 +
83 + if strings.HasPrefix(raddr.String(), "/ip4/0.0.0.0") {
84 + return nil, debugerror.Errorf("Attempted to connect to zero address: %s", raddr)
85 + }
86 +
87 + // get local addr to use.
88 + laddr := pickLocalAddr(d.LocalAddrs, raddr)
89 +
90 + log.Debugf("%s dialing %s -- %s --> %s", d.LocalPeer, remote, laddr, raddr)
91 + if laddr != nil {
92 + // dial using reuseport.Dialer, because we're probably reusing addrs.
93 + // this is optimistic, as the reuseDial may fail to bind the port.
94 + log.Debugf("trying to reuse: %s", laddr)
95 + if nconn, err := d.reuseDial(laddr, raddr); err == nil {
96 + // if it worked, wrap the raw net.Conn with our manet.Conn
97 + log.Debugf("reuse worked! %s %s %s", laddr, nconn.RemoteAddr(), nconn)
98 + return manet.WrapNetConn(nconn)
99 + }
100 + // if not, we fall back to regular Dial without a local addr specified.
101 + }
102 +
103 + // no local addr, or failed to reuse. just dial straight with a new port.
104 + return d.Dialer.Dial(raddr)
105 +}
106 +
107 +func (d *Dialer) reuseDial(laddr, raddr ma.Multiaddr) (net.Conn, error) {
108 + // give reuse.Dialer the manet.Dialer's Dialer.
109 + // (wow, Dialer should've so been an interface...)
110 + rd := reuseport.Dialer{d.Dialer.Dialer}
111 +
112 + // get the local net.Addr manually
113 + var err error
114 + rd.D.LocalAddr, err = manet.ToNetAddr(laddr)
115 + if err != nil {
116 + return nil, err
117 + }
118 +
119 + // get the raddr dial args for rd.dial
120 + network, netraddr, err := manet.DialArgs(raddr)
121 + if err != nil {
122 + return nil, err
123 + }
124 +
125 + // rd.Dial gets us a net.Conn with SO_REUSEPORT and SO_REUSEADDR set.
126 + return rd.Dial(network, netraddr)
127 +}
128 +
129 +func pickLocalAddr(laddrs []ma.Multiaddr, raddr ma.Multiaddr) (laddr ma.Multiaddr) {
130 + if len(laddrs) < 1 {
131 + return nil
132 + }
133 +
134 + laddrs = manet.AddrMatch(raddr, laddrs)
135 + if len(laddrs) < 1 {
136 + return nil
137 + }
138 +
139 + // TODO pick with a good heuristic
140 + // we use a random one for now to prevent bad addresses from making nodes unreachable
141 + // with a random selection, multiple tries may work.
142 + return laddrs[rand.Intn(len(laddrs))]
143 +}
144 +
145 // MultiaddrProtocolsMatch returns whether two multiaddrs match in protocol stacks.
146 func MultiaddrProtocolsMatch(a, b ma.Multiaddr) bool {
147 ap := a.Protocols()
p2p/net/conn/listen.go
+15 -1
@@ -9,7 +9,9 @@ import (
9 ctxgroup "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-ctxgroup"
10 ma "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr"
11 manet "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr-net"
12 + reuseport "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-reuseport"
13 tec "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-temp-err-catcher"
14 +
15 ic "github.com/jbenet/go-ipfs/p2p/crypto"
16 peer "github.com/jbenet/go-ipfs/p2p/peer"
17 )
@@ -123,11 +125,23 @@ func (l *listener) Loggable() map[string]interface{} {
125 // Listen listens on the particular multiaddr, with given peer and peerstore.
126 func Listen(ctx context.Context, addr ma.Multiaddr, local peer.ID, sk ic.PrivKey) (Listener, error) {
127
126 - ml, err := manet.Listen(addr)
128 + network, naddr, err := manet.DialArgs(addr)
129 + if err != nil {
130 + return nil, err
131 + }
132 +
133 + // _ := reuseport.Listen
134 + // ml, err := manet.Listen(addr)
135 + nl, err := reuseport.Listen(network, naddr)
136 if err != nil {
137 return nil, fmt.Errorf("Failed to listen on %s: %s", addr, err)
138 }
139
140 + ml, err := manet.WrapNetListener(nl)
141 + if err != nil {
142 + return nil, err
143 + }
144 +
145 l := &listener{
146 Listener: ml,
147 local: local,