I'm trying to find the fastest way to enqueue and dequeue items concurrently in Go.
There are some restrictions:
- it needs to be unbounded
- Memory allocation should be low
- Multiple producers
- (Single consumer)
This is what I've come up with so far:
package queue
import "sync"
type Queue struct {
buffer []interface{}
head int //index where items are poped
tail int //index where items are pushed
len int //current len of queue
mod int //ring size
lock sync.RWMutex
}
func New() *Queue {
initialSize := 10
return &Queue{
buffer: make([]interface{}, initialSize),
head: 0,
tail: 0,
len: 0,
mod: initialSize,
}
}
func (q *Queue) Push(item interface{}) {
q.lock.Lock()
defer q.lock.Unlock()
q.tail = ((q.tail + 1) % q.mod)
if q.tail == q.head { //if head equals tail, the ring is too small, resize
fillFactor := 2 //how much should the buffer grow
newLen := q.mod * fillFactor
newBuff := make([]interface{}, newLen)
//unwind the current buffer on the new larger buffer
for i := 0; i < q.mod; i++ {
buffIndex := (q.head + i) % q.mod
newBuff[i] = q.buffer[buffIndex]
}
//set the new buffer and reset head and tail
q.buffer = newBuff
q.head = 0
q.tail = q.mod
q.mod *= fillFactor
}
q.len++
q.buffer[q.tail] = item //place the item on the buffer
}
..//Length and Empty method removed
func (q *Queue) Pop() (interface{}, bool) {
q.lock.Lock()
defer q.lock.Unlock()
if q.len == 0 { //early exit if no items
return nil, false
}
//move head to next ring index, consume it and reduce length
q.head = ((q.head + 1) % q.mod)
q.len--
return q.buffer[q.head], true
}
func (q *Queue) PopMany(count int) ([]interface{}, bool) {
q.lock.Lock()
defer q.lock.Unlock()
if q.len == 0 {
return nil, false
}
if count >= q.len {
count = q.len
}
buffer := make([]interface{}, count)
for i := 0; i < count; i++ {
buffer[i] = q.buffer[(q.head+1+i)%q.mod]
}
q.head = (q.head + count) % q.mod
q.len -= count
return buffer, true
}
This implementation works, and is currently the one that gives me the best sustained throughput. Known issues with this one is that it never down size the buffer when the length is shrinking.
I've tried the queue implemented in Go Data Structures, but it simply eats too much memory when under high load.
There is also a lock free ring buffer in the same package, but that one is bounded.
I figured as there is only a single consumer, it might be possible to make the Pop()
lock free (?). as long as there are at least 1 item in the queue, it should be possible to pop it w/o locking, I think. The problem is when the producers cause a resize event to occur, then the Pop needs handle that somehow.
Any feedback is welcome, coding conventions, optimiation tricks etc. I'm fairly new to Go so there might be issues.