goworkerpool

package module

v0.10.2 Latest Latest Go to latest Published: Feb 24, 2022 License: MIT Imports: 6 Imported by: 21

Details

Valid go.mod file
Redistributable license
Tagged version
Stable version
Learn more about best practices

Repository

github.com/enriquebris/goworkerpool

Links

Open Source Insights

README ¶

goworkerpool - Pool of workers

Pool of concurrent workers with the ability of increment / decrement / pause / resume workers on demand.

Features

Enqueue jobs on demand
Multiple ways to wait / block
- Wait until at least a worker is alive
- Wait until n jobs get successfully processed
Add / kill workers on demand
Be notified once a worker is started up / killed
Multiple ways to kill workers:
- On demand
- After currently enqueued jobs get processed
Change the number of workers on demand
Pause / resume all workers

Prerequisites

Golang version >= 1.9

Installation

Execute:

go get github.com/enriquebris/goworkerpool

Documentation

Visit goworkerpool at godoc.org

TODO

Visit the TODO page.

Get started

Simple usage

package main

import (
	"fmt"
	"log"
	"time"

	"github.com/enriquebris/goworkerpool"
)

func main() {
	var (
		maxOperationsInQueue uint = 50
	)
	pool, err := goworkerpool.NewPoolWithOptions(goworkerpool.PoolOptions{
		TotalInitialWorkers:          10,
		MaxWorkers:                   20,
		MaxOperationsInQueue:         maxOperationsInQueue,
		WaitUntilInitialWorkersAreUp: true,
		LogVerbose:                   true,
	})

	if err != nil {
		fmt.Println(err)
		return
	}

	// add the worker handler function
	pool.SetWorkerFunc(func(data interface{}) bool {
		log.Printf("processing %v\n", data)
		// add a 1 second delay (to makes it look as it were processing the job)
		time.Sleep(time.Second)
		log.Printf("processing finished for: %v\n", data)

		// let the pool knows that the worker was able to complete the task
		return true
	})

	// enqueue jobs in a separate goroutine
	go func() {
		for i := 0; i < int(maxOperationsInQueue); i++ {
			pool.AddTask(i)
		}

		// kill all workers after the currently enqueued jobs get processed
		pool.LateKillAllWorkers()
	}()

	// wait while at least one worker is alive
	pool.Wait()
}

Examples

See code examples at examples folder.

How To

Set the worker's handler function

pool.SetWorkerFunc(fn)

This is the way to set the function that will be invoked from a worker each time it pulls a job from the queue. Internally, the worker will pass the job as a parameter to this function.

The function signature (PoolFunc):

func handler(interface{}) bool {
	
}

The handler function should return true to let know that the job was successfully processed, or false in other case.

pool.SetWorkerFunc(func(data interface{}) bool {
		// do the processing

		// let the pool knows that the worker was able to complete the task
		return true
	})

Start up the workers

Start up the workers asynchronously

StartWorkers spins up the workers. The amount of workers to be started up is defined at the Pool instantiation.

pool.StartWorkers()

This is an asynchronous operation, but there is a way to be be notified each time a new worker is started up: through a channel. See SetNewWorkerChan(chan).

Start up the workers synchronously

StartWorkersAndWait spins up the workers and wait until all of them are 100% up. The amount of workers to be started up is defined at the Pool instantiation.

pool.StartWorkersAndWait()

Although this is an synchronous operation, there is a way to be be notified each time a new worker is started up: through a channel. See SetNewWorkerChan(chan). Keep in mind that the channel listener should be running on a different goroutine.

Receive a notification every time a new worker is started up

SetNewWorkerChan(chan) sets a channel to receive notifications every time a new worker is started up.

pool.SetNewWorkerChan(ch chan<- int)

This is optional, no channel is needed to start up new workers. Basically is just a way to give feedback for the worker's start up operation.

Enqueue jobs on demand

Enqueue a simple job

AddTask will enqueue a job into a FIFO queue (a channel).

pool.AddTask(data)

The parameter for the job's data accepts any kind of value (interface{}).

Workers (if alive) will be listening to and picking up jobs from this queue. If no workers are alive nor idle, the job will stay in the queue until any worker will be ready to pick it up and start processing it.

The queue in which this function enqueues the jobs has a limit (it was set up at pool initialization). It means that AddTask will wait for a free queue slot to enqueue a new job in case the queue is at full capacity.

AddTask will return an error if no new tasks could be enqueued at the execution time. No new tasks could be enqueued during a certain amount of time when WaitUntilNSuccesses meets the stop condition.

Enqueue a simple job plus callback function

AddTaskCallback will enqueue a job plus a callback function into a FIFO queue (a channel).

pool.AddTaskCallback(data, callback)

The parameter for the job's data accepts any type (interface{}).

Workers (if alive) will be listening to and picking up jobs from this queue. If no workers are alive nor idle, the job will stay in the queue until any worker will be ready to pick it up and start processing it.

The worker who picks up this job + callback will process the job first and later will invoke the callback function, passing the job's data as a parameter.

The queue in which this function enqueues the jobs has a limit (it was set up at pool initialization). It means that AddTaskCallback will wait for a free queue slot to enqueue a new job in case the queue is at full capacity.

AddTaskCallback will return an error if no new tasks could be enqueued at the execution time. No new tasks could be enqueued during a certain amount of time when WaitUntilNSuccesses meets the stop condition.

Enqueue a callback function

AddCallback will enqueue a callback function into a FIFO queue (a channel).

pool.AddCallback(callback)

Workers (if alive) will be listening to and picking up jobs from this queue. If no workers are alive nor idle, the job will stay in the queue until any worker will be ready to pick it up and start processing it.

The worker who picks up this task will only invoke the callback function, passing nil as a parameter.

The queue in which this function enqueues the jobs has a limit (it was set up at pool initialization). It means that AddCallback will wait for a free queue slot to enqueue a new job in case the queue is at full capacity.

AddCallback will return an error if no new tasks could be enqueued at the execution time. No new tasks could be enqueued during a certain amount of time when WaitUntilNSuccesses meets the stop condition.

Enqueue a job with category and callback

AddComplexTask will enqueue a job into a FIFO queue (a channel).

pool.AddComplexTask(data, category, callback)

This function extends the scope of AddTask adding category and callback.

The job will be grouped based on the given category (for stats purposes).

The callback function (if any) will be invoked just after the job gets processed.

Pass multiple data to be processed by a worker

Let's suppose you have the following struct:

type JobData struct {
	Filename string
	Path     string
	Size     uint64
}

then you can enqueue it as a job (to be processed by a worker):

pool.AddTask(JobData{
	Filename: "file.txt",
	Path:     "/tmp/myfiles/",
	Size:     1500,
})

Keep in mind that the worker's handler function needs to cast the parameter as a JobData (that is on your side).

Wait until at least a worker is alive

Wait blocks until at least one worker is alive.

pool.Wait()

Wait until n jobs get successfully processed

WaitUntilNSuccesses blocks until n jobs get successfully processed.

pool.WaitUntilNSuccesses(n)

Add workers on demand

Add a worker on demand

AddWorker adds a new worker to the pool.

pool.AddWorker()

This is an asynchronous operation, but there is a way to be be notified each time a new worker is started up: through a channel.

Add n workers on demand

AddWorkers adds n new workers to the pool.

pool.AddWorkers(n)

This is an asynchronous operation, but there is a way to be be notified each time a new worker is started up: through a channel.

Multiple ways to kill workers