multiconn in swarm
Juan Batiz-Benet committed
Oct 19, 2014 at 05:05 UTC
63d6ee6daaa9c7a555ab3ed996fe227d9009b197
7 files changed
+100
-59
crypto/key.go
+1
-1
@@ -99,7 +99,7 @@ func GenerateEKeyPair(curveName string) ([]byte, GenSharedKey, error) {
99
}
100
101
pubKey := elliptic.Marshal(curve, x, y)
102
- log.Debug("GenerateEKeyPair %d", len(pubKey))
102
+ // log.Debug("GenerateEKeyPair %d", len(pubKey))
103
104
done := func(theirPub []byte) ([]byte, error) {
105
// Verify and unpack node's public key.
net/conn/multiconn.go
+7
-7
@@ -38,7 +38,7 @@ type MultiConn struct {
38
}
39
40
// NewMultiConn constructs a new connection
41
-func NewMultiConn(ctx context.Context, local, remote *peer.Peer, conns []Conn) (Conn, error) {
41
+func NewMultiConn(ctx context.Context, local, remote *peer.Peer, conns []Conn) (*MultiConn, error) {
42
43
c := &MultiConn{
44
local: local,
@@ -53,13 +53,10 @@ func NewMultiConn(ctx context.Context, local, remote *peer.Peer, conns []Conn) (
53
// must happen before Adds / fanOut
54
c.ContextCloser = NewContextCloser(ctx, c.close)
55
56
- log.Info("adding %d...", len(conns))
56
if conns != nil && len(conns) > 0 {
57
c.Add(conns...)
58
}
59
go c.fanOut()
61
-
62
- log.Info("newMultiConn: %v to %v", local, remote)
60
return c, nil
61
}
62
@@ -72,6 +69,9 @@ func (c *MultiConn) Add(conns ...Conn) {
69
log.Info("MultiConn: adding %s", c2)
70
if c.LocalPeer() != c2.LocalPeer() || c.RemotePeer() != c2.RemotePeer() {
71
log.Error("%s", c2)
72
+ c.Unlock() // ok to unlock (to log). panicing.
73
+ log.Error("%s", c)
74
+ c.Lock() // gotta relock to avoid lock panic from deferring.
75
panic("connection addresses mismatch")
76
}
77
@@ -102,12 +102,12 @@ func (c *MultiConn) Remove(conns ...Conn) {
102
}
103
104
// close all in parallel, but wait for all to be done closing.
105
- CloseConns(conns)
105
+ CloseConns(conns...)
106
}
107
108
// CloseConns closes multiple connections in parallel, and waits for all
109
// to finish closing.
110
-func CloseConns(conns []Conn) {
110
+func CloseConns(conns ...Conn) {
111
var wg sync.WaitGroup
112
for _, child := range conns {
113
@@ -204,7 +204,7 @@ func (c *MultiConn) close() error {
204
c.RUnlock()
205
206
// close underlying connections
207
- CloseConns(conns)
207
+ CloseConns(conns...)
208
return nil
209
}
210
net/conn/multiconn_test.go
+1
-1
@@ -150,7 +150,7 @@ func setupMultiConns(t *testing.T, ctx context.Context) (a, b *MultiConn) {
150
p2l.Close()
151
152
log.Info("did you make multiconns?")
153
- return c1.(*MultiConn), c2.(*MultiConn)
153
+ return c1, c2
154
}
155
156
func TestMulticonnSend(t *testing.T) {
net/swarm/conn.go
+60
-29
@@ -36,7 +36,7 @@ func (s *Swarm) listen() error {
36
// Listen for new connections on the given multiaddr
37
func (s *Swarm) connListen(maddr ma.Multiaddr) error {
38
39
- list, err := conn.Listen(s.ctx, maddr, s.local, s.peers)
39
+ list, err := conn.Listen(s.Context(), maddr, s.local, s.peers)
40
if err != nil {
41
return err
42
}
@@ -50,13 +50,19 @@ func (s *Swarm) connListen(maddr ma.Multiaddr) error {
50
s.listeners = append(s.listeners, list)
51
52
// Accept and handle new connections on this listener until it errors
53
+ // this listener is a child.
54
+ s.Children().Add(1)
55
go func() {
56
+ defer s.Children().Done()
57
+
58
for {
59
select {
56
- case <-s.ctx.Done():
60
+ case <-s.Closing():
61
return
62
63
case conn := <-list.Accept():
64
+ // handler also a child.
65
+ s.Children().Add(1)
66
go s.handleIncomingConn(conn)
67
}
68
}
@@ -67,6 +73,8 @@ func (s *Swarm) connListen(maddr ma.Multiaddr) error {
73
74
// Handle getting ID from this peer, handshake, and adding it into the map
75
func (s *Swarm) handleIncomingConn(nconn conn.Conn) {
76
+ // this handler is a child. added by caller.
77
+ defer s.Children().Done()
78
79
// Setup the new connection
80
_, err := s.connSetup(nconn)
@@ -77,7 +85,7 @@ func (s *Swarm) handleIncomingConn(nconn conn.Conn) {
85
}
86
87
// connSetup adds the passed in connection to its peerMap and starts
80
-// the fanIn routine for that connection
88
+// the fanInSingle routine for that connection
89
func (s *Swarm) connSetup(c conn.Conn) (conn.Conn, error) {
90
if c == nil {
91
return nil, errors.New("Tried to start nil connection.")
@@ -93,28 +101,44 @@ func (s *Swarm) connSetup(c conn.Conn) (conn.Conn, error) {
101
102
// add to conns
103
s.connsLock.Lock()
96
- if c2, ok := s.conns[c.RemotePeer().Key()]; ok {
97
- log.Debug("Conn already open!")
104
+
105
+ mc, ok := s.conns[c.RemotePeer().Key()]
106
+ if !ok {
107
+ // multiconn doesn't exist, make a new one.
108
+ conns := []conn.Conn{c}
109
+ mc, err := conn.NewMultiConn(s.Context(), s.local, c.RemotePeer(), conns)
110
+ if err != nil {
111
+ log.Error("error creating multiconn: %s", err)
112
+ c.Close()
113
+ return nil, err
114
+ }
115
+
116
+ s.conns[c.RemotePeer().Key()] = mc
117
s.connsLock.Unlock()
118
100
- c.Close()
101
- return c2, nil // not error anymore, use existing conn.
102
- // return ErrAlreadyOpen
119
+ log.Debug("added new multiconn: %s", mc)
120
+ } else {
121
+ s.connsLock.Unlock() // unlock before adding new conn
122
+
123
+ mc.Add(c)
124
+ log.Debug("multiconn found: %s", mc)
125
}
104
- s.conns[c.RemotePeer().Key()] = c
105
- log.Debug("Added conn to map!")
106
- s.connsLock.Unlock()
126
+
127
+ log.Debug("multiconn added new conn %s", c)
128
129
// kick off reader goroutine
109
- go s.fanIn(c)
130
+ go s.fanInSingle(c)
131
return c, nil
132
}
133
134
// Handles the unwrapping + sending of messages to the right connection.
135
func (s *Swarm) fanOut() {
136
+ s.Children().Add(1)
137
+ defer s.Children().Done()
138
+
139
for {
140
select {
117
- case <-s.ctx.Done():
141
+ case <-s.Closing():
142
return // told to close.
143
144
case msg, ok := <-s.Outgoing:
@@ -127,9 +151,9 @@ func (s *Swarm) fanOut() {
151
s.connsLock.RUnlock()
152
153
if !found {
130
- e := fmt.Errorf("Sent msg to peer without open conn: %v",
131
- msg.Peer)
154
+ e := fmt.Errorf("Sent msg to peer without open conn: %v", msg.Peer())
155
s.errChan <- e
156
+ log.Error("%s", e)
157
continue
158
}
159
@@ -143,30 +167,37 @@ func (s *Swarm) fanOut() {
167
168
// Handles the receiving + wrapping of messages, per conn.
169
// Consider using reflect.Select with one goroutine instead of n.
146
-func (s *Swarm) fanIn(c conn.Conn) {
170
+func (s *Swarm) fanInSingle(c conn.Conn) {
171
+ s.Children().Add(1)
172
+ c.Children().Add(1) // child of Conn as well.
173
+
174
+ // cleanup all data associated with this child Connection.
175
+ defer func() {
176
+ // remove it from the map.
177
+ s.connsLock.Lock()
178
+ delete(s.conns, c.RemotePeer().Key())
179
+ s.connsLock.Unlock()
180
+
181
+ s.Children().Done()
182
+ c.Children().Done() // child of Conn as well.
183
+ }()
184
+
185
for {
186
select {
149
- case <-s.ctx.Done():
150
- // close Conn.
151
- c.Close()
152
- goto out
187
+ case <-s.Closing(): // Swarm closing
188
+ return
189
+
190
+ case <-c.Closing(): // Conn closing
191
+ return
192
193
case data, ok := <-c.In():
194
if !ok {
156
- e := fmt.Errorf("Error retrieving from conn: %v", c.RemotePeer())
157
- s.errChan <- e
158
- goto out
195
+ return // channel closed.
196
}
160
-
197
// log.Debug("[peer: %s] Received message [from = %s]", s.local, c.Peer)
198
s.Incoming <- msg.New(c.RemotePeer(), data)
199
}
200
}
165
-
166
-out:
167
- s.connsLock.Lock()
168
- delete(s.conns, c.RemotePeer().Key())
169
- s.connsLock.Unlock()
201
}
202
203
// Commenting out because it's platform specific
net/swarm/simul_test.go
-1
@@ -32,7 +32,6 @@ func TestSimultOpen(t *testing.T) {
32
if _, err := s.Dial(cp); err != nil {
33
t.Fatal("error swarm dialing to peer", err)
34
}
35
- log.Info("done?!?")
35
wg.Done()
36
}
37
net/swarm/swarm.go
+26
-19
@@ -56,48 +56,42 @@ type Swarm struct {
56
errChan chan error
57
58
// conns are the open connections the swarm is handling.
59
- conns conn.Map
59
+ // these are MultiConns, which multiplex multiple separate underlying Conns.
60
+ conns conn.MultiConnMap
61
connsLock sync.RWMutex
62
63
// listeners for each network address
64
listeners []conn.Listener
65
65
- // cancel is an internal function used to stop the Swarm's processing.
66
- cancel context.CancelFunc
67
- ctx context.Context
66
+ // ContextCloser
67
+ conn.ContextCloser
68
}
69
70
// NewSwarm constructs a Swarm, with a Chan.
71
func NewSwarm(ctx context.Context, local *peer.Peer, ps peer.Peerstore) (*Swarm, error) {
72
s := &Swarm{
73
Pipe: msg.NewPipe(10),
74
- conns: conn.Map{},
74
+ conns: conn.MultiConnMap{},
75
local: local,
76
peers: ps,
77
errChan: make(chan error, 100),
78
}
79
80
- s.ctx, s.cancel = context.WithCancel(ctx)
80
+ // ContextCloser for proper child management.
81
+ s.ContextCloser = conn.NewContextCloser(ctx, s.close)
82
+
83
go s.fanOut()
84
return s, s.listen()
85
}
86
85
-// Close stops a swarm.
86
-func (s *Swarm) Close() error {
87
- if s.cancel == nil {
88
- return errors.New("Swarm already closed.")
89
- }
90
-
91
- // issue cancel for the context
92
- s.cancel()
93
-
94
- // set cancel to nil to prevent calling Close again, and signal to Listeners
95
- s.cancel = nil
96
-
87
+// close stops a swarm. It's the underlying function called by ContextCloser
88
+func (s *Swarm) close() error {
89
// close listeners
90
for _, list := range s.listeners {
91
list.Close()
92
}
93
+ // close connections
94
+ conn.CloseConns(s.Connections()...)
95
return nil
96
}
97
@@ -132,7 +126,7 @@ func (s *Swarm) Dial(peer *peer.Peer) (conn.Conn, error) {
126
Peerstore: s.peers,
127
}
128
135
- c, err = d.Dial(s.ctx, "tcp", peer)
129
+ c, err = d.Dial(s.Context(), "tcp", peer)
130
if err != nil {
131
return nil, err
132
}
@@ -158,6 +152,19 @@ func (s *Swarm) GetConnection(pid peer.ID) conn.Conn {
152
return c
153
}
154
155
+// Connections returns a slice of all connections.
156
+func (s *Swarm) Connections() []conn.Conn {
157
+ s.connsLock.RLock()
158
+
159
+ conns := make([]conn.Conn, 0, len(s.conns))
160
+ for _, c := range s.conns {
161
+ conns = append(conns, c)
162
+ }
163
+
164
+ s.connsLock.RUnlock()
165
+ return conns
166
+}
167
+
168
// CloseConnection removes a given peer from swarm + closes the connection
169
func (s *Swarm) CloseConnection(p *peer.Peer) error {
170
c := s.GetConnection(p.ID)
net/swarm/swarm_test.go
+5
-1
@@ -85,7 +85,11 @@ func SubtestSwarm(t *testing.T, addrs []string, MsgNum int) {
85
var wg sync.WaitGroup
86
connect := func(s *Swarm, dst *peer.Peer) {
87
// copy for other peer
88
- cp := &peer.Peer{ID: dst.ID}
88
+
89
+ cp, err := s.peers.Get(dst.ID)
90
+ if err != nil {
91
+ cp = &peer.Peer{ID: dst.ID}
92
+ }
93
cp.AddAddress(dst.Addresses[0])
94
95
log.Info("SWARM TEST: %s dialing %s", s.local, dst)