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 (optionally with timeout)

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 the examples subfolder to access a few small example programs that use pooler.

Here's a very basic example:

package main

import (
  "fmt"
  "os"
  "os/signal"
  "runtime"
  "sync/atomic"
  "syscall"
  "time"

  "github.com/segmentio/ksuid"
  "github.com/syncplify/pooler"
)

var tasks *pooler.Pool

var startTime = time.Now()

var counter int32 // we'll use this to simulate some work

// ****************************
// * TASK TO BE EXECUTED      *
// ****************************

type myTask struct {
  TaskID string
}

// In order to be a valid "pooler task" our struct needs to implement the pooler.Runnable interface,
// which means that we need (mandatory) to implement three methods:
// 1. ID() to return the task's unique ID
// 2. CustomData() to return the task's custom data, or nil in case this task has no need for custom data
// 3. Run(routine id) which is the actual func that runs the task

func (t *myTask) ID() string {
  return t.TaskID
}

func (t *myTask) CustomData() interface{} {
  // in this basic example, our task has no custom data, so we simply return nil
  return nil
}

func (t *myTask) Run(routine int) error {
  atomic.AddInt32(&counter, 1)
  return nil
}

// ****************************
// * MAIN PROGRAM             *
// ****************************

func main() {
  runtime.GOMAXPROCS(runtime.NumCPU())

  // Let's create a pool of 64 "workers" with a queue of up to 1 million tasks to execute
  var err error
  tasks, err = pooler.New(64, 1000000)
  if err != nil {
    panic(err)
  }

  // Now we spawn 10 goroutines that simultaneouly enqueue 100,000 tasks each to the pool (tot: 1 million tasks)
  for k := 0; k < 10; k++ {
    go func() {
      for i := 0; i < 100000; i++ {
        // Each task *must* have a unique ID, we use the excellent segmentio/ksuid package for this purpose
        job := &myTask{TaskID: ksuid.New().String()}
        // Let's add the task to the queue of tasks to be executed (enqueue)
        err := tasks.Enqueue(job)
        if err != nil {
          fmt.Println(err)
        }
      }
    }()
  }

  // Now let's just wait for the user to hit Ctrl-C
  quit := make(chan os.Signal)
  signal.Notify(quit, os.Interrupt, os.Kill, syscall.SIGTERM)
  <-quit

  // Shutdown the pool
  tasks.Shutdown()

  // Check the counter
  fmt.Println("Final value of counter:", counter)
}

Benchmark

BenchmarkPooler_Easy-8            705495              1615 ns/op              64 B/op          3 allocs/op

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 int64, maxTasks int64) *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 int64, maxTasks int64) (*Pool, error)
  • func NewWithConfig(config *Config) (*Pool, error)
  • func (p *Pool) ActiveTasks() int64
  • func (p *Pool) ActiveWorkers() int64
  • func (p *Pool) ConfiguredRoutines() int64
  • func (p *Pool) Enqueue(task Runnable) error
  • func (p *Pool) IsShuttingDown() bool
  • func (p *Pool) MaxTasks() int64
  • func (p *Pool) PrepareToWait()
  • func (p *Pool) QueueLen() int
  • func (p *Pool) Resize(newGoroutines int64) error
  • func (p *Pool) Shutdown()
  • func (p *Pool) ShutdownWithTimeout(timeout time.Duration) bool
  • func (p *Pool) Wait()
• 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, for example).

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 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 atomic.Int64
	// MaxTasks is the maximum number of tasks that can be in this pool's queue at any given time
	MaxTasks atomic.Int64
	// 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 int64, maxTasks int64) *Config

NewConfig creates and returns a basic/initial Config struct, with a specified numer of worker `routines` and a specified maximum number of queueable `maxTasks`.

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 int64, maxTasks int64) (*Pool, error)

New creates a new pooler.Pool object without any Callback functions. `routines` is the maximum number of "worker" goroutines that are allowed to run concurrently. `maxTasks` is the 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 pooler.Pool object with a user-provided configuration. `config` is a pointer to a pooler.Config object (see types.go).

func (*Pool) ActiveTasks

func (p *Pool) ActiveTasks() int64

ActiveTasks retuns the number of tasks that are REALLY being executed at this time.

func (*Pool) ActiveWorkers

func (p *Pool) ActiveWorkers() int64

ActiveWorkers returns the number of running goroutines, including the ones that are idle.

func (*Pool) ConfiguredRoutines

func (p *Pool) ConfiguredRoutines() int64

ConfiguredRoutines returns the number of configured goroutines in a thread-safe way

func (*Pool) Enqueue

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

Enqueue adds a task to the queue of tasks waiting to be executed. `task` can be any object that implements the pooler.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) MaxTasks

func (p *Pool) MaxTasks() int64

MaxTasks returns the maximum number of queueable tasks in a thread-safe way

func (*Pool) PrepareToWait

func (p *Pool) PrepareToWait()

func (*Pool) QueueLen

func (p *Pool) QueueLen() int

QueueLen returns the number of tasks currently queued, and waiting to be executed.

func (*Pool) Resize

func (p *Pool) Resize(newGoroutines int64) error

Resize attempts to resize the pool, adding or terminating goroutines as needed. It returns an error if resizing conditions aren't met.

func (*Pool) Shutdown

func (p *Pool) Shutdown()

Shutdown stops all goroutines running all tasks, and shuts down the entire pool. Please note that this method could actually wait forever untill all pending tasks are done.

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 doesn't wait forever, and it always returns on or before `timeout`. It returns true if it times out, and false if it shuts down regularly (before timeout occurs). Please note that if this function returns true (a timeout has occurred) you may still have "orphan" goroutines running; it is, therefore, recommended that this is the among the last methods you call just before your program terminates.

func (*Pool) Wait

func (p *Pool) Wait()

Done returns a channel that can be used to wait for the pool to be shut down.

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.