collections

Documentation

Index

• Constants
• Variables
• type HashedPriorityQueue
  • func NewHashedPriorityQueue[K comparable, T any](indexer IndexerFunc[K, T]) *HashedPriorityQueue[K, T]
  • func (q *HashedPriorityQueue[K, T]) Contains(id K) bool
  • func (q *HashedPriorityQueue[K, T]) Dequeue() *QueueItem[T]
  • func (q *HashedPriorityQueue[K, T]) DequeueWhere(matcher MatchingFunction[T]) *QueueItem[T]
  • func (q *HashedPriorityQueue[K, T]) Enqueue(data T, priority int64)
  • func (q *HashedPriorityQueue[K, T]) IsEmpty() bool
  • func (q *HashedPriorityQueue[K, T]) Len() int
  • func (q *HashedPriorityQueue[K, T]) Peek() *QueueItem[T]
• type IndexerFunc
• type MatchingFunction
• type Pair
  • func NewPair[L any, R any](left L, right R) Pair[L, R]
  • func (p Pair[L, R]) String() string
• type PriorityQueue
  • func NewPriorityQueue[T any]() *PriorityQueue[T]
  • func (pq *PriorityQueue[T]) Dequeue() *QueueItem[T]
  • func (pq *PriorityQueue[T]) DequeueWhere(matcher MatchingFunction[T]) *QueueItem[T]
  • func (pq *PriorityQueue[T]) Enqueue(data T, priority int64)
  • func (pq *PriorityQueue[T]) IsEmpty() bool
  • func (pq *PriorityQueue[T]) Len() int
  • func (pq *PriorityQueue[T]) Peek() *QueueItem[T]
• type PriorityQueueInterface
• type QueueItem
• type ScheduledTask
• type ScheduledTaskHeap
  • func NewScheduledTaskHeap[T any]() *ScheduledTaskHeap[T]
  • func (h *ScheduledTaskHeap[T]) Contains(task ScheduledTask[T]) bool
  • func (h *ScheduledTaskHeap[T]) Length() int
  • func (h *ScheduledTaskHeap[T]) Peek() ScheduledTask[T]
  • func (h *ScheduledTaskHeap[T]) Pop() ScheduledTask[T]
  • func (h *ScheduledTaskHeap[T]) Push(task ScheduledTask[T]) error
  • func (h *ScheduledTaskHeap[T]) Remove(task ScheduledTask[T])
  • func (h *ScheduledTaskHeap[T]) Update(task ScheduledTask[T]) error

Constants

const (
	InitialQueueCapacity = 64
)

Variables

var (
	ErrEmptyQueue error = errors.New("queue is empty")
	ErrNoMatch    error = errors.New("no items matched")
)

Types

type HashedPriorityQueue

type HashedPriorityQueue[K comparable, T any] struct {
	// contains filtered or unexported fields
}

HashedPriorityQueue is a priority queue that maintains only a single item per unique key. It combines the functionality of a hash map and a priority queue to provide efficient operations with the following key characteristics:

1. Single Item Per Key: The queue maintains only the latest version of an item for each unique key. When a new item with an existing key is enqueued, it replaces the old item instead of adding a duplicate.

2. Lazy Dequeuing: Outdated items (those that have been replaced by a newer version) are not immediately removed from the underlying queue. Instead, they are filtered out during dequeue operations. This approach improves enqueue performance at the cost of potentially slower dequeue operations.

3. Higher Enqueue Throughput: By avoiding immediate removal of outdated items, the HashedPriorityQueue achieves higher enqueue throughput. This makes it particularly suitable for scenarios with frequent updates to existing items.

4. Eventually Consistent: The queue becomes consistent over time as outdated items are lazily removed during dequeue operations. This means that the queue's length and the items it contains become accurate as items are dequeued.

5. Memory Consideration: Due to the lazy removal of outdated items, the underlying queue may temporarily hold more items than there are unique keys. This trade-off allows for better performance but may use more memory compared to a strictly consistent queue.

Use HashedPriorityQueue when you need a priority queue that efficiently handles updates to existing items and can tolerate some latency in removing outdated entries in favor of higher enqueue performance.

func NewHashedPriorityQueue

func NewHashedPriorityQueue[K comparable, T any](indexer IndexerFunc[K, T]) *HashedPriorityQueue[K, T]

NewHashedPriorityQueue creates a new PriorityQueue that allows us to check if specific items (indexed by a key field) are present in the queue. The provided IndexerFunc will be used on Enqueue/Dequeue to keep the index up to date.

func (*HashedPriorityQueue[K, T]) Contains

func (q *HashedPriorityQueue[K, T]) Contains(id K) bool

Contains will return true if the provided identifier (of type K) will be found in this queue, false if it is not present.

func (*HashedPriorityQueue[K, T]) Dequeue

func (q *HashedPriorityQueue[K, T]) Dequeue() *QueueItem[T]

Dequeue returns the next highest priority item, returning both the data Enqueued previously, and the priority with which it was enqueued. An err (ErrEmptyQueue) may be returned if the queue is currently empty.

func (*HashedPriorityQueue[K, T]) DequeueWhere

func (q *HashedPriorityQueue[K, T]) DequeueWhere(matcher MatchingFunction[T]) *QueueItem[T]

DequeueWhere allows the caller to iterate through the queue, in priority order, and attempt to match an item using the provided `MatchingFunction`. This method has a high time cost as dequeued but non-matching items must be held and requeued once the process is complete. Luckily, we use the same amount of space (bar a few bytes for the extra PriorityQueue) for the dequeued items.

func (*HashedPriorityQueue[K, T]) Enqueue

func (q *HashedPriorityQueue[K, T]) Enqueue(data T, priority int64)

Enqueue will add the item specified by `data` to the queue with the the priority given by `priority`.

func (*HashedPriorityQueue[K, T]) IsEmpty

func (q *HashedPriorityQueue[K, T]) IsEmpty() bool

IsEmpty returns a boolean denoting whether the queue is currently empty or not.

func (*HashedPriorityQueue[K, T]) Len

func (q *HashedPriorityQueue[K, T]) Len() int

Len returns the number of items currently in the queue

func (*HashedPriorityQueue[K, T]) Peek

func (q *HashedPriorityQueue[K, T]) Peek() *QueueItem[T]

Peek returns the next highest priority item without removing it from the queue. It returns nil if the queue is empty.

type IndexerFunc

type IndexerFunc[K comparable, T any] func(item T) K

IndexerFunc is used to find the key (of type K) from the provided item (T). This will be used for the item lookup in `Contains`

type MatchingFunction

type MatchingFunction[T any] func(possibleMatch T) bool

MatchingFunction can be used when 'iterating' the priority queue to find items with specific properties.

type Pair

type Pair[L any, R any] struct {
	Left  L
	Right R
}

Pair is a generic structure that holds two values of any type.

func NewPair

func NewPair[L any, R any](left L, right R) Pair[L, R]

NewPair creates a new Pair with the given values.

func (Pair[L, R]) String

func (p Pair[L, R]) String() string

String returns a string representation of the Pair.

type PriorityQueue

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

PriorityQueue contains items of type T, and allows you to enqueue and dequeue items with a specific priority. Items are dequeued in highest priority first order.

func NewPriorityQueue

func NewPriorityQueue[T any]() *PriorityQueue[T]

NewPriorityQueue creates a new ptr to a priority queue for type T.

func (*PriorityQueue[T]) Dequeue

func (pq *PriorityQueue[T]) Dequeue() *QueueItem[T]

Dequeue returns the next highest priority item, returning both the data Enqueued previously, and the priority with which it was enqueued. An err (ErrEmptyQueue) may be returned if the queue is currently empty.

func (*PriorityQueue[T]) DequeueWhere

func (pq *PriorityQueue[T]) DequeueWhere(matcher MatchingFunction[T]) *QueueItem[T]

DequeueWhere allows the caller to iterate through the queue, in priority order, and attempt to match an item using the provided `MatchingFunction`. This method has a high time cost as dequeued but non-matching items must be held and requeued once the process is complete. Luckily, we use the same amount of space (bar a few bytes for the extra PriorityQueue) for the dequeued items.

func (*PriorityQueue[T]) Enqueue

func (pq *PriorityQueue[T]) Enqueue(data T, priority int64)

Enqueue will add the item specified by `data` to the queue with the the priority given by `priority`.

func (*PriorityQueue[T]) IsEmpty

func (pq *PriorityQueue[T]) IsEmpty() bool

IsEmpty returns a boolean denoting whether the queue is currently empty or not.

func (*PriorityQueue[T]) Len

func (pq *PriorityQueue[T]) Len() int

Len returns the number of items currently in the queue

func (*PriorityQueue[T]) Peek

func (pq *PriorityQueue[T]) Peek() *QueueItem[T]

Peek returns the next highest priority item without removing it from the queue. It returns nil if the queue is empty.

type PriorityQueueInterface

type PriorityQueueInterface[T any] interface {
	// Enqueue will add the item specified by `data` to the queue with the
	// the priority given by `priority`.
	Enqueue(data T, priority int64)

	// Dequeue returns the next highest priority item, returning both
	// the data Enqueued previously, and the priority with which it was
	// enqueued. An err (ErrEmptyQueue) may be returned if the queue is
	// currently empty.
	Dequeue() *QueueItem[T]

	// DequeueWhere allows the caller to iterate through the queue, in priority order, and
	// attempt to match an item using the provided `MatchingFunction`.  This method has a high
	// time cost as dequeued but non-matching items must be held and requeued once the process
	// is complete.  Luckily, we use the same amount of space (bar a few bytes for the
	// extra PriorityQueue) for the dequeued items.
	DequeueWhere(matcher MatchingFunction[T]) *QueueItem[T]

	// Peek returns the next highest priority item without removing it from the queue.
	// It returns nil if the queue is empty.
	Peek() *QueueItem[T]

	// Len returns the number of items currently in the queue
	Len() int

	// IsEmpty returns a boolean denoting whether the queue is
	// currently empty or not.
	IsEmpty() bool
}

type QueueItem

type QueueItem[T any] struct {
	Value    T
	Priority int64
}

QueueItem encapsulates an item in the queue when we return it from the various dequeue methods

type ScheduledTask

type ScheduledTask[T any] interface {
	Data() T              // The data object
	ID() string           // ID of the object
	WaitUntil() time.Time // Time to wait until
}

ScheduledTask is an interface type implemented by objects stored in the ScheduledTaskHeap

type ScheduledTaskHeap

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

ScheduledTaskHeap wraps a heap and provides deduplication and operations other than Push/Pop. The heap elements are sorted by the time in the WaitUntil field of scheduledHeapNode

func NewScheduledTaskHeap

func NewScheduledTaskHeap[T any]() *ScheduledTaskHeap[T]

func (*ScheduledTaskHeap[T]) Contains

func (h *ScheduledTaskHeap[T]) Contains(task ScheduledTask[T]) bool

func (*ScheduledTaskHeap[T]) Length

func (h *ScheduledTaskHeap[T]) Length() int

func (*ScheduledTaskHeap[T]) Peek

func (h *ScheduledTaskHeap[T]) Peek() ScheduledTask[T]

func (*ScheduledTaskHeap[T]) Pop

func (h *ScheduledTaskHeap[T]) Pop() ScheduledTask[T]

func (*ScheduledTaskHeap[T]) Push

func (h *ScheduledTaskHeap[T]) Push(task ScheduledTask[T]) error

func (*ScheduledTaskHeap[T]) Remove

func (h *ScheduledTaskHeap[T]) Remove(task ScheduledTask[T])

func (*ScheduledTaskHeap[T]) Update

func (h *ScheduledTaskHeap[T]) Update(task ScheduledTask[T]) error