ants Goroutine Pool Tutorial

Introduction to Go ants

ants is a high-performance goroutine pool that implements the scheduling and management of a large number of goroutines, allowing for limitation and reuse of resources, thus achieving more efficient execution of tasks when developing concurrent programs.

Features

• Automatically schedule a massive number of goroutines and reuse them.
• Regularly clean up expired goroutines to save resources further.
• Provides a large number of useful interfaces: task submission, obtaining the number of running goroutines, dynamically adjusting the pool size, releasing the pool, and restarting the pool.
• Gracefully handle panics to prevent program crashes.
• Resource reuse greatly saves memory usage. In the scenario of large-scale batch concurrent tasks, it has higher performance than native goroutine concurrency.
• Non-blocking mechanism

How does ants work

Flow Chart

Animated Images

Installation

Using ants v1 version:

go get -u github.com/panjf2000/ants

Using ants v2 version (enable GO111MODULE=on):

go get -u github.com/panjf2000/ants/v2

Usage

When writing a Go concurrent program that launches a large number of goroutines, it will inevitably consume a large amount of system resources (memory, CPU). By using ants, you can instantiate a goroutine pool to reuse goroutines, saving resources and improving performance:

package main

import (
	"fmt"
	"sync"
	"sync/atomic"
	"time"

	"github.com/panjf2000/ants/v2"
)

var sum int32

func myFunc(i interface{}) {
	n := i.(int32)
	atomic.AddInt32(&sum, n)
	fmt.Printf("run with %d\n", n)
}

func demoFunc() {
	time.Sleep(10 * time.Millisecond)
	fmt.Println("Hello World!")
}

func main() {
	defer ants.Release()

	runTimes := 1000

	// Use the common pool.
	var wg sync.WaitGroup
	syncCalculateSum := func() {
		demoFunc()
		wg.Done()
	}
	for i := 0; i < runTimes; i++ {
		wg.Add(1)
		_ = ants.Submit(syncCalculateSum)
	}
	wg.Wait()
	fmt.Printf("running goroutines: %d\n", ants.Running())
	fmt.Printf("finish all tasks.\n")

	// Use the pool with a function,
	// set 10 to the capacity of goroutine pool and 1 second for expired duration.
	p, _ := ants.NewPoolWithFunc(10, func(i interface{}) {
		myFunc(i)
		wg.Done()
	})
	defer p.Release()
	// Submit tasks one by one.
	for i := 0; i < runTimes; i++ {
		wg.Add(1)
		_ = p.Invoke(int32(i))
	}
	wg.Wait()
	fmt.Printf("running goroutines: %d\n", p.Running())
	fmt.Printf("finish all tasks, result is %d\n", sum)
}

Pool Configuration

// Option represents the optional function.
type Option func(opts *Options)

// Options contains all options which will be applied when instantiating a ants pool.
type Options struct {
	// ExpiryDuration is a period for the scavenger goroutine to clean up those expired workers,
	// the scavenger scans all workers every `ExpiryDuration` and clean up those workers that haven't been
	// used for more than `ExpiryDuration`.
	ExpiryDuration time.Duration

	// PreAlloc indicates whether to make memory pre-allocation when initializing Pool.
	PreAlloc bool

	// Max number of goroutine blocking on pool.Submit.
	// 0 (default value) means no such limit.
	MaxBlockingTasks int

	// When Nonblocking is true, Pool.Submit will never be blocked.
	// ErrPoolOverload will be returned when Pool.Submit cannot be done at once.
	// When Nonblocking is true, MaxBlockingTasks is inoperative.
	Nonblocking bool

	// PanicHandler is used to handle panics from each worker goroutine.
	// if nil, panics will be thrown out again from worker goroutines.
	PanicHandler func(interface{})

	// Logger is the customized logger for logging info, if it is not set,
	// default standard logger from log package is used.
	Logger Logger
}

// WithOptions accepts the whole options config.
func WithOptions(options Options) Option {
	return func(opts *Options) {
		*opts = options
	}
}

// WithExpiryDuration sets up the interval time of cleaning up goroutines.
func WithExpiryDuration(expiryDuration time.Duration) Option {
	return func(opts *Options) {
		opts.ExpiryDuration = expiryDuration
	}
}

// WithPreAlloc indicates whether it should malloc for workers.
func WithPreAlloc(preAlloc bool) Option {
	return func(opts *Options) {
		opts.PreAlloc = preAlloc
	}
}

// WithMaxBlockingTasks sets up the maximum number of goroutines that are blocked when it reaches the capacity of pool.
func WithMaxBlockingTasks(maxBlockingTasks int) Option {
	return func(opts *Options) {
		opts.MaxBlockingTasks = maxBlockingTasks
	}
}

// WithNonblocking indicates that pool will return nil when there is no available workers.
func WithNonblocking(nonblocking bool) Option {
	return func(opts *Options) {
		opts.Nonblocking = nonblocking
	}
}

// WithPanicHandler sets up panic handler.
func WithPanicHandler(panicHandler func(interface{})) Option {
	return func(opts *Options) {
		opts.PanicHandler = panicHandler
	}
}

// WithLogger sets up a customized logger.
func WithLogger(logger Logger) Option {
	return func(opts *Options) {
		opts.Logger = logger
	}
}

By using various optional functions when calling NewPool/NewPoolWithFunc, the values of each configuration item in ants.Options can be set, and then used to customize the goroutine pool.

Custom Pool

ants supports instantiating the user's own Pool with a specific pool capacity; by calling the NewPool method, a new Pool with the specified capacity can be instantiated as follows:

p, _ := ants.NewPool(10000)

Task Submission

Tasks are submitted by calling the ants.Submit(func()) method:

ants.Submit(func(){})

Dynamically Adjusting the Goroutine Pool Capacity

To dynamically adjust the goroutine pool capacity, you can use the Tune(int) method:

pool.Tune(1000) // Tune its capacity to 1000
pool.Tune(100000) // Tune its capacity to 100000

This method is thread-safe.

Pre-allocate goroutine queue memory

The ants allows you to pre-allocate memory for the entire pool, which can improve the performance of the goroutine pool in certain specific scenarios. For example, in a scenario where a pool with a very large capacity is needed and each task within the goroutine is time-consuming, pre-allocating memory for the goroutine queue will reduce unnecessary memory reallocation.

// ants will pre-malloc the whole capacity of pool when you invoke this function
p, _ := ants.NewPool(100000, ants.WithPreAlloc(true))

Release the Pool

pool.Release()

Restart the Pool

// By calling the Reboot() method, you can reactivate a pool that has been previously destroyed and put it back into use.
pool.Reboot()

About the task execution order

ants does not guarantee the order of task execution, and the execution order does not necessarily correspond to the order of submission. This is because ants processes all submitted tasks concurrently, and the tasks will be assigned to workers that are concurrently running, resulting in the execution of tasks in a concurrent and non-deterministic order.