gowp

README

GOWP

Package gowp (Go Worker-Pool) provides concurrency limiting, error propagation, and Context cancellation for a group of workers/goroutines.

Features

• Context cancellation, won't process new tasks if parent context gets cancelled.
• Error propagation, will return the first error encountered.
• Exit on error, if specified, it won't process further tasks if an error is encountered.
• Concurrency limiting

Installation

go get -u github.com/akshaybharambe14/gowp

Why?

Goroutines are cheap in terms of memory, but not free. If you want to achieve extreme performance, you need to limit the number of goroutines you use. Also, in real world applications, you need to take care of the failures. This package does that for you.

This package works best with short bursts of tasks. We have other packages that are meant for running huge number of tasks.

Why yet another worker pool implementation

I wanted to build a perfect worker pool implementation with above specified features. We have other implementations, but I think they do a lot of work in background if used to execute fixed set of tasks. Gowp outperforms some of them (see benchmarks).

Benchmarks

For following benchmarks, a single operation means running 10 tasks over 4 workers. Packages compared (results are in same sequence as packages are listed):

• github.com/akshaybharambe14/gowp
• github.com/gammazero/workerpool
• github.com/alitto/pond

$ go test -bench=. -benchmem github.com/akshaybharambe14/gowp/benchmarks
goos: windows
goarch: amd64
pkg: github.com/akshaybharambe14/gowp/benchmarks
cpu: Intel(R) Core(TM) i5-4200M CPU @ 2.50GHz
Benchmark_simple_gowp-4                    79420             13663 ns/op             704 B/op         13 allocs/op
Benchmark_simple_workerpool-4              40402             29716 ns/op             952 B/op         17 allocs/op
Benchmark_simple_pond-4                    57692             20689 ns/op             968 B/op         20 allocs/op
PASS
ok      github.com/akshaybharambe14/gowp/benchmarks     4.323s

Examples

see package examples

Contact

Akshay Bharambe

---

If this is not something you are looking for, you can check other similar packages on go.libhunt.com.

Do let me know if you have any feedback. Leave a ⭐ if you like this work.

Documentation

Overview

Package gowp provides a pool of workers with limited concurrency.

Example:

// create a pool for 10 tasks with 4 workers that is exists on error
wp := gowp.New(context.TODO(), 10, 4, true)

// add tasks to the pool
for i := 0; i < 10; i++ {
	wp.Submit(func() error {
		// do something
		if i%2 == 0 {
			// return error if something happens
			return errors.New("error")
		}

		return nil
	}()
}

// signal that we are done producing work
wp.Close()

// wait for all the workers to finish their work, check the first error, if any
if err := wp.Wait(); err != nil {
	// handle error
}

Index

• Constants
• type Error
  • func (e Error) Error() string
• type Option
  • func WithContext(ctx context.Context) Option
  • func WithExitOnError(exitOnErr bool) Option
  • func WithNumWorkers(numWorkers int) Option
• type Pool
  • func New(numTasks int, opts ...Option) (*Pool, error)
  • func (p *Pool) IsClosed() bool
  • func (p *Pool) Submit(t Task) error
  • func (p *Pool) Wait() error
• type Task

Examples

• Pool (ExitOnError)

Constants

const (
	ErrPoolClosed  = Error("pool is closed")
	ErrNoBuffer    = Error("insufficient buffer")
	ErrInvalidSend = Error("work sent on closed pool")
	ErrNilTask     = Error("task is nil")
)

processing errors

const (
	ErrInvalidBuffer    = Error("buffer value should be greater than zero")
	ErrInvalidWorkerCnt = Error("worker count should be greater than zero")
	ErrNilContext       = Error("context is nil")
)

validation errors

Types

type Error

type Error string

func (Error) Error

func (e Error) Error() string

type Option

type Option func(o *config)

func WithContext

func WithContext(ctx context.Context) Option

WithContext returns an Option that sets the context for the pool. If the context is canceled the pool will be closed.

func WithExitOnError

func WithExitOnError(exitOnErr bool) Option

WithExitOnError returns an Option that sets the exitOnErr for the pool. If the exitOnErr is true, the pool will be closed when the first error is received.

func WithNumWorkers

func WithNumWorkers(numWorkers int) Option

WithNumWorkers returns an Option that sets the number of workers for the pool. If the number of workers is less than or equal to zero, ErrInvalidWorkerCnt will be returned on Poll initialization.

type Pool

type Pool struct {
	// contains filtered or unexported fields
}

Pool represents a pool of workers that limits concurency as per the provided worker count.

Zero value is not usable. Use New() to create a new Pool.

Example (ExitOnError)

ExitOnError demonstrates the use of a Worker Pool to implement a use-case where a task can fail and the pool should exit and return the error.

package main

import (
	"errors"
	"fmt"
	"time"

	"github.com/akshaybharambe14/gowp"
)

func main() {
	const (
		numJobs    = 50
		numWorkers = 2
		closeOnErr = true
	)

	wp, _ := gowp.New(
		numJobs,
		gowp.WithExitOnError(true),
		gowp.WithNumWorkers(numWorkers),
	)

	for i := 0; i < numJobs; i++ {
		i := i
		_ = wp.Submit(func() error {
			fmt.Println("processing ", i)
			time.Sleep(time.Millisecond)
			if i == 2 {
				return errors.New("can't continue")
			}
			return nil
		})
	}

	if err := wp.Wait(); err != nil {
		fmt.Println("process jobs: ", err)
	}
}

func New

func New(numTasks int, opts ...Option) (*Pool, error)

func (*Pool) IsClosed

func (p *Pool) IsClosed() bool

func (*Pool) Submit

func (p *Pool) Submit(t Task) error

func (*Pool) Wait

func (p *Pool) Wait() error

type Task

type Task func() error

Task is a unit of work that is submitted to the pool by consumers.