README
¶
queue
Package queue implements a mutex-free channel-based generic queue. It is not designed for raw speed, but rather to be used as a data source for things like worker pools where reading input from a channel makes sense.
The queue is backed by a ring buffer with a default initial capacity of 256 elements that grows as needed. A constructor method is provided to set the initial capacity if desired.
Where locking is required, it is done at the ring buffer level using spinlocks to minimize contention.
Features
- Dependency free
- Mutex free
- Concurrent
- Generic
- Channel based implementation
- Backed by a ringbuffer for reduced allocs
- Used by my workerpool implementation
Example
package main
import (
"fmt"
queue "github.com/fireflycons/concurrentqueue"
)
func main() {
q := queue.New[int]()
for i := range 10 {
_ = q.Enqueue(i)
}
// Close the queue to signal no more elements will be added
q.Close()
// Dequeue is a channel, therefore it would block until elements are available
// or the queue is closed and emptied.
for v := range q.Dequeue() {
fmt.Println(v)
}
}
Documentation
¶
Overview ¶
Package queue implements a mutex-free channel-based generic queue. It is not designed for raw speed, but rather to be used as a data source for things like worker pools where reading input from a channel makes sense.
The queue is backed by a ring buffer with a default initial capacity of 256 elements that grows as needed. A constructor method is provided to set the initial capacity if desired.
Where locking is required, it is done at the ring buffer level using spinlocks to minimize contention.
Index ¶
Constants ¶
This section is empty.
Variables ¶
var ErrQueueClosed = errors.New("queue is closed")
ErrQueueClosed is returned by Enqueue if the Close method has been called.
Functions ¶
This section is empty.
Types ¶
type Queue ¶
type Queue[T any] struct { // contains filtered or unexported fields }
Queue is a concurrent queue that supports multiple producers and consumers.
func New ¶
func New[T any](opts ...QueueOptFunc) *Queue[T]
New creates a new queue. You can provide optional configuration functions to customize the queue's behavior.
func (*Queue[T]) Close ¶
func (q *Queue[T]) Close()
Close closes the queue's channels. Items can be dequeued until the queue is empty at which point reading the dequeue channel with v,ok returns false
func (*Queue[T]) Dequeue ¶
func (q *Queue[T]) Dequeue() <-chan T
Dequeue returns the queue's read channel. Reading the channel will block if the queue is empty, or return false on v,ok if the queue has been closed.
func (*Queue[T]) Drain ¶
func (q *Queue[T]) Drain()
Drain empties the queue of any remaining elements and releases resources.
You should call this if you do not expect your consumers to empty the queue themselves, else a goroutine and some memory will be leaked.
If called before Close, Drain will have no effect.
func (*Queue[T]) DrainTo ¶
DrainTo appends all remaining elements in this queue to another queue and closes this queue.
You should call this if you do not expect your consumers to empty the queue themselves, else a goroutine and some memory will be leaked.
If called before Close, DrainTo will have no effect.
type QueueOptFunc ¶
type QueueOptFunc func(*qopts)
QueueOptFunc is a function that configures a Queue.
func WithInitialCapacity ¶
func WithInitialCapacity(capacity int) QueueOptFunc
WithInitialCapacity is a constructor function to set the initial capacity of the queue's internal ring buffer.
Source Files
Directories