workgroup

README

Work Group

This module (github.com/tschaub/workgroup) provides a group for executing tasks with limited concurrency.

Installation

Requires Go >= 1.18

go get github.com/tschaub/workgroup

Use

See the reference documentation for example usage.

Documentation

Index

• Variables
• func NewDefaultQueue[T any]() *defaultQueue[T]
• type Options
• type Queue
• type WorkFunc
• type Worker
  • func New[T any](opts *Options[T]) *Worker[T]
  • func (w *Worker[T]) Add(data T) error
  • func (w *Worker[T]) Context() context.Context
  • func (w *Worker[T]) Wait() error

Examples

• Worker
• Worker (Context)

Variables

var ErrEmptyQueue = errors.New("empty queue")

ErrEmptyEqueue is returned when the queue is empty.

Functions

func NewDefaultQueue

func NewDefaultQueue[T any]() *defaultQueue[T]

NewDefaultQueue returns an in-memory queue. Workgroups created without a queue will get one of these.

Types

type Options

type Options[T any] struct {
	Context context.Context
	Limit   int
	Queue   Queue[T]
	Work    WorkFunc[T]
}

Options for constructing a new worker. Only the Work function is required. No more than the Limit number of tasks will be executed concurrently.

By default, a limit of 1 is used and an in-memory queue is used. You can provide your own queue that implements the Queue interface.

type Queue

type Queue[T any] interface {
	// Add task data to the queue.
	Add(ctx context.Context, data T) error

	// Next task data in the queue.  Returns ErrEmptyQueue if the queue is empty.
	Next(ctx context.Context) (data T, err error)

	// HasNext returns true if the queue has more data.
	HasNext(ctx context.Context) bool
}

Queue holds task data to be executed.

By default, the workgroup uses an in-memory queue. You can provide your own queue that implements this interface.

type WorkFunc

type WorkFunc[T any] func(*Worker[T], T) error

WorkFunc is called by the worker with task data. The work function receives a pointer to the worker so that it can add additional task data.

type Worker

type Worker[T any] struct {
	// contains filtered or unexported fields
}

Worker executes a group of tasks that may grow over time.

Example

Output:

working on abcdef...
working on bcdef...
working on cdef...
working on def...
working on ef...
working on f...

Example (Context)

Output:

working on abcdef...
working on bcdef...
working on cdef...

func New

func New[T any](opts *Options[T]) *Worker[T]

New creates a new worker.

func (*Worker[T]) Add

func (w *Worker[T]) Add(data T) error

Add queues up additional task data.

func (*Worker[T]) Context

func (w *Worker[T]) Context() context.Context

Context returns the group context while tasks are being executed.

func (*Worker[T]) Wait

func (w *Worker[T]) Wait() error

Wait blocks until all tasks have been executed.