pooler

README

Pooler

A minimalistic yet fast worker-pool for Go, with support for custom callback functions.

Features and philosophy

When we designed pooler we had several goals in mind that we wanted to achieve:

• fast worker-pool implementation that only relies on Go channels and atomic
• optional callback functions to receive event notifications from the pool and its goroutines
• optional custom data that can be manipulated from inside the goroutines and/or the callback function
• graceful shutdown of running goroutines

After reviewing several third-party benchmarks, and running a few more of our own, we realized that we wanted to stay away from lists, maps, and mutexes; the key to achieving top speed appeared to be delegating goroutine synchronization entirely to channels, and using sync/atomic for counters and to simulate atomic boolean values.

Istallation

pooler is packed as Go module (Go >= 1.11), but it also works just fine when used with older Go versions (<= 1.10). To install it, you may use the typical go get command.

go get -u github.com/syncplify/pooler

To learn more, you may also want to read the documentation.

Examples

Please take a look at examples to access a few small example programs that use pooler.

License

This project is licensed under the terms of the Apache 2.0 License. See the LICENSE file for the full license text.

Documentation

Overview

Package pooler implements a worker-pool paradigm, relying on channels for all goroutine interoperation in order to achieve high speed an thread-safety. The only other dependency is the sync/atomic package, but it's kept down to a minimum, because we want pooler's operation to be as non-blocking as possible.

Index

• type CallbackFuncQueue
• type CallbackFuncQueueErr
• type CallbackFuncTask
• type CallbackFuncTaskErr
• type CallbackFuncWrk
• type Config
  • func NewConfig(routines, maxTasks int) *Config
  • func (c *Config) OnTaskCrashed(fn CallbackFuncTaskErr) *Config
  • func (c *Config) OnTaskDone(fn CallbackFuncTask) *Config
  • func (c *Config) OnTaskDoneWithError(fn CallbackFuncTaskErr) *Config
  • func (c *Config) OnTaskQueued(fn CallbackFuncQueue) *Config
  • func (c *Config) OnTaskQueuingError(fn CallbackFuncQueueErr) *Config
  • func (c *Config) OnTaskStarted(fn CallbackFuncTask) *Config
  • func (c *Config) OnWorkerCreated(fn CallbackFuncWrk) *Config
  • func (c *Config) OnWorkerShutdown(fn CallbackFuncWrk) *Config
• type Pool
  • func New(routines int, maxTasks int) (*Pool, error)
  • func NewWithConfig(config *Config) (*Pool, error)
  • func (p *Pool) ActiveWorkers() int
  • func (p *Pool) Enqueue(task Runnable) error
  • func (p *Pool) IsShuttingDown() bool
  • func (p *Pool) QueueLen() int
  • func (p *Pool) Shutdown()
  • func (p *Pool) ShutdownWithTimeout(timeout time.Duration) bool
• type Runnable
• type Task

Types

type CallbackFuncQueue

type CallbackFuncQueue func(task *Task)

CallbackFuncQueue is the prototype of a function that will be called by the pool to notify of successful events pertaining the queue (like a task successfully enqueued)

type CallbackFuncQueueErr

type CallbackFuncQueueErr func(task *Task, err error)

CallbackFuncQueueErr is the prototype of a function that will be called by the pool to notify of errors pertaining the queue (queuing errors)

type CallbackFuncTask

type CallbackFuncTask func(routine int, task *Task)

CallbackFuncTask is the prototype of a function that will be called by the pool to notify when workers start/stop/enqueue tasks

type CallbackFuncTaskErr

type CallbackFuncTaskErr func(routine int, task *Task, err error)

CallbackFuncTaskErr is the prototype of a function that will be called by the pool to notify of errors pertaining tasks (typically runtime errors)

type CallbackFuncWrk

type CallbackFuncWrk func(routine int)

CallbackFuncWrk is the prototype of a function that will be called by the pool to notify when workers are created or shutdown

type Config

type Config struct {
	// Routines is the desired number of "worker" goroutines
	Routines int
	// MaxTasks is the maximum number of tasks that can be in this pool's queue at any given time
	MaxTasks int
	// WorkerCreatedCB is an optional callback func that will be called every time a "worker" goroutine is created
	WorkerCreatedCB CallbackFuncWrk
	// WorkerShutdownCB is an optional callback func that will be called every time a "worker" goroutine is shutdown
	WorkerShutdownCB CallbackFuncWrk
	// TaskQueuedCB is an optional callback func that will be called every time a task is successfully added to the pool's queue
	TaskQueuedCB CallbackFuncQueue
	// TaskQueuingErrorCB is an optional callback func that will be called every time there's a problem adding a task to the pool's queue
	TaskQueuingErrorCB CallbackFuncQueueErr
	// TaskStartedCB is an optional callback func that will be called every time a task is picked up by a "worker" routine and its execution begins
	TaskStartedCB CallbackFuncTask
	// TaskDoneCB is an optional callback func that will be called every time a task is done running without errors
	TaskDoneCB CallbackFuncTask
	// TaskDoneWithErrorCB is an optional callback func that will be called every time a task is done running but has returned an error
	TaskDoneWithErrorCB CallbackFuncTaskErr
	// TaskCrashedCB is an optional callback func that will be called every time a `panic` has occurred within the Run() method while a task was running
	TaskCrashedCB CallbackFuncTaskErr
}

Config is the global pool configuration struct

func NewConfig

func NewConfig(routines, maxTasks int) *Config

NewConfig creates and returns a basic/initial Config struct

func (*Config) OnTaskCrashed

func (c *Config) OnTaskCrashed(fn CallbackFuncTaskErr) *Config

OnTaskCrashed sets the callback function that's called when a "worker" goroutine suddenly crashed (panic) while running a task

func (*Config) OnTaskDone

func (c *Config) OnTaskDone(fn CallbackFuncTask) *Config

OnTaskDone sets the callback function that's called when a "worker" goroutine is done running a task, and no error is returned

func (*Config) OnTaskDoneWithError

func (c *Config) OnTaskDoneWithError(fn CallbackFuncTaskErr) *Config

OnTaskDoneWithError sets the callback function that's called when a "worker" goroutine is done running a task, but an error is returned

func (*Config) OnTaskQueued

func (c *Config) OnTaskQueued(fn CallbackFuncQueue) *Config

OnTaskQueued sets the callback function that's called when a new task is successfully added to the pending queue

func (*Config) OnTaskQueuingError

func (c *Config) OnTaskQueuingError(fn CallbackFuncQueueErr) *Config

OnTaskQueuingError sets the callback function that's called when a "worker" goroutine is done running a task, but an error is returned

func (*Config) OnTaskStarted

func (c *Config) OnTaskStarted(fn CallbackFuncTask) *Config

OnTaskStarted sets the callback function that's called when a "worker" goroutine picks up a task from the queue and starts running it

func (*Config) OnWorkerCreated

func (c *Config) OnWorkerCreated(fn CallbackFuncWrk) *Config

OnWorkerCreated sets the callback function that's called when a new "worker" goroutine is created

func (*Config) OnWorkerShutdown

func (c *Config) OnWorkerShutdown(fn CallbackFuncWrk) *Config

OnWorkerShutdown sets the callback function that's called when a new "worker" goroutine is shutdown

type Pool

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

Pool is a container for a pool of goroutines that will run the queued tasks

func New

func New(routines int, maxTasks int) (*Pool, error)

New creates a new Pool object without any Callback functions routines: maximum number of "worker" goroutines that are allowed to run concurrently maxTasks: maximum number of tasks that can be waiting in line to be executed by the next available goroutine

func NewWithConfig

func NewWithConfig(config *Config) (*Pool, error)

NewWithConfig creates a new Pool object with a user-provided configuration config: a pointer to a Config object (see types.go)

func (*Pool) ActiveWorkers

func (p *Pool) ActiveWorkers() int

ActiveWorkers returns the number of goroutines that are actually busy doing something

func (*Pool) Enqueue

func (p *Pool) Enqueue(task Runnable) error

Enqueue adds a task to the queue of tasks waiting to be executed task: any object that implements the Runnable interface (see types.go)

func (*Pool) IsShuttingDown

func (p *Pool) IsShuttingDown() bool

IsShuttingDown returns false during normal operation and true if the pool is shutting down; all tasks should periodically check it inside of their "Run" func.

func (*Pool) QueueLen

func (p *Pool) QueueLen() int

QueueLen returns the number of tasks currently queued

func (*Pool) Shutdown

func (p *Pool) Shutdown()

Shutdown stops all goroutines running all tasks, and shuts down the entire pool

func (*Pool) ShutdownWithTimeout

func (p *Pool) ShutdownWithTimeout(timeout time.Duration) bool

ShutdownWithTimeout stops all goroutines running all tasks, and shuts down the entire pool It returns true if it times out, and false if it shuts down regularly (before timeout occurs)

type Runnable

type Runnable interface {
	ID() string
	Run(routine int) error
	CustomData() interface{}
}

Runnable is the interface that all "runnable" tasks must implement

type Task

type Task struct {
	Runnable
}

Task encapsulates a base struct for objects that implement the Runnable interface