pool/lib.go

// pool with support for:
// - finalizing items
// - monitoring: events and/or stats
// - adaptive shrinking algorithm
// - custom shrinking algorithms
package pool

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

type Stats struct {
	Idle   int
	Active int
	Get    int
	Put    int
	Alloc  int
	Free   int
}

type EventType int

const (
	None         EventType = 0
	GetOperation EventType = 1 << iota
	PutOperation
	AllocateOperation
	FreeOperation
	AllocateError

	AllEvents = GetOperation | PutOperation | AllocateOperation | FreeOperation | AllocateError
)

type Event struct {
	Type  EventType
	Stats Stats
}

type Algo interface {

	// always called
	// desired idle items
	// implementations should consider the cost of freeing the stored resources
	// must support being called from a goroutine other than it was created in
	// a single pool instance only calls it from a single goroutine at a time
	// items need to be allocated always by calling Get
	// second return argument for requested next check
	// the Target function of a single Algo implementation is not called concurrently
	// called on all put, regardless of nextCheck
	// not all nextChecks result in a call if a previously request nextCheck is still pending
	Target(Stats) (target int, nextCheck time.Duration)

	// called when Pool.Load
	// can be used to adjust internal state
	// can be noop if not required
	Load(int)
}

type Options struct {

	// events can be dropped if the consumer is blocked
	Events chan<- Event

	EventMask EventType
	Algo      Algo
	Clock     times.Clock
	TestBus   *syncbus.SyncBus
}

type Pool[R any] struct {
	pool pool[R]
}

var ErrEmpty = errors.New("empty pool")

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&FreeOperation != 0 {
		s = append(s, "free")
	}

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

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

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

func (ev Event) String() string {
	return fmt.Sprintf("%v; %v", ev.Type, ev.Stats)
}

func (s Stats) String() string {
	return fmt.Sprintf(
		"idle: %d, active: %d, get: %d, put: %d, alloc: %d, free: %d",
		s.Idle,
		s.Active,
		s.Get,
		s.Put,
		s.Alloc,
		s.Free,
	)
}

// zero-config
// potential caveats:
// - a caveaat depending on the expectations, since no absolute time input is used, identifies frequent
// spikes from zero and slow grow and shrink from and to zero cycles are considered the same and the pool cleans
// up idle items accordingly. In short: _|_|_|_ = __/\__/\__/\__
func Adaptive() Algo {
	return makeAdaptiveAlgo()
}

// enfoces a max pool size and a timeout for the items
// when adding items to the pool via Put that were not fetched via Get, there can discrepancies occur in which
// items get timed out, but the general pool limitations get still consistently enforced eventually
func MaxTimeout(max int, to time.Duration) Algo {
	return makeMaxTimeout(max, to)
}

// like MaxTimeout but without enforcing timeouts
func Max(max int) Algo {
	return makeMaxTimeout(max, 0)
}

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

// the user code can decide not to put back items to the pool, however, the primary purpose is testing
func NoShrink() Algo {
	return makeMaxTimeout(0, 0)
}

// alloc and free need to support calls from goroutines other than they were created in
// a single pool instance only calls them from a single goroutine at a time
// free happens synchronously, user code may execute it in the background, in which case it is the user code's
// responsibility to ensure that free is fully carried out before the application exits, if that's necessary
func Make[R any](alloc func() (R, error), free func(R), o Options) Pool[R] {
	return Pool[R]{pool: makePool(alloc, free, o)}
}

func (p Pool[R]) Get() (R, error) {
	return p.pool.get()
}

// it is allowed to put items that were not received by get, but the selected algorithm may produce unexpected
// behavior. In most cases, it is recommended to use Load instead, and using Put to put back only those items
// that were received via Get.
func (p Pool[R]) Put(i R) {
	p.pool.put(i)
}

func (p Pool[R]) Load(i []R) {
	p.pool.load(i)
}

func (p Pool[R]) Stats() Stats {
	return p.pool.stats()
}

func (p Pool[R]) Free() {
	p.pool.freePool()
}