peerstream update (gc conns)
Juan Batiz-Benet committed
Jan 7, 2015 at 18:54 UTC
c88a4b2ccc3184a5c14beaeed396d5975225e701
6 files changed
+95
-9
Godeps/Godeps.json
+1
-1
@@ -136,7 +136,7 @@
136
},
137
{
138
"ImportPath": "github.com/jbenet/go-peerstream",
139
- "Rev": "eab3056e47ecbd1bb32b8c8512fe46fc856f0387"
139
+ "Rev": "55792f89d00cf62166668ded3288536cbe6a72cc"
140
},
141
{
142
"ImportPath": "github.com/jbenet/go-random",
Godeps/_workspace/src/github.com/jbenet/go-peerstream/conn.go
+3
-4
@@ -111,8 +111,8 @@ func (c *Conn) Close() error {
111
}
112
113
// close underlying connection
114
- c.netConn.Close()
115
- return c.swarm.removeConn(c)
114
+ c.swarm.removeConn(c)
115
+ return c.pstConn.Close()
116
}
117
118
// ConnsWithGroup narrows down a set of connections to those in a given group.
@@ -234,10 +234,9 @@ func (s *Swarm) removeStream(stream *Stream) error {
234
return stream.pstStream.Close()
235
}
236
237
-func (s *Swarm) removeConn(conn *Conn) error {
237
+func (s *Swarm) removeConn(conn *Conn) {
238
// remove from our maps
239
s.connLock.Lock()
240
delete(s.conns, conn)
241
s.connLock.Unlock()
242
- return nil
242
}
Godeps/_workspace/src/github.com/jbenet/go-peerstream/swarm.go
+66
-3
@@ -4,6 +4,7 @@ import (
4
"errors"
5
"net"
6
"sync"
7
+ "time"
8
9
pst "github.com/jbenet/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-peerstream/transport"
10
)
@@ -11,6 +12,9 @@ import (
12
// fd is a (file) descriptor, unix style
13
type fd uint32
14
15
+// GarbageCollectTimeout governs the periodic connection closer.
16
+var GarbageCollectTimeout = 5 * time.Second
17
+
18
type Swarm struct {
19
// the transport we'll use.
20
transport pst.Transport
@@ -33,10 +37,12 @@ type Swarm struct {
37
connHandler ConnHandler // receives Conns intiated remotely
38
streamHandler StreamHandler // receives Streams initiated remotely
39
selectConn SelectConn // default SelectConn function
40
+
41
+ closed chan struct{}
42
}
43
44
func NewSwarm(t pst.Transport) *Swarm {
39
- return &Swarm{
45
+ s := &Swarm{
46
transport: t,
47
streams: make(map[*Stream]struct{}),
48
conns: make(map[*Conn]struct{}),
@@ -44,7 +50,10 @@ func NewSwarm(t pst.Transport) *Swarm {
50
selectConn: SelectRandomConn,
51
streamHandler: NoOpStreamHandler,
52
connHandler: NoOpConnHandler,
53
+ closed: make(chan struct{}),
54
}
55
+ go s.connGarbageCollect()
56
+ return s
57
}
58
59
// SetStreamHandler assigns the stream handler in the swarm.
@@ -122,7 +131,16 @@ func (s *Swarm) Conns() []*Conn {
131
conns = append(conns, c)
132
}
133
s.connLock.RUnlock()
125
- return conns
134
+
135
+ open := make([]*Conn, 0, len(conns))
136
+ for _, c := range conns {
137
+ if c.pstConn.IsClosed() {
138
+ c.Close()
139
+ } else {
140
+ open = append(open, c)
141
+ }
142
+ }
143
+ return open
144
}
145
146
// Listeners returns all the listeners associated with this Swarm.
@@ -225,6 +243,11 @@ func (s *Swarm) NewStreamWithConn(conn *Conn) (*Stream, error) {
243
return nil, errors.New("connection not associated with swarm")
244
}
245
246
+ if conn.pstConn.IsClosed() {
247
+ go conn.Close()
248
+ return nil, errors.New("conn is closed")
249
+ }
250
+
251
s.connLock.RLock()
252
if _, found := s.conns[conn]; !found {
253
s.connLock.RUnlock()
@@ -251,6 +274,46 @@ func (s *Swarm) StreamsWithGroup(g Group) []*Stream {
274
275
// Close shuts down the Swarm, and it's listeners.
276
func (s *Swarm) Close() error {
254
- // shut down TODO
277
+ // automatically close everything new we get.
278
+ s.SetConnHandler(func(c *Conn) { c.Close() })
279
+ s.SetStreamHandler(func(s *Stream) { s.Close() })
280
+
281
+ var wgl sync.WaitGroup
282
+ for _, l := range s.Listeners() {
283
+ wgl.Add(1)
284
+ go func() {
285
+ l.Close()
286
+ wgl.Done()
287
+ }()
288
+ }
289
+ wgl.Wait()
290
+
291
+ var wgc sync.WaitGroup
292
+ for _, c := range s.Conns() {
293
+ wgc.Add(1)
294
+ go func() {
295
+ c.Close()
296
+ wgc.Done()
297
+ }()
298
+ }
299
+ wgc.Wait()
300
return nil
301
}
302
+
303
+// connGarbageCollect periodically sweeps conns to make sure
304
+// they're still alive. if any are closed, remvoes them.
305
+func (s *Swarm) connGarbageCollect() {
306
+ for {
307
+ select {
308
+ case <-s.closed:
309
+ return
310
+ case <-time.After(GarbageCollectTimeout):
311
+ }
312
+
313
+ for _, c := range s.Conns() {
314
+ if c.pstConn.IsClosed() {
315
+ go c.Close()
316
+ }
317
+ }
318
+ }
319
+}
Godeps/_workspace/src/github.com/jbenet/go-peerstream/transport/muxado/muxado.go
+17
-1
@@ -31,6 +31,8 @@ func (s *stream) Close() error {
31
// Conn is a connection to a remote peer.
32
type conn struct {
33
ms muxado.Session
34
+
35
+ closed chan struct{}
36
}
37
38
func (c *conn) muxadoSession() muxado.Session {
@@ -41,6 +43,15 @@ func (c *conn) Close() error {
43
return c.ms.Close()
44
}
45
46
+func (c *conn) IsClosed() bool {
47
+ select {
48
+ case <-c.closed:
49
+ return true
50
+ default:
51
+ return false
52
+ }
53
+}
54
+
55
// OpenStream creates a new stream.
56
func (c *conn) OpenStream() (pst.Stream, error) {
57
s, err := c.ms.Open()
@@ -76,5 +87,10 @@ func (t transport) NewConn(nc net.Conn, isServer bool) (pst.Conn, error) {
87
} else {
88
s = muxado.Client(nc)
89
}
79
- return &conn{ms: s}, nil
90
+ cl := make(chan struct{})
91
+ go func() {
92
+ s.Wait()
93
+ close(cl)
94
+ }()
95
+ return &conn{ms: s, closed: cl}, nil
96
}
Godeps/_workspace/src/github.com/jbenet/go-peerstream/transport/transport.go
+4
@@ -20,6 +20,10 @@ type StreamHandler func(Stream)
20
type Conn interface {
21
io.Closer
22
23
+ // IsClosed returns whether a connection is fully closed, so it can
24
+ // be garbage collected.
25
+ IsClosed() bool
26
+
27
// OpenStream creates a new stream.
28
OpenStream() (Stream, error)
29
Godeps/_workspace/src/github.com/jbenet/go-peerstream/transport/yamux/yamux.go
+4
@@ -39,6 +39,10 @@ func (c *conn) Close() error {
39
return c.yamuxSession().Close()
40
}
41
42
+func (c *conn) IsClosed() bool {
43
+ return c.yamuxSession().IsClosed()
44
+}
45
+
46
// OpenStream creates a new stream.
47
func (c *conn) OpenStream() (pst.Stream, error) {
48
s, err := c.yamuxSession().OpenStream()