remove head of line blocking from workerpool

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

Author: steeling

pkg/messaging/broker.go

@@ -2,6 +2,7 @@ package messaging
 
 import (
 	"fmt"
+	"sync"
 	"sync/atomic"
 	"time"
 
@@ -92,100 +93,29 @@ func (b *Broker) runWorkqueueProcessor(stopCh <-chan struct{}) {
 
 // runProxyUpdateDispatcher runs the dispatcher responsible for batching
 // proxy update events received in close proximity.
-// It batches proxy update events with the use of 2 timers:
-// 1. Sliding window timer that resets when a proxy update event is received
-// 2. Max window timer that caps the max duration a sliding window can be reset to
-// When either of the above timers expire, the proxy update event is published
-// on the dedicated pub-sub instance.
+// It batches proxy update events with the use of a mutex, and TryLock.
 func (b *Broker) runProxyUpdateDispatcher(stopCh <-chan struct{}) {
-	// batchTimer and maxTimer are updated by the dispatcher routine
-	// when events are processed and timeouts expire. They are initialized
-	// with a large timeout (a decade) so they don't time out till an event
-	// is received.
-	noTimeout := 87600 * time.Hour // A decade
-	slidingTimer := time.NewTimer(noTimeout)
-	maxTimer := time.NewTimer(noTimeout)
-
-	// dispatchPending indicates whether a proxy update event is pending
-	// from being published on the pub-sub. A proxy update event will
-	// be held for 'proxyUpdateSlidingWindow' duration to be able to
-	// coalesce multiple proxy update events within that duration, before
-	// it is dispatched on the pub-sub. The 'proxyUpdateSlidingWindow' duration
-	// is a sliding window, which means each event received within a window
-	// slides the window further ahead in time, up to a max of 'proxyUpdateMaxWindow'.
-	//
-	// This mechanism is necessary to avoid triggering proxy update pub-sub events in
-	// a hot loop, which would otherwise result in CPU spikes on the controller.
-	// We want to coalesce as many proxy update events within the 'proxyUpdateMaxWindow'
-	// duration.
-	dispatchPending := false
+	var coalesce sync.Mutex
 	batchCount := 0 // number of proxy update events batched per dispatch
-
-	var event proxyUpdateEvent
 	for {
 		select {
-		case e, ok := <-b.proxyUpdateCh:
+		case event, ok := <-b.proxyUpdateCh:
 			if !ok {
 				log.Warn().Msgf("Proxy update event chan closed, exiting dispatcher")
 				return
 			}
-			event = e
-
-			if !dispatchPending {
-				// No proxy update events are pending send on the pub-sub.
-				// Reset the dispatch timers. The events will be dispatched
-				// when either of the timers expire.
-				if !slidingTimer.Stop() {
-					<-slidingTimer.C
-				}
-				slidingTimer.Reset(proxyUpdateSlidingWindow)
-				if !maxTimer.Stop() {
-					<-maxTimer.C
-				}
-				maxTimer.Reset(proxyUpdateMaxWindow)
-				dispatchPending = true
-				batchCount++
-				log.Trace().Msgf("Pending dispatch of msg kind %s", event.msg.Kind)
-			} else {
-				// A proxy update event is pending dispatch. Update the sliding window.
-				if !slidingTimer.Stop() {
-					<-slidingTimer.C
-				}
-				slidingTimer.Reset(proxyUpdateSlidingWindow)
-				batchCount++
-				log.Trace().Msgf("Reset sliding window for msg kind %s", event.msg.Kind)
+			batchCount++
+			if coalesce.TryLock() {
+				log.Trace().Msgf("Coalescing proxy update event %s", event.msg.Kind)
+				continue
 			}
+			log.Trace().Msgf("Batching, msg kind %s, batch size %d", event.msg.Kind, batchCount)
 
-		case <-slidingTimer.C:
-			slidingTimer.Reset(noTimeout) // 'slidingTimer' drained in this case statement
-			// Stop and drain 'maxTimer' before Reset()
-			if !maxTimer.Stop() {
-				// Drain channel. Refer to Reset() doc for more info.
-				<-maxTimer.C
-			}
-			maxTimer.Reset(noTimeout)
-			b.proxyUpdatePubSub.Pub(event.msg, event.topic)
 			atomic.AddUint64(&b.totalDispatchedProxyEventCount, 1)
 			metricsstore.DefaultMetricsStore.ProxyBroadcastEventCount.Inc()
-			log.Trace().Msgf("Sliding window expired, msg kind %s, batch size %d", event.msg.Kind, batchCount)
-			dispatchPending = false
-			batchCount = 0
-
-		case <-maxTimer.C:
-			maxTimer.Reset(noTimeout) // 'maxTimer' drained in this case statement
-			// Stop and drain 'slidingTimer' before Reset()
-			if !slidingTimer.Stop() {
-				// Drain channel. Refer to Reset() doc for more info.
-				<-slidingTimer.C
-			}
-			slidingTimer.Reset(noTimeout)
 			b.proxyUpdatePubSub.Pub(event.msg, event.topic)
-			atomic.AddUint64(&b.totalDispatchedProxyEventCount, 1)
-			metricsstore.DefaultMetricsStore.ProxyBroadcastEventCount.Inc()
-			log.Trace().Msgf("Max window expired, msg kind %s, batch size %d", event.msg.Kind, batchCount)
-			dispatchPending = false
 			batchCount = 0
-
+			coalesce.Unlock()
 		case <-stopCh:
 			log.Info().Msg("Proxy update dispatcher received stop signal, exiting")
 			return

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)
-	}
 }