Merge pull request #26 from catmullet/fix_frozen_parent_workers

Cleanup

go.mod

@@ -5,4 +5,5 @@ go 1.15
 require (
 	github.com/panjf2000/ants v1.2.0
 	github.com/stretchr/testify v1.7.0
+	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c
 )

workers.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"golang.org/x/sync/semaphore"
 	"io"
 	"os"
 	"os/signal"
@@ -14,7 +15,6 @@ import (
 
 const (
 	internalBufferFlushLimit = 512
-	minNumberOfWorkersLimit  = 1
 	signalChannelBufferSize  = 1
 )
 
@@ -29,9 +29,10 @@ type Worker struct {
 	Ctx             context.Context
 	workerFunction  WorkerObject
 	err             error
-	numberOfWorkers int
+	numberOfWorkers int64
 	inChan          chan interface{}
 	outChan         chan interface{}
+	sema            *semaphore.Weighted
 	sigChan         chan os.Signal
 	timeout         time.Duration
 	cancel          context.CancelFunc
@@ -42,14 +43,14 @@ type Worker struct {
 }
 
 // NewWorker factory method to return new Worker
-func NewWorker(ctx context.Context, workerFunction WorkerObject, numberOfWorkers int) (worker *Worker) {
+func NewWorker(ctx context.Context, workerFunction WorkerObject, numberOfWorkers int64) (worker *Worker) {
 	c, cancel := context.WithCancel(ctx)
-	numberOfWorkers = getCorrectWorkers(numberOfWorkers)
 	return &Worker{
 		numberOfWorkers: numberOfWorkers,
 		Ctx:             c,
 		workerFunction:  workerFunction,
 		inChan:          make(chan interface{}, numberOfWorkers),
+		sema:            semaphore.NewWeighted(numberOfWorkers),
 		sigChan:         make(chan os.Signal, signalChannelBufferSize),
 		timeout:         time.Duration(0),
 		cancel:          cancel,
@@ -83,12 +84,14 @@ func (iw *Worker) Work() *Worker {
 	if iw.timeout > 0 {
 		iw.Ctx, iw.cancel = context.WithTimeout(iw.Ctx, iw.timeout)
 	}
+	iw.wg.Add(1)
 	go func() {
-		iw.wg.Add(1)
 		defer iw.wg.Done()
+		var wg = new(sync.WaitGroup)
 		for {
 			select {
 			case <-iw.IsDone():
+				wg.Wait()
 				if len(iw.inChan) > 0 {
 					continue
 				}
@@ -97,16 +100,22 @@ func (iw *Worker) Work() *Worker {
 				}
 				return
 			case in := <-iw.inChan:
-				iw.wg.Add(1)
+				err := iw.sema.Acquire(iw.Ctx, 1)
+				if err != nil {
+					return
+				}
+				wg.Add(1)
 				go func(in interface{}) {
-					defer iw.wg.Done()
+					defer wg.Done()
+					defer iw.sema.Release(1)
 					if err := iw.workerFunction.Work(iw, in); err != nil {
 						iw.once.Do(func() {
 							iw.err = err
 							if iw.cancel != nil {
 								iw.cancel()
 							}
 						})
+						return
 					}
 				}(in)
 			}
@@ -144,15 +153,6 @@ func (iw *Worker) waitForSignal(signals ...os.Signal) {
 	}()
 }
 
-// getCorrectWorkers don't let oversizing occur on workers
-func getCorrectWorkers(numberOfWorkers int) int {
-	if numberOfWorkers < minNumberOfWorkersLimit {
-		numberOfWorkers = minNumberOfWorkersLimit
-	}
-
-	return numberOfWorkers
-}
-
 // Wait waits for all the workers to finish up
 func (iw *Worker) Wait() (err error) {
 	iw.wg.Wait()

workers_test.go

@@ -49,16 +49,6 @@ var (
 			workerObject: NewTestWorkerObject(workBasicPrint()),
 			numWorkers:   workerCount,
 		},
-		{
-			name:         "work basic less than minimum worker count",
-			workerObject: NewTestWorkerObject(workBasic()),
-			numWorkers:   0,
-		},
-		{
-			name:         "work basic more than maximum worker count",
-			workerObject: NewTestWorkerObject(workBasic()),
-			numWorkers:   20000,
-		},
 		{
 			name:         "work basic with timeout",
 			timeout:      workerTimeout,
@@ -115,7 +105,7 @@ type workerTest struct {
 	timeout      time.Duration
 	deadline     func() time.Time
 	workerObject WorkerObject
-	numWorkers   int
+	numWorkers   int64
 	testSignal   bool
 	errExpected  bool
 }