Approach: use Dmitry Vyukov’s bounded MPMC ring-buffer (per-slot sequence numbers) to avoid a global lock; producers/consumers use atomic fetch-and-add on head/tail indices and inspect per-slot sequence to decide if slot is ready. For wake-ups, integrate a notification channel or condition variable that producers/consumers signal when they make progress.go
package mpmc
import (
"sync"
"sync/atomic"
)
type slot[T any] struct {
seq uint64
data T
}
type MPMCQueue[T any] struct {
mask uint64
buffer []slot[T]
enqueuePos uint64 // atomic
dequeuePos uint64 // atomic
notifyCh chan struct{}
notifyMu sync.Mutex
}
func NewMPMCQueue[T any](capacity int) *MPMCQueue[T] {
// capacity must be power of two
n := 1
for n < capacity { n <<= 1 }
q := &MPMCQueue[T]{
mask: uint64(n-1),
buffer: make([]slot[T], n),
notifyCh: make(chan struct{}, 1),
}
for i := 0; i < n; i++ {
q.buffer[i].seq = uint64(i)
}
return q
}
// Enqueue returns false if queue full.
func (q *MPMCQueue[T]) Enqueue(v T) bool {
for {
pos := atomic.LoadUint64(&q.enqueuePos)
idx := pos & q.mask
s := &q.buffer[idx]
seq := atomic.LoadUint64(&s.seq)
dif := int64(seq) - int64(pos)
if dif == 0 {
if atomic.CompareAndSwapUint64(&q.enqueuePos, pos, pos+1) {
s.data = v
atomic.StoreUint64(&s.seq, pos+1)
// notify one waiter
select {
case q.notifyCh <- struct{}{}:
default:
}
return true
}
continue
} else if dif < 0 {
// full
return false
} else {
// retry
}
}
}
// Dequeue returns (value, true) or (zero, false) if empty.
func (q *MPMCQueue[T]) Dequeue() (T, bool) {
var zero T
for {
pos := atomic.LoadUint64(&q.dequeuePos)
idx := pos & q.mask
s := &q.buffer[idx]
seq := atomic.LoadUint64(&s.seq)
dif := int64(seq) - int64(pos+1)
if dif == 0 {
if atomic.CompareAndSwapUint64(&q.dequeuePos, pos, pos+1) {
v := s.data
atomic.StoreUint64(&s.seq, pos+uint64(q.mask)+1+1) // pos + capacity
return v, true
}
continue
} else if dif < 0 {
// empty
return zero, false
} else {
// retry
}
}
}
// WaitDequeue blocks until an item is available or ctx canceled (simple version).
func (q *MPMCQueue[T]) WaitDequeue() (T, bool) {
for {
if v, ok := q.Dequeue(); ok { return v, true }
<-q.notifyCh
}
}
