pooler

README

Pooler

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

Features

• A fast worker-pool implementation that only relies on Go channels and atomic
• Optional callback function 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

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

Examples

Please take a look at examples to access a few smal example programs that use pooler. These are good learning sources.

To show how easy-to-use this package is, here's the source of the most 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

// stats prints "what's going on" every 0.5 seconds
func stats() {
  for {
    cnt := atomic.LoadInt32(&counter)
    if cnt < 1000000 {
      fmt.Printf("#\n# Elapsed: %s - Counter: %d\n#\n", time.Now().Sub(startTime).String(), cnt)
    } else {
      fmt.Printf("#\n# Elapsed: %s - Counter: %d\n# Hit Ctrl-C to terminate the program\n#\n", time.Now().Sub(startTime).String(), cnt)
    }
    time.Sleep(time.Second)
  }
}

func main() {
  // Set GOMAXPROCS = NumCPU to use all available cores
  runtime.GOMAXPROCS(runtime.NumCPU())

  // Let's create a pool of 64 "workers" with a queue of up to 1 million tasks to execute
  tasks = pooler.New(64, 1000000)

  // This goroutine only shows "what's going on" every second
  go stats()

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

// ****************************
// * 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{} {
  return nil // in this basic example, our task has no custom data, so we simply return nil
}

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

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 CallbackFunc
• type Event
• type Pool
  • func New(routineNum int, maxTasks int) (pool *Pool)
  • func NewWithCallback(routineNum int, maxTasks int, cbFunc CallbackFunc) (pool *Pool)
  • 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()
• type Runnable
• type Task

Types

type CallbackFunc

type CallbackFunc func(routine int, task *Task, event Event, err error)

CallbackFunc is the prototype of a function that will be called by the pool to notify events

type Event

type Event int

Event is an enumerable that represents task-related events

const (
	// WorkerCreated is when a "worker" goroutine is created
	WorkerCreated Event = iota
	// WorkerShutdown is called when a "worker" goroutine is shut down
	WorkerShutdown Event = iota
	// TaskQueued is when the task is first accepted and inserted in the queue
	TaskQueued Event = iota
	// TaskStarted is when the job is about to be start executing a task
	TaskStarted Event = iota
	// TaskDone is when a task is done running
	TaskDone Event = iota
	// TaskDoneWithError is when a job has finished running but returned an error
	TaskDoneWithError Event = iota
	// TaskCrashed is when a job is interrupted by an unexpected panic
	TaskCrashed Event = iota
	// QueueError is when an attempt to enqueue a task is made while pool is shutting down (panic is caught and gracefully handled)
	QueueError Event = iota
)

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(routineNum int, maxTasks int) (pool *Pool)

New creates a new Pool object without a Callback function routineNum: 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 NewWithCallback

func NewWithCallback(routineNum int, maxTasks int, cbFunc CallbackFunc) (pool *Pool)

NewWithCallback creates a new Pool object with a Callback function routineNum: 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 cfFunc: a callback function that will be called when certain events occur (see its prototype in 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

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