pool/lib.go

// Package pool provides a resource pool implementation that is safe to access from multiple goroutines.
//
// The pool supports finalizing items when shrinking the pool. It helps monitoring the pool usage and state with
// events and statistics. While it implements max idle size and timeout based shrinking algorithms to release
// idle resources from the pool, it also provides a zero-config adaptive algorithm for this purpose that can
// automatically adapt to changing resource usage characteristics. It also accepts custom algorithm
// implementations.
package pool

import (
	"code.squareroundforest.org/arpio/syncbus"
	"code.squareroundforest.org/arpio/times"
	"errors"
	"fmt"
	"strings"
	"time"
)

// Stats provides information about the pool state.
type Stats struct {

	// Idle is the number of resources currently held by the pool.
	Idle int

	// Active is the number of resources that are currently in use as known by the pool.
	Active int

	// Get is the number of get operations during the entire life cycle of the pool.
	Get int

	// Put is the number of put operations during the entire life cycle of the pool.
	Put int

	// Alloc is the number of allocations executed by the pool during the entire life cycle of the pool.
	Alloc int

	// Load is the total number of items explicitly added to the pool via the Load method.
	Load int

	// Free is the number of deallocations executed by the pool during the entire life cycle of the pool.
	Free int
}

// EventType is a binary flag categorizing the reason of an event. The types of events can be combined together,
// e.g. if a get operation requires an allocate operation, then the event type will be
// GetOperation|AllocateOperation.
type EventType int

const (
	// None can be used to mask out all event types and not receiving any events.
	None EventType = 0

	// GetOperation is the type of events sent after a get operation.
	GetOperation EventType = 1 << iota

	// PutOperation is the type of events sent after a put operation.
	PutOperation

	// AllocateOperation is the type of events sent after an allocate operation.
	AllocateOperation

	// LoadOperation is the type of events sent after a load operation.
	LoadOperation

	// FreeOperation is the type of events sent after a free operation.
	FreeOperation

	// AllocateError is the type of events sent after a failed allocation. The error will not be included
	// with the event, but it will be returned by the failed Get function call.
	AllocateError

	// AllEvents can be used as a mask that includes all the event types.
	AllEvents = GetOperation | PutOperation | AllocateOperation | LoadOperation | FreeOperation | AllocateError
)

// Event values are sent by the pool after various operations, if a channel is provided to send the events to.
type Event struct {

	// Type is the binary flag depicting the type of the event.
	Type EventType

	// Stats contains the statistics about the pool at the time of the event.
	Stats Stats
}

// Algo implementations control when the shrinking of the idle items in the pool happens. The pool can be used
// with the implementations provided by this package or custom ones. The implementation can hold internal state,
// the pool guarantees that the algo instance is accessed only from one goroutine at a time.
type Algo interface {

	// Target is called on every Put operation with the current pool state as the input. It is expected to
	// return the target idle count, as dictated by the implementing algorithm. Optionally, a timeout value
	// can be returned (nextCheck), and if it is a positive value, the pool will call Target again after the
	// defined time expires to fetch the next target idle count. In each case, when Target returns a smaller
	// number than the current idle count, it shrinks the pool to the defined target.
	//
	// When using nextCheck, not every returned value results in calling Target by the pool, only the ones
	// that were set after the previous check expired.
	//
	// Implementations should consider that while the recommended way of using the pool is to only call Put
	// with items that were received by calling Get, the pool itself doesn't prohibit calling Put with
	// 'foreign' items.
	Target(Stats) (target int, nextCheck time.Duration)

	// Load is called by the pool when Pool.Load is used, passing in the number of items that were loaded as
	// the result of a 'prewarm' or other preallocation. The number of loaded items is passed in as
	// argument, which can be used to adjust the algorithm's internal state. If the implemented algorithm is
	// stateless, or it is not sensitive to loading items this way, Load can be implemented as a noop.
	Load(int)
}

// Options can be used to configure the pool. Some of the options are provided to support testing various
// scenarios.
type Options struct {

	// Events is a channel that, when set, the pool uses to send events. The channel needs to be used
	// together with a non-default event mask set. When using events, we should consider using a buffered
	// channel. Events can be dropped if the consumer is blocked and the channel is not ready to communicate
	// at the time of the event.
	Events chan<- Event

	// EventMask is a binary flag that defines which events will be sent to the provided channel. The
	// default is no events.
	EventMask EventType

	// Algo is the algorithm implementation used for shrinking the pool. The default is Adaptive().
	Algo Algo

	// Clock is an optional clock meant to be used with testing. The main purpose is to avoid time sensitive
	// tests running for a too long time.
	Clock times.Clock

	// TestBus is an optional signal bus meant to be used with testing. The main purpose is to ensure that
	// specific blocks of code are executed in a predefined order during concurrent tests.
	TestBus *syncbus.SyncBus
}

// Pool is a synchronized pool of resources that are considered expensive to allocate. Initialize the pool with
// the Make() function. Methods of uninitialized Pool instances may block forever. For the usage of the pool,
// see the docs of its methods, initialization options and the provided algorithms.
type Pool[R any] struct {
	pool pool[R]
}

// ErrEmpty is returned on Get calls when the pool is empty and it was initialized without an allocate function.
var ErrEmpty = errors.New("empty pool")

// String returns the string representation of the EventType binary flag, including all the flags that are set.
func (et EventType) String() string {
	var s []string
	if et&GetOperation != 0 {
		s = append(s, "get")
	}

	if et&PutOperation != 0 {
		s = append(s, "put")
	}

	if et&AllocateOperation != 0 {
		s = append(s, "allocate")
	}

	if et&LoadOperation != 0 {
		s = append(s, "load")
	}

	if et&FreeOperation != 0 {
		s = append(s, "free")
	}

	if et&AllocateError != 0 {
		s = append(s, "allocerr")
	}

	if len(s) == 0 {
		return "none"
	}

	return strings.Join(s, "|")
}

// String returns the string representation of an Event value, including the type and statistics.
func (ev Event) String() string {
	return fmt.Sprintf("%v; %v", ev.Type, ev.Stats)
}

// String returns the string representation of a set of statistics about the pool.
func (s Stats) String() string {
	return fmt.Sprintf(
		"idle: %d, active: %d, get: %d, put: %d, alloc: %d, load: %d, free: %d",
		s.Idle,
		s.Active,
		s.Get,
		s.Put,
		s.Alloc,
		s.Load,
		s.Free,
	)
}

// Adaptive creates a zero-config pool shrink algorithm instance. It is the default algorithm used by the pool.
//
// It is based on the exponential moving average of the active items and their deviation. This way it can react
// to, and to some extent overbuild, on the perceived stress. It decays the number of idle items gradually, and
// on very sudden drops in traffic, it ensures the eventual release of all pooled items with an internal
// background job. This job is timed based on the duration of the last active usage session (the period during
// which there were active items). Together with the pool implementation, it always reuses the most recent
// items, as in LIFO for Get and FIFO for Free.
//
// We need to be aware of some potential caveats due to its zero-config nature. Because it relies on the
// sequence of operations rather than wall-clock thresholds for its core logic, the algorithm treats rapid
// spikes and gradual surges similarly if the sequence of pool states is identical. In short, it can happen
// that: __/\__/\__/\__ = _|_|_|_. It prioritizes maintaining sufficient 'headroom' based on observed
// volatility.
func Adaptive() Algo {
	return makeAdaptiveAlgo()
}

// MaxTimeout creates a pool shrink algorithm instance, that releases items whenever the number of idle items
// exceeds max, and it also releases those items that were idle for too long. Together with the pool, it ensures
// that the Get operation is LIFO and the Free operation is FIFO.
//
// If max <= 0, the max pool size is not enforced. If to <= 0, the timeout is not enforced.
func MaxTimeout(max int, to time.Duration) Algo {
	return makeMaxTimeout(max, to)
}

// Max is like MaxTimeout, but without the max idle time.
func Max(max int) Algo {
	return makeMaxTimeout(max, 0)
}

// Timeout is like MaxTimeout, but without the max pool size.
func Timeout(to time.Duration) Algo {
	return makeMaxTimeout(0, to)
}

// NoShrink is a noop shrink algorithm, it doesn't release any idle items. The user code can decide whether to
// put back items in the pool or not. It might be useful in certain testing scenarios.
func NoShrink() Algo {
	return makeMaxTimeout(0, 0)
}

// Make initializes a Pool instance.
//
// The parameter alloc is used on Get operations when the pool is empty. If alloc is nil, and the pool is empty
// at the time of calling Get, Get will return ErrEmpty. If alloc returns an error, the same error is returned
// by Get. If events were configured, alloc triggers an AllocateOperation event. This event is typically the
// same as the GetOperation event.
//
// The parameter free is called when an item is released from the pool, with the item being released as the
// argument. It can be nil for resource types that don't need explicit deallocation. If events were configured,
// releasing an item triggers a FreeOperation event, regardless if the free parameter is nil.
func Make[R any](alloc func() (R, error), free func(R), o Options) Pool[R] {
	return Pool[R]{pool: makePool(alloc, free, o)}
}

// Get returns an item from the pool. If the pool is empty and no allocation function was configured, it returns
// ErrEmpty. If the pool is empty, and the allocation function returns an error, it returns that error. If
// events were configured, Get triggers a GetOperation event.
func (p Pool[R]) Get() (R, error) {
	return p.pool.get()
}

// Put stores an item in the pool. If events were configured, it triggers a PutOperation event.
//
// It is recommended to use it only with items that were received by the Get method. While it is allowed to put
// other items in the pool, it may change the way the shrinking algorithm works. E.g. it can be considered as a
// sudden drop in the number of active items. If the pool needs to be prewarmed, or prepared for an expected
// spike of traffic, consider using the Load method.
func (p Pool[R]) Put(i R) {
	p.pool.put(i)
}

// Load can be used to populate the pool with items that were not allocated as the result of the Get operation.
// It can be useful in scenarios where prewarming or preparing for a sudden traffic spike is necessary. If
// events were configured, it triggers a LoadOperation event.
func (p Pool[R]) Load(i []R) {
	p.pool.load(i)
}

// Stats returns statistics about the current state of the pool. It contains the current number of active/idle
// items, and perpetual counters for the various pool operations.
func (p Pool[R]) Stats() Stats {
	return p.pool.stats()
}

// Free releases all idle items in the pool. While the pool stays operational, Free is meant to be used when the
// pool is not required anymore.
func (p Pool[R]) Free() {
	p.pool.freePool()
}