/
tcp.go
146 lines (120 loc) · 4.31 KB
/
tcp.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
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
138
139
140
141
142
143
144
145
146
package tcp
import (
"context"
"net"
"time"
logging "github.com/ipfs/fs-repo-migrations/ipfs-10-to-11/_vendor/github.com/ipfs/go-log"
"github.com/ipfs/fs-repo-migrations/ipfs-10-to-11/_vendor/github.com/libp2p/go-libp2p-core/peer"
"github.com/ipfs/fs-repo-migrations/ipfs-10-to-11/_vendor/github.com/libp2p/go-libp2p-core/transport"
tptu "github.com/ipfs/fs-repo-migrations/ipfs-10-to-11/_vendor/github.com/libp2p/go-libp2p-transport-upgrader"
rtpt "github.com/ipfs/fs-repo-migrations/ipfs-10-to-11/_vendor/github.com/libp2p/go-reuseport-transport"
ma "github.com/ipfs/fs-repo-migrations/ipfs-10-to-11/_vendor/github.com/multiformats/go-multiaddr"
mafmt "github.com/ipfs/fs-repo-migrations/ipfs-10-to-11/_vendor/github.com/multiformats/go-multiaddr-fmt"
manet "github.com/ipfs/fs-repo-migrations/ipfs-10-to-11/_vendor/github.com/multiformats/go-multiaddr-net"
)
// DefaultConnectTimeout is the (default) maximum amount of time the TCP
// transport will spend on the initial TCP connect before giving up.
var DefaultConnectTimeout = 5 * time.Second
var log = logging.Logger("tcp-tpt")
// try to set linger on the connection, if possible.
func tryLinger(conn net.Conn, sec int) {
type canLinger interface {
SetLinger(int) error
}
if lingerConn, ok := conn.(canLinger); ok {
_ = lingerConn.SetLinger(sec)
}
}
type lingerListener struct {
manet.Listener
sec int
}
func (ll *lingerListener) Accept() (manet.Conn, error) {
c, err := ll.Listener.Accept()
if err != nil {
return nil, err
}
tryLinger(c, ll.sec)
return c, nil
}
// TcpTransport is the TCP transport.
type TcpTransport struct {
// Connection upgrader for upgrading insecure stream connections to
// secure multiplex connections.
Upgrader *tptu.Upgrader
// Explicitly disable reuseport.
DisableReuseport bool
// TCP connect timeout
ConnectTimeout time.Duration
reuse rtpt.Transport
}
var _ transport.Transport = &TcpTransport{}
// 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(upgrader *tptu.Upgrader) *TcpTransport {
return &TcpTransport{Upgrader: upgrader, ConnectTimeout: DefaultConnectTimeout}
}
var dialMatcher = mafmt.And(mafmt.IP, mafmt.Base(ma.P_TCP))
// CanDial returns true if this transport believes it can dial the given
// multiaddr.
func (t *TcpTransport) CanDial(addr ma.Multiaddr) bool {
return dialMatcher.Matches(addr)
}
func (t *TcpTransport) maDial(ctx context.Context, raddr ma.Multiaddr) (manet.Conn, error) {
// Apply the deadline iff applicable
if t.ConnectTimeout > 0 {
deadline := time.Now().Add(t.ConnectTimeout)
if d, ok := ctx.Deadline(); !ok || deadline.Before(d) {
var cancel func()
ctx, cancel = context.WithDeadline(ctx, deadline)
defer cancel()
}
}
if t.UseReuseport() {
return t.reuse.DialContext(ctx, raddr)
}
var d manet.Dialer
return d.DialContext(ctx, raddr)
}
// Dial dials the peer at the remote address.
func (t *TcpTransport) Dial(ctx context.Context, raddr ma.Multiaddr, p peer.ID) (transport.CapableConn, error) {
conn, err := t.maDial(ctx, raddr)
if err != nil {
return nil, err
}
// Set linger to 0 so we never get stuck in the TIME-WAIT state. When
// linger is 0, connections are _reset_ instead of closed with a FIN.
// This means we can immediately reuse the 5-tuple and reconnect.
tryLinger(conn, 0)
return t.Upgrader.UpgradeOutbound(ctx, t, conn, p)
}
// UseReuseport returns true if reuseport is enabled and available.
func (t *TcpTransport) UseReuseport() bool {
return !t.DisableReuseport && ReuseportIsAvailable()
}
func (t *TcpTransport) maListen(laddr ma.Multiaddr) (manet.Listener, error) {
if t.UseReuseport() {
return t.reuse.Listen(laddr)
}
return manet.Listen(laddr)
}
// Listen listens on the given multiaddr.
func (t *TcpTransport) Listen(laddr ma.Multiaddr) (transport.Listener, error) {
list, err := t.maListen(laddr)
if err != nil {
return nil, err
}
list = &lingerListener{list, 0}
return t.Upgrader.UpgradeListener(t, list), nil
}
// Protocols returns the list of terminal protocols this transport can dial.
func (t *TcpTransport) Protocols() []int {
return []int{ma.P_TCP}
}
// Proxy always returns false for the TCP transport.
func (t *TcpTransport) Proxy() bool {
return false
}
func (t *TcpTransport) String() string {
return "TCP"
}