net: have an explicit IdentifyConn on dial
- Make sure we call IdentifyConn on dialed out conns - we wait until the identify is **done** before return - on listening case, we can also wait. - tests now make sure dial does wait. - tests now make sure we can wait on listening case.
Juan Batiz-Benet committed
Dec 22, 2014 at 20:41 UTC
4fe1dd9b62ede2def307d6a025eecbe262a9fec8
5 files changed
+149
-19
net/id.go
+64
-1
@@ -1,6 +1,8 @@
1
package net
2
3
import (
4
+ "sync"
5
+
6
handshake "github.com/jbenet/go-ipfs/net/handshake"
7
pb "github.com/jbenet/go-ipfs/net/handshake/pb"
8
@@ -18,14 +20,54 @@ import (
20
// * Our public Listen Addresses
21
type IDService struct {
22
Network Network
23
+
24
+ // connections undergoing identification
25
+ // for wait purposes
26
+ currid map[Conn]chan struct{}
27
+ currmu sync.RWMutex
28
}
29
30
func NewIDService(n Network) *IDService {
24
- s := &IDService{Network: n}
31
+ s := &IDService{
32
+ Network: n,
33
+ currid: make(map[Conn]chan struct{}),
34
+ }
35
n.SetHandler(ProtocolIdentify, s.RequestHandler)
36
return s
37
}
38
39
+func (ids *IDService) IdentifyConn(c Conn) {
40
+ ids.currmu.Lock()
41
+ if _, found := ids.currid[c]; found {
42
+ ids.currmu.Unlock()
43
+ log.Debugf("IdentifyConn called twice on: %s", c)
44
+ return // already identifying it.
45
+ }
46
+ ids.currid[c] = make(chan struct{})
47
+ ids.currmu.Unlock()
48
+
49
+ s, err := c.NewStreamWithProtocol(ProtocolIdentify)
50
+ if err != nil {
51
+ log.Error("network: unable to open initial stream for %s", ProtocolIdentify)
52
+ log.Event(ids.Network.CtxGroup().Context(), "IdentifyOpenFailed", c.RemotePeer())
53
+ }
54
+
55
+ // ok give the response to our handler.
56
+ ids.ResponseHandler(s)
57
+
58
+ ids.currmu.Lock()
59
+ ch, found := ids.currid[c]
60
+ delete(ids.currid, c)
61
+ ids.currmu.Unlock()
62
+
63
+ if !found {
64
+ log.Errorf("IdentifyConn failed to find channel (programmer error) for %s", c)
65
+ return
66
+ }
67
+
68
+ close(ch) // release everyone waiting.
69
+}
70
+
71
func (ids *IDService) RequestHandler(s Stream) {
72
defer s.Close()
73
c := s.Conn()
@@ -101,6 +143,7 @@ func (ids *IDService) consumeMessage(mes *pb.Handshake3, c Conn) {
143
144
// update our peerstore with the addresses.
145
ids.Network.Peerstore().AddAddresses(p, lmaddrs)
146
+ log.Debugf("%s received listen addrs for %s: %s", c.LocalPeer(), c.RemotePeer(), lmaddrs)
147
148
// get protocol versions
149
pv := *mes.H1.ProtocolVersion
@@ -108,3 +151,23 @@ func (ids *IDService) consumeMessage(mes *pb.Handshake3, c Conn) {
151
ids.Network.Peerstore().Put(p, "ProtocolVersion", pv)
152
ids.Network.Peerstore().Put(p, "AgentVersion", av)
153
}
154
+
155
+// IdentifyWait returns a channel which will be closed once
156
+// "ProtocolIdentify" (handshake3) finishes on given conn.
157
+// This happens async so the connection can start to be used
158
+// even if handshake3 knowledge is not necesary.
159
+// Users **MUST** call IdentifyWait _after_ IdentifyConn
160
+func (ids *IDService) IdentifyWait(c Conn) <-chan struct{} {
161
+ ids.currmu.Lock()
162
+ ch, found := ids.currid[c]
163
+ ids.currmu.Unlock()
164
+ if found {
165
+ return ch
166
+ }
167
+
168
+ // if not found, it means we are already done identifying it, or
169
+ // haven't even started. either way, return a new channel closed.
170
+ ch = make(chan struct{})
171
+ close(ch)
172
+ return ch
173
+}
net/id_test.go
+39
-8
@@ -32,7 +32,7 @@ func DivulgeAddresses(a, b inet.Network) {
32
b.Peerstore().AddAddresses(id, addrs)
33
}
34
35
-func TestIDService(t *testing.T) {
35
+func subtestIDService(t *testing.T, postDialWait time.Duration) {
36
37
// the generated networks should have the id service wired in.
38
ctx := context.Background()
@@ -55,16 +55,26 @@ func TestIDService(t *testing.T) {
55
t.Fatalf("Failed to dial:", err)
56
}
57
58
- // this is shitty. dial should wait for connecting to end
59
- <-time.After(100 * time.Millisecond)
58
+ // we need to wait here if Dial returns before ID service is finished.
59
+ if postDialWait > 0 {
60
+ <-time.After(postDialWait)
61
+ }
62
63
// the IDService should be opened automatically, by the network.
64
// what we should see now is that both peers know about each others listen addresses.
65
testKnowsAddrs(t, n1, n2p, n2.Peerstore().Addresses(n2p)) // has them
64
- testKnowsAddrs(t, n2, n1p, n1.Peerstore().Addresses(n1p)) // has them
66
+ testHasProtocolVersions(t, n1, n2p)
67
+
68
+ // now, this wait we do have to do. it's the wait for the Listening side
69
+ // to be done identifying the connection.
70
+ c := n2.ConnsToPeer(n1.LocalPeer())
71
+ if len(c) < 1 {
72
+ t.Fatal("should have connection by now at least.")
73
+ }
74
+ <-n2.IdentifyProtocol().IdentifyWait(c[0])
75
76
// and the protocol versions.
67
- testHasProtocolVersions(t, n1, n2p)
77
+ testKnowsAddrs(t, n2, n1p, n1.Peerstore().Addresses(n1p)) // has them
78
testHasProtocolVersions(t, n2, n1p)
79
}
80
@@ -82,18 +92,39 @@ func testKnowsAddrs(t *testing.T, n inet.Network, p peer.ID, expected []ma.Multi
92
for _, addr := range expected {
93
if _, found := have[addr.String()]; !found {
94
t.Errorf("%s did not have addr for %s: %s", n.LocalPeer(), p, addr)
85
- panic("ahhhhhhh")
95
+ // panic("ahhhhhhh")
96
}
97
}
98
}
99
100
func testHasProtocolVersions(t *testing.T, n inet.Network, p peer.ID) {
101
v, err := n.Peerstore().Get(p, "ProtocolVersion")
102
+ if v == nil {
103
+ t.Error("no protocol version")
104
+ return
105
+ }
106
if v.(string) != handshake.IpfsVersion.String() {
93
- t.Fatal("protocol mismatch", err)
107
+ t.Error("protocol mismatch", err)
108
}
109
v, err = n.Peerstore().Get(p, "AgentVersion")
110
if v.(string) != handshake.ClientVersion {
97
- t.Fatal("agent version mismatch", err)
111
+ t.Error("agent version mismatch", err)
112
+ }
113
+}
114
+
115
+// TestIDServiceWait gives the ID service 100ms to finish after dialing
116
+// this is becasue it used to be concurrent. Now, Dial wait till the
117
+// id service is done.
118
+func TestIDServiceWait(t *testing.T) {
119
+ N := 3
120
+ for i := 0; i < N; i++ {
121
+ subtestIDService(t, 100*time.Millisecond)
122
+ }
123
+}
124
+
125
+func TestIDServiceNoWait(t *testing.T) {
126
+ N := 3
127
+ for i := 0; i < N; i++ {
128
+ subtestIDService(t, 0)
129
}
130
}
net/interface.go
+8
@@ -88,6 +88,9 @@ type Network interface {
88
// Conns returns the connections in this Netowrk
89
Conns() []Conn
90
91
+ // ConnsToPeer returns the connections in this Netowrk for given peer.
92
+ ConnsToPeer(p peer.ID) []Conn
93
+
94
// BandwidthTotals returns the total number of bytes passed through
95
// the network since it was instantiated
96
BandwidthTotals() (uint64, uint64)
@@ -102,6 +105,11 @@ type Network interface {
105
106
// CtxGroup returns the network's contextGroup
107
CtxGroup() ctxgroup.ContextGroup
108
+
109
+ // IdentifyProtocol returns the instance of the object running the Identify
110
+ // Protocol. This is what runs the ifps handshake-- this should be removed
111
+ // if this abstracted out to its own package.
112
+ IdentifyProtocol() *IDService
113
}
114
115
// Dialer represents a service that can dial out to peers
net/mock/mock_peernet.go
+14
@@ -29,6 +29,7 @@ type peernet struct {
29
30
// needed to implement inet.Network
31
mux inet.Mux
32
+ ids *inet.IDService
33
34
cg ctxgroup.ContextGroup
35
sync.RWMutex
@@ -61,6 +62,11 @@ func newPeernet(ctx context.Context, m *mocknet, k ic.PrivKey,
62
}
63
64
n.cg.SetTeardown(n.teardown)
65
+
66
+ // setup a conn handler that immediately "asks the other side about them"
67
+ // this is ProtocolIdentify.
68
+ n.ids = inet.NewIDService(n)
69
+
70
return n, nil
71
}
72
@@ -158,6 +164,10 @@ func (pn *peernet) remoteOpenedConn(c *conn) {
164
// addConn constructs and adds a connection
165
// to given remote peer over given link
166
func (pn *peernet) addConn(c *conn) {
167
+
168
+ // run the Identify protocol/handshake.
169
+ pn.ids.IdentifyConn(c)
170
+
171
pn.Lock()
172
cs, found := pn.connsByPeer[c.RemotePeer()]
173
if !found {
@@ -327,3 +337,7 @@ func (pn *peernet) NewStream(pr inet.ProtocolID, p peer.ID) (inet.Stream, error)
337
func (pn *peernet) SetHandler(p inet.ProtocolID, h inet.StreamHandler) {
338
pn.mux.SetHandler(p, h)
339
}
340
+
341
+func (pn *peernet) IdentifyProtocol() *inet.IDService {
342
+ return pn.ids
343
+}
net/net.go
+24
-10
@@ -129,21 +129,21 @@ func NewNetwork(ctx context.Context, listen []ma.Multiaddr, local peer.ID,
129
130
func (n *network) newConnHandler(c *swarm.Conn) {
131
cc := (*conn_)(c)
132
- s, err := cc.NewStreamWithProtocol(ProtocolIdentify)
133
- if err != nil {
134
- log.Error("network: unable to open initial stream for %s", ProtocolIdentify)
135
- log.Event(n.CtxGroup().Context(), "IdentifyOpenFailed", c.RemotePeer())
136
- }
137
-
138
- // ok give the response to our handler.
139
- n.ids.ResponseHandler(s)
132
+ n.ids.IdentifyConn(cc)
133
}
134
135
// DialPeer attempts to establish a connection to a given peer.
136
// Respects the context.
137
func (n *network) DialPeer(ctx context.Context, p peer.ID) error {
145
- _, err := n.swarm.Dial(ctx, p)
146
- return err
138
+ sc, err := n.swarm.Dial(ctx, p)
139
+ if err != nil {
140
+ return err
141
+ }
142
+
143
+ // identify the connection before returning.
144
+ n.ids.IdentifyConn((*conn_)(sc))
145
+ log.Debugf("network for %s finished dialing %s", n.local, p)
146
+ return nil
147
}
148
149
func (n *network) Protocols() []ProtocolID {
@@ -185,6 +185,16 @@ func (n *network) Conns() []Conn {
185
return out
186
}
187
188
+// ConnsToPeer returns the connections in this Netowrk for given peer.
189
+func (n *network) ConnsToPeer(p peer.ID) []Conn {
190
+ conns1 := n.swarm.ConnectionsToPeer(p)
191
+ out := make([]Conn, len(conns1))
192
+ for i, c := range conns1 {
193
+ out[i] = (*conn_)(c)
194
+ }
195
+ return out
196
+}
197
+
198
// ClosePeer connection to peer
199
func (n *network) ClosePeer(p peer.ID) error {
200
return n.swarm.CloseConnection(p)
@@ -254,6 +264,10 @@ func (n *network) SetHandler(p ProtocolID, h StreamHandler) {
264
n.mux.SetHandler(p, h)
265
}
266
267
+func (n *network) IdentifyProtocol() *IDService {
268
+ return n.ids
269
+}
270
+
271
func WriteProtocolHeader(pr ProtocolID, s Stream) error {
272
if pr != "" { // only write proper protocol headers
273
if err := WriteLengthPrefix(s, string(pr)); err != nil {