workerpool

README

workerpool

This package offers a simple and minimalist solution for managing concurrent job execution via a worker pool. It implements a concurrency-limiting goroutine pool, effectively constraining the concurrency of job execution.

Installation

go get github.com/aoliveti/workerpool

Usage

package main

import (
	"context"
	"fmt"
	"log"

	"github.com/aoliveti/workerpool"
)

func main() {
	pool := workerpool.New(2)

	jobs := make(chan workerpool.Job)
	go func() {
		defer close(jobs)

		for i := 0; i < 10; i++ {
			num := i
			// Define a job function
			job := func(ctx context.Context) error {
				_ = ctx
				fmt.Printf("Job number: %d done!\n", num)
				return nil
			}

			jobs <- job
		}
	}()

	if err := pool.Run(context.Background(), jobs); err != nil {
		log.Fatal("pool:", err)
	}
}

In case of an error or panic occurring within the job, the default pool halts execution and propagates the error. To disable this functionality, you need to pass the WithErrorPropagationDisabled() option to the constructor:

pool := workerpool.New(2, workerpool.WithErrorPropagationDisabled())

Benchmark

goos: darwin
goarch: amd64
pkg: github.com/aoliveti/workerpool
cpu: Intel(R) Core(TM) i5-7360U CPU @ 2.30GHz
BenchmarkWorkerPool/BenchmarkWorkerPool_1w_100j-4                  45645             23979 ns/op               0 B/op          0 allocs/op
BenchmarkWorkerPool/BenchmarkWorkerPool_2w_100j-4                  42039             28131 ns/op               0 B/op          0 allocs/op
BenchmarkWorkerPool/BenchmarkWorkerPool_4w_100j-4                  39540             30274 ns/op               0 B/op          0 allocs/op
BenchmarkWorkerPool/BenchmarkWorkerPool_1w_1000j-4                  4921            246290 ns/op               0 B/op          0 allocs/op
BenchmarkWorkerPool/BenchmarkWorkerPool_2w_1000j-4                  4364            288214 ns/op               0 B/op          0 allocs/op
BenchmarkWorkerPool/BenchmarkWorkerPool_4w_1000j-4                  3973            295878 ns/op               0 B/op          0 allocs/op
BenchmarkWorkerPool/BenchmarkWorkerPool_1w_100000j-4                  49          24046569 ns/op               4 B/op          0 allocs/op
BenchmarkWorkerPool/BenchmarkWorkerPool_2w_100000j-4                  44          28404396 ns/op               6 B/op          0 allocs/op
BenchmarkWorkerPool/BenchmarkWorkerPool_4w_100000j-4                  40          29593806 ns/op              49 B/op          0 allocs/op
PASS
ok      github.com/aoliveti/workerpool  14.701s

License

The library is released under the MIT license. See LICENSE file.

Documentation

Overview

Package workerpool implements a worker pool for concurrent job execution.

Index

• type Job
• type Option
  • func WithErrorPropagationDisabled() Option
• type Pool
  • func New(workers int, opts ...Option) *Pool
  • func (p *Pool) Run(ctx context.Context, jobs <-chan Job) error

Types

type Job

type Job func(ctx context.Context) error

Job represents a function that can be executed as a job in the worker pool.

type Option

type Option func(p *Pool)

Option represents an option function that can be used to configure a Pool.

func WithErrorPropagationDisabled

func WithErrorPropagationDisabled() Option

WithErrorPropagationDisabled returns an option function that disables error propagation in the pool.

type Pool

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

Pool represents a worker pool for concurrent job execution.

func New

func New(workers int, opts ...Option) *Pool

New creates a new worker pool with the specified number of workers and options.

func (*Pool) Run

func (p *Pool) Run(ctx context.Context, jobs <-chan Job) error

Run starts the worker pool with the provided context and job channel.