Help us shape the future of Go! Take the 2026 Go Developer Survey

queue

package
v0.0.0-...-b293f34 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Sep 24, 2026 License: MIT Imports: 16 Imported by: 0

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var (
	ErrQueueFull   = errors.New("queue is full")
	ErrQueueClosed = errors.New("queue is closed")
)
View Source 
var ErrNoFeasibleCapacity = errors.New("no feasible worker capacity")

Functions

func FindCForSLO 

func FindCForSLO(lambda, tau, wTarget, ca, cs float64, cMax int) (int, float64, error)

FindCForSLO finds the smallest stable capacity whose estimated mean total latency is at most wTarget seconds, subject to cMax.

func MmcWait 

func MmcWait(lambda, mu float64, c int, ca, cs float64) (wq, w float64, err error)

MmcWait returns mean queue wait and total latency. It applies the Allen-Cunneen variability factor, floored at the M/M/c baseline.

func RecommendCFromObserved 

func RecommendCFromObserved(qs Stats, wp WorkerPoolStats, wqTarget time.Duration, ca, cs float64) (int, float64, error)

RecommendCFromObserved computes a recommended worker count from observed queue stats and worker pool stats. It estimates lambda from enqueue rate, tau from per-worker throughput, and targets a total latency of wqTarget+tau.

Types

type AdaptiveQueue 

type AdaptiveQueue struct {
	// contains filtered or unexported fields
}

AdaptiveQueue is a thread-safe, fixed-size circular queue whose core uses the lock-free Vyukov bounded MPMC algorithm. Metrics use short mutex sections. Each cell carries a sequence number that establishes a happens-before edge between the producer's write and the consumer's read, so a consumer can never observe a slot before its job has been published (the previous implementation advanced tail before writing the slot, which could hand a nil/stale job to a concurrent consumer).

func NewAdaptiveQueue 

func NewAdaptiveQueue(capacity uint64) (*AdaptiveQueue, error)

NewAdaptiveQueue creates a new AdaptiveQueue with the given capacity. Capacity must be a power of 2 and at least 2.

func (*AdaptiveQueue) Close 

func (q *AdaptiveQueue) Close()

Close marks the queue as closed.

func (*AdaptiveQueue) Dequeue 

func (q *AdaptiveQueue) Dequeue() (jobs.Job, error)

Dequeue removes and returns a single job from the queue.

func (*AdaptiveQueue) DequeueBatch 

func (q *AdaptiveQueue) DequeueBatch(maxSize int) ([]jobs.Job, error)

DequeueBatch removes and returns a batch of jobs from the queue.

func (*AdaptiveQueue) Enqueue 

func (q *AdaptiveQueue) Enqueue(job jobs.Job) error

Enqueue adds a single job to the queue.

func (*AdaptiveQueue) EnqueueBatch 

func (q *AdaptiveQueue) EnqueueBatch(jobsInterface []interface{}) error

EnqueueBatch adds a batch of jobs to the queue.

func (*AdaptiveQueue) EnsureCapacity 

func (q *AdaptiveQueue) EnsureCapacity(targetCap int)

EnsureCapacity is a no-op for AdaptiveQueue as it has a fixed capacity.

func (*AdaptiveQueue) IsEmpty 

func (q *AdaptiveQueue) IsEmpty() bool

IsEmpty checks if the queue is empty.

func (*AdaptiveQueue) Stats 

func (q *AdaptiveQueue) Stats() Stats

Stats returns the current statistics for the queue.

type DropPolicy 

type DropPolicy int

DropPolicy defines how the hybrid queue behaves once both the ring and overflow paths are saturated. 

const (
	// DropPolicyReject rejects new jobs when the queue is full.
	DropPolicyReject DropPolicy = iota
	// DropPolicyDropNewest drops the just-arrived job when capacity is exceeded.
	DropPolicyDropNewest
	// DropPolicyDropOldest evicts the oldest overflow job to admit a new job.
	DropPolicyDropOldest
)

func (DropPolicy) String 

func (p DropPolicy) String() string

type DynamicWorkerPool 

type DynamicWorkerPool struct {
	// contains filtered or unexported fields
}

DynamicWorkerPool manages a pool of workers that execute jobs from a queue. It can dynamically adjust the number of workers based on load.

func NewDynamicWorkerPool 

func NewDynamicWorkerPool(q Queue, config WorkerPoolConfig, logger *log.Logger) (*DynamicWorkerPool, error)

NewDynamicWorkerPool creates a new dynamic worker pool.

func (*DynamicWorkerPool) CancelWork 

func (p *DynamicWorkerPool) CancelWork()

DrainAndStop waits for outstanding tasks to finish before stopping the worker pool. CancelWork cancels operation contexts without dropping buffered results. The owner must still join DrainAndStop and consume result routes.

func (*DynamicWorkerPool) DrainAndStop 

func (p *DynamicWorkerPool) DrainAndStop()

func (*DynamicWorkerPool) GetRouter 

func (p *DynamicWorkerPool) GetRouter() *ResultRouter

GetRouter returns the result router for accessing type-specific result channels.

func (*DynamicWorkerPool) GetVariabilityCoefficients 

func (p *DynamicWorkerPool) GetVariabilityCoefficients() (ca, cs float64)

GetVariabilityCoefficients returns interarrival and execution-time CVs. Unknown variability uses the M/M/c baseline of one. A measured zero is valid.

func (*DynamicWorkerPool) Pause 

func (p *DynamicWorkerPool) Pause()

Pause temporarily stops the worker pool from processing new tasks.

func (*DynamicWorkerPool) Queue 

func (p *DynamicWorkerPool) Queue() Queue

func (*DynamicWorkerPool) ReplaceQueue 

func (p *DynamicWorkerPool) ReplaceQueue(newQueue Queue) error

ReplaceQueue replaces the current queue with a new one. This is used for dynamic queue switching (e.g., from Workiva to Adaptive). Any jobs still in the old queue are drained and re-enqueued into the new queue so no in-flight work is lost.

func (*DynamicWorkerPool) Resume 

func (p *DynamicWorkerPool) Resume()

Resume resumes worker pool processing after a pause.

func (*DynamicWorkerPool) SetArrivalRate 

func (p *DynamicWorkerPool) SetArrivalRate(lambda float64)

SetArrivalRate feeds the autoscaler the job arrival rate (lambda, jobs/sec). When set (>0), desiredCapacity sizes the pool from lambda instead of the throttled enqueue rate, so the pool can scale up even when the queue is empty.

func (*DynamicWorkerPool) SetSLOFeedback 

func (p *DynamicWorkerPool) SetSLOFeedback(recorder *slo.Recorder, config runtimeconfig.SLO)

func (*DynamicWorkerPool) SetServiceTime 

func (p *DynamicWorkerPool) SetServiceTime(tau float64)

SetServiceTime feeds the autoscaler the per-job service time (tau, seconds). It is a startup estimate; observed execution times take precedence.

func (*DynamicWorkerPool) SetSizingPolicy 

func (p *DynamicWorkerPool) SetSizingPolicy(totalLatency time.Duration, headroom float64)

SetSizingPolicy sets the mean total-latency target and fractional headroom. Zero values select TargetQueueLatency plus service time and 15% headroom.

func (*DynamicWorkerPool) SetTargetWorkers 

func (p *DynamicWorkerPool) SetTargetWorkers(n int)

SetTargetWorkers explicitly sets the worker pool capacity to n (clamped to the configured min/max) and records the target. It is used to apply the M/M/c-derived initial sizing computed at startup.

func (*DynamicWorkerPool) Start 

func (p *DynamicWorkerPool) Start()

Start begins the worker pool's operations. It is idempotent: a second call is a no-op, so Start cannot spawn duplicate goroutines or corrupt wg.

func (*DynamicWorkerPool) Stats 

func (p *DynamicWorkerPool) Stats() WorkerPoolStats

Stats returns runtime statistics for the worker pool.

type Handle 

type Handle struct {
	// contains filtered or unexported fields
}

Handle gives producers, workers and observers the same stable queue identity. Replacement is allowed only when empty, before controller execution begins.

func NewHandle 

func NewHandle(q Queue) *Handle

func (*Handle) Close 

func (h *Handle) Close()

func (*Handle) Dequeue 

func (h *Handle) Dequeue() (jobs.Job, error)

func (*Handle) DequeueBatch 

func (h *Handle) DequeueBatch(n int) ([]jobs.Job, error)

func (*Handle) Enqueue 

func (h *Handle) Enqueue(j jobs.Job) error

func (*Handle) EnqueueBatch 

func (h *Handle) EnqueueBatch(js []interface{}) error

func (*Handle) ReplaceEmpty 

func (h *Handle) ReplaceEmpty(q Queue) error

func (*Handle) Stats 

func (h *Handle) Stats() Stats

type History 

type History[T any] struct {
	// contains filtered or unexported fields
}

History is a fixed-size, thread-safe ring buffer of timestamped samples. It retains a rolling window of queue and worker-pool statistics for the web API's history endpoints. Writers (the sampler goroutine) and readers (HTTP handlers) may run concurrently.

func NewHistory 

func NewHistory[T any](capacity int) *History[T]

NewHistory returns a History that retains up to capacity samples.

func (*History[T]) Len 

func (h *History[T]) Len() int

Len returns the number of retained samples.

func (*History[T]) Record 

func (h *History[T]) Record(v T)

Record appends a sample with the current wall-clock time.

func (*History[T]) RecordAt 

func (h *History[T]) RecordAt(t time.Time, v T)

RecordAt appends a sample with an explicit timestamp.

func (*History[T]) Snapshot 

func (h *History[T]) Snapshot() []Sample[T]

Snapshot returns the retained samples in chronological order (oldest first).

type HybridQueue 

type HybridQueue struct {
	// contains filtered or unexported fields
}

HybridQueue combines a lock-free xsync ring buffer with a mutex-protected overflow slice. The ring handles steady-state work while the overflow absorbs bursts before optional dropping.

func NewHybridQueue 

func NewHybridQueue(config HybridQueueConfig) (*HybridQueue, error)

NewHybridQueue builds a HybridQueue using the supplied configuration.

func (*HybridQueue) Close 

func (q *HybridQueue) Close()

Close prevents new jobs from being enqueued.

func (*HybridQueue) Dequeue 

func (q *HybridQueue) Dequeue() (jobs.Job, error)

Dequeue alternates preference between overflow and ring to keep both moving.

func (*HybridQueue) DequeueBatch 

func (q *HybridQueue) DequeueBatch(maxSize int) ([]jobs.Job, error)

DequeueBatch reserves progress for both paths while either holds work.

func (*HybridQueue) Enqueue 

func (q *HybridQueue) Enqueue(job jobs.Job) error

Enqueue adds a single job to the queue, preferring the lock-free ring fast path.

func (*HybridQueue) EnqueueBatch 

func (q *HybridQueue) EnqueueBatch(items []interface{}) error

EnqueueBatch inserts a slice of jobs in FIFO order.

func (*HybridQueue) OverflowDepth 

func (q *HybridQueue) OverflowDepth() int

OverflowDepth returns the current overflow-slice depth (debug/metrics).

func (*HybridQueue) OverflowEvents 

func (q *HybridQueue) OverflowEvents() uint64

OverflowEvents returns the total number of overflow enqueue events (debug/metrics).

func (*HybridQueue) RingDepth 

func (q *HybridQueue) RingDepth() int

RingDepth returns the current ring-buffer depth (debug/metrics).

func (*HybridQueue) Stats 

func (q *HybridQueue) Stats() Stats

Stats returns observable metrics for monitoring.

type HybridQueueConfig 

type HybridQueueConfig struct {
	Name             string
	RingCapacity     int
	OverflowCapacity int
	SoftWatermark    float64
	HardWatermark    float64
	DropPolicy       DropPolicy
	Logger           *zap.Logger
}

HybridQueueConfig controls the behaviour of a HybridQueue instance.

func DefaultHybridQueueConfig 

func DefaultHybridQueueConfig() HybridQueueConfig

DefaultHybridQueueConfig returns the recommended production defaults.

type Queue 

type Queue interface {
	// Enqueue adds a single job to the queue.
	Enqueue(job jobs.Job) error

	// EnqueueBatch adds jobs in order and can accept a prefix before returning
	// an error. Callers that need exact admission accounting must use Enqueue
	// per job; retrying the entire batch can duplicate already accepted work.
	EnqueueBatch(jobs []interface{}) error

	// Dequeue removes and returns a single job from the queue.
	Dequeue() (jobs.Job, error)

	// DequeueBatch removes and returns a batch of jobs from the queue.
	DequeueBatch(maxSize int) ([]jobs.Job, error)

	// Close shuts down the queue and prevents new jobs from being enqueued.
	Close()

	// Stats returns statistics about the queue's performance.
	Stats() Stats
}

Queue defines the interface for a generic, thread-safe queue system. This allows the controller and systems to be decoupled from a specific queue implementation.

func NewQueue 

func NewQueue(config QueueConfig) (Queue, error)

NewQueue creates a new queue based on the provided configuration

func NewWorkivaQueue 

func NewWorkivaQueue(capacity int) Queue

NewWorkivaQueue creates a new expanding queue backed by Workiva RingBuffers.

type QueueConfig 

type QueueConfig struct {
	Name         string
	Type         QueueType
	Capacity     int
	HybridConfig HybridQueueConfig
}

QueueConfig holds configuration for queue creation

func DefaultQueueConfig 

func DefaultQueueConfig() QueueConfig

DefaultQueueConfig returns the default queue configuration

type QueueType 

type QueueType string

QueueType represents the type of queue to create 

const (
	QueueTypeAdaptive QueueType = "adaptive"
	QueueTypeWorkiva  QueueType = "workiva"
	QueueTypeHybrid   QueueType = "hybrid"
)

type ResultRouter 

type ResultRouter struct {
	PulseResultChan        chan []jobs.Result
	InterventionResultChan chan []jobs.Result
	CodeResultChan         chan []jobs.Result
	// contains filtered or unexported fields
}

ResultRouter handles routing of job results to type-specific channels. This enables decoupling result processing from the main worker pool.

func NewResultRouter 

func NewResultRouter(config WorkerPoolConfig, logger *log.Logger) *ResultRouter

NewResultRouter creates a new result router with buffered channels.

func (*ResultRouter) Close 

func (r *ResultRouter) Close()

Close closes all result channels. It is idempotent.

func (*ResultRouter) RouteResults 

func (r *ResultRouter) RouteResults(results []jobs.Result)

RouteResults takes a batch of mixed results and routes them to appropriate channels.

type Sample 

type Sample[T any] struct {
	Timestamp time.Time `json:"timestamp"`
	Value     T         `json:"value"`
}

Sample is a timestamped value stored in a History ring buffer.

type Stats 

type Stats struct {
	ArrivalCV      float64       `json:"arrival_cv"`
	ArrivalSamples int           `json:"arrival_samples"`
	LastEnqueue    time.Time     `json:"last_enqueue"`
	LastDequeue    time.Time     `json:"last_dequeue"`
	AvgQueueTime   time.Duration `json:"avg_queue_time"`
	Dequeued       int64         `json:"dequeued"`
	Dropped        int64         `json:"dropped"`
	MaxQueueTime   time.Duration `json:"max_queue_time"`
	QueueDepth     int           `json:"queue_depth"`
	MaxJobLatency  time.Duration `json:"max_job_latency"`
	AvgJobLatency  time.Duration `json:"avg_job_latency"`
	EnqueueRate    float64       `json:"enqueue_rate"`
	DequeueRate    float64       `json:"dequeue_rate"`
	Enqueued       int64         `json:"enqueued"`
	Capacity       int           `json:"capacity"`
	SampleWindow   time.Duration `json:"sample_window"`
	// OldestPendingAge measures a still-pending admitted copy, never an active
	// executor. A missing source or empty queue has no age measurement.
	OldestPendingAge       time.Duration `json:"oldest_pending_age"`
	OldestPendingAvailable bool          `json:"oldest_pending_available"`
	OldestPendingReason    string        `json:"oldest_pending_reason,omitempty"`
}

Stats holds performance metrics for a queue.

type WorkerPoolConfig 

type WorkerPoolConfig struct {
	MinWorkers         int
	MaxWorkers         int
	AdjustmentInterval time.Duration
	ResultBatchSize    int
	ResultBatchTimeout time.Duration
	ResultChannelDepth int
	// NumShards is the number of result-routing shards. 0 (default) means
	// runtime.GOMAXPROCS(0), capped at 64. Each shard has its own channel and
	// resultProcessor goroutine so results are routed in parallel.
	NumShards          int
	TargetQueueLatency time.Duration
	// DrainTimeout cancels accepted operations after this grace period.
	DrainTimeout time.Duration
	// Ants-specific options
	PreAlloc         bool
	NonBlocking      bool
	MaxBlockingTasks int
	ExpiryDuration   time.Duration
}

WorkerPoolConfig holds configuration for the DynamicWorkerPool.

func DefaultWorkerPoolConfig 

func DefaultWorkerPoolConfig() WorkerPoolConfig

DefaultWorkerPoolConfig returns a default configuration for the worker pool.

type WorkerPoolStats 

type WorkerPoolStats struct {
	Attempts        int64         `json:"attempts"`
	AttemptTimeouts int64         `json:"attempt_timeouts"`
	AttemptDuration time.Duration `json:"attempt_duration"`
	RetryDelay      time.Duration `json:"retry_delay"`
	SLOCondition    string        `json:"slo_condition"`
	ServiceTime     time.Duration `json:"service_time"`
	ServiceCV       float64       `json:"service_cv"`
	ServiceSamples  int           `json:"service_samples"`
	ArrivalRate     float64       `json:"arrival_rate"`
	SizingModel     string        `json:"sizing_model"`
	LastScaleTime   time.Time     `json:"last_scale_time"`
	MinWorkers      int           `json:"min_workers"`
	MaxWorkers      int           `json:"max_workers"`
	CurrentCapacity int           `json:"current_capacity"`
	RunningWorkers  int           `json:"running_workers"`
	WaitingTasks    int           `json:"waiting_tasks"`
	TargetWorkers   int           `json:"target_workers"`
	TasksSubmitted  int64         `json:"tasks_submitted"`
	TasksCompleted  int64         `json:"tasks_completed"`
	ScalingEvents   int64         `json:"scaling_events"`
	PendingResults  int           `json:"pending_results"`
}

WorkerPoolStats exposes runtime metrics for the dynamic worker pool.

type WorkivaQueue 

type WorkivaQueue struct {
	// contains filtered or unexported fields
}

WorkivaQueue is a capacity-expanding MPMC queue using Workiva ring buffers. Producers serialize per segment so retired segments cannot receive new jobs.

func (*WorkivaQueue) Close 

func (q *WorkivaQueue) Close()

func (*WorkivaQueue) Dequeue 

func (q *WorkivaQueue) Dequeue() (jobs.Job, error)

Dequeue removes and returns a job. Returns (nil, nil) if empty.

func (*WorkivaQueue) DequeueBatch 

func (q *WorkivaQueue) DequeueBatch(maxSize int) ([]jobs.Job, error)

DequeueBatch removes up to maxSize jobs. Returns quickly if empty.

func (*WorkivaQueue) Enqueue 

func (q *WorkivaQueue) Enqueue(job jobs.Job) error

Enqueue adds a single job; expands capacity if the current segment is full.

func (*WorkivaQueue) EnqueueBatch 

func (q *WorkivaQueue) EnqueueBatch(items []interface{}) error

EnqueueBatch validates the item types before admitting jobs one at a time.

func (*WorkivaQueue) IsEmpty 

func (q *WorkivaQueue) IsEmpty() bool

func (*WorkivaQueue) Stats 

func (q *WorkivaQueue) Stats() Stats

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.