Merge pull request #1 from joeybloggs/v1

Fix recovery race condition.

Author: Dean Karn

README.md

@@ -81,7 +81,7 @@ func main() {
 
 	for result := range p.Results() {
 
-		err, ok := result.(error)
+		err, ok := result.(*pool.ErrRecovery)
 		if ok {
 			// there was some sort of panic that
 			// was recovered, in this scenario
@@ -139,8 +139,8 @@ func main() {
 
 	for result := range p.Results() {
 		switch result.(type) {
-		case error:
-			err := result.(error)
+		case *pool.ErrRecovery:
+			err := result.(*pool.ErrRecovery)
 			// do what you want with error or cancel the pool here p.Cancel()
 			fmt.Println(err)
 		default:
@@ -158,11 +158,10 @@ Benchmarks
 ```go
 $ go test -cpu=4 -bench=. -benchmem=true
 PASS
-BenchmarkSmallRun-4           	       1	3009120497 ns/op	    3360 B/op	      65 allocs/op
-BenchmarkSmallCancel-4        	       1	2003173598 ns/op	    3696 B/op	      81 allocs/op
-BenchmarkLargeCancel-4        	       1	2001222531 ns/op	  106784 B/op	    3028 allocs/op
-BenchmarkOverconsumeLargeRun-4	       1	4004509778 ns/op	   36528 B/op	     661 allocs/op
-ok  	github.com/joeybloggs/pool	14.230s
+BenchmarkSmallRun-4           	       1	3000201819 ns/op	    2272 B/op	      58 allocs/op
+BenchmarkSmallCancel-4        	       1	2002207036 ns/op	    2928 B/op	      79 allocs/op
+BenchmarkLargeCancel-4        	       1	2000774880 ns/op	  106656 B/op	    3026 allocs/op
+BenchmarkOverconsumeLargeRun-4	       1	4003364358 ns/op	   29872 B/op	     557 allocs/op
 ```
 To put these benchmarks in perspective:
 

examples/error/error.go

@@ -33,8 +33,8 @@ func main() {
 
 	for result := range p.Results() {
 		switch result.(type) {
-		case error:
-			err := result.(error)
+		case *pool.ErrRecovery:
+			err := result.(*pool.ErrRecovery)
 			// do what you want with error or cancel the pool here p.Cancel()
 			fmt.Println(err)
 		default:

examples/struct/struct.go

@@ -43,7 +43,7 @@ func main() {
 
 	for result := range p.Results() {
 
-		err, ok := result.(error)
+		err, ok := result.(*pool.ErrRecovery)
 		if ok {
 			// there was some sort of panic that
 			// was recovered, in this scenario

pool.go

@@ -1,9 +1,7 @@
 package pool
 
 import (
-	"errors"
 	"fmt"
-	"log"
 	"reflect"
 	"runtime"
 	"sync"
@@ -13,6 +11,14 @@ const (
 	errRecoveryString = "recovering from panic: %+v\nStack Trace:\n %s"
 )
 
+type ErrRecovery struct {
+	s string
+}
+
+func (e *ErrRecovery) Error() string {
+	return e.s
+}
+
 // Pool Contains all information for the pool instance
 type Pool struct {
 	jobs       chan *Job
@@ -65,10 +71,12 @@ func NewPool(consumers int, jobs int) *Pool {
 				if err := recover(); err != nil {
 					trace := make([]byte, 1<<16)
 					n := runtime.Stack(trace, true)
-					err := errors.New(fmt.Sprintf(errRecoveryString, err, trace[:n]))
-					log.Println(err)
-					p.results <- err
+					rerr := &ErrRecovery{
+						s: fmt.Sprintf(errRecoveryString, err, trace[:n]),
+					}
+					p.results <- rerr
 					p.Cancel()
+					p.wg.Done()
 				}
 			}(p)
 
@@ -78,8 +86,9 @@ func NewPool(consumers int, jobs int) *Pool {
 					if reflect.ValueOf(j).IsNil() {
 						return
 					}
-					defer p.wg.Done()
+
 					j.fn(j)
+					p.wg.Done()
 				case <-p.cancel:
 					return
 				}

pool_test.go

@@ -132,9 +132,10 @@ func TestPanicRecovery(t *testing.T) {
 	var count int
 
 	for result := range pool.Results() {
-		_, ok := result.(error)
+		err, ok := result.(*ErrRecovery)
 		if ok {
 			count++
+			err.Error()
 		}
 	}