queue

README

queue

---

Overview

The queue package provides multiple thread-safe generic queue implementations in Go.

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

Queues implemented in this package are designed to be easy to use and provide a consistent API.

Benchmarks and Example tests can be found in this package.

Features

The queue package provides two types of queues:

• Blocking Queue: FIFO Ordering, provides blocking and non-blocking methods. The non-blocking methods return an error. Implemented using sync.Cond from the standard library.

• Priority Queue: Order based on the less function provided at construction. Implemented using container/heap standard library package.

Installation

To add this package as a dependency to your project, run:

go get -u github.com/adrianbrad/queue

Import

To use this package in your project, you can import it as follows:

import "github.com/adrianbrad/queue"

Usage

Queue Interface

// Queue is a generic queue interface, defining the methods that all queues must implement.
type Queue[T comparable] interface {
	Get() (T, error)
	Offer(T) error
	Reset()
	Peek() (T, error)
	Size() int
	IsEmpty() bool
	Iterator() <-chan T
	Clear() []T
}

Quick Start

Blocking Queue

package main

import (
	"fmt"

	"github.com/adrianbrad/queue"
)

func main() {
	elems := []int{2, 3}

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

	containsTwo := blockingQueue.Contains(2)
	fmt.Println(containsTwo) // true

	size := blockingQueue.Size()
	fmt.Println(size) // 2

	empty := blockingQueue.IsEmpty()
	fmt.Println(empty) // false

	if err := blockingQueue.Offer(1); err != nil {
		// handle err
	}

	elem, err := blockingQueue.Get()
	if err != nil {
		// handle err
	}

	fmt.Println("elem: ", elem) // elem: 2
}

Priority Queue

package main

import (
	"fmt"

	"github.com/adrianbrad/queue"
)

func main() {
	elems := []int{2, 3, 4}

	priorityQueue := queue.NewPriority(
		elems, 
		func(elem, otherElem int) bool { return elem < otherElem },
    )

	containsTwo := priorityQueue.Contains(2)
	fmt.Println(containsTwo) // true

	size := priorityQueue.Size()
	fmt.Println(size) // 3

	empty := priorityQueue.IsEmpty()
	fmt.Println(empty) // false

	if err := priorityQueue.Offer(1); err != nil {
		// handle err
	}

	elem, err := priorityQueue.Get()
	if err != nil {
		// handle err
	}

	fmt.Printf("elem: %d\n", elem) // elem: 1
}

Benchmarks

Results as of 3rd of February 2023.

BenchmarkBlockingQueue/Peek-12          63275360                19.44 ns/op            0 B/op          0 allocs/op
BenchmarkBlockingQueue/Get_Offer-12     19066974                69.67 ns/op           40 B/op          0 allocs/op
BenchmarkBlockingQueue/Offer-12         36569245                37.86 ns/op           41 B/op          0 allocs/op
BenchmarkPriorityQueue/Peek-12          66765319                15.86 ns/op            0 B/op          0 allocs/op
BenchmarkPriorityQueue/Get_Offer-12     16677442                71.33 ns/op            0 B/op          0 allocs/op
BenchmarkPriorityQueue/Offer-12         20044909                58.58 ns/op           55 B/op          0 allocs/op

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 comparable](elems []T, opts ...Option) *Blocking[T]
  • func (bq *Blocking[T]) Clear() []T
  • func (bq *Blocking[T]) Contains(elem T) bool
  • func (bq *Blocking[T]) Get() (v T, _ error)
  • func (bq *Blocking[T]) GetWait() (v T)
  • func (bq *Blocking[T]) IsEmpty() bool
  • func (bq *Blocking[T]) Iterator() <-chan 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 Option
  • func WithCapacity(capacity int) Option
• type Priority
  • func NewPriority[T comparable](elems []T, lessFunc func(elem, otherElem T) bool, opts ...Option) *Priority[T]
  • func (pq *Priority[T]) Clear() []T
  • func (pq *Priority[T]) Contains(a T) bool
  • func (pq *Priority[T]) Get() (elem T, _ error)
  • func (pq *Priority[T]) IsEmpty() bool
  • func (pq *Priority[T]) Iterator() <-chan T
  • 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 comparable] 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, Reset and Contains 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

package main

import (
	"fmt"
	"time"

	"github.com/adrianbrad/queue"
)

func main() {
	elems := []int{1, 2, 3}

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

	containsThree := blockingQueue.Contains(3)
	fmt.Println("Contains 3:", containsThree)

	size := blockingQueue.Size()
	fmt.Println("Size:", size)

	empty := blockingQueue.IsEmpty()
	fmt.Println("Empty before clear:", empty)

	clearElems := blockingQueue.Clear()
	fmt.Println("Clear:", clearElems)

	empty = blockingQueue.IsEmpty()
	fmt.Println("Empty after clear:", empty)

	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.
	if err := blockingQueue.Offer(4); err != nil {
		fmt.Println("Offer err:", err)
		return
	}

	nextElem, err := blockingQueue.Peek()
	if err != nil {
		fmt.Println("Peek err:", err)
		return
	}

	fmt.Println("Peeked elem:", nextElem)

	<-done

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

}

Output:

Contains 3: true
Size: 3
Empty before clear: false
Clear: [1 2 3]
Empty after clear: true
Peeked elem: 4
Elem 4 received after 1ms

func NewBlocking

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

NewBlocking returns a new Blocking Queue containing the given elements.

func (*Blocking[T]) Clear

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

Clear removes and returns all elements from the queue.

func (*Blocking[T]) Contains

func (bq *Blocking[T]) Contains(elem T) bool

Contains returns true if the queue contains the given element.

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]) IsEmpty

func (bq *Blocking[T]) IsEmpty() bool

IsEmpty returns true if the queue is empty.

func (*Blocking[T]) Iterator

func (bq *Blocking[T]) Iterator() <-chan T

Iterator returns an iterator over the elements in this queue. Iterator returns an iterator over the elements in the queue. It removes the elements from the queue.

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 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 comparable] 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"
)

func main() {
	elems := []int{2, 4, 1}

	priorityQueue := queue.NewPriority(
		elems,
		func(elem, otherElem int) bool {
			return elem < otherElem
		},
		queue.WithCapacity(4),
	)

	containsTwo := priorityQueue.Contains(2)
	fmt.Println("Contains 3:", containsTwo)

	size := priorityQueue.Size()
	fmt.Println("Size:", size)

	if err := priorityQueue.Offer(3); err != nil {
		fmt.Println("Offer err: ", err)
		return
	}

	empty := priorityQueue.IsEmpty()
	fmt.Println("Empty before clear:", empty)

	clearElems := priorityQueue.Clear()
	fmt.Println("Clear:", clearElems)

	empty = priorityQueue.IsEmpty()
	fmt.Println("Empty after clear:", empty)

	if err := priorityQueue.Offer(5); err != nil {
		fmt.Println("Offer err: ", err)
		return
	}

	elem, err := priorityQueue.Get()
	if err != nil {
		fmt.Println("Get err: ", err)
		return
	}

	fmt.Println("Get:", elem)

}

Output:

Contains 3: true
Size: 3
Empty before clear: false
Clear: [1 2 3 4]
Empty after clear: true
Get: 5

func NewPriority

func NewPriority[T comparable](
	elems []T,
	lessFunc func(elem, otherElem T) bool,
	opts ...Option,
) *Priority[T]

NewPriority creates a new Priority Queue containing the given elements. It panics if lessFunc is nil.

func (*Priority[T]) Clear

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

Clear removes all elements from the queue.

func (*Priority[T]) Contains

func (pq *Priority[T]) Contains(a T) bool

Contains returns true if the queue contains the element, false otherwise.

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]) IsEmpty

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

IsEmpty returns true if the queue is empty, false otherwise.

func (*Priority[T]) Iterator

func (pq *Priority[T]) Iterator() <-chan T

Iterator returns an iterator over the elements in the queue. It removes the elements from the queue.

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 comparable] interface {
	// 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()

	// Contains returns true if the queue contains the element.
	Contains(T) bool

	// 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

	// IsEmpty returns true if the queue is empty.
	IsEmpty() bool

	// Iterator returns a channel that will be filled with the elements
	Iterator() <-chan T

	// Clear removes all elements from the queue.
	Clear() []T
}

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.