Jeffail/tunny

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.

go get github.com/Jeffail/tunny

Or, using dep:

dep ensure -add github.com/Jeffail/tunny

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)
}

You can also use the context from the request (or any other context) to handle timeouts and deadlines. Simply replace the Process call to the following:

result, err := pool.ProcessCtx(r.Context(), input)
if err == context.DeadlineExceeded {
	http.Error(w, "Request timed out", http.StatusRequestTimeout)
}

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.

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.

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.