remove head of line blocking from workerpool

Signed-off-by: Sean Teeling <[email protected]>

Author: steeling

pkg/workerpool/workerpool.go

@@ -9,30 +9,17 @@ import (
 	"github.com/openservicemesh/osm/pkg/logger"
 )
 
-const (
-	// Size of the job queue per worker
-	maxJobPerWorker = 4096
-)
-
 var (
 	log = logger.New("workerpool")
 )
 
-// worker context for a worker routine
-type worker struct {
-	id            int
-	jobs          chan Job        // Job queue
-	stop          chan struct{}   // Stop channel
-	wg            *sync.WaitGroup // Pointer to WorkerPool wg
-	jobsProcessed uint64          // Jobs processed by this worker
-}
-
 // WorkerPool object representation
 type WorkerPool struct {
 	wg            sync.WaitGroup // Sync group, to stop workers if needed
-	workerContext []*worker      // Worker contexts
 	nWorkers      uint64         // Number of workers. Uint64 for easier mod hash later
-	rRobinCounter uint64         // Used only by the round robin api. Modified atomically on API.
+	jobsProcessed uint64         // Jobs processed by this worker
+	jobs          chan Job
+	stop          chan struct{} // Stop channel
 }
 
 // Job is a runnable interface to queue jobs on a WorkerPool
@@ -62,72 +49,56 @@ func NewWorkerPool(nWorkers int) *WorkerPool {
 
 	log.Info().Msgf("New worker pool setting up %d workers", nWorkers)
 
-	var workPool WorkerPool
+	workPool := &WorkerPool{
+		nWorkers: uint64(nWorkers),
+		jobs:     make(chan Job, nWorkers),
+		stop:     make(chan struct{}),
+	}
 	for i := 0; i < nWorkers; i++ {
-		workPool.workerContext = append(workPool.workerContext,
-			&worker{
-				id:            i,
-				jobs:          make(chan Job, maxJobPerWorker),
-				stop:          make(chan struct{}, 1),
-				wg:            &workPool.wg,
-				jobsProcessed: 0,
-			},
-		)
-		workPool.wg.Add(1)
-		workPool.nWorkers++
-
-		go (workPool.workerContext[i]).work()
+		i := i
+		go workPool.work(i)
 	}
 
-	return &workPool
+	return workPool
 }
 
 // AddJob posts the job on a worker queue
 // Uses Hash underneath to choose worker to post the job to
 func (wp *WorkerPool) AddJob(job Job) <-chan struct{} {
-	wp.workerContext[job.Hash()%wp.nWorkers].jobs <- job
+	wp.jobs <- job
 	return job.GetDoneCh()
 }
 
-// AddJobRoundRobin adds a job in round robin to the queues
-// Concurrent calls to AddJobRoundRobin are thread safe and fair
-// between each other
-func (wp *WorkerPool) AddJobRoundRobin(jobs Job) {
-	added := atomic.AddUint64(&wp.rRobinCounter, 1)
-	wp.workerContext[added%wp.nWorkers].jobs <- jobs
-}
-
 // GetWorkerNumber get number of queues/workers
 func (wp *WorkerPool) GetWorkerNumber() int {
 	return int(wp.nWorkers)
 }
 
 // Stop stops the workerpool
 func (wp *WorkerPool) Stop() {
-	for _, worker := range wp.workerContext {
-		worker.stop <- struct{}{}
-	}
+	close(wp.stop)
 	wp.wg.Wait()
 }
 
-func (workContext *worker) work() {
-	defer workContext.wg.Done()
+func (wp *WorkerPool) work(id int) {
+	wp.wg.Add(1)
+	defer wp.wg.Done()
+
+	log.Info().Msgf("Worker %d running", id)
 
-	log.Info().Msgf("Worker %d running", workContext.id)
 	for {
 		select {
-		case j := <-workContext.jobs:
+		case j := <-wp.jobs:
 			t := time.Now()
-			log.Debug().Msgf("work[%d]: Starting %v", workContext.id, j.JobName())
+			log.Debug().Msgf("work[%d]: Starting %v", id, j.JobName())
 
 			// Run current job
 			j.Run()
 
-			log.Debug().Msgf("work[%d][%s] : took %v", workContext.id, j.JobName(), time.Since(t))
-			workContext.jobsProcessed++
-
-		case <-workContext.stop:
-			log.Debug().Msgf("work[%d]: Stopped", workContext.id)
+			log.Debug().Msgf("work[%d][%s] : took %v", id, j.JobName(), time.Since(t))
+			atomic.AddUint64(&wp.jobsProcessed, 1)
+		case <-wp.stop:
+			log.Debug().Msgf("work[%d]: Stopped", id)
 			return
 		}
 	}

pkg/workerpool/workerpool_test.go

@@ -44,8 +44,6 @@ func (tj *testJob) Hash() uint64 {
 
 // Uses AddJob, which relies on job hash for queue assignment
 func TestAddJob(t *testing.T) {
-	assert := tassert.New(t)
-
 	njobs := 10 // also worker routines
 	wp := NewWorkerPool(njobs)
 	joblist := make([]testJob, njobs)
@@ -66,41 +64,4 @@ func TestAddJob(t *testing.T) {
 	}
 
 	wp.Stop()
-
-	// Verify all the workers processed 1 job (as expected by the static hash)
-	for i := 0; i < njobs; i++ {
-		assert.Equal(uint64(1), wp.workerContext[i].jobsProcessed)
-	}
-}
-
-// Uses AddJobRoundRobin, which relies on round robin for queue assignment
-func TestAddJobRoundRobin(t *testing.T) {
-	assert := tassert.New(t)
-
-	njobs := 10 // also worker routines
-	wp := NewWorkerPool(njobs)
-	joblist := make([]testJob, njobs)
-
-	// Create and add jobs
-	for i := 0; i < njobs; i++ {
-		joblist[i] = testJob{
-			jobDone: make(chan struct{}, 1),
-			hash:    uint64(i),
-		}
-
-		wp.AddJobRoundRobin(&joblist[i])
-	}
-
-	// Verify all jobs ran through the workers
-	for i := 0; i < njobs; i++ {
-		<-joblist[i].jobDone
-	}
-
-	wp.Stop()
-
-	// Verify all the workers processed 1 job (round-robbined)
-	assert.Equal(uint64(njobs), wp.rRobinCounter)
-	for i := 0; i < njobs; i++ {
-		assert.Equal(uint64(1), wp.workerContext[i].jobsProcessed)
-	}
 }