Merge branch 'reboot'
diff --git a/TODO b/TODO
index 7eaf300..df90ae2 100644
--- a/TODO
+++ b/TODO
@@ -1,41 +1,23 @@
-- The consumer code needs a much cleaner abstraction
- In the end, it's two-phase, e.g. receive message, ack message, perhaps we make this a
- first-class citizen in the API on the consumer side where the consumer calls:
- Receive() and gets the sequence of the message in question (or a specific value for
- gating/idling?) and then calls Ack/Confirm etc with the message sequence?
-
- Wireup DSL
+
- Rewrite all code test first (including diamond pattern)
-- Benchmark code (go test -bench=.) (with GOMAXPROCS=2 if env GOMAXPROCS=1)
+
- Squeeze little bits of performance here and there by trying a few different things
e.g. pointers vs structs, padding, etc.
-- Investigate ways to utilize an int32 under the hood but have it be exposed as an int64?
- e.g. utilizing two int32s? This would allow us to remove the atomic statements surrounding
- i386 and ARM architecutes, but may be more effort than it's worth for our particular
- use case...
-- Understand why "slower" consumer code is faster and more consistent when performing single-item
- publishes
-- Cursors should probably all be created at the same time to keep them as close together as possible:
+- Cursors should probably all be created at the same time in wireup to keep them as close together as possible:
https://news.ycombinator.com/item?id=7800825
-
-- I need to get some help doing some low-level profiling the application to find out where
- the bottlenecks are and determining how the Go scheduler is hindering performance, if at all.
-
-- More assertive tests on CPU ordering, e.g. claim a slot which has multiple values
- perform a computation on the sequence in some deterministic way and store each computation
- as one of the values. Then on the reader side assert that the struct values are correct
- for that given sequence.
-
- Integration suite that can exercise the Disruptor code on:
1. Multiple Go runtime versions, e.g. (1.1, 1.2, 1.3, etc.) (plus point releases and the latest tip)
2. Multiple CPU architectures: ARM, i386, AMD64
3. Multiple operation systems: Linux, OSX, Windows, etc.
Ideally each of the above combinations would be an individual exercise
-- contributors
-- license
-- readme (Github markdown?)
+- Contributors
+
+- License
+
+- Readme (Github markdown?)
- Website with cool images and documentation? (e.g. GoConvey)
\ No newline at end of file
diff --git a/barrier.go b/barrier.go
index 3fcecb2..03d0bcf 100644
--- a/barrier.go
+++ b/barrier.go
@@ -1,5 +1,5 @@
package disruptor
type Barrier interface {
- LoadBarrier(int64) int64
+ Read(int64) int64
}
diff --git a/barrier_test.go b/barrier_test.go
deleted file mode 100644
index a4e5904..0000000
--- a/barrier_test.go
+++ /dev/null
@@ -1,15 +0,0 @@
-package disruptor
-
-import "testing"
-
-func BenchmarkBarrierLoadSingle(b *testing.B) {
- barrier := NewCursor()
-
- iterations := int64(b.N)
- b.ReportAllocs()
- b.ResetTimer()
-
- for i := int64(0); i < iterations; i++ {
- barrier.LoadBarrier(0)
- }
-}
diff --git a/benchmark-channels/blocking_channel_test.go b/benchmark-channels/blocking_channel_test.go
new file mode 100644
index 0000000..f2783ca
--- /dev/null
+++ b/benchmark-channels/blocking_channel_test.go
@@ -0,0 +1,46 @@
+package benchmarks
+
+import (
+ "runtime"
+ "testing"
+)
+
+func BenchmarkBlockingOneGoroutine(b *testing.B) {
+ runtime.GOMAXPROCS(1)
+ benchmarkBlocking(b, 1)
+}
+
+func BenchmarkBlockingTwoGoroutines(b *testing.B) {
+ runtime.GOMAXPROCS(2)
+ benchmarkBlocking(b, 1)
+ runtime.GOMAXPROCS(1)
+}
+
+func BenchmarkBlockingThreeGoroutinesWithContendedWrite(b *testing.B) {
+ runtime.GOMAXPROCS(3)
+ benchmarkBlocking(b, 2)
+ runtime.GOMAXPROCS(1)
+}
+
+func benchmarkBlocking(b *testing.B, writers int64) {
+ iterations := int64(b.N)
+ channel := make(chan int64, 1024*16)
+
+ b.ReportAllocs()
+ b.ResetTimer()
+
+ for x := int64(0); x < writers; x++ {
+ go func() {
+ for i := int64(0); i < iterations; i++ {
+ channel <- i
+ }
+ }()
+ }
+
+ for i := int64(0); i < iterations*writers; i++ {
+ msg := <-channel
+ if writers == 1 && msg != i {
+ panic("Out of sequence")
+ }
+ }
+}
diff --git a/benchmark-channels/non_blocking_channel_test.go b/benchmark-channels/non_blocking_channel_test.go
new file mode 100644
index 0000000..cdf3e66
--- /dev/null
+++ b/benchmark-channels/non_blocking_channel_test.go
@@ -0,0 +1,54 @@
+package benchmarks
+
+import (
+ "runtime"
+ "testing"
+)
+
+func BenchmarkNonBlockingOneGoroutine(b *testing.B) {
+ runtime.GOMAXPROCS(1)
+ benchmarkNonBlocking(b, 1)
+}
+
+func BenchmarkNonBlockingTwoGoroutines(b *testing.B) {
+ runtime.GOMAXPROCS(2)
+ benchmarkNonBlocking(b, 1)
+ runtime.GOMAXPROCS(1)
+}
+func BenchmarkNonBlockingThreeGoroutinesWithContendedWrite(b *testing.B) {
+ runtime.GOMAXPROCS(3)
+ benchmarkNonBlocking(b, 2)
+ runtime.GOMAXPROCS(1)
+}
+
+func benchmarkNonBlocking(b *testing.B, writers int64) {
+ iterations := int64(b.N)
+ channel := make(chan int64, 1024*16)
+
+ b.ReportAllocs()
+ b.ResetTimer()
+
+ for x := int64(0); x < writers; x++ {
+ go func() {
+ for i := int64(0); i < iterations; {
+ select {
+ case channel <- i:
+ i++
+ default:
+ continue
+ }
+ }
+ }()
+ }
+
+ for i := int64(0); i < iterations*writers; i++ {
+ select {
+ case msg := <-channel:
+ if writers == 1 && msg != i {
+ // panic("Out of sequence")
+ }
+ default:
+ continue
+ }
+ }
+}
diff --git a/benchmark-disruptor/init_test.go b/benchmark-disruptor/init_test.go
new file mode 100644
index 0000000..8d4945b
--- /dev/null
+++ b/benchmark-disruptor/init_test.go
@@ -0,0 +1,15 @@
+package benchmarks
+
+import "runtime"
+
+const (
+ RingBufferSize = 1024 * 64
+ RingBufferMask = RingBufferSize - 1
+ ReserveOne = 1
+ ReserveMany = 16
+ ReserveManyDelta = ReserveMany - 1
+)
+
+func init() {
+ runtime.GOMAXPROCS(2)
+}
diff --git a/benchmark-disruptor/sample_consumer_test.go b/benchmark-disruptor/sample_consumer_test.go
new file mode 100644
index 0000000..61c5a0e
--- /dev/null
+++ b/benchmark-disruptor/sample_consumer_test.go
@@ -0,0 +1,19 @@
+package benchmarks
+
+import "fmt"
+
+type SampleConsumer struct {
+ ringBuffer *[RingBufferSize]int64
+}
+
+func (this SampleConsumer) Consume(lower, upper int64) {
+ for lower <= upper {
+ message := this.ringBuffer[lower&RingBufferMask]
+ if message != lower {
+ warning := fmt.Sprintf("\nRace condition--Sequence: %d, Message: %d\n", lower, message)
+ fmt.Printf(warning)
+ panic(warning)
+ }
+ lower++
+ }
+}
diff --git a/benchmark-disruptor/shared_writer_test.go b/benchmark-disruptor/shared_writer_test.go
new file mode 100644
index 0000000..fcd7122
--- /dev/null
+++ b/benchmark-disruptor/shared_writer_test.go
@@ -0,0 +1,56 @@
+package benchmarks
+
+import (
+ "testing"
+
+ "github.com/smartystreets/go-disruptor"
+)
+
+func BenchmarkSharedWriterReserveOne(b *testing.B) {
+ ringBuffer := [RingBufferSize]int64{}
+ written, read := disruptor.NewCursor(), disruptor.NewCursor()
+ shared := disruptor.NewSharedWriterBarrier(written, RingBufferSize)
+ reader := disruptor.NewReader(read, written, shared, SampleConsumer{&ringBuffer})
+ writer := disruptor.NewSharedWriter(shared, read)
+ reader.Start()
+
+ iterations := int64(b.N)
+ b.ReportAllocs()
+ b.ResetTimer()
+
+ sequence := disruptor.InitialSequenceValue
+ for sequence < iterations {
+ sequence = writer.Reserve(ReserveOne)
+ ringBuffer[sequence&RingBufferMask] = sequence
+ writer.Commit(sequence, sequence)
+ }
+
+ reader.Stop()
+}
+
+func BenchmarkSharedWriterReserveMany(b *testing.B) {
+ ringBuffer := [RingBufferSize]int64{}
+ written, read := disruptor.NewCursor(), disruptor.NewCursor()
+ shared := disruptor.NewSharedWriterBarrier(written, RingBufferSize)
+ reader := disruptor.NewReader(read, written, shared, SampleConsumer{&ringBuffer})
+ writer := disruptor.NewSharedWriter(shared, read)
+ reader.Start()
+
+ iterations := int64(b.N)
+ b.ReportAllocs()
+ b.ResetTimer()
+
+ previous, current := disruptor.InitialSequenceValue, disruptor.InitialSequenceValue
+ for current < iterations {
+ current = writer.Reserve(ReserveMany)
+
+ for i := previous + 1; i <= current; i++ {
+ ringBuffer[i&RingBufferMask] = i
+ }
+
+ writer.Commit(previous+1, current)
+ previous = current
+ }
+
+ reader.Stop()
+}
diff --git a/benchmark-disruptor/writer_await_test.go b/benchmark-disruptor/writer_await_test.go
new file mode 100644
index 0000000..81f453e
--- /dev/null
+++ b/benchmark-disruptor/writer_await_test.go
@@ -0,0 +1,54 @@
+package benchmarks
+
+import (
+ "testing"
+
+ "github.com/smartystreets/go-disruptor"
+)
+
+func BenchmarkWriterAwaitOne(b *testing.B) {
+ ringBuffer := [RingBufferSize]int64{}
+ written, read := disruptor.NewCursor(), disruptor.NewCursor()
+ reader := disruptor.NewReader(read, written, written, SampleConsumer{&ringBuffer})
+ writer := disruptor.NewWriter(written, read, RingBufferSize)
+ reader.Start()
+
+ iterations := int64(b.N)
+ b.ReportAllocs()
+ b.ResetTimer()
+
+ sequence := disruptor.InitialSequenceValue
+ for sequence < iterations {
+ sequence += ReserveOne
+ writer.Await(sequence)
+ ringBuffer[sequence&RingBufferMask] = sequence
+ writer.Commit(sequence, sequence)
+ }
+
+ reader.Stop()
+}
+func BenchmarkWriterAwaitMany(b *testing.B) {
+ ringBuffer := [RingBufferSize]int64{}
+ written, read := disruptor.NewCursor(), disruptor.NewCursor()
+ reader := disruptor.NewReader(read, written, written, SampleConsumer{&ringBuffer})
+ writer := disruptor.NewWriter(written, read, RingBufferSize)
+ reader.Start()
+
+ iterations := int64(b.N)
+ b.ReportAllocs()
+ b.ResetTimer()
+
+ sequence := disruptor.InitialSequenceValue
+ for sequence < iterations {
+ sequence += ReserveMany
+ writer.Await(sequence)
+
+ for i := sequence - ReserveManyDelta; i <= sequence; i++ {
+ ringBuffer[i&RingBufferMask] = i
+ }
+
+ writer.Commit(sequence, sequence)
+ }
+
+ reader.Stop()
+}
diff --git a/benchmark-disruptor/writer_reservation_multiple_readers_test.go b/benchmark-disruptor/writer_reservation_multiple_readers_test.go
new file mode 100644
index 0000000..2bd8775
--- /dev/null
+++ b/benchmark-disruptor/writer_reservation_multiple_readers_test.go
@@ -0,0 +1,62 @@
+package benchmarks
+
+import (
+ "testing"
+
+ "github.com/smartystreets/go-disruptor"
+)
+
+func BenchmarkWriterReserveOneMultipleReaders(b *testing.B) {
+ ringBuffer := [RingBufferSize]int64{}
+ written, read1, read2 := disruptor.NewCursor(), disruptor.NewCursor(), disruptor.NewCursor()
+ reader1 := disruptor.NewReader(read1, written, written, SampleConsumer{&ringBuffer})
+ reader2 := disruptor.NewReader(read2, written, written, SampleConsumer{&ringBuffer})
+ barrier := disruptor.NewCompositeBarrier(read1, read2)
+ writer := disruptor.NewWriter(written, barrier, RingBufferSize)
+
+ reader1.Start()
+ reader2.Start()
+
+ iterations := int64(b.N)
+ b.ReportAllocs()
+ b.ResetTimer()
+
+ sequence := disruptor.InitialSequenceValue
+ for sequence < iterations {
+ sequence = writer.Reserve(ReserveOne)
+ ringBuffer[sequence&RingBufferMask] = sequence
+ writer.Commit(sequence, sequence)
+ }
+
+ reader1.Stop()
+ reader2.Stop()
+}
+func BenchmarkWriterReserveManyMultipleReaders(b *testing.B) {
+ ringBuffer := [RingBufferSize]int64{}
+ written, read1, read2 := disruptor.NewCursor(), disruptor.NewCursor(), disruptor.NewCursor()
+ reader1 := disruptor.NewReader(read1, written, written, SampleConsumer{&ringBuffer})
+ reader2 := disruptor.NewReader(read2, written, written, SampleConsumer{&ringBuffer})
+ barrier := disruptor.NewCompositeBarrier(read1, read2)
+ writer := disruptor.NewWriter(written, barrier, RingBufferSize)
+
+ reader1.Start()
+ reader2.Start()
+
+ iterations := int64(b.N)
+ b.ReportAllocs()
+ b.ResetTimer()
+
+ sequence := disruptor.InitialSequenceValue
+ for sequence < iterations {
+ sequence = writer.Reserve(ReserveMany)
+
+ for i := sequence - ReserveManyDelta; i <= sequence; i++ {
+ ringBuffer[i&RingBufferMask] = i
+ }
+
+ writer.Commit(sequence, sequence)
+ }
+
+ reader1.Stop()
+ reader2.Stop()
+}
diff --git a/benchmark-disruptor/writer_reservation_test.go b/benchmark-disruptor/writer_reservation_test.go
new file mode 100644
index 0000000..4be4238
--- /dev/null
+++ b/benchmark-disruptor/writer_reservation_test.go
@@ -0,0 +1,52 @@
+package benchmarks
+
+import (
+ "testing"
+
+ "github.com/smartystreets/go-disruptor"
+)
+
+func BenchmarkWriterReserveOne(b *testing.B) {
+ ringBuffer := [RingBufferSize]int64{}
+ written, read := disruptor.NewCursor(), disruptor.NewCursor()
+ reader := disruptor.NewReader(read, written, written, SampleConsumer{&ringBuffer})
+ writer := disruptor.NewWriter(written, read, RingBufferSize)
+ reader.Start()
+
+ iterations := int64(b.N)
+ b.ReportAllocs()
+ b.ResetTimer()
+
+ sequence := disruptor.InitialSequenceValue
+ for sequence < iterations {
+ sequence = writer.Reserve(ReserveOne)
+ ringBuffer[sequence&RingBufferMask] = sequence
+ writer.Commit(sequence, sequence)
+ }
+
+ reader.Stop()
+}
+func BenchmarkWriterReserveMany(b *testing.B) {
+ ringBuffer := [RingBufferSize]int64{}
+ written, read := disruptor.NewCursor(), disruptor.NewCursor()
+ reader := disruptor.NewReader(read, written, written, SampleConsumer{&ringBuffer})
+ writer := disruptor.NewWriter(written, read, RingBufferSize)
+ reader.Start()
+
+ iterations := int64(b.N)
+ b.ReportAllocs()
+ b.ResetTimer()
+
+ sequence := disruptor.InitialSequenceValue
+ for sequence < iterations {
+ sequence = writer.Reserve(ReserveMany)
+
+ for i := sequence - ReserveManyDelta; i <= sequence; i++ {
+ ringBuffer[i&RingBufferMask] = i
+ }
+
+ writer.Commit(sequence, sequence)
+ }
+
+ reader.Stop()
+}
diff --git a/benchmarks/blocking_channel_test.go b/benchmarks/blocking_channel_test.go
deleted file mode 100644
index 2223efb..0000000
--- a/benchmarks/blocking_channel_test.go
+++ /dev/null
@@ -1,23 +0,0 @@
-package benchmarks
-
-import "testing"
-
-func BenchmarkBlockingChannel(b *testing.B) {
- iterations := int64(b.N)
- b.ReportAllocs()
- b.ResetTimer()
-
- channel := make(chan int64, 1024*16)
- go func() {
- for i := int64(0); i < iterations; i++ {
- channel <- i
- }
- }()
-
- for i := int64(0); i < iterations; i++ {
- msg := <-channel
- if msg != i {
- panic("Out of sequence")
- }
- }
-}
diff --git a/benchmarks/multi_writer_test.go b/benchmarks/multi_writer_test.go
deleted file mode 100644
index 46355a8..0000000
--- a/benchmarks/multi_writer_test.go
+++ /dev/null
@@ -1,68 +0,0 @@
-package benchmarks
-
-import (
- "math"
- "runtime"
- "testing"
-
- "github.com/smartystreets/go-disruptor"
-)
-
-const multiWriterRingBufferSize = 1024 * 16
-const multiWriterRingBufferMask = multiWriterRingBufferSize - 1
-
-func BenchmarkDisruptorSharedWriterClaimSingle(b *testing.B) {
- benchmarkMultiWriter(b, 1)
-}
-func BenchmarkDisruptorSharedWriterClaimMultiple(b *testing.B) {
- benchmarkMultiWriter(b, 32)
-}
-
-func benchmarkMultiWriter(b *testing.B, maxClaim int64) {
- written, read := disruptor.NewCursor(), disruptor.NewCursor()
- shared := disruptor.NewSharedWriterBarrier(written, multiWriterRingBufferSize)
- reader := disruptor.NewReader(read, written, shared)
- writer := disruptor.NewSharedWriter(shared, read)
-
- iterations := int64(b.N)
- ringBuffer := [multiWriterRingBufferSize]int64{}
- claim := (int64(math.Log2(float64(iterations))) + 1)
- if claim >= maxClaim {
- claim = maxClaim
- }
-
- b.ReportAllocs()
- b.ResetTimer()
-
- go func() {
- sequence := int64(0)
- for sequence < iterations {
- if lower, upper := writer.Reserve(claim); upper >= 0 {
- for ; sequence <= upper; sequence++ {
- ringBuffer[sequence&multiWriterRingBufferMask] = sequence
- }
-
- writer.Commit(lower, upper)
- sequence = upper
- }
- }
- }()
-
- sequence := int64(0)
- for sequence < iterations {
- if _, upper := reader.Receive(); upper >= 0 {
- for ; sequence <= upper; sequence++ {
- if sequence != ringBuffer[sequence&multiWriterRingBufferMask] {
- panic("Out of sequence")
- }
- }
-
- reader.Commit(upper)
- sequence = upper
- }
- }
-}
-
-func init() {
- runtime.GOMAXPROCS(2)
-}
diff --git a/benchmarks/non_blocking_channel_test.go b/benchmarks/non_blocking_channel_test.go
deleted file mode 100644
index 0867462..0000000
--- a/benchmarks/non_blocking_channel_test.go
+++ /dev/null
@@ -1,34 +0,0 @@
-package benchmarks
-
-import "testing"
-
-func BenchmarkNonBlockingChannel(b *testing.B) {
- iterations := int64(b.N)
- b.ReportAllocs()
- b.ResetTimer()
-
- channel := make(chan int64, 1024*16)
- go func() {
- for i := int64(0); i < iterations; {
- select {
- case channel <- i:
- i++
- default:
- continue
- }
- }
- }()
-
- for i := int64(0); i < iterations; {
- select {
- case msg := <-channel:
- if msg != i {
- panic("Out of sequence")
- } else {
- i++
- }
- default:
- continue
- }
- }
-}
diff --git a/benchmarks/single_writer_test.go b/benchmarks/single_writer_test.go
deleted file mode 100644
index 096e20d..0000000
--- a/benchmarks/single_writer_test.go
+++ /dev/null
@@ -1,67 +0,0 @@
-package benchmarks
-
-import (
- "math"
- "runtime"
- "testing"
-
- "github.com/smartystreets/go-disruptor"
-)
-
-const singleWriterRingBufferSize = 1024 * 16
-const singleWriterRingBufferMask = singleWriterRingBufferSize - 1
-
-func BenchmarkDisruptorWriterClaimSingle(b *testing.B) {
- benchmarkSingleWriter(b, 1)
-}
-func BenchmarkDisruptorWriterClaimMultiple(b *testing.B) {
- benchmarkSingleWriter(b, 32)
-}
-
-func benchmarkSingleWriter(b *testing.B, maxClaim int64) {
- written, read := disruptor.NewCursor(), disruptor.NewCursor()
- reader := disruptor.NewReader(read, written, written)
- writer := disruptor.NewWriter(written, read, singleWriterRingBufferSize)
-
- iterations := int64(b.N)
- ringBuffer := [singleWriterRingBufferSize]int64{}
- claim := (int64(math.Log2(float64(iterations))) + 1)
- if claim >= maxClaim {
- claim = maxClaim
- }
-
- b.ReportAllocs()
- b.ResetTimer()
-
- go func() {
- sequence := int64(0)
- for sequence < iterations {
- if lower, upper := writer.Reserve(claim); upper >= 0 {
- for ; sequence <= upper; sequence++ {
- ringBuffer[sequence&singleWriterRingBufferMask] = sequence
- }
-
- writer.Commit(lower, upper)
- sequence = upper
- }
- }
- }()
-
- sequence := int64(0)
- for sequence < iterations {
- if _, upper := reader.Receive(); upper >= 0 {
- for ; sequence <= upper; sequence++ {
- if sequence != ringBuffer[sequence&singleWriterRingBufferMask] {
- panic("Out of sequence")
- }
- }
-
- reader.Commit(upper)
- sequence = upper
- }
- }
-}
-
-func init() {
- runtime.GOMAXPROCS(2)
-}
diff --git a/composite_barrier.go b/composite_barrier.go
index e9d2211..ace1d19 100644
--- a/composite_barrier.go
+++ b/composite_barrier.go
@@ -1,22 +1,20 @@
package disruptor
-type CompositeBarrier struct {
- cursors []*Cursor
-}
+type CompositeBarrier []*Cursor
-func NewCompositeBarrier(upstream ...*Cursor) *CompositeBarrier {
+func NewCompositeBarrier(upstream ...*Cursor) CompositeBarrier {
if len(upstream) == 0 {
panic("At least one upstream cursor is required.")
}
cursors := make([]*Cursor, len(upstream))
copy(cursors, upstream)
- return &CompositeBarrier{cursors}
+ return CompositeBarrier(cursors)
}
-func (this *CompositeBarrier) LoadBarrier(current int64) int64 {
+func (this CompositeBarrier) Read(noop int64) int64 {
minimum := MaxSequenceValue
- for _, item := range this.cursors {
+ for _, item := range this {
sequence := item.Load()
if sequence < minimum {
minimum = sequence
diff --git a/composite_barrier_test.go b/composite_barrier_test.go
index 1a72733..9696429 100644
--- a/composite_barrier_test.go
+++ b/composite_barrier_test.go
@@ -2,7 +2,7 @@
import "testing"
-func BenchmarkCompositeBarrierLoad(b *testing.B) {
+func BenchmarkCompositeBarrierRead(b *testing.B) {
barrier := NewCompositeBarrier(NewCursor(), NewCursor(), NewCursor(), NewCursor())
iterations := int64(b.N)
@@ -10,6 +10,6 @@
b.ResetTimer()
for i := int64(0); i < iterations; i++ {
- barrier.LoadBarrier(0)
+ barrier.Read(0)
}
}
diff --git a/consumer.go b/consumer.go
index 6e6998b..9d55e7c 100644
--- a/consumer.go
+++ b/consumer.go
@@ -1,5 +1,6 @@
package disruptor
+// TODO: investigate performance impact of Consume(sequence, remaining int64) (consumed int64)
type Consumer interface {
- Consume(sequence, remaining int64)
+ Consume(lower, upper int64)
}
diff --git a/cursor.go b/cursor.go
index 62990df..d69bedd 100644
--- a/cursor.go
+++ b/cursor.go
@@ -1,11 +1,9 @@
package disruptor
const (
- cpuCacheLinePadding = 7
- InitialSequenceValue int64 = -1
MaxSequenceValue int64 = (1 << 63) - 1
- Gating = InitialSequenceValue - 1
- Idling = Gating - 1
+ InitialSequenceValue int64 = -1
+ cpuCacheLinePadding = 7
)
// TODO: research aligned read/write:
@@ -14,9 +12,10 @@
// https://gist.github.com/anachronistic/7495541
// http://blog.chewxy.com/2013/12/10/pointer-tagging-in-go/
// http://www.goinggo.net/2014/01/concurrency-goroutines-and-gomaxprocs.html
+
type Cursor struct {
sequence int64
- padding [cpuCacheLinePadding]int64 // https://github.com/fmstephe/fatomic/blob/master/paddedint.go
+ padding [cpuCacheLinePadding]int64
}
func NewCursor() *Cursor {
diff --git a/cursor_386.go b/cursor_386.go
index 96d60b9..e93a3f1 100644
--- a/cursor_386.go
+++ b/cursor_386.go
@@ -5,9 +5,11 @@
func (this *Cursor) Store(sequence int64) {
atomic.StoreInt64(&this.sequence, sequence)
}
+
func (this *Cursor) Load() int64 {
return atomic.LoadInt64(&this.sequence)
}
-func (this *Cursor) LoadBarrier(next int64) int64 {
+
+func (this *Cursor) Read(noop int64) int64 {
return atomic.LoadInt64(&this.sequence)
}
diff --git a/cursor_amd64.go b/cursor_amd64.go
index 19dbb63..dd021a3 100644
--- a/cursor_amd64.go
+++ b/cursor_amd64.go
@@ -3,9 +3,11 @@
func (this *Cursor) Store(sequence int64) {
this.sequence = sequence
}
+
func (this *Cursor) Load() int64 {
return this.sequence
}
-func (this *Cursor) LoadBarrier(next int64) int64 {
+
+func (this *Cursor) Read(noop int64) int64 {
return this.sequence
}
diff --git a/cursor_arm.go b/cursor_arm.go
index 96d60b9..e93a3f1 100644
--- a/cursor_arm.go
+++ b/cursor_arm.go
@@ -5,9 +5,11 @@
func (this *Cursor) Store(sequence int64) {
atomic.StoreInt64(&this.sequence, sequence)
}
+
func (this *Cursor) Load() int64 {
return atomic.LoadInt64(&this.sequence)
}
-func (this *Cursor) LoadBarrier(next int64) int64 {
+
+func (this *Cursor) Read(noop int64) int64 {
return atomic.LoadInt64(&this.sequence)
}
diff --git a/cursor_test.go b/cursor_test.go
index 4bc1e0d..e491a4a 100644
--- a/cursor_test.go
+++ b/cursor_test.go
@@ -2,9 +2,23 @@
import "testing"
-func BenchmarkCursorLoad(b *testing.B) {
- cursor := NewCursor()
+func BenchmarkCursorStore(b *testing.B) {
iterations := int64(b.N)
+
+ cursor := NewCursor()
+
+ b.ReportAllocs()
+ b.ResetTimer()
+
+ for i := int64(0); i < iterations; i++ {
+ cursor.Store(i)
+ }
+}
+func BenchmarkCursorLoad(b *testing.B) {
+ iterations := int64(b.N)
+
+ cursor := NewCursor()
+
b.ReportAllocs()
b.ResetTimer()
@@ -12,14 +26,27 @@
cursor.Load()
}
}
+func BenchmarkCursorRead(b *testing.B) {
+ iterations := int64(b.N)
-func BenchmarkCursorStore(b *testing.B) {
cursor := NewCursor()
+
+ b.ReportAllocs()
+ b.ResetTimer()
+
+ for i := int64(0); i < iterations; i++ {
+ cursor.Read(i)
+ }
+}
+
+func BenchmarkCursorReadAsBarrier(b *testing.B) {
+ var barrier Barrier = NewCursor()
+
iterations := int64(b.N)
b.ReportAllocs()
b.ResetTimer()
for i := int64(0); i < iterations; i++ {
- cursor.Store(i)
+ barrier.Read(0)
}
}
diff --git a/example/amd64.sh b/example/amd64.sh
deleted file mode 100755
index 42be16e..0000000
--- a/example/amd64.sh
+++ /dev/null
@@ -1,4 +0,0 @@
-#!/bin/bash
-set -e
-
-go install && example
diff --git a/example/android.sh b/example/android.sh
deleted file mode 100755
index b332b59..0000000
--- a/example/android.sh
+++ /dev/null
@@ -1,6 +0,0 @@
-#!/bin/bash
-set -e
-
-GOOS=linux GOARCH=arm GOARM=7 go build -o go-disruptor
-adb push go-disruptor /data/local/tmp
-adb shell "cd /data/local/tmp; chmod 755 go-disruptor; ./go-disruptor"
diff --git a/example/example_consumer.go b/example/example_consumer.go
deleted file mode 100644
index f6eac20..0000000
--- a/example/example_consumer.go
+++ /dev/null
@@ -1,56 +0,0 @@
-package main
-
-import (
- "fmt"
- "time"
-
- "github.com/smartystreets/go-disruptor"
-)
-
-func consume0(reader *disruptor.SimpleReader) {
- for {
- reader.Receive()
- }
-}
-func consume1(reader *disruptor.Reader) {
- started := time.Now()
-
- for {
- if lower, upper := reader.Receive(); upper >= 0 {
- for sequence := lower; sequence <= upper; sequence++ {
- if sequence%ReportingFrequency == 0 {
- finished := time.Now()
- fmt.Println(sequence, finished.Sub(started))
- started = time.Now()
- }
-
- message := ringBuffer[sequence&RingMask]
- if sequence != message {
- alert := fmt.Sprintf("***Race Condition***\tSequence: %d, Message %d\n", sequence, message)
- fmt.Println(alert)
- panic(alert)
- }
-
- ringBuffer[sequence&RingMask] = sequence % 2
- }
-
- reader.Commit(upper)
- }
- }
-}
-
-func consume2(reader *disruptor.Reader) {
- for {
- if lower, upper := reader.Receive(); upper >= 0 {
- for sequence := lower; sequence <= upper; sequence++ {
- message := ringBuffer[sequence&RingMask]
- if message != sequence%2 {
- alert := fmt.Sprintf("Race Condition (Layer 2)::Sequence: %d, Message %d\n", sequence, message)
- fmt.Print(alert)
- panic(alert)
- }
- }
- reader.Commit(upper)
- }
- }
-}
diff --git a/example/example_consumer_handler.go b/example/example_consumer_handler.go
deleted file mode 100644
index 7823b58..0000000
--- a/example/example_consumer_handler.go
+++ /dev/null
@@ -1,31 +0,0 @@
-package main
-
-import (
- "fmt"
- "time"
-
- "github.com/smartystreets/go-disruptor"
-)
-
-type ExampleConsumerHandler struct {
- started time.Time
-}
-
-func NewExampleConsumerHandler() disruptor.Consumer {
- return &ExampleConsumerHandler{started: time.Now()}
-}
-
-func (this *ExampleConsumerHandler) Consume(sequence, remaining int64) {
- if sequence%ReportingFrequency == 0 {
- finished := time.Now()
- fmt.Println(sequence, finished.Sub(this.started))
- this.started = time.Now()
- }
-
- if sequence != ringBuffer[sequence&RingMask] {
- message := ringBuffer[sequence&RingMask]
- alert := fmt.Sprintf("Race Condition::Sequence: %d, Message %d\n", sequence, message)
- fmt.Print(alert)
- panic(alert)
- }
-}
diff --git a/example/example_producer.go b/example/example_producer.go
deleted file mode 100644
index ac85259..0000000
--- a/example/example_producer.go
+++ /dev/null
@@ -1,15 +0,0 @@
-package main
-
-import "github.com/smartystreets/go-disruptor"
-
-func publish(writer *disruptor.SharedWriter) {
-
- for {
- if lower, upper := writer.Reserve(ItemsToPublish); upper != disruptor.Gating {
- for sequence := lower; sequence <= upper; sequence++ {
- ringBuffer[sequence&RingMask] = sequence
- }
- writer.Commit(lower, upper)
- }
- }
-}
diff --git a/example/main.go b/example/main.go
index 5428882..d375917 100644
--- a/example/main.go
+++ b/example/main.go
@@ -1,80 +1,84 @@
package main
import (
+ "fmt"
"runtime"
+ "time"
"github.com/smartystreets/go-disruptor"
)
const (
- MaxConsumersPerGroup = 1
- MaxConsumerGroups = 2
- MaxProducers = 2
- ItemsToPublish = 4
- ReportingFrequency = 1000000 * 10 // 1 million * N
- RingSize = 1024 * 16
- RingMask = RingSize - 1
+ BufferSize = 1024 * 64
+ BufferMask = BufferSize - 1
+ Iterations = 1000000 * 100
)
-var ringBuffer [RingSize]int64
+var ringBuffer = [BufferSize]int64{}
func main() {
- runtime.GOMAXPROCS(MaxConsumerGroups*MaxConsumersPerGroup + MaxProducers)
+ runtime.GOMAXPROCS(2)
- // written := disruptor.NewCursor()
- // upstream := startConsumerGroups(written, written)
- // writer := disruptor.NewWriter(written, upstream, RingSize)
- // startExclusiveProducer(writer)
+ written, read := disruptor.NewCursor(), disruptor.NewCursor()
+ reader := disruptor.NewReader(read, written, written, SampleConsumer{})
- written := disruptor.NewCursor()
- shared := disruptor.NewSharedWriterBarrier(written, RingSize)
- upstream := startConsumerGroups(shared, written)
- writer := disruptor.NewSharedWriter(shared, upstream)
- startSharedProducers(writer)
+ started := time.Now()
+ reader.Start()
+ publish(written, read)
+ reader.Stop()
+ finished := time.Now()
+ fmt.Println(Iterations, finished.Sub(started))
}
-// func startExclusiveProducer(writer *disruptor.Writer) {
-// publish(writer)
+const Reservations = 1
+
+func publish(written *disruptor.Cursor, upstream disruptor.Barrier) {
+ sequence := disruptor.InitialSequenceValue
+ writer := disruptor.NewWriter(written, upstream, BufferSize)
+ for sequence <= Iterations {
+ sequence += Reservations
+ writer.Await(sequence)
+ ringBuffer[sequence&BufferMask] = sequence
+ writer.Commit(sequence, sequence)
+ }
+}
+
+// func publish(written *disruptor.Cursor, upstream disruptor.Barrier) {
+// sequence := disruptor.InitialSequenceValue
+// writer := disruptor.NewWriter(written, upstream, BufferSize)
+// for sequence <= Iterations {
+// sequence = writer.Reserve(Reservations)
+// ringBuffer[sequence&BufferMask] = sequence
+// writer.Commit(sequence, sequence)
+// }
// }
-func startSharedProducers(writer *disruptor.SharedWriter) {
- for i := 0; i < MaxProducers-1; i++ {
- go publish(writer)
- }
+// func publish(written *disruptor.Cursor, upstream disruptor.Barrier) {
+// previous := disruptor.InitialSequenceValue
+// gate := disruptor.InitialSequenceValue
- publish(writer)
-}
+// for previous <= Iterations {
+// next := previous + 1
-func startConsumerGroups(upstream disruptor.Barrier, written *disruptor.Cursor) disruptor.Barrier {
- for i := 0; i < MaxConsumerGroups; i++ {
- upstream = startConsumerGroup(i, upstream, written)
- }
+// for next-BufferSize > gate {
+// gate = upstream.Read(next)
+// }
- return upstream
-}
-func startConsumerGroup(group int, upstream disruptor.Barrier, written *disruptor.Cursor) disruptor.Barrier {
- cursors := []*disruptor.Cursor{}
+// ringBuffer[next&BufferMask] = next
+// written.Store(next)
+// previous = next
+// }
+// }
- for i := 0; i < MaxConsumersPerGroup; i++ {
- read := disruptor.NewCursor()
- cursors = append(cursors, read)
- reader := disruptor.NewReader(read, written, upstream)
+type SampleConsumer struct{}
- // constant time regardless of the number of items
- // go consume0(disruptor.NewSimpleReader(reader, NewExampleConsumerHandler()))
-
- // TODO: wildly sporadic latency for single-item publish, e.g. 2 seconds, 65 ms, etc.
- // faster for 2-3+ items per publish
- // go consume1(reader)
-
- if group == 0 {
- go consume1(reader)
- } else if group == 1 {
- go consume2(reader)
- } else {
- panic("only two consumer groups currently supported.")
+func (this SampleConsumer) Consume(lower, upper int64) {
+ for lower <= upper {
+ message := ringBuffer[lower&BufferMask]
+ if message != lower {
+ fmt.Println("Race condition", message, lower)
+ panic("Race condition")
}
+ lower++
}
-
- return disruptor.NewCompositeBarrier(cursors...)
}
diff --git a/reader.go b/reader.go
index e60ff31..4ffda19 100644
--- a/reader.go
+++ b/reader.go
@@ -1,28 +1,59 @@
package disruptor
+import "time"
+
type Reader struct {
read *Cursor
written *Cursor
upstream Barrier
+ consumer Consumer
+ ready bool
}
-func NewReader(read, written *Cursor, upstream Barrier) *Reader {
+func NewReader(read, written *Cursor, upstream Barrier, consumer Consumer) *Reader {
return &Reader{
read: read,
written: written,
upstream: upstream,
+ consumer: consumer,
+ ready: false,
}
}
-func (this *Reader) Receive() (int64, int64) {
- lower := this.read.Load() + 1
- upper := this.upstream.LoadBarrier(lower)
+func (this *Reader) Start() {
+ this.ready = true
+ go this.receive()
+}
+func (this *Reader) Stop() {
+ this.ready = false
+}
- if lower <= upper {
- return lower, upper
- } else if gate := this.written.Load(); lower <= gate {
- return InitialSequenceValue, Gating
- } else {
- return InitialSequenceValue, Idling
+func (this *Reader) receive() {
+ previous := this.read.Load()
+ idling, gating := 0, 0
+
+ for {
+ lower := previous + 1
+ upper := this.upstream.Read(lower)
+
+ if lower <= upper {
+ this.consumer.Consume(lower, upper)
+ this.read.Store(upper)
+ previous = upper
+ } else if upper = this.written.Load(); lower <= upper {
+ // Gating--TODO: wait strategy (provide gating count to wait strategy for phased backoff)
+ gating++
+ idling = 0
+ } else if this.ready {
+ // Idling--TODO: wait strategy (provide idling count to wait strategy for phased backoff)
+ idling++
+ gating = 0
+ } else {
+ break
+ }
+
+ // sleeping increases the batch size which reduces number of writes required to store the sequence
+ // reducing the number of writes allows the CPU to optimize the pipeline without prediction failures
+ time.Sleep(time.Microsecond)
}
}
diff --git a/reader_386.go b/reader_386.go
deleted file mode 100644
index f1df97c..0000000
--- a/reader_386.go
+++ /dev/null
@@ -1,5 +0,0 @@
-package disruptor
-
-func (this *Reader) Commit(sequence int64) {
- this.read.Store(sequence)
-}
diff --git a/reader_amd64.go b/reader_amd64.go
deleted file mode 100644
index 36236b2..0000000
--- a/reader_amd64.go
+++ /dev/null
@@ -1,5 +0,0 @@
-package disruptor
-
-func (this *Reader) Commit(sequence int64) {
- this.read.sequence = sequence
-}
diff --git a/reader_arm.go b/reader_arm.go
deleted file mode 100644
index f1df97c..0000000
--- a/reader_arm.go
+++ /dev/null
@@ -1,5 +0,0 @@
-package disruptor
-
-func (this *Reader) Commit(sequence int64) {
- this.read.Store(sequence)
-}
diff --git a/reader_test.go b/reader_test.go
deleted file mode 100644
index 6c39d84..0000000
--- a/reader_test.go
+++ /dev/null
@@ -1,21 +0,0 @@
-package disruptor
-
-import "testing"
-
-func BenchmarkReader(b *testing.B) {
- written := NewCursor()
- read := NewCursor()
- reader := NewReader(written, written, read)
-
- written.Store(1)
-
- iterations := int64(b.N)
- b.ReportAllocs()
- b.ResetTimer()
-
- for i := int64(0); i < iterations; i++ {
- _, upper := reader.Receive()
- reader.Commit(upper)
- read.Store(0)
- }
-}
diff --git a/readme.md b/readme.md
index f5783ec..d27b400 100644
--- a/readme.md
+++ b/readme.md
@@ -15,14 +15,21 @@
Each of the following benchmark tests sends an incrementing sequence message from one goroutine to another. The receiving goroutine asserts that the message is received is the expected incrementing sequence value. Any failures cause a panic. All tests were run using `GOMAXPROCS=2`.
Scenario | Per Operation Time
--------- | ------------------
-Channel: Non-blocking | 681 ns/op
-Channel: Blocking | 86.6 ns/op|
-Disruptor: SharedWriter (single claim) | 55.2 ns/op
-Disruptor: SharedWriter (multi claim) | 2.59 ns/op
-Disruptor: Writer (single claim) | 18.6 ns/op
-Disruptor: Writer (multi claim) | 1.12 ns/op
-
+-------- | ------------------
+Channels: Buffered, Blocking, GOMAXPROCS=1 | 58.6 ns/op
+Channels: Buffered, Blocking, GOMAXPROCS=2 | 86.6 ns/op
+Channels: Buffered, Blocking, GOMAXPROCS=3, Contended Write | 194 ns/op
+Channels: Buffered, Non-blocking, GOMAXPROCS=1| 73.9 ns/op
+Channels: Buffered, Non-blocking, GOMAXPROCS=2| 72.3 ns/op
+Channels: Buffered, Non-blocking, GOMAXPROCS=3, Contended Write | 259 ns/op
+Disruptor: Writer, Reserve One | 4.3 ns/op
+Disruptor: Writer, Reserve Many | 1.1 ns/op
+Disruptor: Writer, Await One | 3.5 ns/op
+Disruptor: Writer, Await Many | 1.0 ns/op
+Disruptor: SharedWriter, Reserve One | 14.5 ns/op
+Disruptor: SharedWriter, Reserve Many | 2.5 ns/op
+Disruptor: SharedWriter, Reserve One, Contended Write | nn.n ns/op
+Disruptor: SharedWriter, Reserve Many, Contended Write | nn.n ns/op
When In Doubt, Use Channels
----------------------------
diff --git a/shared_writer.go b/shared_writer.go
index e9d07a0..e0820d5 100644
--- a/shared_writer.go
+++ b/shared_writer.go
@@ -3,53 +3,52 @@
import "sync/atomic"
type SharedWriter struct {
+ written *Cursor
+ upstream Barrier
capacity int64
gate *Cursor
mask int64
shift uint8
committed []int32
- upstream Barrier
- written *Cursor
}
func NewSharedWriter(write *SharedWriterBarrier, upstream Barrier) *SharedWriter {
return &SharedWriter{
+ written: write.written,
+ upstream: upstream,
capacity: write.capacity,
gate: NewCursor(),
mask: write.mask,
shift: write.shift,
committed: write.committed,
- upstream: upstream,
- written: write.written,
}
}
-func (this *SharedWriter) Reserve(count int64) (int64, int64) {
+func (this *SharedWriter) Reserve(count int64) int64 {
for {
previous := this.written.Load()
upper := previous + count
- wrap := upper - this.capacity
- if wrap > this.gate.Load() {
- min := this.upstream.LoadBarrier(0)
- if wrap > min {
- return InitialSequenceValue, Gating
- }
-
- this.gate.Store(min)
+ for upper-this.capacity > this.gate.Load() {
+ this.gate.Store(this.upstream.Read(0))
}
if atomic.CompareAndSwapInt64(&this.written.sequence, previous, upper) {
- return previous + 1, upper
+ return upper
}
}
}
func (this *SharedWriter) Commit(lower, upper int64) {
- // POTENTIAL TODO: start from upper and work toward lower
- // this may have the effect of keeping a batch together which
- // might otherwise be split up...
- for shift, mask := this.shift, this.mask; lower <= upper; lower++ {
- this.committed[lower&mask] = int32(lower >> shift)
+ if lower == upper {
+ this.committed[upper&this.mask] = int32(upper >> this.shift)
+ } else {
+ // working down the array keeps all items in the commit together
+ // otherwise the reader(s) could split up the group
+ for upper >= lower {
+ this.committed[upper&this.mask] = int32(upper >> this.shift)
+ upper--
+ }
+
}
}
diff --git a/shared_writer_barrier.go b/shared_writer_barrier.go
index a2d4204..0283e59 100644
--- a/shared_writer_barrier.go
+++ b/shared_writer_barrier.go
@@ -3,22 +3,22 @@
import "math"
type SharedWriterBarrier struct {
+ written *Cursor
+ committed []int32
capacity int64
mask int64
shift uint8
- committed []int32
- written *Cursor
}
func NewSharedWriterBarrier(written *Cursor, capacity int64) *SharedWriterBarrier {
assertPowerOfTwo(capacity)
return &SharedWriterBarrier{
+ written: written,
+ committed: prepareCommitBuffer(capacity),
capacity: capacity,
mask: capacity - 1,
shift: uint8(math.Log2(float64(capacity))),
- committed: prepareCommitBuffer(capacity),
- written: written,
}
}
func prepareCommitBuffer(capacity int64) []int32 {
@@ -29,7 +29,7 @@
return buffer
}
-func (this *SharedWriterBarrier) LoadBarrier(lower int64) int64 {
+func (this *SharedWriterBarrier) Read(lower int64) int64 {
shift, mask := this.shift, this.mask
upper := this.written.Load()
diff --git a/simple_reader.go b/simple_reader.go
deleted file mode 100644
index 8fc5b20..0000000
--- a/simple_reader.go
+++ /dev/null
@@ -1,24 +0,0 @@
-package disruptor
-
-type SimpleReader struct {
- reader *Reader
- callback Consumer
-}
-
-func NewSimpleReader(reader *Reader, callback Consumer) *SimpleReader {
- return &SimpleReader{reader: reader, callback: callback}
-}
-
-func (this *SimpleReader) Receive() (int64, int64) {
- lower, upper := this.reader.Receive()
-
- if lower <= upper {
- for sequence := lower; sequence <= upper; sequence++ {
- this.callback.Consume(sequence, upper-sequence)
- }
-
- this.reader.Commit(upper)
- }
-
- return lower, upper
-}
diff --git a/writer.go b/writer.go
index 093f8c0..5f44bc6 100644
--- a/writer.go
+++ b/writer.go
@@ -3,9 +3,9 @@
type Writer struct {
written *Cursor
upstream Barrier
+ capacity int64
previous int64
gate int64
- capacity int64
}
func NewWriter(written *Cursor, upstream Barrier, capacity int64) *Writer {
@@ -14,32 +14,29 @@
return &Writer{
upstream: upstream,
written: written,
+ capacity: capacity,
previous: InitialSequenceValue,
gate: InitialSequenceValue,
- capacity: capacity,
}
}
func assertPowerOfTwo(value int64) {
if value > 0 && (value&(value-1)) != 0 {
- // http://en.wikipedia.org/wiki/Power_of_two#Fast_algorithm_to_check_if_a_positive_number_is_a_power_of_two
+ // Wikipedia entry: http://bit.ly/1krhaSB
panic("The ring capacity must be a power of two, e.g. 2, 4, 8, 16, 32, 64, etc.")
}
}
-func (this *Writer) Reserve(count int64) (int64, int64) {
- upper := this.previous + count
- wrap := upper - this.capacity
-
- if wrap > this.gate {
- min := this.upstream.LoadBarrier(0)
- if wrap > min {
- return InitialSequenceValue, Gating
- }
-
- this.gate = min
+func (this *Writer) Reserve(count int64) int64 {
+ this.previous += count
+ for this.previous-this.capacity > this.gate {
+ this.gate = this.upstream.Read(0)
}
+ return this.previous
+}
- this.previous = upper
- return upper - count + 1, upper
+func (this *Writer) Await(next int64) {
+ for next-this.capacity > this.gate {
+ this.gate = this.upstream.Read(0)
+ }
}
diff --git a/writer_test.go b/writer_test.go
index 46cfeb3..30edd90 100644
--- a/writer_test.go
+++ b/writer_test.go
@@ -2,36 +2,21 @@
import "testing"
-func BenchmarkWriterCommit(b *testing.B) {
- writer := NewWriter(NewCursor(), nil, 1024)
- iterations := int64(b.N)
- b.ReportAllocs()
- b.ResetTimer()
-
- for i := int64(0); i < iterations; i++ {
- writer.Commit(i, i)
- }
-}
-
func BenchmarkWriterReserve(b *testing.B) {
- read := NewCursor()
- written := NewCursor()
-
+ read, written := NewCursor(), NewCursor()
writer := NewWriter(written, read, 1024)
iterations := int64(b.N)
b.ReportAllocs()
b.ResetTimer()
for i := int64(0); i < iterations; i++ {
- claimed, _ := writer.Reserve(1)
- read.Store(claimed)
+ sequence := writer.Reserve(1)
+ read.Store(sequence)
}
}
func BenchmarkWriterNextWrapPoint(b *testing.B) {
- read := NewCursor()
- written := NewCursor()
-
+ read, written := NewCursor(), NewCursor()
writer := NewWriter(written, read, 1024)
iterations := int64(b.N)
b.ReportAllocs()
@@ -42,3 +27,27 @@
writer.Reserve(1)
}
}
+
+func BenchmarkWriterAwait(b *testing.B) {
+ written, read := NewCursor(), NewCursor()
+ writer := NewWriter(written, read, 1024*64)
+ iterations := int64(b.N)
+ b.ReportAllocs()
+ b.ResetTimer()
+
+ for i := int64(0); i < iterations; i++ {
+ writer.Await(i)
+ read.Store(i)
+ }
+}
+
+func BenchmarkWriterCommit(b *testing.B) {
+ writer := NewWriter(NewCursor(), nil, 1024)
+ iterations := int64(b.N)
+ b.ReportAllocs()
+ b.ResetTimer()
+
+ for i := int64(0); i < iterations; i++ {
+ writer.Commit(i, i)
+ }
+}
