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.