make it pretty

Author: pltr

Diff for: bulk_test.go

@@ -20,7 +20,7 @@ const (
 )
 
 func BenchmarkResponseTimesRing(b *testing.B) {
-	var ring = New{Size: 8192, MaxBatch: 127}.SPSC()
+	var ring = New{Size: 8192, BatchSize: 127}.SPSC()
 	var wg sync.WaitGroup
 	wg.Add(2)
 	var diffs = make([]int64, (b.N/sampleTimes)+1)

Diff for: defs.go

@@ -1,4 +1,4 @@
 package onering
 
 // default max batch size for batched consumers
-const DefaultMaxBatch = 1023
+const DefaultMaxBatch = (1 << 8) - 1

Diff for: mpmc.go

@@ -12,8 +12,8 @@ type MPMC struct {
 	multi
 }
 
-func (r *MPMC) init(size uint32) {
-	r.multi.init(size)
+func (r *MPMC) init(n *New) {
+	r.multi.init(n)
 	for i := range r.seq {
 		r.seq[i] = -int64(i)
 	}

Diff for: mpsc.go

@@ -8,11 +8,6 @@ type MPSC struct {
 	multi
 }
 
-func (r *MPSC) init(size uint32) {
-	r.multi.init(size)
-	r.rc = 1
-}
-
 func (r *MPSC) Get(i interface{}) bool {
 	var (
 		rc        = r.rc

Diff for: queue.go

@@ -43,30 +43,27 @@ type New struct {
 	// Size (Capacity) of the queue
 	Size uint32
 	// Maximum number of batched messages
-	MaxBatch int32
+	BatchSize uint32
 }
 
 // SPSC constructs a Single Producer/Single Consumer queue
 func (n New) SPSC() Queue {
 	var spsc = new(SPSC)
-	spsc.init(n.Size)
-	spsc.maxbatch = n.BatchSize()
+	spsc.init(&n)
 	return spsc
 }
 
 // MPSC constructs a Multi-Producer/Single Consumer queue
 func (n New) MPSC() Queue {
 	var mpsc = new(MPSC)
-	mpsc.init(n.Size)
-	mpsc.maxbatch = n.BatchSize()
+	mpsc.init(&n)
 	return mpsc
 }
 
 // SPMC constructs a Single Producer/Multi-Consumer queue
 func (n New) SPMC() Queue {
 	var spmc = new(SPMC)
-	spmc.init(n.Size)
-	spmc.maxbatch = n.BatchSize()
+	spmc.init(&n)
 	return spmc
 }
 
@@ -75,18 +72,10 @@ func (n New) SPMC() Queue {
 // However it will not implement many of the optimizations available to other queue types
 func (n New) MPMC() Queue {
 	var mpmc = new(MPMC)
-	mpmc.init(n.Size)
-	mpmc.maxbatch = n.BatchSize()
+	mpmc.init(&n)
 	return mpmc
 }
 
-func (n *New) BatchSize() int64 {
-	if n.MaxBatch > 0 {
-		return int64(n.MaxBatch)
-	}
-	return DefaultMaxBatch
-}
-
 //type Waiter interface {
 //	Wait()
 //	Signal()

Diff for: ring.go

@@ -1,7 +1,6 @@
 package onering
 
 import (
-	"math/bits"
 	"runtime"
 	"sync/atomic"
 	"unsafe"
@@ -24,9 +23,15 @@ type ring struct {
 	done     int32
 }
 
-func (r *ring) init(size uint32) {
-	r.data = make([]unsafe.Pointer, 1<<uint(32-bits.LeadingZeros32(size-1)))
+func (r *ring) init(n *New) {
+	r.data = make([]unsafe.Pointer, roundUp2(n.Size))
 	r.mask = int64(len(r.data) - 1)
+
+	var bs = n.BatchSize
+	if bs == 0 {
+		bs = DefaultMaxBatch
+	}
+	r.maxbatch = int64(roundUp2(bs) - 1)
 }
 
 func (r *ring) Close() {
@@ -54,8 +59,8 @@ type multi struct {
 	seq []int64
 }
 
-func (c *multi) init(size uint32) {
-	c.ring.init(size)
+func (c *multi) init(n *New) {
+	c.ring.init(n)
 	c.size = int64(len(c.data))
 	c.seq = make([]int64, len(c.data))
 	c.wp = 1 // just to avoid 0-awkwardness with seq

Diff for: util.go

@@ -28,3 +28,13 @@ func inject(i interface{}, ptr unsafe.Pointer) {
 	var v = (*unsafe.Pointer)((*iface)(unsafe.Pointer(&i)).d)
 	*v = ptr
 }
+
+func roundUp2(v uint32) uint32 {
+	v--
+	v |= v >> 1
+	v |= v >> 2
+	v |= v >> 4
+	v |= v >> 8
+	v |= v >> 16
+	return v + 1
+}