queue

README

queue

---

Queue is a Go package that provides multiple thread-safe generic queue implementations.

A queue is a sequence of entities that is open at both ends where he elements are added(enqueued) to the tail(back) of the queue and removed(dequeued) from the head(front) of the queue.

Notable characteristics

• Usable zero values.
• The queues can be reset to their initial state.
• Implementations satisfy the Queue interface.
• Example tests are provided.

Blocking Queue

• FIFO Ordering
• Waits for the queue have elements available before retrieving from it.
• Implemented using sync.Cond from standard library.

Priority Queue

• Order based on the Less method implemented by elements.
• Waits for the queue have elements available before retrieving from it.
• Implemented using container/heap standard library package.

Documentation

Overview

Package queue provides multiple thread-safe generic queue implementations. Currently, there are 2 available implementations:

A blocking queue, which provides methods that wait for the queue to have available elements when attempting to retrieve an element, and waits for a free slot when attempting to insert an element.

A priority queue based on a container.Heap. The elements in the queue must implement the Lesser interface, and are ordered based on the Less method. The head of the queue is always the highest priority element.

Index

• Variables
• type Blocking
  • func NewBlocking[T any](elems []T, opts ...Option) *Blocking[T]
  • func (bq *Blocking[T]) Get() (v T, _ error)
  • func (bq *Blocking[T]) GetWait() (v T)
  • func (bq *Blocking[T]) Offer(elem T) error
  • func (bq *Blocking[T]) OfferWait(elem T)
  • func (bq *Blocking[T]) Peek() (v T, _ error)
  • func (bq *Blocking[T]) PeekWait() T
  • func (bq *Blocking[T]) Reset()
  • func (bq *Blocking[T]) Size() int
• type Lesser
• type Option
  • func WithCapacity(capacity int) Option
• type Priority
  • func NewPriority[T Lesser](elems []T, opts ...Option) *Priority[T]
  • func (pq *Priority[T]) Get() (elem T, _ error)
  • func (pq *Priority[T]) Offer(elem T) error
  • func (pq *Priority[T]) Peek() (elem T, _ error)
  • func (pq *Priority[T]) Reset()
  • func (pq *Priority[T]) Size() int
• type Queue

Examples

• Blocking
• Priority

Variables

var (
	// ErrNoElementsAvailable is an error returned whenever there are no
	// elements available to be extracted from a queue.
	ErrNoElementsAvailable = errors.New("no elements available in the queue")
	// ErrQueueIsFull is an error returned whenever the queue is full and there
	// is an attempt to add an element to it.
	ErrQueueIsFull = errors.New("queue is full")
)

Types

type Blocking

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

Blocking is a Queue implementation that additionally supports operations that wait for the queue to have available items, and wait for a slot to become available in case the queue is full. ! The Blocking Queue shares most functionality with channels. If you do not make use of Peek or Reset methods you are safe to use channels instead.

It supports operations for retrieving and adding elements to a FIFO queue. If there are no elements available the retrieve operations wait until elements are added to the queue.

Example

elems := []int{1, 2, 3}

blockingQueue := queue.NewBlocking(elems, queue.WithCapacity(4))

fmt.Println(drainQueue[int](blockingQueue))

var (
	elem  int
	after time.Duration
)

done := make(chan struct{})

start := time.Now()

// this function waits for a new element to be available in the queue.
go func() {
	defer close(done)

	elem = blockingQueue.GetWait()
	after = time.Since(start)
}()

time.Sleep(time.Millisecond)

// insert a new element into the queue.
fmt.Println("Offer err:", blockingQueue.Offer(4))

<-done

fmt.Printf(
	"Elem %d received after %s",
	elem,
	after.Round(time.Millisecond),
)

Output:

[1 2 3]
Offer err: <nil>
Elem 4 received after 1ms

func NewBlocking

func NewBlocking[T any](
	elems []T,
	opts ...Option,
) *Blocking[T]

NewBlocking returns a new Blocking Queue containing the given elements.

func (*Blocking[T]) Get

func (bq *Blocking[T]) Get() (v T, _ error)

Get removes and returns the head of the elements queue. If no element is available it returns an ErrNoElementsAvailable error.

It does not actually remove elements from the elements slice, but it's incrementing the underlying index.

func (*Blocking[T]) GetWait

func (bq *Blocking[T]) GetWait() (v T)

GetWait removes and returns the head of the elements queue. If no element is available it waits until the queue has an element available.

It does not actually remove elements from the elements slice, but it's incrementing the underlying index.

func (*Blocking[T]) Offer

func (bq *Blocking[T]) Offer(elem T) error

Offer inserts the element to the tail the queue. If the queue is full it returns the ErrQueueIsFull error.

func (*Blocking[T]) OfferWait

func (bq *Blocking[T]) OfferWait(elem T)

OfferWait inserts the element to the tail the queue. It waits for necessary space to become available.

func (*Blocking[T]) Peek

func (bq *Blocking[T]) Peek() (v T, _ error)

Peek retrieves but does not return the head of the queue. If no element is available it returns an ErrNoElementsAvailable error.

func (*Blocking[T]) PeekWait

func (bq *Blocking[T]) PeekWait() T

PeekWait retrieves but does not return the head of the queue. If no element is available it waits until the queue has an element available.

func (*Blocking[T]) Reset

func (bq *Blocking[T]) Reset()

Reset sets the queue elements index to 0. The queue will be in its initial state.

func (*Blocking[T]) Size

func (bq *Blocking[T]) Size() int

Size returns the number of elements in the queue.

type Lesser

type Lesser interface {
	// Less compares the caller to the other
	Less(other any) bool
}

Lesser is the interface that wraps basic Less method.

type Option

type Option interface {
	// contains filtered or unexported methods
}

An Option configures a Queue using the functional options paradigm.

func WithCapacity

func WithCapacity(capacity int) Option

WithCapacity specifies a fixed capacity for a queue.

type Priority

type Priority[T Lesser] struct {
	// contains filtered or unexported fields
}

Priority is a Queue implementation. ! The elements must implement the Lesser interface.

The ordering is given by the Lesser.Less method implementation. The head of the queue is always the highest priority element.

! If capacity is provided and is less than the number of elements provided, the elements slice is sorted and trimmed to fit the capacity.

For ordered types (types that support the operators < <= >= >), the order can be defined by using the following operators: > - for ascending order < - for descending order.

Example

package main

import (
	"fmt"

	"github.com/adrianbrad/queue"
)

var _ queue.Lesser = (*intValAscending)(nil)

type intValAscending int

func (i intValAscending) Less(other any) bool {
	return i < other.(intValAscending)
}

var _ queue.Lesser = (*intValDescending)(nil)

type intValDescending int

func (i intValDescending) Less(other any) bool {
	return i > other.(intValDescending)
}

func main() {
	fmt.Println("Ascending:")

	elemsAsc := []intValAscending{2, 4, 1}

	pAsc := queue.NewPriority(elemsAsc, queue.WithCapacity(4))

	if err := pAsc.Offer(3); err != nil {
		fmt.Printf("offer err: %s\n", err)
		return
	}

	fmt.Println(pAsc.Offer(5))
	fmt.Println(drainQueue[intValAscending](pAsc))
	fmt.Println(pAsc.Get())

	fmt.Printf("\nDescending:\n")

	elemsDesc := []intValDescending{2, 4, 1}

	pDesc := queue.NewPriority(elemsDesc, queue.WithCapacity(4))

	if err := pDesc.Offer(3); err != nil {
		fmt.Printf("offer err: %s\n", err)
		return
	}

	fmt.Println(drainQueue[intValDescending](pDesc))

}

func drainQueue[T any](q queue.Queue[T]) []T {
	size := q.Size()

	elems := make([]T, size)

	var err error

	for i := 0; i < size; i++ {
		elems[i], err = q.Get()
		if err != nil {
			return nil
		}
	}

	return elems
}

Output:

Ascending:
queue is full
[1 2 3 4]
0 no elements available in the queue

Descending:
[4 3 2 1]

func NewPriority

func NewPriority[T Lesser](
	elems []T,
	opts ...Option,
) *Priority[T]

NewPriority returns a new Priority Queue containing the given elements.

func (*Priority[T]) Get

func (pq *Priority[T]) Get() (elem T, _ error)

Get removes and returns the head of the queue. If no element is available it returns an ErrNoElementsAvailable error.

func (*Priority[T]) Offer

func (pq *Priority[T]) Offer(elem T) error

Offer inserts the element into the queue. If the queue is full it returns the ErrQueueIsFull error.

func (*Priority[T]) Peek

func (pq *Priority[T]) Peek() (elem T, _ error)

Peek retrieves but does not return the head of the queue.

func (*Priority[T]) Reset

func (pq *Priority[T]) Reset()

Reset sets the queue to its initial stat, by replacing the current elements with the elements provided at creation.

func (*Priority[T]) Size

func (pq *Priority[T]) Size() int

Size returns the number of elements in the queue.

type Queue

type Queue[T any] interface {
	// Peek retrieves but does not remove the head of the queue.
	Peek() (T, error)

	// Size returns the number of elements in the queue.
	Size() int

	// Get retrieves and removes the head of the queue.
	Get() (T, error)

	// Offer inserts the element to the tail of the queue.
	Offer(T) error

	// Reset sets the queue to its initial state.
	Reset()
}

Queue is a collection that orders elements in a FIFO order. This interface provides basic methods for adding and extracting elements from the queue. Items are extracted from the head of the queue and added to the tail of the queue.