queue

package module
v0.1.0 Latest Latest
Warning

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

 Go to latest Published: Aug 18, 2026 License: MIT Imports: 10 Imported by: 0

README

queue

A durable, distributed task queue with priority scheduling, CPU-aware and goroutine-pool-aware dispatch, crash recovery, and consumer concurrency control.

Architecture

Producer → Enqueue → Queue Backend (memory / redis)
                              ↓
                    Scheduler (worker pool)
                     ↓              ↓
               CPU Adaptive    Fixed Pool
                     ↓
                   Workers → Handler(task)

Backends

memory

In-process priority queue — fast, single-node, no persistence.

import memoryqueue "github.com/LingByte/ling-base/queue/memory"

q := memoryqueue.New("my-tasks")
redis

Redis-backed distributed queue using sorted sets for priority scheduling. Supports multiple consumer nodes, crash recovery, and persistent storage.

import redisqueue "github.com/LingByte/ling-base/queue/redis"

client := redis.NewClient(&redis.Options{Addr: "localhost:6379"})
q := redisqueue.New("my-tasks", client)

Scheduler

The scheduler pulls tasks from the queue and dispatches them to workers.

Fixed worker pool (default)
scheduler, _ := queue.NewScheduler(queue.SchedulerConfig{
    Queue:       q,
    Handler:     myHandler,
    WorkerCount: 4,
})
scheduler.Start()
defer scheduler.Stop()
CPU-adaptive scheduling

Workers scale based on CPU usage — more workers when CPU is idle, fewer when busy:

scheduler, _ := queue.NewScheduler(queue.SchedulerConfig{
    Queue:            q,
    Handler:          myHandler,
    Mode:             queue.ModeCPUAdaptive,
    WorkerCount:      4,
    MinWorkers:       1,
    MaxWorkers:       16,
    CPUHighThreshold: 80,  // reduce workers above 80% CPU
    CPULowThreshold:  30,  // add workers below 30% CPU
    CPUCheckInterval: 5 * time.Second,
})
External worker pool

Use an external pool.WorkerPool for dispatch:

wp := pool.NewWorkerPool(8, 100)
wp.Start()

scheduler, _ := queue.NewScheduler(queue.SchedulerConfig{
    Queue:      q,
    Handler:    myHandler,
    WorkerPool: wp,
})

Task lifecycle

pending → running → success
                  → failed (if max retries exceeded)
                  → retry (if retries remaining)
pending → canceled

Tasks support:

  • Priority — higher priority tasks are dequeued first
  • Retries — configurable MaxRetries with automatic requeue
  • Cancel — remove pending tasks
  • Progress — track execution progress

Persistence & recovery

When using the Redis backend, tasks survive process restarts:

  1. On Enqueue, the task is persisted to Redis
  2. On Dequeue, the task is moved from pending to running set
  3. On Ack, the task is marked terminal and cleaned up
  4. On Start(), the scheduler calls Recover() — all running tasks (from a crashed process) are reset to pending and re-queued

API

Queue interface
type Queue interface {
    Enqueue(ctx, task) error
    Dequeue(ctx, timeout) (*Task, error)
    Ack(ctx, taskID, status, errMsg) error
    Requeue(ctx, taskID) error
    Get(ctx, taskID) (*Task, error)
    Cancel(ctx, taskID) error
    Recover(ctx) ([]*Task, error)
    Stats(ctx) (QueueStats, error)
    Close() error
    Name() string
}
Submitting tasks
payload, _ := queue.EncodePayload(MyJob{URL: "https://example.com"})

task := &queue.Task{
    ID:         idgen.ShortID(),
    Queue:      "my-tasks",
    Priority:   5,
    Payload:    payload,
    MaxRetries: 3,
}

_ = q.Enqueue(ctx, task)
Handler
func handler(ctx context.Context, task *queue.Task) error {
    job, err := queue.DecodePayload[MyJob](task)
    if err != nil {
        return err
    }
    return processJob(ctx, job)
}

Capacity-aware scheduler

For the scenario "limited computing power, large number of async jobs — queuing, resource preemption, priority-based scheduling", use CapacityScheduler:

scheduler, _ := queue.NewCapacityScheduler(queue.CapacitySchedulerConfig{
    Queue:            q,
    Handler:          myHandler,
    WorkerCount:      4,
    Capacity:         15,     // max total weight of running tasks
    Strategy:         queue.StrategyPreemptive,
    EnablePreemption: true,
    AgingThreshold:   30 * time.Second,
})
Features

Capacity limiting — each task has a Weight (resource cost). The total weight of concurrently running tasks cannot exceed Capacity. Tasks that would exceed capacity wait in the queue.

Job grouping — tasks with the same JobID run sequentially in submission order. Tasks with different JobIDs run concurrently. This models the common pattern of multi-step jobs where steps within a job are ordered but jobs are independent.

Priority scheduling — higher-priority tasks are dispatched first. Four strategies are available:

Strategy Description
StrategyFIFO First in, first out (ignores priority)
StrategyPriority Highest priority first (may starve low priority)
StrategyWeightedFair Fair sharing with aging (no starvation)
StrategyPreemptive Priority + preemption of running tasks

Aging — tasks that wait longer than AgingThreshold get a priority boost proportional to their wait time, preventing starvation of low-priority tasks.

Preemption — with StrategyPreemptive + EnablePreemption, when capacity is full and a high-priority task arrives, the scheduler preempts lower-priority running tasks (cooperative cancellation via context) and re-queues them. The preempted task's OnPreempt callback is invoked.

Task fields for capacity scheduling
task := &queue.Task{
    ID:           "task-1",
    JobID:        "job-42",      // same JobID = sequential
    Priority:     10,            // higher = more important
    Weight:       5,             // resource cost (capacity units)
    Preemptible:  true,          // can be preempted
    Payload:      payload,
    MaxRetries:   3,
}
Progress reporting & execution logging

The CapacityScheduler supports a RichHandler that receives a TaskContext for live progress updates and execution log recording:

scheduler, _ := queue.NewCapacityScheduler(queue.CapacitySchedulerConfig{
    Queue:       q,
    Capacity:    10,
    RichHandler: func(tctx queue.TaskContext, task *queue.Task) error {
        tctx.Log(queue.LogLevelInfo, "starting work")
        tctx.SetProgress(10)
        // ... do work, check tctx.Err() for cancellation ...
        tctx.SetProgress(100)
        tctx.Log(queue.LogLevelInfo, "done")
        return nil
    },
})

// Query a task's queue position (0 = next to dispatch).
pos, _ := scheduler.Position(ctx, "task-1")

// Retrieve execution logs (newest first).
logs, _ := scheduler.ListLogs(ctx, "task-1", 50)

Logs are stored in the backend:

  • memory: in-process slice per task
  • redis: Redis list (LPUSH/LRANGE)
Duplicate prevention

Both backends reject duplicate task IDs with queue.ErrDuplicateTask:

  • memory: checks an in-memory ID map
  • redis: uses atomic SETNX — if the task key already exists, the enqueue is rejected without overwriting
Restart recovery

On Start(), the scheduler calls Recover() on the backend:

  • memory: returns nothing (no persistence)
  • redis: returns all pending + running tasks (running tasks are reset to pending), which are re-ingested into the scheduler

License

MIT

Documentation

Overview

Package queue provides a durable, distributed task queue with pluggable backends (in-memory, Redis), priority scheduling, CPU-aware and goroutine-pool-aware dispatch, crash recovery, and consumer concurrency control.

Architecture

  • Task: a unit of work with ID, priority, payload, status, and metadata.
  • Queue: pluggable backend (memory, Redis) that stores and dispatches tasks.
  • Scheduler: manages a worker pool, pulls tasks from the queue, and executes them with optional CPU-based or goroutine-count-based throttling.

Backends

  • memory/ — in-process priority queue (single-node, fast, no persistence)
  • redis/ — Redis-backed distributed queue (multi-node, persistent)

Scheduling modes

  • GoroutinePool: fixed worker count (default)
  • CPUAdaptive: worker count scales with CPU usage (more workers when CPU is idle, fewer when busy)

Persistence & recovery

When a persistent backend (Redis or DB) is used, tasks survive process restarts. On startup, the scheduler recovers pending/running tasks and re-queues them.

Index

Constants

View Source 
const (
	LogLevelInfo  = "info"
	LogLevelWarn  = "warn"
	LogLevelError = "error"
)

LogLevel constants for TaskLogEntry.

Variables

View Source 
var (
	// ErrQueueEmpty is returned when no tasks are available.
	ErrQueueEmpty = errors.New("queue: empty")

	// ErrQueueClosed is returned when the queue is closed.
	ErrQueueClosed = errors.New("queue: closed")

	// ErrTaskNotFound is returned when a task ID does not exist.
	ErrTaskNotFound = errors.New("queue: task not found")

	// ErrDuplicateTask is returned when a task ID already exists.
	ErrDuplicateTask = errors.New("queue: duplicate task")
)

Sentinel errors.

Functions

func DecodePayload 

func DecodePayload[T any](task *Task) (T, error)

DecodePayload unmarshals Task.Payload into a value.

func EncodePayload 

func EncodePayload(v any) (json.RawMessage, error)

EncodePayload marshals a value to JSON for Task.Payload.

Types

type CapacityScheduler 

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

CapacityScheduler is a scheduler that enforces a total capacity limit on concurrently running tasks, supports job grouping (same JobID runs sequentially), priority-based scheduling, aging, and optional preemption.

It is designed for the scenario: "limited computing power, large number of async jobs — how to queue, preempt resources, and schedule by priority."

func NewCapacityScheduler 

func NewCapacityScheduler(cfg CapacitySchedulerConfig) (*CapacityScheduler, error)

NewCapacityScheduler creates a capacity-aware scheduler.

func (*CapacityScheduler) AppendLog 

func (s *CapacityScheduler) AppendLog(ctx context.Context, entry *TaskLogEntry) error

AppendLog appends an execution log entry for a task.

func (*CapacityScheduler) ListLogs 

func (s *CapacityScheduler) ListLogs(ctx context.Context, taskID string, limit int) ([]*TaskLogEntry, error)

ListLogs returns execution log entries for a task, newest first. limit of 0 means no limit (backend may apply its own cap).

func (*CapacityScheduler) Position 

func (s *CapacityScheduler) Position(ctx context.Context, taskID string) (int, error)

Position returns the queue position of a pending task (0 = next to dispatch). Returns -1 if the task is not pending. This checks the scheduler's internal pending list first, then falls back to the backend.

func (*CapacityScheduler) Start 

func (s *CapacityScheduler) Start() error

Start launches the scheduler. Recovers pending tasks from the backend, then starts worker goroutines and the dispatch loop.

func (*CapacityScheduler) Stats 

func (s *CapacityScheduler) Stats() CapacitySchedulerStats

Stats returns a snapshot of scheduler statistics.

func (*CapacityScheduler) Stop 

func (s *CapacityScheduler) Stop() error

Stop gracefully shuts down the scheduler.

func (*CapacityScheduler) UpdateProgress 

func (s *CapacityScheduler) UpdateProgress(ctx context.Context, taskID string, progress int) error

UpdateProgress updates the execution progress of a running task (0-100). This is typically called from inside a RichHandler via TaskContext, but can also be called externally.

type CapacitySchedulerConfig 

type CapacitySchedulerConfig struct {
	// Queue is the backend queue (memory or Redis).
	Queue Queue

	// Handler processes each task. If RichHandler is also set, RichHandler
	// takes precedence (it receives a TaskContext for progress + logging).
	Handler Handler

	// RichHandler is an optional handler that receives a TaskContext for
	// progress reporting and execution logging. If set, it takes precedence
	// over Handler.
	RichHandler RichHandler

	// Capacity is the maximum total weight of concurrently running tasks.
	// A task's Weight represents its resource cost. If Weight is 0, it
	// counts as 1. The scheduler will not dispatch a task if doing so
	// would exceed Capacity. Default: 0 (unlimited).
	Capacity int

	// WorkerCount is the maximum number of concurrent worker goroutines.
	// Default: runtime.NumCPU().
	WorkerCount int

	// Strategy is the scheduling strategy.
	// Default: StrategyPriority.
	Strategy SchedulingStrategy

	// AgingThreshold is the wait duration after which a task's priority
	// is boosted to prevent starvation. Default: 30s.
	// Set to 0 to disable aging.
	AgingThreshold time.Duration

	// DequeueTimeout is how long Dequeue blocks waiting for tasks.
	// Default: 500ms.
	DequeueTimeout time.Duration

	// EnablePreemption allows high-priority tasks to preempt
	// lower-priority running tasks when capacity is full.
	// Only effective with StrategyPreemptive. Default: false.
	EnablePreemption bool

	// WorkerPool is an optional external worker pool.
	WorkerPool *pool.WorkerPool

	// OnTaskStart is called when a task begins executing.
	OnTaskStart func(task *Task)

	// OnTaskComplete is called when a task finishes.
	OnTaskComplete func(task *Task, err error)

	// OnPreempt is called when a task is preempted.
	OnPreempt func(task *Task)

	// OnRecover is called after crash recovery.
	OnRecover func(count int)
}

CapacitySchedulerConfig configures the capacity-aware scheduler.

type CapacitySchedulerMetrics 

type CapacitySchedulerMetrics struct {
	Dispatched   atomic.Int64
	Succeeded    atomic.Int64
	Failed       atomic.Int64
	Retried      atomic.Int64
	Preempted    atomic.Int64
	Recovered    atomic.Int64
	CapacityFull atomic.Int64
}

CapacitySchedulerMetrics holds atomic counters.

type CapacitySchedulerStats 

type CapacitySchedulerStats struct {
	Pending      int    `json:"pending"`
	Running      int    `json:"running"`
	ActiveJobs   int    `json:"active_jobs"`
	UsedWeight   int    `json:"used_weight"`
	Capacity     int    `json:"capacity"`
	Workers      int    `json:"workers"`
	Strategy     string `json:"strategy"`
	Dispatched   int64  `json:"dispatched"`
	Succeeded    int64  `json:"succeeded"`
	Failed       int64  `json:"failed"`
	Retried      int64  `json:"retried"`
	Preempted    int64  `json:"preempted"`
	Recovered    int64  `json:"recovered"`
	CapacityFull int64  `json:"capacity_full"`
}

CapacitySchedulerStats is a point-in-time snapshot.

type Handler 

type Handler func(ctx context.Context, task *Task) error

Handler processes a task. The implementation is responsible for unmarshalling the payload and performing the work.

type Queue 

type Queue interface {
	// Enqueue adds a task to the queue. Returns ErrDuplicateTask if the
	// task ID already exists.
	Enqueue(ctx context.Context, task *Task) error

	// Dequeue removes and returns the highest-priority pending task.
	// Returns ErrQueueEmpty if no tasks are available. Blocks up to
	// timeout if non-zero.
	Dequeue(ctx context.Context, timeout time.Duration) (*Task, error)

	// Ack marks a task as completed (success or failure).
	Ack(ctx context.Context, taskID string, status TaskStatus, errMsg string) error

	// Requeue moves a task back to pending status.
	Requeue(ctx context.Context, taskID string) error

	// Get retrieves a task by ID.
	Get(ctx context.Context, taskID string) (*Task, error)

	// Cancel removes a pending task or marks a running task as canceled.
	Cancel(ctx context.Context, taskID string) error

	// Position returns the queue position of a pending task (0 = next to
	// dispatch). Returns -1 if the task is not in the pending queue.
	Position(ctx context.Context, taskID string) (int, error)

	// UpdateProgress updates the execution progress of a running task.
	// progress should be 0-100.
	UpdateProgress(ctx context.Context, taskID string, progress int) error

	// AppendLog appends an execution log entry for a task.
	AppendLog(ctx context.Context, entry *TaskLogEntry) error

	// ListLogs returns execution log entries for a task, newest first.
	// limit of 0 means no limit.
	ListLogs(ctx context.Context, taskID string, limit int) ([]*TaskLogEntry, error)

	// Recover returns all pending and interrupted (running) tasks for
	// restart recovery. Running tasks are reset to pending.
	Recover(ctx context.Context) ([]*Task, error)

	// Stats returns current queue statistics.
	Stats(ctx context.Context) (QueueStats, error)

	// Close releases all backend resources.
	Close() error

	// Name returns the queue name.
	Name() string
}

Queue is the backend interface for storing and dispatching tasks. Implementations must be safe for concurrent use.

type QueueStats 

type QueueStats struct {
	Queue     string `json:"queue"`
	Pending   int64  `json:"pending"`
	Running   int64  `json:"running"`
	Completed int64  `json:"completed"`
	Failed    int64  `json:"failed"`
	Total     int64  `json:"total"`
}

QueueStats is a snapshot of queue metrics.

type RichHandler 

type RichHandler func(tctx TaskContext, task *Task) error

RichHandler is a handler that receives a TaskContext for progress and logging. Schedulers that support rich handlers will call this instead of the plain Handler when configured.

type Scheduler 

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

Scheduler pulls tasks from a Queue and dispatches them to workers.

func NewScheduler 

func NewScheduler(cfg SchedulerConfig) (*Scheduler, error)

NewScheduler creates a scheduler with the given config.

func (*Scheduler) Start 

func (s *Scheduler) Start() error

Start launches worker goroutines and begins processing tasks. If a persistent backend is used, pending tasks are recovered first.

func (*Scheduler) Stats 

func (s *Scheduler) Stats() SchedulerStats

Stats returns current scheduler statistics.

func (*Scheduler) Stop 

func (s *Scheduler) Stop() error

Stop gracefully shuts down the scheduler, waiting for in-flight tasks.

type SchedulerConfig 

type SchedulerConfig struct {
	// Queue is the backend queue (memory or Redis).
	Queue Queue

	// Handler processes each task.
	Handler Handler

	// WorkerCount is the initial/fixed number of workers.
	// Default: runtime.NumCPU().
	WorkerCount int

	// Mode is the scheduling mode.
	// Default: ModeFixedPool.
	Mode SchedulingMode

	// MinWorkers is the minimum workers in CPUAdaptive mode.
	// Default: 1.
	MinWorkers int

	// MaxWorkers is the maximum workers in CPUAdaptive mode.
	// Default: WorkerCount * 4.
	MaxWorkers int

	// CPUHighThreshold is the CPU usage % above which workers are reduced.
	// Default: 80.
	CPUHighThreshold float64

	// CPULowThreshold is the CPU usage % below which workers are added.
	// Default: 30.
	CPULowThreshold float64

	// CPUCheckInterval is how often CPU usage is evaluated.
	// Default: 5s.
	CPUCheckInterval time.Duration

	// DequeueTimeout is how long Dequeue blocks waiting for tasks.
	// Default: 1s.
	DequeueTimeout time.Duration

	// WorkerPool is an optional external worker pool. If set, the
	// scheduler dispatches tasks to this pool instead of spawning its
	// own worker goroutines.
	WorkerPool *pool.WorkerPool

	// OnTaskStart is called when a task begins executing.
	OnTaskStart func(task *Task)

	// OnTaskComplete is called when a task finishes.
	OnTaskComplete func(task *Task, err error)

	// OnRecover is called after crash recovery with the number of recovered tasks.
	OnRecover func(count int)
}

SchedulerConfig configures the scheduler.

type SchedulerMetrics 

type SchedulerMetrics struct {
	Dequeued  atomic.Int64
	Succeeded atomic.Int64
	Failed    atomic.Int64
	Retried   atomic.Int64
	Recovered atomic.Int64
}

SchedulerMetrics holds atomic counters.

type SchedulerStats 

type SchedulerStats struct {
	Workers   int     `json:"workers"`
	Running   int     `json:"running"`
	Dequeued  int64   `json:"dequeued"`
	Succeeded int64   `json:"succeeded"`
	Failed    int64   `json:"failed"`
	Retried   int64   `json:"retried"`
	Recovered int64   `json:"recovered"`
	CPUUsage  float64 `json:"cpu_usage"`
}

SchedulerStats is a point-in-time snapshot.

type SchedulingMode 

type SchedulingMode int

SchedulingMode controls how the scheduler manages worker concurrency. 

const (
	// ModeFixedPool uses a fixed number of workers (default).
	ModeFixedPool SchedulingMode = iota
	// ModeCPUAdaptive scales workers based on CPU usage.
	ModeCPUAdaptive
)

type SchedulingStrategy 

type SchedulingStrategy int

SchedulingStrategy determines how pending tasks are ordered for dispatch. 

const (
	// StrategyFIFO dispatches tasks in submission order (oldest first).
	// Simple and fair, but ignores priority and weight.
	StrategyFIFO SchedulingStrategy = iota

	// StrategyPriority dispatches highest-priority tasks first.
	// Tasks with the same priority are ordered by submission time.
	// May cause starvation of low-priority tasks.
	StrategyPriority

	// StrategyWeightedFair dispatches tasks using a weighted fair queuing
	// algorithm. Each priority level gets a proportional share of
	// execution slots. Combined with aging to prevent starvation.
	StrategyWeightedFair

	// StrategyPreemptive is like StrategyPriority but also supports
	// preemption: when capacity is full and a high-priority task arrives,
	// a lower-priority running task is preempted (paused and re-queued).
	StrategyPreemptive
)

func (SchedulingStrategy) String 

func (s SchedulingStrategy) String() string

String returns a human-readable strategy name.

type Task 

type Task struct {
	ID         string          `json:"id"`
	Queue      string          `json:"queue"`
	Kind       string          `json:"kind,omitempty"`
	JobID      string          `json:"job_id,omitempty"`
	Priority   int             `json:"priority"`
	Weight     int             `json:"weight,omitempty"`
	Payload    json.RawMessage `json:"payload"`
	Status     TaskStatus      `json:"status"`
	Progress   int             `json:"progress"`
	RetryCount int             `json:"retry_count"`
	MaxRetries int             `json:"max_retries"`
	ErrorMsg   string          `json:"error_msg,omitempty"`
	SubmitTime time.Time       `json:"submit_time"`
	StartedAt  *time.Time      `json:"started_at,omitempty"`
	FinishedAt *time.Time      `json:"finished_at,omitempty"`
	WorkerID   string          `json:"worker_id,omitempty"`

	// Preemptible indicates whether this task can be preempted by a
	// higher-priority task when capacity is full. Default: true.
	Preemptible bool `json:"preemptible,omitempty"`
}

Task is a unit of work in the queue.

type TaskContext 

type TaskContext interface {
	context.Context
	Task() *Task
	SetProgress(progress int) error
	Log(level string, message string) error
}

TaskContext is passed to handlers when using a scheduler that supports progress reporting and execution logging. It wraps the base context and task, and provides methods to update progress and append log entries to the queue backend.

Usage: 

func myHandler(tctx queue.TaskContext, task *queue.Task) error {
    tctx.SetProgress(10)
    tctx.Log(queue.LogLevelInfo, "starting work")
    // ... do work ...
    tctx.SetProgress(100)
    return nil
}

type TaskLogEntry 

type TaskLogEntry struct {
	TaskID    string    `json:"task_id"`
	Level     string    `json:"level"` // "info", "warn", "error"
	Message   string    `json:"message"`
	Timestamp time.Time `json:"timestamp"`
}

TaskLogEntry is a single execution log line for a task.

type TaskResult 

type TaskResult struct {
	TaskID  string
	Status  TaskStatus
	Error   error
	Elapsed time.Duration
}

TaskResult holds the outcome of a task execution.

type TaskStatus 

type TaskStatus string

TaskStatus is the lifecycle state of a task. 

const (
	StatusPending  TaskStatus = "pending"
	StatusRunning  TaskStatus = "running"
	StatusSuccess  TaskStatus = "success"
	StatusFailed   TaskStatus = "failed"
	StatusCanceled TaskStatus = "canceled"
	StatusRetry    TaskStatus = "retry"
)

func (TaskStatus) IsTerminal 

func (s TaskStatus) IsTerminal() bool

IsTerminal reports whether the status is a final state.

func (TaskStatus) String 

func (s TaskStatus) String() string

String returns the status as a string.

Source Files

View all Source files

Directories

Path Synopsis
memory module

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.