queue

Documentation

Index

• Variables
• type Circular
  • func NewCircular[T any, P Pointer[T]](maxSize uint64) *Circular[T, P]
  • func (q *Circular[T, P]) Close()
  • func (q *Circular[T, P]) Drain() (values []P)
  • func (q *Circular[T, P]) IsClosed() (closed bool)
  • func (q *Circular[T, P]) IsEmpty() (empty bool)
  • func (q *Circular[T, P]) IsFull() (full bool)
  • func (q *Circular[T, P]) Length() (size int)
  • func (q *Circular[T, P]) Pop() (p P, err error)
  • func (q *Circular[T, P]) Push(p P) error
• type LockFree
  • func NewLockFree[T any, P Pointer[T]](size uint64) *LockFree[T, P]
  • func (q *LockFree[T, P]) Close()
  • func (q *LockFree[T, P]) Drain() []P
  • func (q *LockFree[T, P]) IsClosed() bool
  • func (q *LockFree[T, P]) Length() int
  • func (q *LockFree[T, P]) Pop() (P, error)
  • func (q *LockFree[T, P]) Push(item P) error
• type NonBlocking
  • func NewNonBlocking[T any, P Pointer[T]](maxSize uint64) *NonBlocking[T, P]
  • func (q *NonBlocking[T, P]) Close()
  • func (q *NonBlocking[T, P]) Drain() (values []P)
  • func (q *NonBlocking[T, P]) IsClosed() (closed bool)
  • func (q *NonBlocking[T, P]) IsEmpty() (empty bool)
  • func (q *NonBlocking[T, P]) IsFull() (full bool)
  • func (q *NonBlocking[T, P]) Length() (size int)
  • func (q *NonBlocking[T, P]) Pop() (p P, err error)
  • func (q *NonBlocking[T, P]) Push(p P) error
• type Pointer

Variables

var (
	Closed     = errors.New("queue is closed")
	FullError  = errors.New("queue is full")
	EmptyError = errors.New("queue is empty")
)

Types

type Circular

type Circular[T any, P Pointer[T]] struct {
	// contains filtered or unexported fields
}

Circular is a circular sized FIFO queue that uses an array of fixed size to store the elements.

It is thread safe and extremely performant, however it is a blocking queue and will block the caller if the queue is full or if it is empty.

func NewCircular

func NewCircular[T any, P Pointer[T]](maxSize uint64) *Circular[T, P]

NewCircular creates a new circular queue with the given size.

func (*Circular[T, P]) Close

func (q *Circular[T, P]) Close()

Close closes the queue permanently.

The Drain method can be used to drain the queue after it is closed.

func (*Circular[T, P]) Drain

func (q *Circular[T, P]) Drain() (values []P)

Drain removes all elements from the queue. and returns them in a slice.

This function should only be called after the queue is closed.

func (*Circular[T, P]) IsClosed

func (q *Circular[T, P]) IsClosed() (closed bool)

IsClosed returns true if the queue is Closed

The Drain method can be used to drain the queue after it is closed.

func (*Circular[T, P]) IsEmpty

func (q *Circular[T, P]) IsEmpty() (empty bool)

IsEmpty returns true if the queue is empty.

func (*Circular[T, P]) IsFull

func (q *Circular[T, P]) IsFull() (full bool)

IsFull returns true if the queue is full.

func (*Circular[T, P]) Length

func (q *Circular[T, P]) Length() (size int)

Length returns the number of elements in the queue.

func (*Circular[T, P]) Pop

func (q *Circular[T, P]) Pop() (p P, err error)

Pop removes an element from the queue.

func (*Circular[T, P]) Push

func (q *Circular[T, P]) Push(p P) error

Push adds an element to the queue.

type LockFree

type LockFree[T any, P Pointer[T]] struct {
	// contains filtered or unexported fields
}

LockFree is the struct used to store a blocking or non-blocking FIFO queue of type *packet.Packet

In it's non-blocking form it acts as a ringbuffer, overwriting old data when new data arrives. In its blocking form it waits for a space in the queue to open up before it adds the item to the LockFree.

func NewLockFree

func NewLockFree[T any, P Pointer[T]](size uint64) *LockFree[T, P]

NewLockFree creates a new LockFree with blocking or non-blocking behavior

func (*LockFree[T, P]) Close

func (q *LockFree[T, P]) Close()

Close marks the LockFree as closed, returns any waiting Pop() calls, and blocks all future Push calls from occurring.

func (*LockFree[T, P]) Drain

func (q *LockFree[T, P]) Drain() []P

Drain drains all the current packets in the queue and returns them to the caller.

It is an unsafe function that should only be used once, only after the queue has been closed, and only while there are no producers writing to it. If used incorrectly it has the potential to infinitely block the caller. If used correctly, it allows a single caller to drain any remaining packets in the queue after the queue has been closed.

func (*LockFree[T, P]) IsClosed

func (q *LockFree[T, P]) IsClosed() bool

IsClosed returns whether the LockFree has been closed

func (*LockFree[T, P]) Length

func (q *LockFree[T, P]) Length() int

Length is the current number of items in the LockFree

func (*LockFree[T, P]) Pop

func (q *LockFree[T, P]) Pop() (P, error)

Pop removes an item from the start of the LockFree and returns it to the caller. This method blocks until an item is available, but unblocks when the LockFree is closed. This allows for long-term listeners to wait on the LockFree until either an item is available or the LockFree is closed.

This method is safe to be used concurrently and is even optimized for the SPMC use case.

func (*LockFree[T, P]) Push

func (q *LockFree[T, P]) Push(item P) error

Push appends an item of type *packet.Packet to the LockFree, and will block until the item is pushed succ›essfully (with the blocking function depending on whether this is a blocking LockFree).

This method is not meant to be used concurrently, and the LockFree is meant to operate as an SPMC LockFree with one producer operating at a time. If we want to use this as an MPMC LockFree we can modify this Push function by replacing the existing `default` switch case with the following snippet: ``` default:

head, err = q.overflow()
if err != nil {
	return err
}

```

type NonBlocking

type NonBlocking[T any, P Pointer[T]] struct {
	// contains filtered or unexported fields
}

NonBlocking is a circular sized FIFO queue that uses an array of fixed size to store the elements.

It is thread safe and extremely performant, however it is a blocking queue and will block the caller if the queue is full or if it is empty.

func NewNonBlocking

func NewNonBlocking[T any, P Pointer[T]](maxSize uint64) *NonBlocking[T, P]

NewNonBlocking creates a new circular queue with the given size.

func (*NonBlocking[T, P]) Close

func (q *NonBlocking[T, P]) Close()

Close closes the queue permanently.

The Drain method can be used to drain the queue after it is closed.

func (*NonBlocking[T, P]) Drain

func (q *NonBlocking[T, P]) Drain() (values []P)

Drain removes all elements from the queue. and returns them in a slice.

This function should only be called after the queue is closed.

func (*NonBlocking[T, P]) IsClosed

func (q *NonBlocking[T, P]) IsClosed() (closed bool)

IsClosed returns true if the queue is Closed

The Drain method can be used to drain the queue after it is closed.

func (*NonBlocking[T, P]) IsEmpty

func (q *NonBlocking[T, P]) IsEmpty() (empty bool)

IsEmpty returns true if the queue is empty.

func (*NonBlocking[T, P]) IsFull

func (q *NonBlocking[T, P]) IsFull() (full bool)

IsFull returns true if the queue is full.

func (*NonBlocking[T, P]) Length

func (q *NonBlocking[T, P]) Length() (size int)

Length returns the number of elements in the queue.

func (*NonBlocking[T, P]) Pop

func (q *NonBlocking[T, P]) Pop() (p P, err error)

Pop removes an element from the queue.

func (*NonBlocking[T, P]) Push

func (q *NonBlocking[T, P]) Push(p P) error

Push adds an element to the queue.

type Pointer

type Pointer[T any] interface {
	*T
}