workgroup

README

workgroup - wraps sync.WaitGroup

Package workgroup is a Go client library providing a wrapper for implementing a sync.WaitGroup. It attempts to eliminate all the boilerplate often required when writing sync.WaitGroup workers.

Status

Installation

go get -v github.com/Spatially/go-workgroup

Documentation

See GoDoc or Go Walker for automatically generated documentation.

Usage

1. Define a Worker

This is a Worker function. The workgroup will start however many of these you specify. In this example, it will start one for each CPU (see 3 below).

workhorse := func(worker int, work workgroup.Work) {
	log.Printf("%d %+v Done : +%v", worker, work, time.Now())
}

2. Define a Work-Generator function

This is a Work-Generator. It simply feeds work to each workhorse goroutine as each is ready for Work. Although the Workers are goroutines, a workgroup uses sync.WaitGroup internally so this goroutine will block the out channel until a Worker reads from the channel. The completion of this signals the workgroup's cleanup process (all the Workers will complete their work).

workUnits := workgroup.Generator(func(out chan<- workgroup.Work) {
	for i := 0; i < 100; i++ {
		out <- i
	}
})

3. Initiate the WorkGroup with 8 workers

This configures and initates the workgroup.

• FanOut specifies how many Workers to start. 8 in this case.
• Drain specifies the Generator function.
• With provides the Worker function.

workgroup.FanOut(8).Drain(workUnits).With(workhorse).Go()

Documentation

Index

• type Generator
  • func (g Generator) Pump(wc chan<- Work)
• type Initializer
• type Monitor
• type Options
• type Source
• type Work
• type WorkGroup
  • func FanIn(parameters ...Options) WorkGroup
  • func FanOut(workers int, name ...string) WorkGroup
  • func New(parameters ...Options) WorkGroup
• type Worker

Types

type Generator

type Generator func(chan<- Work)

This is the Work generator.

func (Generator) Pump

func (g Generator) Pump(wc chan<- Work)

A Generator function is a Source.

type Initializer

type Initializer func(int)

Called once per Worker

type Monitor

type Monitor interface {
	Work() Work
	Elapsed() time.Duration
}

type Options

type Options struct {
	Workers int
	Name    string
	Timeout time.Duration
	Timing  bool
}

For specifying parameters to New.

type Source

type Source interface {
	Pump(chan<- Work)
}

Sources of work implement Pump to send work on its provided channel.

type Work

type Work interface{}

Anything that can Perform() can be a Work.

type WorkGroup

type WorkGroup interface {
	// A WorkGroup is a Source
	Source
	// A WorkGroup can Drain a Source
	Drain(Source) WorkGroup
	// A WorkGroup assigns a worker
	Configure(Initializer) WorkGroup
	// A WorkGroup assigns a worker
	With(Worker) WorkGroup
	// A WorkGroup can be started
	Go()
	// contains filtered or unexported methods
}

The things a WorkGroup can do.

func FanIn

func FanIn(parameters ...Options) WorkGroup

Creates a WorkGroup with 1 running worker.

func FanOut

func FanOut(workers int, name ...string) WorkGroup

Creates a WorkGroup with more than 1 running worker.

func New

func New(parameters ...Options) WorkGroup

If workers is <= 0, the number of workers will be set to the number of numCPU. When workers > 1, the workgroup is a fan-out. Maybe just use FanOut(). When workers == 1, the workgroup is a fan-in. Maybe just use FanIn().

type Worker

type Worker func(int, Work)