p2p/net/swarm: notifications
Juan Batiz-Benet committed
Jan 24, 2015 at 07:14 UTC
4ae01e7a5ec3eddf183eb01f01553194b665c298
8 files changed
+520
-8
p2p/net/interface.go
+17
@@ -111,6 +111,10 @@ type Dialer interface {
111
112
// ConnsToPeer returns the connections in this Netowrk for given peer.
113
ConnsToPeer(p peer.ID) []Conn
114
+
115
+ // Notify/StopNotify register and unregister a notifiee for signals
116
+ Notify(Notifiee)
117
+ StopNotify(Notifiee)
118
}
119
120
// Connectedness signals the capacity for a connection with a given node.
@@ -131,3 +135,16 @@ const (
135
// (should signal "made effort, failed")
136
CannotConnect
137
)
138
+
139
+// Notifiee is an interface for an object wishing to receive
140
+// notifications from a Network.
141
+type Notifiee interface {
142
+ Connected(Network, Conn) // called when a connection opened
143
+ Disconnected(Network, Conn) // called when a connection closed
144
+ OpenedStream(Network, Stream) // called when a stream opened
145
+ ClosedStream(Network, Stream) // called when a stream closed
146
+
147
+ // TODO
148
+ // PeerConnected(Network, peer.ID) // called when a peer connected
149
+ // PeerDisconnected(Network, peer.ID) // called when a peer disconnected
150
+}
p2p/net/mock/mock_conn.go
+9
@@ -37,6 +37,9 @@ func (c *conn) Close() error {
37
s.Close()
38
}
39
c.net.removeConn(c)
40
+ c.net.notifyAll(func(n inet.Notifiee) {
41
+ n.Disconnected(c.net, c)
42
+ })
43
return nil
44
}
45
@@ -73,11 +76,17 @@ func (c *conn) allStreams() []inet.Stream {
76
func (c *conn) remoteOpenedStream(s *stream) {
77
c.addStream(s)
78
c.net.handleNewStream(s)
79
+ c.net.notifyAll(func(n inet.Notifiee) {
80
+ n.OpenedStream(c.net, s)
81
+ })
82
}
83
84
func (c *conn) openStream() *stream {
85
sl, sr := c.link.newStreamPair()
86
c.addStream(sl)
87
+ c.net.notifyAll(func(n inet.Notifiee) {
88
+ n.OpenedStream(c.net, sl)
89
+ })
90
c.rconn.remoteOpenedStream(sr)
91
return sl
92
}
p2p/net/mock/mock_notif_test.go
new
+198
@@ -0,0 +1,198 @@
1
+package mocknet
2
+
3
+import (
4
+ "testing"
5
+ "time"
6
+
7
+ inet "github.com/jbenet/go-ipfs/p2p/net"
8
+
9
+ context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
10
+)
11
+
12
+func TestNotifications(t *testing.T) {
13
+ t.Parallel()
14
+
15
+ mn, err := FullMeshLinked(context.Background(), 5)
16
+ if err != nil {
17
+ t.Fatal(err)
18
+ }
19
+
20
+ timeout := 5 * time.Second
21
+
22
+ // signup notifs
23
+ nets := mn.Nets()
24
+ notifiees := make([]*netNotifiee, len(nets))
25
+ for i, pn := range nets {
26
+ n := newNetNotifiee()
27
+ pn.Notify(n)
28
+ notifiees[i] = n
29
+ }
30
+
31
+ // connect all
32
+ for _, n1 := range nets {
33
+ for _, n2 := range nets {
34
+ if n1 == n2 {
35
+ continue
36
+ }
37
+ if _, err := mn.ConnectNets(n1, n2); err != nil {
38
+ t.Fatal(err)
39
+ }
40
+ }
41
+ }
42
+
43
+ // test everyone got the correct connection opened calls
44
+ for i, s := range nets {
45
+ n := notifiees[i]
46
+ for _, s2 := range nets {
47
+ cos := s.ConnsToPeer(s2.LocalPeer())
48
+ func() {
49
+ for i := 0; i < len(cos); i++ {
50
+ var c inet.Conn
51
+ select {
52
+ case c = <-n.connected:
53
+ case <-time.After(timeout):
54
+ t.Fatal("timeout")
55
+ }
56
+ for _, c2 := range cos {
57
+ if c == c2 {
58
+ t.Log("got notif for conn")
59
+ return
60
+ }
61
+ }
62
+ t.Error("connection not found")
63
+ }
64
+ }()
65
+ }
66
+ }
67
+
68
+ complement := func(c inet.Conn) (inet.Network, *netNotifiee, *conn) {
69
+ for i, s := range nets {
70
+ for _, c2 := range s.Conns() {
71
+ if c2.(*conn).rconn == c {
72
+ return s, notifiees[i], c2.(*conn)
73
+ }
74
+ }
75
+ }
76
+ t.Fatal("complementary conn not found", c)
77
+ return nil, nil, nil
78
+ }
79
+
80
+ testOCStream := func(n *netNotifiee, s inet.Stream) {
81
+ var s2 inet.Stream
82
+ select {
83
+ case s2 = <-n.openedStream:
84
+ t.Log("got notif for opened stream")
85
+ case <-time.After(timeout):
86
+ t.Fatal("timeout")
87
+ }
88
+ if s != nil && s != s2 {
89
+ t.Fatalf("got incorrect stream %p %p", s, s2)
90
+ }
91
+
92
+ select {
93
+ case s2 = <-n.closedStream:
94
+ t.Log("got notif for closed stream")
95
+ case <-time.After(timeout):
96
+ t.Fatal("timeout")
97
+ }
98
+ if s != nil && s != s2 {
99
+ t.Fatalf("got incorrect stream %p %p", s, s2)
100
+ }
101
+ }
102
+
103
+ streams := make(chan inet.Stream)
104
+ for _, s := range nets {
105
+ s.SetStreamHandler(func(s inet.Stream) {
106
+ streams <- s
107
+ s.Close()
108
+ })
109
+ }
110
+
111
+ // there's one stream per conn that we need to drain....
112
+ // unsure where these are coming from
113
+ for i, _ := range nets {
114
+ n := notifiees[i]
115
+ testOCStream(n, nil)
116
+ testOCStream(n, nil)
117
+ testOCStream(n, nil)
118
+ testOCStream(n, nil)
119
+ }
120
+
121
+ // open a streams in each conn
122
+ for i, s := range nets {
123
+ conns := s.Conns()
124
+ for _, c := range conns {
125
+ _, n2, c2 := complement(c)
126
+ st1, err := c.NewStream()
127
+ if err != nil {
128
+ t.Error(err)
129
+ } else {
130
+ t.Logf("%s %s <--%p--> %s %s", c.LocalPeer(), c.LocalMultiaddr(), st1, c.RemotePeer(), c.RemoteMultiaddr())
131
+ // st1.Write([]byte("hello"))
132
+ st1.Close()
133
+ st2 := <-streams
134
+ t.Logf("%s %s <--%p--> %s %s", c2.LocalPeer(), c2.LocalMultiaddr(), st2, c2.RemotePeer(), c2.RemoteMultiaddr())
135
+ testOCStream(notifiees[i], st1)
136
+ testOCStream(n2, st2)
137
+ }
138
+ }
139
+ }
140
+
141
+ // close conns
142
+ for i, s := range nets {
143
+ n := notifiees[i]
144
+ for _, c := range s.Conns() {
145
+ _, n2, c2 := complement(c)
146
+ c.(*conn).Close()
147
+ c2.Close()
148
+
149
+ var c3, c4 inet.Conn
150
+ select {
151
+ case c3 = <-n.disconnected:
152
+ case <-time.After(timeout):
153
+ t.Fatal("timeout")
154
+ }
155
+ if c != c3 {
156
+ t.Fatal("got incorrect conn", c, c3)
157
+ }
158
+
159
+ select {
160
+ case c4 = <-n2.disconnected:
161
+ case <-time.After(timeout):
162
+ t.Fatal("timeout")
163
+ }
164
+ if c2 != c4 {
165
+ t.Fatal("got incorrect conn", c, c2)
166
+ }
167
+ }
168
+ }
169
+}
170
+
171
+type netNotifiee struct {
172
+ connected chan inet.Conn
173
+ disconnected chan inet.Conn
174
+ openedStream chan inet.Stream
175
+ closedStream chan inet.Stream
176
+}
177
+
178
+func newNetNotifiee() *netNotifiee {
179
+ return &netNotifiee{
180
+ connected: make(chan inet.Conn),
181
+ disconnected: make(chan inet.Conn),
182
+ openedStream: make(chan inet.Stream),
183
+ closedStream: make(chan inet.Stream),
184
+ }
185
+}
186
+
187
+func (nn *netNotifiee) Connected(n inet.Network, v inet.Conn) {
188
+ nn.connected <- v
189
+}
190
+func (nn *netNotifiee) Disconnected(n inet.Network, v inet.Conn) {
191
+ nn.disconnected <- v
192
+}
193
+func (nn *netNotifiee) OpenedStream(n inet.Network, v inet.Stream) {
194
+ nn.openedStream <- v
195
+}
196
+func (nn *netNotifiee) ClosedStream(n inet.Network, v inet.Stream) {
197
+ nn.closedStream <- v
198
+}
p2p/net/mock/mock_peernet.go
+38
-2
@@ -31,6 +31,9 @@ type peernet struct {
31
streamHandler inet.StreamHandler
32
connHandler inet.ConnHandler
33
34
+ notifmu sync.RWMutex
35
+ notifs map[inet.Notifiee]struct{}
36
+
37
cg ctxgroup.ContextGroup
38
sync.RWMutex
39
}
@@ -58,6 +61,8 @@ func newPeernet(ctx context.Context, m *mocknet, k ic.PrivKey,
61
62
connsByPeer: map[peer.ID]map[*conn]struct{}{},
63
connsByLink: map[*link]map[*conn]struct{}{},
64
+
65
+ notifs: make(map[inet.Notifiee]struct{}),
66
}
67
68
n.cg.SetTeardown(n.teardown)
@@ -163,6 +168,9 @@ func (pn *peernet) openConn(r peer.ID, l *link) *conn {
168
lc, rc := l.newConnPair(pn)
169
log.Debugf("%s opening connection to %s", pn.LocalPeer(), lc.RemotePeer())
170
pn.addConn(lc)
171
+ pn.notifyAll(func(n inet.Notifiee) {
172
+ n.Connected(pn, lc)
173
+ })
174
rc.net.remoteOpenedConn(rc)
175
return lc
176
}
@@ -171,6 +179,9 @@ func (pn *peernet) remoteOpenedConn(c *conn) {
179
log.Debugf("%s accepting connection from %s", pn.LocalPeer(), c.RemotePeer())
180
pn.addConn(c)
181
pn.handleNewConn(c)
182
+ pn.notifyAll(func(n inet.Notifiee) {
183
+ n.Connected(pn, c)
184
+ })
185
}
186
187
// addConn constructs and adds a connection
@@ -201,13 +212,13 @@ func (pn *peernet) removeConn(c *conn) {
212
213
cs, found := pn.connsByLink[c.link]
214
if !found || len(cs) < 1 {
204
- panic("attempting to remove a conn that doesnt exist")
215
+ panic(fmt.Sprintf("attempting to remove a conn that doesnt exist %p", c.link))
216
}
217
delete(cs, c)
218
219
cs, found = pn.connsByPeer[c.remote]
220
if !found {
210
- panic("attempting to remove a conn that doesnt exist")
221
+ panic(fmt.Sprintf("attempting to remove a conn that doesnt exist %p", c.remote))
222
}
223
delete(cs, c)
224
}
@@ -360,3 +371,28 @@ func (pn *peernet) SetConnHandler(h inet.ConnHandler) {
371
pn.connHandler = h
372
pn.Unlock()
373
}
374
+
375
+// Notify signs up Notifiee to receive signals when events happen
376
+func (pn *peernet) Notify(f inet.Notifiee) {
377
+ pn.notifmu.Lock()
378
+ pn.notifs[f] = struct{}{}
379
+ pn.notifmu.Unlock()
380
+}
381
+
382
+// StopNotify unregisters Notifiee fromr receiving signals
383
+func (pn *peernet) StopNotify(f inet.Notifiee) {
384
+ pn.notifmu.Lock()
385
+ delete(pn.notifs, f)
386
+ pn.notifmu.Unlock()
387
+}
388
+
389
+// notifyAll runs the notification function on all Notifiees
390
+func (pn *peernet) notifyAll(notification func(f inet.Notifiee)) {
391
+ pn.notifmu.RLock()
392
+ for n := range pn.notifs {
393
+ // make sure we dont block
394
+ // and they dont block each other.
395
+ go notification(n)
396
+ }
397
+ pn.notifmu.RUnlock()
398
+}
p2p/net/mock/mock_stream.go
+4
-1
@@ -19,8 +19,11 @@ func (s *stream) Close() error {
19
r.Close()
20
}
21
if w, ok := (s.Writer).(io.Closer); ok {
22
- return w.Close()
22
+ w.Close()
23
}
24
+ s.conn.net.notifyAll(func(n inet.Notifiee) {
25
+ n.ClosedStream(s.conn.net, s)
26
+ })
27
return nil
28
}
29
p2p/net/swarm/swarm.go
+58
-5
@@ -4,6 +4,7 @@ package swarm
4
5
import (
6
"fmt"
7
+ "sync"
8
"time"
9
10
inet "github.com/jbenet/go-ipfs/p2p/net"
@@ -38,6 +39,9 @@ type Swarm struct {
39
backf dialbackoff
40
dialT time.Duration // mainly for tests
41
42
+ notifmu sync.RWMutex
43
+ notifs map[inet.Notifiee]ps.Notifiee
44
+
45
cg ctxgroup.ContextGroup
46
}
47
@@ -54,11 +58,12 @@ func NewSwarm(ctx context.Context, listenAddrs []ma.Multiaddr,
58
}
59
60
s := &Swarm{
57
- swarm: ps.NewSwarm(PSTransport),
58
- local: local,
59
- peers: peers,
60
- cg: ctxgroup.WithContext(ctx),
61
- dialT: DialTimeout,
61
+ swarm: ps.NewSwarm(PSTransport),
62
+ local: local,
63
+ peers: peers,
64
+ cg: ctxgroup.WithContext(ctx),
65
+ dialT: DialTimeout,
66
+ notifs: make(map[inet.Notifiee]ps.Notifiee),
67
}
68
69
// configure Swarm
@@ -177,3 +182,51 @@ func (s *Swarm) Peers() []peer.ID {
182
func (s *Swarm) LocalPeer() peer.ID {
183
return s.local
184
}
185
+
186
+// Notify signs up Notifiee to receive signals when events happen
187
+func (s *Swarm) Notify(f inet.Notifiee) {
188
+ // wrap with our notifiee, to translate function calls
189
+ n := &ps2netNotifee{net: (*Network)(s), not: f}
190
+
191
+ s.notifmu.Lock()
192
+ s.notifs[f] = n
193
+ s.notifmu.Unlock()
194
+
195
+ // register for notifications in the peer swarm.
196
+ s.swarm.Notify(n)
197
+}
198
+
199
+// StopNotify unregisters Notifiee fromr receiving signals
200
+func (s *Swarm) StopNotify(f inet.Notifiee) {
201
+ s.notifmu.Lock()
202
+ n, found := s.notifs[f]
203
+ if found {
204
+ delete(s.notifs, f)
205
+ }
206
+ s.notifmu.Unlock()
207
+
208
+ if found {
209
+ s.swarm.StopNotify(n)
210
+ }
211
+}
212
+
213
+type ps2netNotifee struct {
214
+ net *Network
215
+ not inet.Notifiee
216
+}
217
+
218
+func (n *ps2netNotifee) Connected(c *ps.Conn) {
219
+ n.not.Connected(n.net, inet.Conn((*Conn)(c)))
220
+}
221
+
222
+func (n *ps2netNotifee) Disconnected(c *ps.Conn) {
223
+ n.not.Disconnected(n.net, inet.Conn((*Conn)(c)))
224
+}
225
+
226
+func (n *ps2netNotifee) OpenedStream(s *ps.Stream) {
227
+ n.not.OpenedStream(n.net, inet.Stream((*Stream)(s)))
228
+}
229
+
230
+func (n *ps2netNotifee) ClosedStream(s *ps.Stream) {
231
+ n.not.ClosedStream(n.net, inet.Stream((*Stream)(s)))
232
+}
p2p/net/swarm/swarm_net.go
+10
@@ -154,3 +154,13 @@ func (n *Network) SetConnHandler(h inet.ConnHandler) {
154
func (n *Network) String() string {
155
return fmt.Sprintf("<Network %s>", n.LocalPeer())
156
}
157
+
158
+// Notify signs up Notifiee to receive signals when events happen
159
+func (n *Network) Notify(f inet.Notifiee) {
160
+ n.Swarm().Notify(f)
161
+}
162
+
163
+// StopNotify unregisters Notifiee fromr receiving signals
164
+func (n *Network) StopNotify(f inet.Notifiee) {
165
+ n.Swarm().StopNotify(f)
166
+}
p2p/net/swarm/swarm_notif_test.go
new
+186
@@ -0,0 +1,186 @@
1
+package swarm
2
+
3
+import (
4
+ "testing"
5
+ "time"
6
+
7
+ inet "github.com/jbenet/go-ipfs/p2p/net"
8
+
9
+ context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/code.google.com/p/go.net/context"
10
+)
11
+
12
+func TestNotifications(t *testing.T) {
13
+ t.Parallel()
14
+
15
+ ctx := context.Background()
16
+ swarms := makeSwarms(ctx, t, 5)
17
+ defer func() {
18
+ for _, s := range swarms {
19
+ s.Close()
20
+ }
21
+ }()
22
+
23
+ timeout := 5 * time.Second
24
+
25
+ // signup notifs
26
+ notifiees := make([]*netNotifiee, len(swarms))
27
+ for i, swarm := range swarms {
28
+ n := newNetNotifiee()
29
+ swarm.Notify(n)
30
+ notifiees[i] = n
31
+ }
32
+
33
+ connectSwarms(t, ctx, swarms)
34
+
35
+ <-time.After(time.Millisecond)
36
+ // should've gotten 5 by now.
37
+
38
+ // test everyone got the correct connection opened calls
39
+ for i, s := range swarms {
40
+ n := notifiees[i]
41
+ for _, s2 := range swarms {
42
+ if s == s2 {
43
+ continue
44
+ }
45
+
46
+ cos := s.ConnectionsToPeer(s2.LocalPeer())
47
+ func() {
48
+ for i := 0; i < len(cos); i++ {
49
+ var c inet.Conn
50
+ select {
51
+ case c = <-n.connected:
52
+ case <-time.After(timeout):
53
+ t.Fatal("timeout")
54
+ }
55
+ for _, c2 := range cos {
56
+ if c == c2 {
57
+ t.Log("got notif for conn", c)
58
+ return
59
+ }
60
+ }
61
+ t.Error("connection not found", c)
62
+ }
63
+ }()
64
+ }
65
+ }
66
+
67
+ complement := func(c inet.Conn) (*Swarm, *netNotifiee, *Conn) {
68
+ for i, s := range swarms {
69
+ for _, c2 := range s.Connections() {
70
+ if c.LocalMultiaddr().Equal(c2.RemoteMultiaddr()) &&
71
+ c2.LocalMultiaddr().Equal(c.RemoteMultiaddr()) {
72
+ return s, notifiees[i], c2
73
+ }
74
+ }
75
+ }
76
+ t.Fatal("complementary conn not found", c)
77
+ return nil, nil, nil
78
+ }
79
+
80
+ testOCStream := func(n *netNotifiee, s inet.Stream) {
81
+ var s2 inet.Stream
82
+ select {
83
+ case s2 = <-n.openedStream:
84
+ t.Log("got notif for opened stream")
85
+ case <-time.After(timeout):
86
+ t.Fatal("timeout")
87
+ }
88
+ if s != s2 {
89
+ t.Fatal("got incorrect stream", s.Conn(), s2.Conn())
90
+ }
91
+
92
+ select {
93
+ case s2 = <-n.closedStream:
94
+ t.Log("got notif for closed stream")
95
+ case <-time.After(timeout):
96
+ t.Fatal("timeout")
97
+ }
98
+ if s != s2 {
99
+ t.Fatal("got incorrect stream", s.Conn(), s2.Conn())
100
+ }
101
+ }
102
+
103
+ streams := make(chan inet.Stream)
104
+ for _, s := range swarms {
105
+ s.SetStreamHandler(func(s inet.Stream) {
106
+ streams <- s
107
+ s.Close()
108
+ })
109
+ }
110
+
111
+ // open a streams in each conn
112
+ for i, s := range swarms {
113
+ for _, c := range s.Connections() {
114
+ _, n2, _ := complement(c)
115
+
116
+ st1, err := c.NewStream()
117
+ if err != nil {
118
+ t.Error(err)
119
+ } else {
120
+ st1.Write([]byte("hello"))
121
+ st1.Close()
122
+ testOCStream(notifiees[i], st1)
123
+ st2 := <-streams
124
+ testOCStream(n2, st2)
125
+ }
126
+ }
127
+ }
128
+
129
+ // close conns
130
+ for i, s := range swarms {
131
+ n := notifiees[i]
132
+ for _, c := range s.Connections() {
133
+ _, n2, c2 := complement(c)
134
+ c.Close()
135
+ c2.Close()
136
+
137
+ var c3, c4 inet.Conn
138
+ select {
139
+ case c3 = <-n.disconnected:
140
+ case <-time.After(timeout):
141
+ t.Fatal("timeout")
142
+ }
143
+ if c != c3 {
144
+ t.Fatal("got incorrect conn", c, c3)
145
+ }
146
+
147
+ select {
148
+ case c4 = <-n2.disconnected:
149
+ case <-time.After(timeout):
150
+ t.Fatal("timeout")
151
+ }
152
+ if c2 != c4 {
153
+ t.Fatal("got incorrect conn", c, c2)
154
+ }
155
+ }
156
+ }
157
+}
158
+
159
+type netNotifiee struct {
160
+ connected chan inet.Conn
161
+ disconnected chan inet.Conn
162
+ openedStream chan inet.Stream
163
+ closedStream chan inet.Stream
164
+}
165
+
166
+func newNetNotifiee() *netNotifiee {
167
+ return &netNotifiee{
168
+ connected: make(chan inet.Conn),
169
+ disconnected: make(chan inet.Conn),
170
+ openedStream: make(chan inet.Stream),
171
+ closedStream: make(chan inet.Stream),
172
+ }
173
+}
174
+
175
+func (nn *netNotifiee) Connected(n inet.Network, v inet.Conn) {
176
+ nn.connected <- v
177
+}
178
+func (nn *netNotifiee) Disconnected(n inet.Network, v inet.Conn) {
179
+ nn.disconnected <- v
180
+}
181
+func (nn *netNotifiee) OpenedStream(n inet.Network, v inet.Stream) {
182
+ nn.openedStream <- v
183
+}
184
+func (nn *netNotifiee) ClosedStream(n inet.Network, v inet.Stream) {
185
+ nn.closedStream <- v
186
+}