"circular queue"

circular_queue.go

package core

import "sync"

func NewCircularQueue[T any](capacity int) *CircularQueue[T] {
	m := &sync.Mutex{}
	return &CircularQueue[T]{
		data:  make([]T, capacity),
		Mutex: m,
		cond:  sync.NewCond(m),
	}
}

// CircularQueue is a data structure that is intended to behave simliar to buffered
// channels. Except it will only block when trying to receive from an empty
// queue. Sending is not supposed to block and will overwrite the oldest entry
// in the queue.
type CircularQueue[T any] struct {
	pos  int
	size int
	data []T
	*sync.Mutex
	cond *sync.Cond
}

func (q *CircularQueue[T]) Cap() int { return len(q.data) }

func (q *CircularQueue[T]) Push(x T) {
	q.Lock()
	defer q.Unlock()
	// Set tail (may overwrite last element if at capacity).
	q.data[(q.pos+q.size)%len(q.data)] = x
	// Increase size if not at capacity.
	if q.size == len(q.data) {
		q.pos = (q.pos + 1) % len(q.data)
	}
	if q.size < len(q.data) {
		q.size++
	}
	q.cond.Signal()
}

func (q *CircularQueue[T]) Pop() T {
	q.Lock()
	defer q.Unlock()
	// Wait queue to be greater than zero
	if q.size == 0 {
		q.cond.Wait()
	}
	// Get head
	ret := q.data[q.pos]
	// Increment head.
	q.pos = (q.pos + 1) % len(q.data)
	// Decrement size.
	q.size--
	return ret
}