manager --> worker pool

Author: gas

README.md

@@ -14,22 +14,18 @@ type BalancingStrategy interface {
 API:  
 ```go
 // worker pool struct private api
-(wp *workerPool) getCurrentJobsNumber() int64
-(wp *workerPool) getWorkerStats(workerId int) (Stats, error)
-(wp *workerPool) terminateWorker(workerId int) error
-(wp *workerPool) reloadWorker(workerId int) error
+(wp *WorkerPool) getCurrentJobsNumber() int64
+(wp *WorkerPool) getWorkerStats(workerId int) (Stats, error)
+(wp *WorkerPool) terminateWorker(workerId int) error
+(wp *WorkerPool) reloadWorker(workerId int) error
 
-// key public structs and methods
-// Config holds worker pool params, like 
-// it's size and max jobs per worker (without blocking)
-type Config struct {
-	NWorkers int
-	MaxJobs  int64
-}
-// New creates new instance of workers pool manager
-New(config Config, balancer BalancingStrategy) *Manager
+// worker pool public methods
+// New creates new instance of workers pool
+// params defining it's size and max jobs per worker (without blocking)
+NewWorkerPool(nWorkers, maxJobs uint, balancer BalancingStrategy) *WorkerPool
 // ScheduleJob puts new job in some worker queue
-(m *Manager) ScheduleJob(f JobFunc) (chan Result, error)
+(wp *WorkerPool) ScheduleJob(f JobFunc) (chan Result, error)
+
 // NewRoundRobin creates new instance of the "load balancer"
 NewRoundRobin() *RoundRobin
 // NextWorkerId returns worker id selected by some
@@ -63,13 +59,12 @@ func exampleJob(inp int) wp.JobFunc {
 }
 
 func main() {
-    config := wp.Config{
-        NWorkers: 3,  // NOTE: Number of workers to spawn
-        MaxJobs:  10, // NOTE: Max length of the buffered 
-                      //       channel with submitted jobs (per worker)
-    }
-    balancer := wp.NewRoundRobin()
     pool := wp.New(config, balancer)
+	pool := NewWorkerPool(
+		3,
+		10,
+		NewRoundRobin(),
+	)
 
     // NOTE: if number of jobs per woker will be > MaxJobs, 
     //       ScheduleJob func will block

workerpool.go

@@ -27,14 +27,13 @@ type Job struct {
 
 // Stats holds basic stats for the worker
 type Stats struct {
-	ProcessedJobs          int64
+	ProcessedJobs          uint
 	TotalJobsExecutionTime int64 // NOTE: nanoseconds
 }
 
-// Config holds needed data to start worker pool
-type Config struct {
-	NWorkers int
-	MaxJobs  int64
+// Strategy interface that holds implementation of balancing strategy
+type BalancingStrategy interface {
+	NextWorkerId(workersStats []Stats) int
 }
 
 type workerStats struct {
@@ -43,12 +42,36 @@ type workerStats struct {
 }
 
 // workerPool holds data needed for pool operation
-type workerPool struct {
+type WorkerPool struct {
 	mx            sync.RWMutex
-	config        Config
+	nWorkers      uint
 	workersChan   []chan Job
 	workersStats  []*workerStats
 	terminateChan []chan bool
+	balancer      BalancingStrategy
+}
+
+// NewWorkerPool creates new worker pool
+func NewWorkerPool(nWorkers, maxJobs uint, balancer BalancingStrategy) *WorkerPool {
+	pool := &WorkerPool{
+		nWorkers:      nWorkers,
+		workersChan:   make([]chan Job, nWorkers),
+		workersStats:  make([]*workerStats, nWorkers),
+		terminateChan: make([]chan bool, nWorkers),
+		balancer:      balancer,
+	}
+	var w uint
+	for w = 0; w < nWorkers; w++ {
+		pool.workersChan[w] = make(chan Job, maxJobs)
+		pool.terminateChan[w] = make(chan bool)
+		pool.workersStats[w] = &workerStats{}
+		go worker(
+			pool.workersChan[w],
+			pool.terminateChan[w],
+			pool.workersStats[w],
+		)
+	}
+	return pool
 }
 
 func worker(jobs chan Job, terminate chan bool, s *workerStats) {
@@ -74,7 +97,7 @@ func worker(jobs chan Job, terminate chan bool, s *workerStats) {
 }
 
 // getWorkerStats by specified worker id
-func (wp *workerPool) getWorkerStats(workerId int) (Stats, error) {
+func (wp *WorkerPool) getWorkerStats(workerId int) (Stats, error) {
 	resStats := Stats{}
 	if workerId > len(wp.workersChan) || workerId < 0 {
 		return resStats, workerIndexIsOutOfBoundsErr
@@ -91,7 +114,7 @@ func (wp *workerPool) getWorkerStats(workerId int) (Stats, error) {
 }
 
 // terminateWorker sends termination signal to the specified worker
-func (wp *workerPool) terminateWorker(workerId int) error {
+func (wp *WorkerPool) terminateWorker(workerId int) error {
 	if workerId > len(wp.terminateChan) || workerId < 0 {
 		return workerIndexIsOutOfBoundsErr
 	}
@@ -102,7 +125,7 @@ func (wp *workerPool) terminateWorker(workerId int) error {
 }
 
 // reloadWorker terminates worker by id, and spawns new one
-func (wp *workerPool) reloadWorker(workerId int) error {
+func (wp *WorkerPool) reloadWorker(workerId int) error {
 	if workerId > len(wp.workersChan) || workerId < 0 {
 		return workerIndexIsOutOfBoundsErr
 	}
@@ -121,57 +144,20 @@ func (wp *workerPool) reloadWorker(workerId int) error {
 	return nil
 }
 
-// Strategy interface that holds implementation of balancing strategy
-type BalancingStrategy interface {
-	NextWorkerId(workersStats []Stats) int
-}
-
-// Manager spawns workers and schedule jobs
-type Manager struct {
-	wp       *workerPool
-	balancer BalancingStrategy
-}
-
-// New creates new worker pool
-func New(config Config, balancer BalancingStrategy) *Manager {
-	pool := &workerPool{
-		config:        config,
-		workersChan:   make([]chan Job, config.NWorkers),
-		workersStats:  make([]*workerStats, config.NWorkers),
-		terminateChan: make([]chan bool, config.NWorkers),
-	}
-	for w := 0; w < config.NWorkers; w++ {
-		pool.workersChan[w] = make(chan Job, config.MaxJobs)
-		pool.terminateChan[w] = make(chan bool)
-		pool.workersStats[w] = &workerStats{}
-		go worker(
-			pool.workersChan[w],
-			pool.terminateChan[w],
-			pool.workersStats[w],
-		)
-	}
-	manager := &Manager{
-		wp:       pool,
-		balancer: balancer,
-	}
-	// TODO: add routine that reloads workers
-	return manager
-}
-
 // ScheduleJob puts job in a queue
-func (m *Manager) ScheduleJob(f JobFunc) chan Result {
-	m.wp.mx.RLock()
-	config := m.wp.config
-	workersStats := make([]Stats, config.NWorkers)
-	for i, s := range m.wp.workersStats {
+func (wp *WorkerPool) ScheduleJob(f JobFunc) chan Result {
+	wp.mx.RLock()
+	nWorkers := wp.nWorkers
+	workersStats := make([]Stats, nWorkers)
+	for i, s := range wp.workersStats {
 		s.mx.RLock()
 		workersStats[i] = s.Stats
 		s.mx.RUnlock()
 	}
-	m.wp.mx.RUnlock()
+	wp.mx.RUnlock()
 	ch := make(chan Result, 1)
-	nextWorkerId := m.balancer.NextWorkerId(workersStats)
-	m.wp.workersChan[nextWorkerId] <- Job{f, ch}
+	nextWorkerId := wp.balancer.NextWorkerId(workersStats)
+	wp.workersChan[nextWorkerId] <- Job{f, ch}
 	return ch
 }
 

workerpool_test.go

@@ -24,21 +24,21 @@ func exampleJob(inp int) JobFunc {
 }
 
 func TestPool(t *testing.T) {
-	config := Config{
-		NWorkers: 3,
-		MaxJobs:  10,
-	}
-	balancer := NewRoundRobin()
-	roundRobinPool := New(config, balancer)
-	wp := roundRobinPool.wp
+	nWorkers := 3
+	maxJobs := 10
+	wp := NewWorkerPool(
+		uint(nWorkers),
+		uint(maxJobs),
+		NewRoundRobin(),
+	)
 
 	var actualProcessedJobs float64 = 0.0
 	nJobs := 50
 
 	t.Run("RunJobsSync", func(t *testing.T) {
 		results := make([]chan Result, 0)
 		for i := 0; i < nJobs; i++ {
-			ch := roundRobinPool.ScheduleJob(exampleJob(i))
+			ch := wp.ScheduleJob(exampleJob(i))
 			results = append(results, ch)
 		}
 
@@ -57,7 +57,7 @@ func TestPool(t *testing.T) {
 
 	t.Run("CheckStats", func(t *testing.T) {
 		var averageProcessedJobs float64 = 0.0
-		for w := 0; w < config.NWorkers; w++ {
+		for w := 0; w < nWorkers; w++ {
 			s, err := wp.getWorkerStats(w)
 			if err != nil {
 				t.Fatal(err)
@@ -68,8 +68,8 @@ func TestPool(t *testing.T) {
 			averageProcessedJobs += float64(s.ProcessedJobs)
 			t.Logf("Worker %v stats: %v\n", w, s) // Worker 0 stats: {4 1335} ...
 		}
-		averageProcessedJobs /= float64(config.NWorkers)
-		idealAverageProcessedJobs := actualProcessedJobs / float64(config.NWorkers)
+		averageProcessedJobs /= float64(nWorkers)
+		idealAverageProcessedJobs := actualProcessedJobs / float64(nWorkers)
 		t.Log("Average processed jobs per worker:", averageProcessedJobs, idealAverageProcessedJobs)
 		if math.Abs(averageProcessedJobs-idealAverageProcessedJobs) > 1 {
 			t.Error(jobsDistributionNotEvenErr)
@@ -95,7 +95,7 @@ func TestPool(t *testing.T) {
 	})
 
 	t.Run("RunJobsConcurrent", func(t *testing.T) {
-		for w := 0; w < config.NWorkers; w++ {
+		for w := 0; w < nWorkers; w++ {
 			err := wp.reloadWorker(w)
 			if err != nil {
 				t.Fatal(err)
@@ -108,7 +108,7 @@ func TestPool(t *testing.T) {
 			wg.Add(1)
 			go func(idx int, wg *sync.WaitGroup) {
 				defer wg.Done()
-				ch := roundRobinPool.ScheduleJob(exampleJob(idx))
+				ch := wp.ScheduleJob(exampleJob(idx))
 				jobs <- ch
 			}(i, &wg)
 		}
@@ -122,7 +122,7 @@ func TestPool(t *testing.T) {
 			}
 		}
 
-		for w := 0; w < config.NWorkers; w++ {
+		for w := 0; w < nWorkers; w++ {
 			s, err := wp.getWorkerStats(w)
 			if err != nil {
 				t.Fatal(err)
@@ -135,9 +135,9 @@ func TestPool(t *testing.T) {
 
 	t.Run("CheckStatsConcurrent", func(t *testing.T) {
 		wg := sync.WaitGroup{}
-		stats := make(chan Stats, config.NWorkers)
-		errs := make(chan error, config.NWorkers)
-		for w := 0; w < config.NWorkers; w++ {
+		stats := make(chan Stats, nWorkers)
+		errs := make(chan error, nWorkers)
+		for w := 0; w < nWorkers; w++ {
 			wg.Add(1)
 			go func(w int, wg *sync.WaitGroup) {
 				defer wg.Done()
@@ -167,8 +167,8 @@ func TestPool(t *testing.T) {
 
 	t.Run("ReloadWorkerConcurrent", func(t *testing.T) {
 		wg := sync.WaitGroup{}
-		errs := make(chan error, config.NWorkers)
-		for w := 0; w < config.NWorkers; w++ {
+		errs := make(chan error, nWorkers)
+		for w := 0; w < nWorkers; w++ {
 			wg.Add(1)
 			go func(w int, wg *sync.WaitGroup) {
 				defer wg.Done()
@@ -185,7 +185,7 @@ func TestPool(t *testing.T) {
 			}
 		}
 
-		for w := 0; w < config.NWorkers; w++ {
+		for w := 0; w < nWorkers; w++ {
 			s, err := wp.getWorkerStats(w)
 			if err != nil {
 				t.Fatal(err)