workerpool

README

Worker Pool

A worker pool allows you to setup a pool of finite worker goroutines to perform jobs. This paradigm is especially useful where you need to limit work in progress. For example, if you need to run concurrent HTTP requests, you should limit the number of simultaneous requests because the OS may have limits, and the receieving server may also have limits.

Example

package main

import (
	"fmt"
	"time"

	"github.com/app-nerds/kit/v6/workerpool"
)

type Job struct {
	Index int
}

func (j *Job) Work(workerID int) {
	fmt.Printf("Worker %d sleeping on index %d...\n", workerID, j.Index)
	time.Sleep(2 * time.Second)
}

func main() {
	var pool workerpool.IPool

	pool = workerpool.NewPool(workerpool.PoolConfig{
		MaxJobQueue:       100,
		MaxWorkers:        10,
		MaxWorkerWaitTime: 3 * time.Second,
	})

	pool.Start()

	for index := 0; index < 30; index++ {
		job := &Job{Index: index}
		pool.QueueJob(job)
	}

	pool.Wait()
	pool.Shutdown()
}

Documentation

Overview

* Copyright (c) 2021. App Nerds LLC. All rights reserved

* Copyright (c) 2021. App Nerds LLC. All rights reserved

* Copyright (c) 2021. App Nerds LLC. All rights reserved

* Copyright (c) 2021. App Nerds LLC. All rights reserved

* Copyright (c) 2021. App Nerds LLC. All rights reserved

Index

• type ErrNoAvaialableWorkers
  • func (e ErrNoAvaialableWorkers) Error() string
  • func (e ErrNoAvaialableWorkers) GetJob() Job
• type IPool
• type IWorker
• type Job
• type JobError
• type Pool
  • func NewPool(config PoolConfig) *Pool
  • func (p *Pool) PutWorkerInTheQueue(worker IWorker)
  • func (p *Pool) QueueJob(job Job)
  • func (p *Pool) Shutdown()
  • func (p *Pool) Start()
  • func (p *Pool) Wait()
• type PoolConfig
• type Worker
  • func (w *Worker) DoJob(job Job)
  • func (w *Worker) GetID() int
  • func (w *Worker) RejoinWorkerPool()

Types

type ErrNoAvaialableWorkers

type ErrNoAvaialableWorkers struct {
	Job Job
}

ErrNoAvaialableWorkers is used to describe a situation where there are no workers available to work a job

func (ErrNoAvaialableWorkers) Error

func (e ErrNoAvaialableWorkers) Error() string

func (ErrNoAvaialableWorkers) GetJob

func (e ErrNoAvaialableWorkers) GetJob() Job

GetJob returns the job associated with this error

type IPool

type IPool interface {
	PutWorkerInTheQueue(worker IWorker)
	Shutdown()
	Start()
	QueueJob(job Job)
	Wait()
}

IPool describes an interface for managing a pool of workers who perform jobs

type IWorker

type IWorker interface {
	DoJob(job Job)
	GetID() int
	RejoinWorkerPool()
}

IWorker interface describes a struct that performs a job

type Job

type Job interface {
	Work(workerID int)
}

A Job is an interface structs must implement which actually executes the work to be done by a worker in the pool. The workerID is the identifier of the worker performing the job

type JobError

type JobError interface {
	Error() string
	GetJob() Job
}

JobError is an interface to describe an error that has job information attached

type Pool

type Pool struct {
	ErrorQueue chan JobError
	// contains filtered or unexported fields
}

A Pool provides methods for managing a pool of workers who perform jobs. A pool can be configured to have a maximum number of available workers, and will wait up to a configurable amount of time for a worker to become available before returning an error

func NewPool

func NewPool(config PoolConfig) *Pool

NewPool creates a new Pool

func (*Pool) PutWorkerInTheQueue

func (p *Pool) PutWorkerInTheQueue(worker IWorker)

PutWorkerInTheQueue puts a worker in the worker queue

func (*Pool) QueueJob

func (p *Pool) QueueJob(job Job)

QueueJob adds a job to the work queue

func (*Pool) Shutdown

func (p *Pool) Shutdown()

Shutdown closes the job queue and waits for current workers to finish

func (*Pool) Start

func (p *Pool) Start()

Start hires workers and waits for jobs

func (*Pool) Wait

func (p *Pool) Wait()

Wait waits for active jobs to finish

type PoolConfig

type PoolConfig struct {
	MaxJobQueue       int
	MaxWorkers        int
	MaxWorkerWaitTime time.Duration
}

PoolConfig provides the ability to configure the worker pool. MaxWorkers specifies the maximum number of workers available. This essentially sets the channel size. MaxWorkerWaitTime is a duration that tells the pool how long it will wait before timing out when a client requests a worker.

type Worker

type Worker struct {
	Pool      IPool
	WaitGroup *sync.WaitGroup
	WorkerID  int
}

A Worker is someone that performs a job. There are a finite number of workers in the pool

func (*Worker) DoJob

func (w *Worker) DoJob(job Job)

DoJob executes the provided job. When the work is complete this worker will put itself back in the queue as available. This method execute a goroutine

func (*Worker) GetID

func (w *Worker) GetID() int

GetID returns this worker's ID

func (*Worker) RejoinWorkerPool

func (w *Worker) RejoinWorkerPool()

RejoinWorkerPool puts this worker back in the worker queue of the pool. A worker will rejoin the queue when she has finished the job