queue

Documentation

Overview

Package queue provides several implementations of the amboy.Queue interface capable of processing amboy.Job implementations.

Local Shuffled Queue

The shuffled queue is functionally similar to the LocalUnordered Queue (which is, in fact, a FIFO queue as a result of its implementation); however, the shuffled queue dispatches tasks randomized, using the properties of Go's map type, which is not dependent on insertion order.

Additionally this implementation does not using locking, which may improve performance for some workloads. Intentionally, the implementation retains pointers to all completed tasks, and does not cap the number of pending tasks.

Index

• func NewAdaptiveOrderedLocalQueue(opts *FixedSizeQueueOptions) amboy.Queue
• func NewLocalLimitedSize(opts *FixedSizeQueueOptions) amboy.Queue
• func NewLocalOrdered(workers int) amboy.Queue
• func NewLocalPriorityQueue(opts *FixedSizeQueueOptions) amboy.Queue
• func NewLocalQueueGroup(ctx context.Context, opts LocalQueueGroupOptions) (amboy.QueueGroup, error)
• func NewShuffledLocal(opts *FixedSizeQueueOptions) amboy.Queue
• type Dispatcher
  • func NewDispatcher(q amboy.Queue, log grip.Journaler) Dispatcher
• type FixedSizeQueueOptions
• type GroupCache
  • func NewCacheWithCleanupHook(ttl time.Duration, hook func(ctx context.Context, id string) error) GroupCache
  • func NewGroupCache(ttl time.Duration) GroupCache
• type LocalQueueGroupOptions
• type ScopeManager
  • func NewLocalScopeManager() ScopeManager

Functions

func NewAdaptiveOrderedLocalQueue

func NewAdaptiveOrderedLocalQueue(opts *FixedSizeQueueOptions) amboy.Queue

NewAdaptiveOrderedLocalQueue provides a queue implementation that stores jobs in memory, and dispatches tasks based on the dependency information.

Use this implementation rather than LocalOrderedQueue when you need to add jobs *after* starting the queue, and when you want to avoid the higher potential overhead of the remote-backed queues.

Like other ordered in memory queues, this implementation does not support scoped locks.

func NewLocalLimitedSize

func NewLocalLimitedSize(opts *FixedSizeQueueOptions) amboy.Queue

NewLocalLimitedSize constructs a LocalLimitedSize queue instance with the specified number of workers and capacity.

func NewLocalOrdered

func NewLocalOrdered(workers int) amboy.Queue

NewLocalOrdered constructs an LocalOrdered object. The "workers" argument is passed to a default pool.SimplePool object.

The ordered queue requires that users add all tasks to the queue before starting it, and does not accept tasks after starting.

Like other ordered in memory queues, this implementation does not support scoped locks.

func NewLocalPriorityQueue

func NewLocalPriorityQueue(opts *FixedSizeQueueOptions) amboy.Queue

NewLocalPriorityQueue constructs a new priority queue instance and initializes a local worker queue with the specified number of worker processes.

func NewLocalQueueGroup

func NewLocalQueueGroup(ctx context.Context, opts LocalQueueGroupOptions) (amboy.QueueGroup, error)

NewLocalQueueGroup constructs a new local queue group. If ttl is 0, the queues will not be TTLed except when the client explicitly calls Prune.

func NewShuffledLocal

func NewShuffledLocal(opts *FixedSizeQueueOptions) amboy.Queue

NewShuffledLocal provides a queue implementation that shuffles the order of jobs, relative the insertion order.

Types

type Dispatcher

type Dispatcher interface {
	Dispatch(context.Context, amboy.Job) error
	Release(context.Context, amboy.Job)
	Complete(context.Context, amboy.Job) error
	Close(context.Context) error
}

Dispatcher provides a common mechanism shared between queue implementations to handle job locking to prevent multiple workers from running the same job.

func NewDispatcher

func NewDispatcher(q amboy.Queue, log grip.Journaler) Dispatcher

NewDispatcher constructs a default dispatching implementation.

type FixedSizeQueueOptions

type FixedSizeQueueOptions struct {
	Workers  int
	Capacity int
	Logger   grip.Journaler
}

type GroupCache

type GroupCache interface {
	Set(string, amboy.Queue, time.Duration) error
	Get(string) amboy.Queue
	Remove(context.Context, string) error
	Prune(context.Context) error
	Close(context.Context) error
	Names() []string
	Len() int
}

GroupCache provides a common mechanism for managing collections of queues, for use in specific group cache situations

func NewCacheWithCleanupHook

func NewCacheWithCleanupHook(ttl time.Duration, hook func(ctx context.Context, id string) error) GroupCache

NewCacheWithCleanupHook defines a cache but allows implementations to add additional cleanup logic to the prune and Close operations.

func NewGroupCache

func NewGroupCache(ttl time.Duration) GroupCache

NewGroupCache produces a GroupCache implementation that supports a default TTL setting, and supports cloning and closing operations.

type LocalQueueGroupOptions

type LocalQueueGroupOptions struct {
	Constructor func(ctx context.Context) (amboy.Queue, error)
	TTL         time.Duration
	Logger      grip.Journaler
}

LocalQueueGroupOptions describe options passed to NewLocalQueueGroup.

type ScopeManager

type ScopeManager interface {
	Acquire(string, []string) error
	Release(string, []string) error
}

ScopeManager provides a service to queue implementation to support additional locking semantics for queues that cannot push that into their backing storage.

func NewLocalScopeManager

func NewLocalScopeManager() ScopeManager

NewLocalScopeManager constructs a ScopeManager implementation suitable for use in most local (in memory) queue implementations.