tunny

README

Tunny is a Golang library for spawning and managing a goroutine pool, allowing you to limit work coming from any number of goroutines with a synchronous API.

A fixed goroutine pool is helpful when you have work coming from an arbitrary number of asynchronous sources, but a limited capacity for parallel processing. For example, when processing jobs from HTTP requests that are CPU heavy you can create a pool with a size that matches your CPU count.

Install

go get github.com/Jeffail/tunny

Or, using dep:

dep ensure -add github.com/Jeffail/tunny

Use

For most cases your heavy work can be expressed in a simple func(), where you can use NewFunc. Let's see how this looks using our HTTP requests to CPU count example:

package main

import (
	"io/ioutil"
	"net/http"
	"runtime"

	"github.com/Jeffail/tunny"
)

func main() {
	numCPUs := runtime.NumCPU()

	pool := tunny.NewFunc(numCPUs, func(payload interface{}) interface{} {
		var result []byte

		// TODO: Something CPU heavy with payload

		return result
	})
	defer pool.Close()

	http.HandleFunc("/work", func(w http.ResponseWriter, r *http.Request) {
		input, err := ioutil.ReadAll(r.Body)
		if err != nil {
			http.Error(w, "Internal error", http.StatusInternalServerError)
		}
		defer r.Body.Close()

		// Funnel this work into our pool. This call is synchronous and will
		// block until the job is completed.
		result := pool.Process(input)

		w.Write(result.([]byte))
	})

	http.ListenAndServe(":8080", nil)
}

Tunny also supports timeouts. You can replace the Process call above to the following:

result, err := pool.ProcessTimed(input, time.Second*5)
if err == tunny.ErrJobTimedOut {
	http.Error(w, "Request timed out", http.StatusRequestTimeout)
}

Changing Pool Size

The size of a Tunny pool can be changed at any time with SetSize(int):

pool.SetSize(10) // 10 goroutines
pool.SetSize(100) // 100 goroutines

This is safe to perform from any goroutine even if others are still processing.

Goroutines With State

Sometimes each goroutine within a Tunny pool will require its own managed state. In this case you should implement tunny.Worker, which includes calls for terminating, interrupting (in case a job times out and is no longer needed) and blocking the next job allocation until a condition is met.

When creating a pool using Worker types you will need to provide a constructor function for spawning your custom implementation:

pool := tunny.New(poolSize, func() Worker {
	// TODO: Any per-goroutine state allocation here.
	return newCustomWorker()
})

This allows Tunny to create and destroy Worker types cleanly when the pool size is changed.

Ordering

Backlogged jobs are not guaranteed to be processed in order. Due to the current implementation of channels and select blocks a stack of backlogged jobs will be processed as a FIFO queue. However, this behaviour is not part of the spec and should not be relied upon.

Documentation

Index

• Variables
• type Pool
  • func New(n int, ctor func() Worker) *Pool
  • func NewCallback(n int) *Pool
  • func NewFunc(n int, f func(interface{}) interface{}) *Pool
  • func (p *Pool) Close()
  • func (p *Pool) GetSize() int
  • func (p *Pool) Process(payload interface{}) interface{}
  • func (p *Pool) ProcessTimed(payload interface{}, timeout time.Duration) (interface{}, error)
  • func (p *Pool) QueueLength() int64
  • func (p *Pool) SetSize(n int)
• type Worker

Variables

var (
	ErrPoolNotRunning = errors.New("the pool is not running")
	ErrJobNotFunc     = errors.New("generic worker not given a func()")
	ErrWorkerClosed   = errors.New("worker was closed")
	ErrJobTimedOut    = errors.New("job request timed out")
)

Errors that are used throughout the Tunny API.

Types

type Pool

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

Pool is a struct that manages a collection of workers, each with their own goroutine. The Pool can initialize, expand, compress and close the workers, as well as processing jobs with the workers synchronously.

func New

func New(n int, ctor func() Worker) *Pool

New creates a new Pool of workers that starts with n workers. You must provide a constructor function that creates new Worker types and when you change the size of the pool the constructor will be called to create each new Worker.

func NewCallback

func NewCallback(n int) *Pool

NewCallback creates a new Pool of workers where workers cast the job payload into a func() and runs it, or returns ErrNotFunc if the cast failed.

func NewFunc

func NewFunc(n int, f func(interface{}) interface{}) *Pool

NewFunc creates a new Pool of workers where each worker will process using the provided func.

func (*Pool) Close

func (p *Pool) Close()

Close will terminate all workers and close the job channel of this Pool.

func (*Pool) GetSize

func (p *Pool) GetSize() int

GetSize returns the current size of the pool.

func (*Pool) Process

func (p *Pool) Process(payload interface{}) interface{}

Process will use the Pool to process a payload and synchronously return the result. Process can be called safely by any goroutines, but will panic if the Pool has been stopped.

func (*Pool) ProcessTimed

func (p *Pool) ProcessTimed(
	payload interface{},
	timeout time.Duration,
) (interface{}, error)

ProcessTimed will use the Pool to process a payload and synchronously return the result. If the timeout occurs before the job has finished the worker will be interrupted and ErrJobTimedOut will be returned. ProcessTimed can be called safely by any goroutines.

func (*Pool) QueueLength

func (p *Pool) QueueLength() int64

QueueLength returns the current count of pending queued jobs.

func (*Pool) SetSize

func (p *Pool) SetSize(n int)

SetSize changes the total number of workers in the Pool. This can be called by any goroutine at any time unless the Pool has been stopped, in which case a panic will occur.

type Worker

type Worker interface {
	// Process will synchronously perform a job and return the result.
	Process(interface{}) interface{}

	// BlockUntilReady is called before each job is processed and must block the
	// calling goroutine until the Worker is ready to process the next job.
	BlockUntilReady()

	// Interrupt is called when a job is cancelled. The worker is responsible
	// for unblocking the Process implementation.
	Interrupt()

	// Terminate is called when a Worker is removed from the processing pool
	// and is responsible for cleaning up any held resources.
	Terminate()
}

Worker is an interface representing a Tunny working agent. It will be used to block a calling goroutine until ready to process a job, process that job synchronously, interrupt its own process call when jobs are abandoned, and clean up its resources when being removed from the pool.

Each of these duties are implemented as a single method and can be averted when not needed by simply implementing an empty func.