p2p/host: nat manager
this commit moves management of the nat to its own object. perhaps this can be general enough to work with any host (not just BasicHost), but for now keeping here. the nat manager: - discovers and sets up the nat asynchronously. - adds any port mappings necessary if/after nat has been found. - listens to the network Listen() changes, adding/closing mappings
Juan Batiz-Benet committed
Jan 30, 2015 at 20:23 UTC
763cc945c01eba903249a9ce23a6c37bbe7f9e0d
2 files changed
+237
-44
p2p/host/basic/basic_host.go
+13
-44
@@ -1,15 +1,12 @@
1
package basichost
2
3
import (
4
- "sync"
5
-
4
context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
5
ma "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr"
6
goprocess "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/goprocess"
7
8
eventlog "github.com/jbenet/go-ipfs/thirdparty/eventlog"
9
12
- inat "github.com/jbenet/go-ipfs/p2p/nat"
10
inet "github.com/jbenet/go-ipfs/p2p/net"
11
peer "github.com/jbenet/go-ipfs/p2p/peer"
12
protocol "github.com/jbenet/go-ipfs/p2p/protocol"
@@ -42,9 +39,7 @@ type BasicHost struct {
39
mux *protocol.Mux
40
ids *identify.IDService
41
relay *relay.RelayService
45
-
46
- natmu sync.Mutex
47
- nat *inat.NAT
42
+ natmgr *natManager
43
44
proc goprocess.Process
45
}
@@ -57,6 +52,10 @@ func New(net inet.Network, opts ...Option) *BasicHost {
52
}
53
54
h.proc = goprocess.WithTeardown(func() error {
55
+ if h.natmgr != nil {
56
+ h.natmgr.Close()
57
+ }
58
+
59
return h.Network().Close()
60
})
61
@@ -70,45 +69,13 @@ func New(net inet.Network, opts ...Option) *BasicHost {
69
for _, o := range opts {
70
switch o {
71
case NATPortMap:
73
- h.setupNATPortMap()
72
+ h.natmgr = newNatManager(h)
73
}
74
}
75
76
return h
77
}
78
80
-func (h *BasicHost) setupNATPortMap() {
81
- // do this asynchronously to avoid blocking daemon startup
82
-
83
- h.proc.Go(func(worker goprocess.Process) {
84
- nat := inat.DiscoverNAT()
85
- if nat == nil { // no nat, or failed to get it.
86
- return
87
- }
88
-
89
- select {
90
- case <-worker.Closing():
91
- nat.Close()
92
- return
93
- default:
94
- }
95
-
96
- // wire up the nat to close when proc closes.
97
- h.proc.AddChild(nat.Process())
98
-
99
- h.natmu.Lock()
100
- h.nat = nat
101
- h.natmu.Unlock()
102
-
103
- addrs := h.Network().ListenAddresses()
104
- nat.PortMapAddrs(addrs)
105
- mapAddrs := nat.ExternalAddrs()
106
- if len(mapAddrs) > 0 {
107
- log.Infof("NAT mapping addrs: %s", mapAddrs)
108
- }
109
- })
110
-}
111
-
79
// newConnHandler is the remote-opened conn handler for inet.Network
80
func (h *BasicHost) newConnHandler(c inet.Conn) {
81
h.ids.IdentifyConn(c)
@@ -214,17 +181,19 @@ func (h *BasicHost) dialPeer(ctx context.Context, p peer.ID) error {
181
return nil
182
}
183
184
+// Addrs returns all the addresses of BasicHost at this moment in time.
185
+// It's ok to not include addresses if they're not available to be used now.
186
func (h *BasicHost) Addrs() []ma.Multiaddr {
187
addrs, err := h.Network().InterfaceListenAddresses()
188
if err != nil {
189
log.Debug("error retrieving network interface addrs")
190
}
191
223
- h.natmu.Lock()
224
- nat := h.nat
225
- h.natmu.Unlock()
226
- if nat != nil {
227
- addrs = append(addrs, nat.ExternalAddrs()...)
192
+ if h.natmgr != nil { // natmgr is nil if we do not use nat option.
193
+ nat := h.natmgr.NAT()
194
+ if nat != nil { // nat is nil if not ready, or no nat is available.
195
+ addrs = append(addrs, nat.ExternalAddrs()...)
196
+ }
197
}
198
199
return addrs
p2p/host/basic/natmgr.go
new
+224
@@ -0,0 +1,224 @@
1
+package basichost
2
+
3
+import (
4
+ "sync"
5
+
6
+ context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
7
+ ma "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-multiaddr"
8
+ goprocess "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/goprocess"
9
+
10
+ inat "github.com/jbenet/go-ipfs/p2p/nat"
11
+ inet "github.com/jbenet/go-ipfs/p2p/net"
12
+ lgbl "github.com/jbenet/go-ipfs/util/eventlog/loggables"
13
+)
14
+
15
+// natManager takes care of adding + removing port mappings to the nat.
16
+// Initialized with the host if it has a NATPortMap option enabled.
17
+// natManager receives signals from the network, and check on nat mappings:
18
+// * natManager listens to the network and adds or closes port mappings
19
+// as the network signals Listen() or ListenClose().
20
+// * closing the natManager closes the nat and its mappings.
21
+type natManager struct {
22
+ host *BasicHost
23
+ natmu sync.RWMutex // guards nat (ready could obviate this mutex, but safety first.)
24
+ nat *inat.NAT
25
+
26
+ ready chan struct{} // closed once the nat is ready to process port mappings
27
+ proc goprocess.Process // natManager has a process + children. can be closed.
28
+}
29
+
30
+func newNatManager(host *BasicHost) *natManager {
31
+ nmgr := &natManager{
32
+ host: host,
33
+ ready: make(chan struct{}),
34
+ proc: goprocess.WithParent(host.proc),
35
+ }
36
+
37
+ // teardown
38
+ nmgr.proc = goprocess.WithTeardown(func() error {
39
+ // on closing, unregister from network notifications.
40
+ host.Network().StopNotify((*nmgrNetNotifiee)(nmgr))
41
+ return nil
42
+ })
43
+
44
+ // host is our parent. close when host closes.
45
+ host.proc.AddChild(nmgr.proc)
46
+
47
+ // discover the nat.
48
+ nmgr.discoverNAT()
49
+ return nmgr
50
+}
51
+
52
+// Close closes the natManager, closing the underlying nat
53
+// and unregistering from network events.
54
+func (nmgr *natManager) Close() error {
55
+ return nmgr.proc.Close()
56
+}
57
+
58
+// Ready returns a channel which will be closed when the NAT has been found
59
+// and is ready to be used, or the search process is done.
60
+func (nmgr *natManager) Ready() <-chan struct{} {
61
+ return nmgr.ready
62
+}
63
+
64
+func (nmgr *natManager) discoverNAT() {
65
+
66
+ nmgr.proc.Go(func(worker goprocess.Process) {
67
+ // inat.DiscoverNAT blocks until the nat is found or a timeout
68
+ // is reached. we unfortunately cannot specify timeouts-- the
69
+ // library we're using just blocks.
70
+ //
71
+ // Note: on early shutdown, there may be a case where we're trying
72
+ // to close before DiscoverNAT() returns. Since we cant cancel it
73
+ // (library) we can choose to (1) drop the result and return early,
74
+ // or (2) wait until it times out to exit. For now we choose (2),
75
+ // to avoid leaking resources in a non-obvious way. the only case
76
+ // this affects is when the daemon is being started up and _immediately_
77
+ // asked to close. other services are also starting up, so ok to wait.
78
+ nat := inat.DiscoverNAT()
79
+ if nat == nil { // no nat, or failed to get it.
80
+ return
81
+ }
82
+
83
+ // by this point -- after finding the NAT -- we may have already
84
+ // be closing. if so, just exit.
85
+ select {
86
+ case <-worker.Closing():
87
+ nat.Close()
88
+ return
89
+ default:
90
+ }
91
+
92
+ // wire up the nat to close when nmgr closes.
93
+ // nmgr.proc is our parent, and waiting for us.
94
+ nmgr.proc.AddChild(nat.Process())
95
+
96
+ // set the nat.
97
+ nmgr.natmu.Lock()
98
+ nmgr.nat = nat
99
+ nmgr.natmu.Unlock()
100
+
101
+ // signal that we're ready to process nat mappings:
102
+ close(nmgr.ready)
103
+
104
+ // sign natManager up for network notifications
105
+ // we need to sign up here to avoid missing some notifs
106
+ // before the NAT has been found.
107
+ nmgr.host.Network().Notify((*nmgrNetNotifiee)(nmgr))
108
+
109
+ // if any interfaces were brought up while we were setting up
110
+ // the nat, now is the time to setup port mappings for them.
111
+ // we release ready, then grab them to avoid losing any. adding
112
+ // a port mapping is idempotent, so its ok to add the same twice.
113
+ addrs := nmgr.host.Network().ListenAddresses()
114
+ for _, addr := range addrs {
115
+ // we do it async because it's slow and we may want to close beforehand
116
+ go addPortMapping(nmgr, addr)
117
+ }
118
+ })
119
+}
120
+
121
+// NAT returns the natManager's nat object. this may be nil, if
122
+// (a) the search process is still ongoing, or (b) the search process
123
+// found no nat. Clients must check whether the return value is nil.
124
+func (nmgr *natManager) NAT() *inat.NAT {
125
+ nmgr.natmu.Lock()
126
+ defer nmgr.natmu.Unlock()
127
+ return nmgr.nat
128
+}
129
+
130
+func addPortMapping(nmgr *natManager, intaddr ma.Multiaddr) {
131
+ nat := nmgr.NAT()
132
+ if nat == nil {
133
+ panic("natManager addPortMapping called without a nat.")
134
+ }
135
+
136
+ // first, check if the port mapping already exists.
137
+ for _, mapping := range nat.Mappings() {
138
+ if mapping.InternalAddr().Equal(intaddr) {
139
+ return // it exists! return.
140
+ }
141
+ }
142
+
143
+ ctx := context.TODO()
144
+ lm := make(lgbl.DeferredMap)
145
+ lm["internalAddr"] = func() interface{} { return intaddr.String() }
146
+
147
+ defer log.EventBegin(ctx, "natMgrAddPortMappingWait", lm).Done()
148
+
149
+ select {
150
+ case <-nmgr.proc.Closing():
151
+ lm["outcome"] = "cancelled"
152
+ return // no use.
153
+ case <-nmgr.ready: // wait until it's ready.
154
+ }
155
+
156
+ // actually start the port map (sub-event because waiting may take a while)
157
+ defer log.EventBegin(ctx, "natMgrAddPortMapping", lm).Done()
158
+
159
+ // get the nat
160
+ m, err := nat.NewMapping(intaddr)
161
+ if err != nil {
162
+ lm["outcome"] = "failure"
163
+ lm["error"] = err
164
+ return
165
+ }
166
+
167
+ extaddr, err := m.ExternalAddr()
168
+ if err != nil {
169
+ lm["outcome"] = "failure"
170
+ lm["error"] = err
171
+ return
172
+ }
173
+
174
+ lm["outcome"] = "success"
175
+ lm["externalAddr"] = func() interface{} { return extaddr.String() }
176
+ log.Infof("established nat port mapping: %s <--> %s", intaddr, extaddr)
177
+}
178
+
179
+func rmPortMapping(nmgr *natManager, intaddr ma.Multiaddr) {
180
+ nat := nmgr.NAT()
181
+ if nat == nil {
182
+ panic("natManager rmPortMapping called without a nat.")
183
+ }
184
+
185
+ // list the port mappings (it may be gone on it's own, so we need to
186
+ // check this list, and not store it ourselves behind the scenes)
187
+
188
+ // close mappings for this internal address.
189
+ for _, mapping := range nat.Mappings() {
190
+ if mapping.InternalAddr().Equal(intaddr) {
191
+ mapping.Close()
192
+ }
193
+ }
194
+}
195
+
196
+// nmgrNetNotifiee implements the network notification listening part
197
+// of the natManager. this is merely listening to Listen() and ListenClose()
198
+// events.
199
+type nmgrNetNotifiee natManager
200
+
201
+func (nn *nmgrNetNotifiee) natManager() *natManager {
202
+ return (*natManager)(nn)
203
+}
204
+
205
+func (nn *nmgrNetNotifiee) Listen(n inet.Network, addr ma.Multiaddr) {
206
+ if nn.natManager().NAT() == nil {
207
+ return // not ready or doesnt exist.
208
+ }
209
+
210
+ addPortMapping(nn.natManager(), addr)
211
+}
212
+
213
+func (nn *nmgrNetNotifiee) ListenClose(n inet.Network, addr ma.Multiaddr) {
214
+ if nn.natManager().NAT() == nil {
215
+ return // not ready or doesnt exist.
216
+ }
217
+
218
+ rmPortMapping(nn.natManager(), addr)
219
+}
220
+
221
+func (nn *nmgrNetNotifiee) Connected(inet.Network, inet.Conn) {}
222
+func (nn *nmgrNetNotifiee) Disconnected(inet.Network, inet.Conn) {}
223
+func (nn *nmgrNetNotifiee) OpenedStream(inet.Network, inet.Stream) {}
224
+func (nn *nmgrNetNotifiee) ClosedStream(inet.Network, inet.Stream) {}