Bump goprocess
License: MIT Signed-off-by: rht <rhtbot@gmail.com>
rht committed
Jul 15, 2015 at 09:40 UTC
3089ebaba7db14d3017508896e17dfde2f490411
6 files changed
+104
-21
Godeps/Godeps.json
+1
-1
@@ -200,7 +200,7 @@
200
},
201
{
202
"ImportPath": "github.com/jbenet/goprocess",
203
- "Rev": "a6650d0b69f2aa0fe7c9685baf0b4d7ecc8766bf"
203
+ "Rev": "788dcf5ca3517f243d276394545ca6b3b4ac32d5"
204
},
205
{
206
"ImportPath": "github.com/kardianos/osext",
Godeps/_workspace/src/github.com/jbenet/goprocess/.travis.yml
+2
-2
@@ -1,10 +1,10 @@
1
+sudo: false
2
+
3
language: go
4
5
go:
6
- 1.3
7
- 1.4
6
- - release
7
- - tip
8
9
script:
10
- go test -race -cpu=5 -v ./...
Godeps/_workspace/src/github.com/jbenet/goprocess/context/context.go
+7
-7
@@ -23,14 +23,8 @@ func WithContext(ctx context.Context) goprocess.Process {
23
// WithContextAndTeardown is a helper function to set teardown at initiation
24
// of WithContext
25
func WithContextAndTeardown(ctx context.Context, tf goprocess.TeardownFunc) goprocess.Process {
26
- if ctx == nil {
27
- panic("nil Context")
28
- }
26
p := goprocess.WithTeardown(tf)
30
- go func() {
31
- <-ctx.Done()
32
- p.Close()
33
- }()
27
+ CloseAfterContext(p, ctx)
28
return p
29
}
30
@@ -61,6 +55,12 @@ func CloseAfterContext(p goprocess.Process, ctx context.Context) {
55
panic("nil Context")
56
}
57
58
+ // context.Background(). if ctx.Done() is nil, it will never be done.
59
+ // we check for this to avoid wasting a goroutine forever.
60
+ if ctx.Done() == nil {
61
+ return
62
+ }
63
+
64
go func() {
65
<-ctx.Done()
66
p.Close()
Godeps/_workspace/src/github.com/jbenet/goprocess/goprocess.go
+1
-1
@@ -114,7 +114,7 @@ type Process interface {
114
//
115
// It is useful to construct simple asynchronous workers, children of p.
116
Go(f ProcessFunc) Process
117
-
117
+
118
// SetTeardown sets the process's teardown to tf.
119
SetTeardown(tf TeardownFunc)
120
Godeps/_workspace/src/github.com/jbenet/goprocess/goprocess_test.go
+67
@@ -1,6 +1,8 @@
1
package goprocess
2
3
import (
4
+ "fmt"
5
+ "runtime"
6
"syscall"
7
"testing"
8
"time"
@@ -515,6 +517,71 @@ func TestWithSignals(t *testing.T) {
517
testClosed(t, p)
518
}
519
520
+func TestMemoryLeak(t *testing.T) {
521
+ iters := 100
522
+ fanout := 10
523
+ P := newProcess(nil)
524
+ var memories []float32
525
+
526
+ measure := func(str string) float32 {
527
+ s := new(runtime.MemStats)
528
+ runtime.ReadMemStats(s)
529
+ //fmt.Printf("%d ", s.HeapObjects)
530
+ //fmt.Printf("%d ", len(P.children))
531
+ //fmt.Printf("%d ", runtime.NumGoroutine())
532
+ //fmt.Printf("%s: %dk\n", str, s.HeapAlloc/1000)
533
+ return float32(s.HeapAlloc) / 1000
534
+ }
535
+
536
+ spawn := func() []Process {
537
+ var ps []Process
538
+ // Spawn processes
539
+ for i := 0; i < fanout; i++ {
540
+ p := WithParent(P)
541
+ ps = append(ps, p)
542
+
543
+ for i := 0; i < fanout; i++ {
544
+ p2 := WithParent(p)
545
+ ps = append(ps, p2)
546
+
547
+ for i := 0; i < fanout; i++ {
548
+ p3 := WithParent(p2)
549
+ ps = append(ps, p3)
550
+ }
551
+ }
552
+ }
553
+ return ps
554
+ }
555
+
556
+ // Read initial memory stats
557
+ measure("initial")
558
+ for i := 0; i < iters; i++ {
559
+ ps := spawn()
560
+ //measure("alloc") // read after alloc
561
+
562
+ // Close all processes
563
+ for _, p := range ps {
564
+ p.Close()
565
+ <-p.Closed()
566
+ }
567
+ ps = nil
568
+
569
+ //measure("dealloc") // read after dealloc, but before gc
570
+
571
+ // wait until all/most goroutines finish
572
+ <-time.After(time.Millisecond)
573
+
574
+ // Run GC
575
+ runtime.GC()
576
+ memories = append(memories, measure("gc")) // read after gc
577
+ }
578
+
579
+ memoryInit := memories[10]
580
+ percentGrowth := 100 * (memories[len(memories)-1] - memoryInit) / memoryInit
581
+ fmt.Printf("Memory growth after %d iteration with each %d processes: %.2f%% after %dk\n", iters, fanout*fanout*fanout, percentGrowth, int(memoryInit))
582
+
583
+}
584
+
585
func testClosing(t *testing.T, p Process) {
586
select {
587
case <-p.Closing():
Godeps/_workspace/src/github.com/jbenet/goprocess/impl-mutex.go
+26
-10
@@ -6,9 +6,9 @@ import (
6
7
// process implements Process
8
type process struct {
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.
9
+ children map[*processLink]struct{} // process to close with us
10
+ waitfors map[*processLink]struct{} // 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.
@@ -33,6 +33,8 @@ func newProcess(tf TeardownFunc) *process {
33
teardown: tf,
34
closed: make(chan struct{}),
35
closing: make(chan struct{}),
36
+ waitfors: make(map[*processLink]struct{}),
37
+ children: make(map[*processLink]struct{}),
38
}
39
}
40
@@ -50,7 +52,7 @@ func (p *process) WaitFor(q Process) {
52
}
53
54
pl := newProcessLink(p, q)
53
- p.waitfors = append(p.waitfors, pl)
55
+ p.waitfors[pl] = struct{}{}
56
p.Unlock()
57
go pl.AddToChild()
58
}
@@ -71,7 +73,7 @@ func (p *process) AddChildNoWait(child Process) {
73
}
74
75
pl := newProcessLink(p, child)
74
- p.children = append(p.children, pl)
76
+ p.children[pl] = struct{}{}
77
p.Unlock()
78
go pl.AddToChild()
79
}
@@ -92,8 +94,12 @@ func (p *process) AddChild(child Process) {
94
}
95
96
pl := newProcessLink(p, child)
95
- p.waitfors = append(p.waitfors, pl)
96
- p.children = append(p.children, pl)
97
+ if p.waitfors != nil { // if p.waitfors hasn't been set nil
98
+ p.waitfors[pl] = struct{}{}
99
+ }
100
+ if p.children != nil { // if p.children hasn't been set nil
101
+ p.children[pl] = struct{}{}
102
+ }
103
p.Unlock()
104
go pl.AddToChild()
105
}
@@ -167,7 +173,7 @@ func (p *process) doClose() {
173
close(p.closing) // signal that we're shutting down (Closing)
174
175
for len(p.children) > 0 || len(p.waitfors) > 0 {
170
- for _, plc := range p.children {
176
+ for plc, _ := range p.children {
177
child := plc.Child()
178
if child != nil { // check because child may already have been removed.
179
go child.Close() // force all children to shut down
@@ -180,7 +186,7 @@ func (p *process) doClose() {
186
// change under our feet.
187
wf := p.waitfors
188
p.waitfors = nil // clear them. release memory.
183
- for _, w := range wf {
189
+ for w, _ := range wf {
190
// Here, we wait UNLOCKED, so that waitfors who are in the middle of
191
// adding a child to us can finish. we will immediately close the child.
192
p.Unlock()
@@ -197,8 +203,18 @@ func (p *process) doClose() {
203
go func(waiters []*processLink) {
204
for _, pl := range waiters {
205
pl.ClearChild()
206
+ pr, ok := pl.Parent().(*process)
207
+ if !ok {
208
+ // parent has already been called to close
209
+ continue
210
+ }
211
+ pr.Lock()
212
+ delete(pr.waitfors, pl)
213
+ delete(pr.children, pl)
214
+ pr.Unlock()
215
}
216
}(p.waiters) // pass in so
217
+ p.waiters = nil // clear them. release memory.
218
}
219
220
// We will only wait on the children we have now.
@@ -223,7 +239,7 @@ func (p *process) CloseAfterChildren() error {
239
nextToWaitFor := func() Process {
240
p.Lock()
241
defer p.Unlock()
226
- for _, e := range p.waitfors {
242
+ for e, _ := range p.waitfors {
243
c := e.Child()
244
if c == nil {
245
continue