updated goprocess to deal with memory leak
Juan Batiz-Benet committed
Apr 12, 2015 at 04:40 UTC
dbef4efd38293eb9268da4a167842dad5cac88d6
5 files changed
+166
-19
Godeps/Godeps.json
+1
-1
@@ -199,7 +199,7 @@
199
},
200
{
201
"ImportPath": "github.com/jbenet/goprocess",
202
- "Rev": "c877297c00ffe09f8213ceec3bbb0ab40871f8d4"
202
+ "Rev": "ea63e9540cd19cb39e0e4c4442b9c27664287bb8"
203
},
204
{
205
"ImportPath": "github.com/kardianos/osext",
Godeps/_workspace/src/github.com/jbenet/goprocess/goprocess_test.go
+5
-4
@@ -71,10 +71,6 @@ func TestChildFunc(t *testing.T) {
71
wait2 := make(chan struct{})
72
wait3 := make(chan struct{})
73
wait4 := make(chan struct{})
74
- go func() {
75
- a.Close()
76
- wait4 <- struct{}{}
77
- }()
74
75
a.Go(func(process Process) {
76
wait1 <- struct{}{}
@@ -82,6 +78,11 @@ func TestChildFunc(t *testing.T) {
78
wait3 <- struct{}{}
79
})
80
81
+ go func() {
82
+ a.Close()
83
+ wait4 <- struct{}{}
84
+ }()
85
+
86
<-wait1
87
select {
88
case <-wait3:
Godeps/_workspace/src/github.com/jbenet/goprocess/impl-mutex.go
+38
-13
@@ -6,8 +6,10 @@ import (
6
7
// process implements Process
8
type process struct {
9
- children []Process // process to close with us
10
- waitfors []Process // process to only wait for
9
+ children []*processLink // process to close with us
10
+ waitfors []*processLink // process to only wait for
11
+ waiters []*processLink // processes that wait for us. for gc.
12
+
13
teardown TeardownFunc // called to run the teardown logic.
14
waiting chan struct{} // closed when CloseAfterChildrenClosed is called.
15
closing chan struct{} // closed once close starts.
@@ -47,8 +49,10 @@ func (p *process) WaitFor(q Process) {
49
default:
50
}
51
50
- p.waitfors = append(p.waitfors, q)
52
+ pl := newProcessLink(p, q)
53
+ p.waitfors = append(p.waitfors, pl)
54
p.Unlock()
55
+ go pl.AddToChild()
56
}
57
58
func (p *process) AddChildNoWait(child Process) {
@@ -66,8 +70,10 @@ func (p *process) AddChildNoWait(child Process) {
70
default:
71
}
72
69
- p.children = append(p.children, child)
73
+ pl := newProcessLink(p, child)
74
+ p.children = append(p.children, pl)
75
p.Unlock()
76
+ go pl.AddToChild()
77
}
78
79
func (p *process) AddChild(child Process) {
@@ -85,9 +91,11 @@ func (p *process) AddChild(child Process) {
91
default:
92
}
93
88
- p.waitfors = append(p.waitfors, child)
89
- p.children = append(p.children, child)
94
+ pl := newProcessLink(p, child)
95
+ p.waitfors = append(p.waitfors, pl)
96
+ p.children = append(p.children, pl)
97
p.Unlock()
98
+ go pl.AddToChild()
99
}
100
101
func (p *process) Go(f ProcessFunc) Process {
@@ -141,26 +149,38 @@ func (p *process) doClose() {
149
close(p.closing) // signal that we're shutting down (Closing)
150
151
for len(p.children) > 0 || len(p.waitfors) > 0 {
144
- for _, c := range p.children {
145
- go c.Close() // force all children to shut down
152
+ for _, plc := range p.children {
153
+ child := plc.Child()
154
+ if child != nil { // check because child may already have been removed.
155
+ go child.Close() // force all children to shut down
156
+ }
157
+ plc.ParentClear()
158
}
147
- p.children = nil // clear them
159
+ p.children = nil // clear them. release memory.
160
161
// we must be careful not to iterate over waitfors directly, as it may
162
// change under our feet.
163
wf := p.waitfors
152
- p.waitfors = nil // clear them
164
+ p.waitfors = nil // clear them. release memory.
165
for _, w := range wf {
166
// Here, we wait UNLOCKED, so that waitfors who are in the middle of
167
// adding a child to us can finish. we will immediately close the child.
168
p.Unlock()
157
- <-w.Closed() // wait till all waitfors are fully closed (before teardown)
169
+ <-w.ChildClosed() // wait till all waitfors are fully closed (before teardown)
170
p.Lock()
171
+ w.ParentClear()
172
}
173
}
174
175
p.closeErr = p.teardown() // actually run the close logic (ok safe to teardown)
176
close(p.closed) // signal that we're shut down (Closed)
177
+
178
+ // go remove all the parents from the process links. optimization.
179
+ go func(waiters []*processLink) {
180
+ for _, pl := range waiters {
181
+ pl.ClearChild()
182
+ }
183
+ }(p.waiters) // pass in so
184
}
185
186
// We will only wait on the children we have now.
@@ -186,10 +206,15 @@ func (p *process) CloseAfterChildren() error {
206
p.Lock()
207
defer p.Unlock()
208
for _, e := range p.waitfors {
209
+ c := e.Child()
210
+ if c == nil {
211
+ continue
212
+ }
213
+
214
select {
190
- case <-e.Closed():
215
+ case <-c.Closed():
216
default:
192
- return e
217
+ return c
218
}
219
}
220
return nil
Godeps/_workspace/src/github.com/jbenet/goprocess/link.go
new
+121
@@ -0,0 +1,121 @@
1
+package goprocess
2
+
3
+import (
4
+ "sync"
5
+)
6
+
7
+// closedCh is an alread-closed channel. used to return
8
+// in cases where we already know a channel is closed.
9
+var closedCh chan struct{}
10
+
11
+func init() {
12
+ closedCh = make(chan struct{})
13
+ close(closedCh)
14
+}
15
+
16
+// a processLink is an internal bookkeeping datastructure.
17
+// it's used to form a relationship between two processes.
18
+// It is mostly for keeping memory usage down (letting
19
+// children close and be garbage-collected).
20
+type processLink struct {
21
+ // guards all fields.
22
+ // DO NOT HOLD while holding process locks.
23
+ // it may be slow, and could deadlock if not careful.
24
+ sync.Mutex
25
+ parent Process
26
+ child Process
27
+}
28
+
29
+func newProcessLink(p, c Process) *processLink {
30
+ return &processLink{
31
+ parent: p,
32
+ child: c,
33
+ }
34
+}
35
+
36
+// Closing returns whether the child is closing
37
+func (pl *processLink) ChildClosing() <-chan struct{} {
38
+ // grab a hold of it, and unlock, as .Closing may block.
39
+ pl.Lock()
40
+ child := pl.child
41
+ pl.Unlock()
42
+
43
+ if child == nil { // already closed? memory optimization.
44
+ return closedCh
45
+ }
46
+ return child.Closing()
47
+}
48
+
49
+func (pl *processLink) ChildClosed() <-chan struct{} {
50
+ // grab a hold of it, and unlock, as .Closed may block.
51
+ pl.Lock()
52
+ child := pl.child
53
+ pl.Unlock()
54
+
55
+ if child == nil { // already closed? memory optimization.
56
+ return closedCh
57
+ }
58
+ return child.Closed()
59
+}
60
+
61
+func (pl *processLink) ChildClose() {
62
+ // grab a hold of it, and unlock, as .Closed may block.
63
+ pl.Lock()
64
+ child := pl.child
65
+ pl.Unlock()
66
+
67
+ if child != nil { // already closed? memory optimization.
68
+ child.Close()
69
+ }
70
+}
71
+
72
+func (pl *processLink) ClearChild() {
73
+ pl.Lock()
74
+ pl.child = nil
75
+ pl.Unlock()
76
+}
77
+
78
+func (pl *processLink) ParentClear() {
79
+ pl.Lock()
80
+ pl.parent = nil
81
+ pl.Unlock()
82
+}
83
+
84
+func (pl *processLink) Child() Process {
85
+ pl.Lock()
86
+ defer pl.Unlock()
87
+ return pl.child
88
+}
89
+
90
+func (pl *processLink) Parent() Process {
91
+ pl.Lock()
92
+ defer pl.Unlock()
93
+ return pl.parent
94
+}
95
+
96
+func (pl *processLink) AddToChild() {
97
+ cp := pl.Child()
98
+
99
+ // is it a *process ? if not... panic.
100
+ c, ok := cp.(*process)
101
+ if !ok {
102
+ panic("goprocess does not yet support other process impls.")
103
+ }
104
+
105
+ // first, is it Closed?
106
+ c.Lock()
107
+ select {
108
+ case <-c.Closed():
109
+ c.Unlock()
110
+
111
+ // already closed. must not add.
112
+ // we must clear it, though. do so without the lock.
113
+ pl.ClearChild()
114
+ return
115
+
116
+ default:
117
+ // put the process link into q's waiters
118
+ c.waiters = append(c.waiters, pl)
119
+ c.Unlock()
120
+ }
121
+}
Godeps/_workspace/src/github.com/jbenet/goprocess/periodic/periodic_test.go
+1
-1
@@ -4,8 +4,8 @@ import (
4
"testing"
5
"time"
6
7
- ci "github.com/jbenet/go-cienv"
7
gp "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/goprocess"
8
+ ci "github.com/jbenet/go-cienv"
9
)
10
11
var (