README
¶
queue 
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.
Benchmarks and Example tests can be found in this package.
All queues implement the Queue interface
Installation
go get -u github.com/adrianbrad/queue
Blocking Queue
- FIFO Ordering
- Provides blocking and non-blocking methods. The non-blocking methods return an error.
- Implemented using
sync.Condfrom standard library.
Quick start
package main
import (
"fmt"
"github.com/adrianbrad/queue"
)
func main() {
elems := []int{2, 3}
priorityQueue := queue.NewBlocking(elems, queue.WithCapacity(3))
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: 2
}
Priority Queue
- Order based on the
lessmethod provided at construction. - Implemented using
container/heapstandard library package.
Quick Start
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
})
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 (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 Option
- type Priority
- type Queue
Examples ¶
Constants ¶
This section is empty.
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") )
Functions ¶
This section is empty.
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 ¶
NewBlocking returns a new Blocking Queue containing the given elements.
func (*Blocking[T]) Get ¶ added in v0.6.0
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 ¶ added in v0.8.0
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 ¶ added in v0.7.0
Offer inserts the element to the tail the queue. If the queue is full it returns the ErrQueueIsFull error.
func (*Blocking[T]) OfferWait ¶ added in v0.8.0
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 ¶ added in v0.3.0
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 ¶ added in v0.8.0
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.
type Option ¶ added in v0.6.0
type Option interface {
// contains filtered or unexported methods
}
An Option configures a Queue using the functional options paradigm.
func WithCapacity ¶ added in v0.6.0
WithCapacity specifies a fixed capacity for a queue.
type Priority ¶ added in v0.8.0
type Priority[T any] 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 ¶
fmt.Println("Ascending:")
elemsAsc := []int{2, 4, 1}
pAsc := queue.NewPriority(
elemsAsc,
func(elem, otherElem int) bool {
return elem < otherElem
},
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[int](pAsc))
fmt.Println(pAsc.Get())
fmt.Printf("\nDescending:\n")
elemsDesc := []int{2, 4, 1}
pDesc := queue.NewPriority(
elemsDesc,
func(elem, otherElem int) bool {
return elem > otherElem
},
queue.WithCapacity(4),
)
if err := pDesc.Offer(3); err != nil {
fmt.Printf("offer err: %s\n", err)
return
}
fmt.Println(drainQueue[int](pDesc))
Output: Ascending: queue is full [1 2 3 4] 0 no elements available in the queue Descending: [4 3 2 1]
func NewPriority ¶ added in v0.8.0
func NewPriority[T any]( 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]) Get ¶ added in v0.8.0
Get removes and returns the head of the queue. If no element is available it returns an ErrNoElementsAvailable error.
func (*Priority[T]) Offer ¶ added in v0.8.0
Offer inserts the element into the queue. If the queue is full it returns the ErrQueueIsFull error.
func (*Priority[T]) Peek ¶ added in v0.8.0
Peek retrieves but does not return the head of the queue.
type Queue ¶ added in v0.6.0
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.
Source Files