tcp.go 5.49 KB
Newer Older
1 2 3 4 5 6 7 8 9 10
package tcp

import (
	"context"
	"fmt"
	"net"
	"sync"

	logging "github.com/ipfs/go-log"
	tpt "github.com/libp2p/go-libp2p-transport"
Steven Allen's avatar
Steven Allen committed
11
	reuseport "github.com/libp2p/go-reuseport"
Jeromy's avatar
Jeromy committed
12 13
	ma "github.com/multiformats/go-multiaddr"
	manet "github.com/multiformats/go-multiaddr-net"
14 15 16 17 18 19 20 21 22 23 24 25 26
	mafmt "github.com/whyrusleeping/mafmt"
)

var log = logging.Logger("tcp-tpt")

type TcpTransport struct {
	dlock   sync.Mutex
	dialers map[string]tpt.Dialer

	llock     sync.Mutex
	listeners map[string]tpt.Listener
}

27 28
var _ tpt.Transport = &TcpTransport{}

29 30 31 32 33 34 35 36 37 38
// NewTCPTransport creates a tcp transport object that tracks dialers and listeners
// created. It represents an entire tcp stack (though it might not necessarily be)
func NewTCPTransport() *TcpTransport {
	return &TcpTransport{
		dialers:   make(map[string]tpt.Dialer),
		listeners: make(map[string]tpt.Listener),
	}
}

func (t *TcpTransport) Dialer(laddr ma.Multiaddr, opts ...tpt.DialOpt) (tpt.Dialer, error) {
Jeromy's avatar
Jeromy committed
39 40 41 42 43 44 45
	if laddr == nil {
		zaddr, err := ma.NewMultiaddr("/ip4/0.0.0.0/tcp/0")
		if err != nil {
			return nil, err
		}
		laddr = zaddr
	}
46 47 48 49 50 51 52 53 54
	t.dlock.Lock()
	defer t.dlock.Unlock()
	s := laddr.String()
	d, found := t.dialers[s]
	if found {
		return d, nil
	}
	var base manet.Dialer

55 56 57 58 59 60
	la, err := manet.ToNetAddr(laddr)
	if err != nil {
		return nil, err // something wrong with laddr.
	}
	base.Dialer.LocalAddr = la

61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137
	var doReuse bool
	for _, o := range opts {
		switch o := o.(type) {
		case tpt.ReuseportOpt:
			doReuse = bool(o)
		default:
			return nil, fmt.Errorf("unrecognized option: %#v", o)
		}
	}

	tcpd, err := t.newTcpDialer(base, laddr, doReuse)
	if err != nil {
		return nil, err
	}

	t.dialers[s] = tcpd
	return tcpd, nil
}

func (t *TcpTransport) Listen(laddr ma.Multiaddr) (tpt.Listener, error) {
	if !t.Matches(laddr) {
		return nil, fmt.Errorf("tcp transport cannot listen on %q", laddr)
	}

	t.llock.Lock()
	defer t.llock.Unlock()
	s := laddr.String()
	l, found := t.listeners[s]
	if found {
		return l, nil
	}

	list, err := manetListen(laddr)
	if err != nil {
		return nil, err
	}

	tlist := &tcpListener{
		list:      list,
		transport: t,
	}

	t.listeners[s] = tlist
	return tlist, nil
}

func manetListen(addr ma.Multiaddr) (manet.Listener, error) {
	network, naddr, err := manet.DialArgs(addr)
	if err != nil {
		return nil, err
	}

	if ReuseportIsAvailable() {
		nl, err := reuseport.Listen(network, naddr)
		if err == nil {
			// hey, it worked!
			return manet.WrapNetListener(nl)
		}
		// reuseport is available, but we failed to listen. log debug, and retry normally.
		log.Debugf("reuseport available, but failed to listen: %s %s, %s", network, naddr, err)
	}

	// either reuseport not available, or it failed. try normally.
	return manet.Listen(addr)
}

func (t *TcpTransport) Matches(a ma.Multiaddr) bool {
	return mafmt.TCP.Matches(a)
}

type tcpDialer struct {
	laddr ma.Multiaddr

	doReuse bool

	rd       reuseport.Dialer
	madialer manet.Dialer
Jeromy's avatar
Jeromy committed
138
	pattern  mafmt.Pattern
139 140 141 142

	transport tpt.Transport
}

143 144
var _ tpt.Dialer = &tcpDialer{}

145
func (t *TcpTransport) newTcpDialer(base manet.Dialer, laddr ma.Multiaddr, doReuse bool) (*tcpDialer, error) {
Jeromy's avatar
Jeromy committed
146 147 148 149 150 151 152 153 154
	var pattern mafmt.Pattern
	if TCP4.Matches(laddr) {
		pattern = TCP4
	} else if TCP6.Matches(laddr) {
		pattern = TCP6
	} else {
		return nil, fmt.Errorf("local addr did not match TCP4 or TCP6: %s", laddr)
	}

155 156
	if doReuse && ReuseportIsAvailable() {
		rd := reuseport.Dialer{
157
			D: base.Dialer,
158 159 160 161 162 163 164 165
		}

		return &tcpDialer{
			doReuse:   true,
			laddr:     laddr,
			rd:        rd,
			madialer:  base,
			transport: t,
Jeromy's avatar
Jeromy committed
166
			pattern:   pattern,
167 168 169 170 171 172
		}, nil
	}

	return &tcpDialer{
		doReuse:   false,
		laddr:     laddr,
173
		pattern:   pattern,
174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195
		madialer:  base,
		transport: t,
	}, nil
}

func (d *tcpDialer) Dial(raddr ma.Multiaddr) (tpt.Conn, error) {
	return d.DialContext(context.Background(), raddr)
}

func (d *tcpDialer) DialContext(ctx context.Context, raddr ma.Multiaddr) (tpt.Conn, error) {
	var c manet.Conn
	var err error
	if d.doReuse {
		c, err = d.reuseDial(ctx, raddr)
	} else {
		c, err = d.madialer.DialContext(ctx, raddr)
	}

	if err != nil {
		return nil, err
	}

196
	return &tcpConn{
197
		Conn: c,
198
		t:    d.transport,
199 200 201 202 203 204 205 206 207
	}, nil
}

func (d *tcpDialer) reuseDial(ctx context.Context, raddr ma.Multiaddr) (manet.Conn, error) {
	network, netraddr, err := manet.DialArgs(raddr)
	if err != nil {
		return nil, err
	}

208 209 210 211
	rpev := log.EventBegin(ctx, "tptDialReusePort", logging.LoggableMap{
		"raddr": raddr,
	})

212
	con, err := d.rd.DialContext(ctx, network, netraddr)
213 214 215 216
	if err == nil {
		rpev.Done()
		return manet.WrapNetConn(con)
	}
217 218
	rpev.SetError(err)
	rpev.Done()
219 220 221 222 223 224 225 226

	if !ReuseErrShouldRetry(err) {
		return nil, err
	}

	return d.madialer.DialContext(ctx, raddr)
}

Jeromy's avatar
Jeromy committed
227 228 229
var TCP4 = mafmt.And(mafmt.Base(ma.P_IP4), mafmt.Base(ma.P_TCP))
var TCP6 = mafmt.And(mafmt.Base(ma.P_IP6), mafmt.Base(ma.P_TCP))

230
func (d *tcpDialer) Matches(a ma.Multiaddr) bool {
Jeromy's avatar
Jeromy committed
231
	return d.pattern.Matches(a)
232 233 234 235 236 237 238
}

type tcpListener struct {
	list      manet.Listener
	transport tpt.Transport
}

239 240
var _ tpt.Listener = &tcpListener{}

241 242 243 244 245 246
func (d *tcpListener) Accept() (tpt.Conn, error) {
	c, err := d.list.Accept()
	if err != nil {
		return nil, err
	}

247
	return &tcpConn{
248
		Conn: c,
249
		t:    d.transport,
250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267
	}, nil
}

func (d *tcpListener) Addr() net.Addr {
	return d.list.Addr()
}

func (t *tcpListener) Multiaddr() ma.Multiaddr {
	return t.list.Multiaddr()
}

func (t *tcpListener) NetListener() net.Listener {
	return t.list.NetListener()
}

func (d *tcpListener) Close() error {
	return d.list.Close()
}
268 269 270 271 272 273 274 275 276 277 278

type tcpConn struct {
	manet.Conn
	t tpt.Transport
}

var _ tpt.Conn = &tcpConn{}

func (c *tcpConn) Transport() tpt.Transport {
	return c.t
}