add channel version and common interface (#4)

* add channel version and common interface. refactor tests/benchmark to run both implementations

* improve doc

* avoid extra fragmentation/allocs

* fix godoc

Author: ldemailly

cb/cb.go

@@ -1,30 +1,45 @@
-// FIFO Queue with fixed capacity.
+// First In First Out (FIFO) [Queue] with fixed capacity.
 // Circular Buffer implementation in Go with both
 // pub/sub thread safe blocking API and pure FIFO queue with set capacity
-// unsynchronized base.
+// unsynchronized base. Two versions of the same [Queue] interface:
+// one actually a circular buffer [CircularBuffer], the other using a channel
+// [CircularBufferChan], created using [cb.New] or [cb.NewC] respectively.
 package cb
 
 import "sync"
 
+type Queue[T any] interface {
+	Empty() bool
+	Full() bool
+	Size() int
+	Capacity() int
+	Push(item T) bool
+	Pop() (value T, ok bool)
+	PushBlocking(item T)
+	PopBlocking() (value T)
+}
+
+// FIFO [Queue] with fixed capacity. Fixed array implementation.
 type CircularBuffer[T any] struct {
 	buffer []T
 	head   int
 	tail   int
 	size   int
 	mu     sync.Mutex
-	full   *sync.Cond
-	empty  *sync.Cond
+	full   sync.Cond
+	empty  sync.Cond
 }
 
+// New returns the fixed array version of 0 alloc fixed capacity (optionally blocking) [Queue].
 func New[T any](capacity int) *CircularBuffer[T] {
 	cb := &CircularBuffer[T]{
 		buffer: make([]T, capacity),
 		head:   0,
 		tail:   0,
 		size:   0,
 	}
-	cb.full = sync.NewCond(&cb.mu)
-	cb.empty = sync.NewCond(&cb.mu)
+	cb.full.L = &cb.mu
+	cb.empty.L = &cb.mu
 	return cb
 }
 

cb/cb_channel.go

@@ -0,0 +1,70 @@
+package cb
+
+// FIFO [Queue] with fixed capacity.
+// Channel / go idiomatic version (blocking, multi thread safe),
+// use [CircularBufferChan] for low/no contention cases as it is faster.
+type CircularBufferChan[T any] struct {
+	buffer chan T
+}
+
+// NewC returns a channel ([CircularBufferChan]) version of 0 alloc pub/sub fixed capacity blocking queue.
+func NewC[T any](capacity int) *CircularBufferChan[T] {
+	cb := &CircularBufferChan[T]{
+		buffer: make(chan T, capacity),
+	}
+	return cb
+}
+
+func (cb *CircularBufferChan[T]) Empty() bool {
+	return len(cb.buffer) == 0
+}
+
+func (cb *CircularBufferChan[T]) Full() bool {
+	return len(cb.buffer) == cap(cb.buffer)
+}
+
+func (cb *CircularBufferChan[T]) Size() int {
+	return len(cb.buffer)
+}
+
+func (cb *CircularBufferChan[T]) Capacity() int {
+	return cap(cb.buffer)
+}
+
+// Push adds an item to the queue. returns false if queue is full.
+// Note: might block and not return false at times. Use PushBlocking for
+// correct version.
+func (cb *CircularBufferChan[T]) Push(item T) bool {
+	// Note: this is for equivalent with the array/conditional variable version
+	// but isn't correct, as in Full can return false and yet the push can block
+	// if another producer enqueued in the meanwhile.
+	if cb.Full() {
+		return false
+	}
+	cb.buffer <- item
+	return true
+}
+
+// Pop removes an item from the queue. returns false if queue is empty.
+// Note: might block and not return false at times. Use PopBlocking for
+// correct version.
+func (cb *CircularBufferChan[T]) Pop() (value T, ok bool) {
+	if cb.Empty() {
+		return
+	}
+	ok = true
+	value = <-cb.buffer
+	return
+}
+
+// Thread safe blocking versions:
+
+// Push adds an item to the queue. blocks if queue is full.
+func (cb *CircularBufferChan[T]) PushBlocking(item T) {
+	cb.buffer <- item
+}
+
+// Pop removes an item from the queue. blocks if queue is empty.
+func (cb *CircularBufferChan[T]) PopBlocking() T {
+	return <-cb.buffer
+}

cb/cb_test.go

@@ -7,10 +7,24 @@ import (
 	"fortio.org/memstore/cb"
 )
 
+const capacity = 5
+
 func TestBoundaryConditions(t *testing.T) {
-	const capacity = 5
-	buffer := cb.New[int](capacity)
+	testCases := []struct {
+		name string
+		cb   cb.Queue[int]
+	}{
+		{name: "CircBuffer", cb: cb.New[int](capacity)},
+		{name: "Channel", cb: cb.NewC[int](capacity)},
+	}
+	for _, tc := range testCases {
+		t.Run(tc.name, func(tt *testing.T) {
+			testBoundaryConditions(tt, tc.cb)
+		})
+	}
+}
 
+func testBoundaryConditions(t *testing.T, buffer cb.Queue[int]) {
 	// Test empty buffer
 	if !buffer.Empty() {
 		t.Error("Buffer should be empty")
@@ -91,7 +105,22 @@ func TestBoundaryConditions(t *testing.T) {
 }
 
 func BenchmarkCircularBuffer(b *testing.B) {
-	c := cb.New[int](100)
+	capacity := 100
+	testCases := []struct {
+		name string
+		cb   cb.Queue[int]
+	}{
+		{name: "CircBuffer", cb: cb.New[int](capacity)},
+		{name: "Channel", cb: cb.NewC[int](capacity)},
+	}
+	for _, tc := range testCases {
+		b.Run(tc.name, func(bb *testing.B) {
+			benchmarkCircularBuffer(bb, tc.cb)
+		})
+	}
+}
+
+func benchmarkCircularBuffer(b *testing.B, c cb.Queue[int]) {
 	var x int
 	var ok bool
 	for i := 0; i < b.N; i++ {
@@ -105,8 +134,20 @@ func BenchmarkCircularBuffer(b *testing.B) {
 }
 
 func TestProducerConsumerScenario(t *testing.T) {
-	buffer := cb.New[int](10)
-
+	testCases := []struct {
+		name string
+		cb   cb.Queue[int]
+	}{
+		{name: "CircBuffer", cb: cb.New[int](10)},
+		{name: "Channel", cb: cb.NewC[int](10)},
+	}
+	for _, tc := range testCases {
+		t.Run(tc.name, func(tt *testing.T) {
+			testProducerConsumerScenario(tt, tc.cb)
+		})
+	}
+}
+func testProducerConsumerScenario(t *testing.T, buffer cb.Queue[int]) {
 	var wg sync.WaitGroup
 	wg.Add(11) // 10 producers + 1 consumer
 
@@ -143,7 +184,21 @@ func TestProducerConsumerScenario(t *testing.T) {
 }
 
 func BenchmarkCircularBufferBlocking(b *testing.B) {
-	c := cb.New[int](100)
+	capacity := 100
+	testCases := []struct {
+		name string
+		cb   cb.Queue[int]
+	}{
+		{name: "CircBuffer", cb: cb.New[int](capacity)},
+		{name: "Channel", cb: cb.NewC[int](capacity)},
+	}
+	for _, tc := range testCases {
+		b.Run(tc.name, func(bb *testing.B) {
+			benchmarkCircularBufferBlocking(bb, tc.cb)
+		})
+	}
+}
+func benchmarkCircularBufferBlocking(b *testing.B, c cb.Queue[int]) {
 	var x int
 	for i := 0; i < b.N; i++ {
 		c.PushBlocking(i)
@@ -155,9 +210,7 @@ func BenchmarkCircularBufferBlocking(b *testing.B) {
 	b.Logf("x=%d", x)
 }
 
-func benchmarkPushBlocking(b *testing.B, numProducers, numConsumers int) {
-	buffer := cb.New[int](20) // small queue, higher contention
-
+func benchmarkPushBlocking(b *testing.B, buffer cb.Queue[int], numProducers, numConsumers int) {
 	var wg sync.WaitGroup
 	wg.Add(numProducers + numConsumers)
 	prodN := b.N * numConsumers
@@ -191,11 +244,20 @@ func benchmarkPushBlocking(b *testing.B, numProducers, numConsumers int) {
 	wg.Wait()
 }
 
-func BenchmarkPushBlocking(b *testing.B) {
+func BenchmarkHighContention(b *testing.B) {
 	numProducers := 13
 	numConsumers := 7
 
-	b.Run("BenchmarkPushBlocking 7,5", func(b *testing.B) {
-		benchmarkPushBlocking(b, numProducers, numConsumers)
-	})
+	testCases := []struct {
+		name string
+		cb   cb.Queue[int]
+	}{
+		{name: "CircBuffer", cb: cb.New[int](capacity)},
+		{name: "Channel", cb: cb.NewC[int](capacity)},
+	}
+	for _, tc := range testCases {
+		b.Run(tc.name, func(bb *testing.B) {
+			benchmarkPushBlocking(bb, tc.cb, numProducers, numConsumers)
+		})
+	}
 }