updated goprocess
Juan Batiz-Benet committed
Jan 10, 2015 at 12:29 UTC
c0cc9511187a50665675d33335a1717b8918b594
5 files changed
+152
-31
Godeps/Godeps.json
+1
-1
@@ -156,7 +156,7 @@
156
},
157
{
158
"ImportPath": "github.com/jbenet/goprocess",
159
- "Rev": "162148a58668ca38b0b8f0459ccc6ca88e32f1f4"
159
+ "Rev": "7f96033e206c3cd4e79d1c61cbdfff57869feaf8"
160
},
161
{
162
"ImportPath": "github.com/kr/binarydist",
Godeps/_workspace/src/github.com/jbenet/goprocess/goprocess.go
+18
-2
@@ -113,7 +113,7 @@ type Process interface {
113
// }
114
//
115
// It is useful to construct simple asynchronous workers, children of p.
116
- Go(f ProcessFunc)
116
+ Go(f ProcessFunc) Process
117
118
// Close ends the process. Close blocks until the process has completely
119
// shut down, and any teardown has run _exactly once_. The returned error
@@ -121,6 +121,10 @@ type Process interface {
121
// If the process has already been closed, Close returns immediately.
122
Close() error
123
124
+ // CloseAfterChildren calls Close _after_ its children have Closed
125
+ // normally (i.e. it _does not_ attempt to close them).
126
+ CloseAfterChildren() error
127
+
128
// Closing is a signal to wait upon. The returned channel is closed
129
// _after_ Close has been called at least once, but teardown may or may
130
// not be done yet. The primary use case of Closing is for children who
@@ -167,7 +171,19 @@ var nilProcessFunc = func(Process) {}
171
//
172
// This is because having the process you
173
func Go(f ProcessFunc) Process {
170
- return GoChild(Background(), f)
174
+ // return GoChild(Background(), f)
175
+
176
+ // we use two processes, one for communication, and
177
+ // one for ensuring we wait on the function (unclosable from the outside).
178
+ p := newProcess(nil)
179
+ waitFor := newProcess(nil)
180
+ p.WaitFor(waitFor) // prevent p from closing
181
+ go func() {
182
+ f(p)
183
+ waitFor.Close() // allow p to close.
184
+ p.Close() // ensure p closes.
185
+ }()
186
+ return p
187
}
188
189
// GoChild is like Go, but it registers the returned Process as a child of parent,
Godeps/_workspace/src/github.com/jbenet/goprocess/goprocess_test.go
+76
-5
@@ -289,7 +289,6 @@ func TestAddChild(t *testing.T) {
289
func TestGoChildrenClose(t *testing.T) {
290
291
var a, b, c, d, e Process
292
-
292
var ready = make(chan struct{})
293
var bWait = make(chan struct{})
294
var cWait = make(chan struct{})
@@ -335,10 +334,85 @@ func TestGoChildrenClose(t *testing.T) {
334
go a.Close()
335
testNone(t, Q)
336
337
+ bWait <- struct{}{} // relase b
338
go b.Close()
339
testNone(t, Q)
340
341
+ cWait <- struct{}{} // relase c
342
+ <-c.Closed()
343
+ <-b.Closed()
344
+ testStrs(t, Q, "b", "c")
345
+ testStrs(t, Q, "b", "c")
346
+
347
+ eWait <- struct{}{} // release e
348
+ <-e.Closed()
349
+ testStrs(t, Q, "e")
350
+
351
+ dWait <- struct{}{} // releasse d
352
+ <-d.Closed()
353
+ <-a.Closed()
354
+ testStrs(t, Q, "a", "d")
355
+ testStrs(t, Q, "a", "d")
356
+}
357
+
358
+func TestCloseAfterChildren(t *testing.T) {
359
+
360
+ var a, b, c, d, e Process
361
+
362
+ var ready = make(chan struct{})
363
+
364
+ a = WithParent(Background())
365
+ a.Go(func(p Process) {
366
+ b = p
367
+ b.Go(func(p Process) {
368
+ c = p
369
+ ready <- struct{}{}
370
+ <-p.Closing() // wait till we're told to close (parents mustnt)
371
+ })
372
+ ready <- struct{}{}
373
+ })
374
+ a.Go(func(p Process) {
375
+ d = p
376
+ d.Go(func(p Process) {
377
+ e = p
378
+ ready <- struct{}{}
379
+ <-p.Closing() // wait till we're told to close (parents mustnt)
380
+ })
381
+ ready <- struct{}{}
382
+ })
383
+
384
+ <-ready
385
+ <-ready
386
+ <-ready
387
+ <-ready
388
+
389
+ Q := make(chan string, 5)
390
+
391
+ go onClosedStr(Q, "a", a)
392
+ go onClosedStr(Q, "b", b)
393
+ go onClosedStr(Q, "c", c)
394
+ go onClosedStr(Q, "d", d)
395
+ go onClosedStr(Q, "e", e)
396
+
397
+ aDone := make(chan struct{})
398
+ bDone := make(chan struct{})
399
+
400
+ testNone(t, Q)
401
+ go func() {
402
+ a.CloseAfterChildren()
403
+ aDone <- struct{}{}
404
+ }()
405
+ testNone(t, Q)
406
+
407
+ go func() {
408
+ b.CloseAfterChildren()
409
+ bDone <- struct{}{}
410
+ }()
411
+ testNone(t, Q)
412
+
413
c.Close()
414
+ <-bDone
415
+ <-b.Closed()
416
testStrs(t, Q, "b", "c")
417
testStrs(t, Q, "b", "c")
418
@@ -346,6 +420,7 @@ func TestGoChildrenClose(t *testing.T) {
420
testStrs(t, Q, "e")
421
422
d.Close()
423
+ <-aDone
424
<-a.Closed()
425
testStrs(t, Q, "a", "d")
426
testStrs(t, Q, "a", "d")
@@ -354,11 +429,7 @@ func TestGoChildrenClose(t *testing.T) {
429
func TestBackground(t *testing.T) {
430
// test it hangs indefinitely:
431
b := Background()
357
-
432
go b.Close()
359
- go func() {
360
- b.Close()
361
- }()
433
434
select {
435
case <-b.Closing():
Godeps/_workspace/src/github.com/jbenet/goprocess/impl-mutex.go
+51
-2
@@ -9,6 +9,7 @@ type process struct {
9
children []Process // process to close with us
10
waitfors []Process // process to only wait for
11
teardown TeardownFunc // called to run the teardown logic.
12
+ waiting chan struct{} // closed when CloseAfterChildrenClosed is called.
13
closing chan struct{} // closed once close starts.
14
closed chan struct{} // closed once close is done.
15
closeErr error // error to return to clients of Close()
@@ -73,13 +74,18 @@ func (p *process) AddChild(child Process) {
74
p.Unlock()
75
}
76
76
-func (p *process) Go(f ProcessFunc) {
77
+func (p *process) Go(f ProcessFunc) Process {
78
child := newProcess(nil)
79
p.AddChild(child)
80
+
81
+ waitFor := newProcess(nil)
82
+ child.WaitFor(waitFor) // prevent child from closing
83
go func() {
84
f(child)
81
- child.Close() // close to tear down.
85
+ waitFor.Close() // allow child to close.
86
+ child.Close() // close to tear down.
87
}()
88
+ return child
89
}
90
91
// Close is the external close function.
@@ -125,3 +131,46 @@ func (p *process) doClose() {
131
p.closeErr = p.teardown() // actually run the close logic (ok safe to teardown)
132
close(p.closed) // signal that we're shut down (Closed)
133
}
134
+
135
+// We will only wait on the children we have now.
136
+// We will not wait on children added subsequently.
137
+// this may change in the future.
138
+func (p *process) CloseAfterChildren() error {
139
+ p.Lock()
140
+ select {
141
+ case <-p.Closed():
142
+ p.Unlock()
143
+ return p.Close() // get error. safe, after p.Closed()
144
+ case <-p.waiting: // already called it.
145
+ p.Unlock()
146
+ <-p.Closed()
147
+ return p.Close() // get error. safe, after p.Closed()
148
+ default:
149
+ }
150
+ p.Unlock()
151
+
152
+ // here only from one goroutine.
153
+
154
+ nextToWaitFor := func() Process {
155
+ p.Lock()
156
+ defer p.Unlock()
157
+ for _, e := range p.waitfors {
158
+ select {
159
+ case <-e.Closed():
160
+ default:
161
+ return e
162
+ }
163
+ }
164
+ return nil
165
+ }
166
+
167
+ // wait for all processes we're waiting for are closed.
168
+ // the semantics here are simple: we will _only_ close
169
+ // if there are no processes currently waiting for.
170
+ for next := nextToWaitFor(); next != nil; next = nextToWaitFor() {
171
+ <-next.Closed()
172
+ }
173
+
174
+ // YAY! we're done. close
175
+ return p.Close()
176
+}
Godeps/_workspace/src/github.com/jbenet/goprocess/ratelimit/ratelimit.go
+6
-21
@@ -45,29 +45,14 @@ func NewRateLimiter(parent process.Process, limit int) *RateLimiter {
45
func (rl *RateLimiter) LimitedGo(f process.ProcessFunc) {
46
47
<-rl.limiter
48
- rl.Go(func(child process.Process) {
48
+ p := rl.Go(f)
49
50
- // call the function as rl.Go would.
51
- f(child)
52
-
53
- // this close is here because the child may have spawned
54
- // children of its own, and our rate limiter should capture that.
55
- // we have two options:
56
- // * this approach (which is what process.Go itself does), or
57
- // * spawn another goroutine that waits on <-child.Closed()
58
- //
59
- // go func() {
60
- // <-child.Closed()
61
- // rl.limiter <- struct{}{}
62
- // }()
63
- //
64
- // This approach saves a goroutine. It is fine to call child.Close()
65
- // multiple times.
66
- child.Close()
67
-
68
- // after it's done.
50
+ // this <-closed() is here because the child may have spawned
51
+ // children of its own, and our rate limiter should capture that.
52
+ go func() {
53
+ <-p.Closed()
54
rl.limiter <- struct{}{}
70
- })
55
+ }()
56
}
57
58
// LimitChan returns a rate-limiting channel. it is the usual, simple,