Clean slate.
diff --git a/TODO b/TODO
deleted file mode 100644
index 7eaf300..0000000
--- a/TODO
+++ /dev/null
@@ -1,41 +0,0 @@
-- 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:
- 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?)
-
-- Website with cool images and documentation? (e.g. GoConvey)
\ No newline at end of file
diff --git a/barrier.go b/barrier.go
deleted file mode 100644
index 3fcecb2..0000000
--- a/barrier.go
+++ /dev/null
@@ -1,5 +0,0 @@
-package disruptor
-
-type Barrier interface {
- LoadBarrier(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/batch_reader.go b/batch_reader.go
deleted file mode 100644
index 82a7aef..0000000
--- a/batch_reader.go
+++ /dev/null
@@ -1,50 +0,0 @@
-package disruptor
-
-type BatchReader struct {
- started bool
- read *Cursor
- written *Cursor
- upstream Barrier
- consumer Consumer
- waiter Waiter
-}
-
-func NewBatchReader(read, written *Cursor, upstream Barrier, consumer Consumer, waiter Waiter) *BatchReader {
- return &BatchReader{
- started: false,
- read: read,
- written: written,
- upstream: upstream,
- consumer: consumer,
- waiter: waiter,
- }
-}
-
-func (this *BatchReader) Start() {
- this.started = true
- go this.receive()
-}
-func (this *BatchReader) Stop() {
- this.started = false
-}
-
-func (this *BatchReader) receive() {
- sequence := this.read.Load()
-
- for {
- lower := sequence + 1
- upper := this.upstream.LoadBarrier(lower)
-
- if lower <= upper {
- this.consumer.Consume(lower, upper)
- sequence = upper
- this.read.Store(sequence)
- } else if gate := this.written.Load(); lower <= gate {
- // time.Sleep(time.Millisecond) // TODO: use another method from the wait strategy?
- } else if this.started {
- // this.waiter.Wait()
- } else {
- break
- }
- }
-}
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
deleted file mode 100644
index e9d2211..0000000
--- a/composite_barrier.go
+++ /dev/null
@@ -1,27 +0,0 @@
-package disruptor
-
-type CompositeBarrier struct {
- cursors []*Cursor
-}
-
-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}
-}
-
-func (this *CompositeBarrier) LoadBarrier(current int64) int64 {
- minimum := MaxSequenceValue
- for _, item := range this.cursors {
- sequence := item.Load()
- if sequence < minimum {
- minimum = sequence
- }
- }
-
- return minimum
-}
diff --git a/composite_barrier_test.go b/composite_barrier_test.go
deleted file mode 100644
index 1a72733..0000000
--- a/composite_barrier_test.go
+++ /dev/null
@@ -1,15 +0,0 @@
-package disruptor
-
-import "testing"
-
-func BenchmarkCompositeBarrierLoad(b *testing.B) {
- barrier := NewCompositeBarrier(NewCursor(), NewCursor(), NewCursor(), NewCursor())
-
- iterations := int64(b.N)
- b.ReportAllocs()
- b.ResetTimer()
-
- for i := int64(0); i < iterations; i++ {
- barrier.LoadBarrier(0)
- }
-}
diff --git a/consumer.go b/consumer.go
deleted file mode 100644
index c1971c8..0000000
--- a/consumer.go
+++ /dev/null
@@ -1,5 +0,0 @@
-package disruptor
-
-type Consumer interface {
- Consume(lower, upper int64)
-}
diff --git a/cursor.go b/cursor.go
deleted file mode 100644
index 62990df..0000000
--- a/cursor.go
+++ /dev/null
@@ -1,24 +0,0 @@
-package disruptor
-
-const (
- cpuCacheLinePadding = 7
- InitialSequenceValue int64 = -1
- MaxSequenceValue int64 = (1 << 63) - 1
- Gating = InitialSequenceValue - 1
- Idling = Gating - 1
-)
-
-// TODO: research aligned read/write:
-// https://github.com/fmstephe/fatomic/blob/master/slice.go
-// https://groups.google.com/forum/#!topic/golang-nuts/XDfQUn4U_g8
-// 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
-}
-
-func NewCursor() *Cursor {
- return &Cursor{sequence: InitialSequenceValue}
-}
diff --git a/cursor_386.go b/cursor_386.go
deleted file mode 100644
index 96d60b9..0000000
--- a/cursor_386.go
+++ /dev/null
@@ -1,13 +0,0 @@
-package disruptor
-
-import "sync/atomic"
-
-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 {
- return atomic.LoadInt64(&this.sequence)
-}
diff --git a/cursor_amd64.go b/cursor_amd64.go
deleted file mode 100644
index 19dbb63..0000000
--- a/cursor_amd64.go
+++ /dev/null
@@ -1,11 +0,0 @@
-package disruptor
-
-func (this *Cursor) Store(sequence int64) {
- this.sequence = sequence
-}
-func (this *Cursor) Load() int64 {
- return this.sequence
-}
-func (this *Cursor) LoadBarrier(next int64) int64 {
- return this.sequence
-}
diff --git a/cursor_arm.go b/cursor_arm.go
deleted file mode 100644
index 96d60b9..0000000
--- a/cursor_arm.go
+++ /dev/null
@@ -1,13 +0,0 @@
-package disruptor
-
-import "sync/atomic"
-
-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 {
- return atomic.LoadInt64(&this.sequence)
-}
diff --git a/cursor_test.go b/cursor_test.go
deleted file mode 100644
index 4bc1e0d..0000000
--- a/cursor_test.go
+++ /dev/null
@@ -1,25 +0,0 @@
-package disruptor
-
-import "testing"
-
-func BenchmarkCursorLoad(b *testing.B) {
- cursor := NewCursor()
- iterations := int64(b.N)
- b.ReportAllocs()
- b.ResetTimer()
-
- for i := int64(0); i < iterations; i++ {
- cursor.Load()
- }
-}
-
-func BenchmarkCursorStore(b *testing.B) {
- cursor := NewCursor()
- iterations := int64(b.N)
- b.ReportAllocs()
- b.ResetTimer()
-
- for i := int64(0); i < iterations; i++ {
- cursor.Store(i)
- }
-}
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 2c4057e..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 cedf859..0000000
--- a/example/example_consumer_handler.go
+++ /dev/null
@@ -1,35 +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(lower, upper int64) {
- sequence := lower
- //for sequence := lower; sequence <= upper; sequence++ {
-
- 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 5ca9939..0000000
--- a/example/example_producer.go
+++ /dev/null
@@ -1,16 +0,0 @@
-package main
-
-import "github.com/smartystreets/go-disruptor"
-
-func publish(writer *disruptor.Writer) {
-
- for {
- if lower, upper := writer.Reserve(ItemsToPublish); upper != disruptor.Gating {
- // for sequence := lower; sequence <= upper; sequence++ {
- // ringBuffer[sequence&RingMask] = sequence
- // }
- // ringBuffer[lower&RingMask] = lower
- writer.Commit(lower, upper)
- }
- }
-}
diff --git a/example/main.go b/example/main.go
deleted file mode 100644
index 2b58e25..0000000
--- a/example/main.go
+++ /dev/null
@@ -1,85 +0,0 @@
-package main
-
-import (
- "runtime"
-
- "github.com/smartystreets/go-disruptor"
-)
-
-const (
- MaxConsumersPerGroup = 1
- MaxConsumerGroups = 1
- MaxProducers = 1
- ItemsToPublish = 1
- ReportingFrequency = 1000000 * 100 // 1 million * N
- RingSize = 1024 * 16
- RingMask = RingSize - 1
-)
-
-var ringBuffer [RingSize]int64
-
-func main() {
- runtime.GOMAXPROCS(MaxConsumerGroups*MaxConsumersPerGroup + MaxProducers)
-
- written := disruptor.NewCursor()
- upstream := startConsumerGroups(written, written)
- writer := disruptor.NewWriter(written, upstream, RingSize)
- startExclusiveProducer(writer)
-
- // written := disruptor.NewCursor()
- // shared := disruptor.NewSharedWriterBarrier(written, RingSize)
- // upstream := startConsumerGroups(shared, written)
- // writer := disruptor.NewSharedWriter(shared, upstream)
- // startSharedProducers(writer)
-}
-
-func startExclusiveProducer(writer *disruptor.Writer) {
- publish(writer)
-}
-
-// func startSharedProducers(writer *disruptor.SharedWriter) {
-// for i := 0; i < MaxProducers-1; i++ {
-// go publish(writer)
-// }
-
-// publish(writer)
-// }
-
-func startConsumerGroups(upstream disruptor.Barrier, written *disruptor.Cursor) disruptor.Barrier {
- for i := 0; i < MaxConsumerGroups; i++ {
- upstream = startConsumerGroup(i, upstream, written)
- }
-
- return upstream
-}
-func startConsumerGroup(group int, upstream disruptor.Barrier, written *disruptor.Cursor) disruptor.Barrier {
- cursors := []*disruptor.Cursor{}
-
- for i := 0; i < MaxConsumersPerGroup; i++ {
- read := disruptor.NewCursor()
- cursors = append(cursors, read)
- // reader := disruptor.NewReader(read, written, upstream)
-
- consumer := NewExampleConsumerHandler()
- waiter := SleepWaiter{}
-
- disruptor.NewBatchReader(read, written, upstream, consumer, waiter).Start()
-
- // 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.")
- // }
- }
-
- return disruptor.NewCompositeBarrier(cursors...)
-}
diff --git a/example/sleep_waiter.go b/example/sleep_waiter.go
deleted file mode 100644
index 941fd36..0000000
--- a/example/sleep_waiter.go
+++ /dev/null
@@ -1,6 +0,0 @@
-package main
-
-type SleepWaiter struct{}
-
-func (this SleepWaiter) Wait() {
-}
diff --git a/reader.go b/reader.go
deleted file mode 100644
index e60ff31..0000000
--- a/reader.go
+++ /dev/null
@@ -1,28 +0,0 @@
-package disruptor
-
-type Reader struct {
- read *Cursor
- written *Cursor
- upstream Barrier
-}
-
-func NewReader(read, written *Cursor, upstream Barrier) *Reader {
- return &Reader{
- read: read,
- written: written,
- upstream: upstream,
- }
-}
-
-func (this *Reader) Receive() (int64, int64) {
- lower := this.read.Load() + 1
- upper := this.upstream.LoadBarrier(lower)
-
- if lower <= upper {
- return lower, upper
- } else if gate := this.written.Load(); lower <= gate {
- return InitialSequenceValue, Gating
- } else {
- return InitialSequenceValue, Idling
- }
-}
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
deleted file mode 100644
index f5783ec..0000000
--- a/readme.md
+++ /dev/null
@@ -1,41 +0,0 @@
-Notes
-=====
-
-Disruptor Overview
-----------------------------
-
-This is a port of the [LMAX Disruptor](https://github.com/LMAX-Exchange/disruptor) into the Go programming language. It retains the essence and spirit of the Disruptor and utilizes a lot of the same abstractions and concepts, but does not maintain the same API.
-
-On my MacBook Pro (Intel Core i7 3740QM @ 2.7 Ghz) using Go 1.2.1, I was able to push over **700 million messages per second** (yes, you read that right) from one goroutine to another goroutine. The message being transferred between the two CPU cores was simply the incrementing sequence number, but literally could be anything. Note that your mileage may vary and that different operating systems can introduce significant “jitter” into the application by taking control of the CPU. Linux and Windows have the ability to assign a given process to specific CPU cores which reduces jitter significantly. Parenthetically, when the Disruptor code is compiled and run on a Nexus 5, it can push about 15-20 million messages per second.
-
-Once initialized and running, one of the preeminent design considerations of the Disruptor is to produce no garbage thus avoiding the need for GC altogether and to avoid locks at all costs. The current channel implementation maintains a big, fat lock around enqueue/dequeue operations and maxes out on the aforementioned hardware at about 25M messages per second for uncontended access-—more than an order of magnitude slower when compared to the Disruptor. The same channel, when contended between OS threads (`GOMAXPROCS=2` or more) only pushes about 7 million messages per second.
-
-Benchmarks
-----------------------------
-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
-
-
-When In Doubt, Use Channels
-----------------------------
-Despite Go channels being significantly slower than the Disruptor, channels should still be considered the best and most desirable choice for the vast majority of all use cases. The Disruptor's target use case is ultra-low latency environments where application response times are measured in nanoseconds and where stable, consistent latency is paramount.
-
-Pre-Alpha
----------
-This code is pre-Alpha stage and is not supported or recommended for production environments. That being said, it has been run non-stop for days without exposing any race conditions. Also, it does not yet contain any unit tests and is meant to be spike code to serve as a proof of concept that the Disruptor is, in fact possible, on the Go runtime despite some of the limits imposed by the Go memory model. The goal is to have an alpha release by mid June 2014 and a series of beta releases each month thereafter until we are satisfied. Following this, a release will be created and supported moving forward.
-
-We are very interested to receive feedback on this project and how performance can be improved using subtle techniques such as additional cache line padding, utilizing a pointer vs a struct in a given location, replacing less optimal techniques with more optimal ones, especially in the performance critical paths of `Reserve`/`Commit` in the various `Writer`s and `Receive`/`Commit` in the `Reader`
-
-Caveats
--------
-One last caveat worth noting. In the Java-based Disruptor implementation, a ring buffer is created, preallocated, and prepopulated with instances of the class which serve as the message type to be transferred between threads. Because Go lacks generics, we have opted to not interact with ring buffers at all within the library code. This has the benefit of avoiding an unnecessary type conversion ("cast") during the receipt of a given message from type `interface{}` to a concrete type. It also means that it is the responsibility of the application developer to create and populate their particular ring buffer during application wireup. Prepopulating the ring buffer at startup should ensure contiguous memory allocation for all items in the various ring buffer slots, whereas on-the-fly creation may introduce gaps in the memory allocation and subsequent CPU cache misses.
-
-The reference to the ring buffer can easily be scoped as a package-level variable. The reason for this is that any given application should have very few Disruptor instances. The instances are designed to be created at startup and stopped during shutdown. They are not typically meant to be created adhoc and passed around. In any case, it is the responsibility of the application developer to manage references to the ring buffer instances such that the producer can push messages in and the consumers can receive messages out.
\ No newline at end of file
diff --git a/shared_writer.go b/shared_writer.go
deleted file mode 100644
index e9d07a0..0000000
--- a/shared_writer.go
+++ /dev/null
@@ -1,55 +0,0 @@
-package disruptor
-
-import "sync/atomic"
-
-type SharedWriter struct {
- capacity int64
- gate *Cursor
- mask int64
- shift uint8
- committed []int32
- upstream Barrier
- written *Cursor
-}
-
-func NewSharedWriter(write *SharedWriterBarrier, upstream Barrier) *SharedWriter {
- return &SharedWriter{
- 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) {
- 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)
- }
-
- if atomic.CompareAndSwapInt64(&this.written.sequence, previous, upper) {
- return previous + 1, 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)
- }
-}
diff --git a/shared_writer_barrier.go b/shared_writer_barrier.go
deleted file mode 100644
index a2d4204..0000000
--- a/shared_writer_barrier.go
+++ /dev/null
@@ -1,43 +0,0 @@
-package disruptor
-
-import "math"
-
-type SharedWriterBarrier struct {
- capacity int64
- mask int64
- shift uint8
- committed []int32
- written *Cursor
-}
-
-func NewSharedWriterBarrier(written *Cursor, capacity int64) *SharedWriterBarrier {
- assertPowerOfTwo(capacity)
-
- return &SharedWriterBarrier{
- capacity: capacity,
- mask: capacity - 1,
- shift: uint8(math.Log2(float64(capacity))),
- committed: prepareCommitBuffer(capacity),
- written: written,
- }
-}
-func prepareCommitBuffer(capacity int64) []int32 {
- buffer := make([]int32, capacity)
- for i := range buffer {
- buffer[i] = int32(InitialSequenceValue)
- }
- return buffer
-}
-
-func (this *SharedWriterBarrier) LoadBarrier(lower int64) int64 {
- shift, mask := this.shift, this.mask
- upper := this.written.Load()
-
- for ; lower <= upper; lower++ {
- if this.committed[lower&mask] != int32(lower>>shift) {
- return lower - 1
- }
- }
-
- return upper
-}
diff --git a/waiter.go b/waiter.go
deleted file mode 100644
index f284021..0000000
--- a/waiter.go
+++ /dev/null
@@ -1,5 +0,0 @@
-package disruptor
-
-type Waiter interface {
- Wait()
-}
diff --git a/writer.go b/writer.go
deleted file mode 100644
index 093f8c0..0000000
--- a/writer.go
+++ /dev/null
@@ -1,45 +0,0 @@
-package disruptor
-
-type Writer struct {
- written *Cursor
- upstream Barrier
- previous int64
- gate int64
- capacity int64
-}
-
-func NewWriter(written *Cursor, upstream Barrier, capacity int64) *Writer {
- assertPowerOfTwo(capacity)
-
- return &Writer{
- upstream: upstream,
- written: written,
- 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
- 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
- }
-
- this.previous = upper
- return upper - count + 1, upper
-}
diff --git a/writer_386.go b/writer_386.go
deleted file mode 100644
index a2a3d76..0000000
--- a/writer_386.go
+++ /dev/null
@@ -1,5 +0,0 @@
-package disruptor
-
-func (this *Writer) Commit(lower, upper int64) {
- this.written.Store(upper)
-}
diff --git a/writer_amd64.go b/writer_amd64.go
deleted file mode 100644
index f3d3fa3..0000000
--- a/writer_amd64.go
+++ /dev/null
@@ -1,5 +0,0 @@
-package disruptor
-
-func (this *Writer) Commit(lower, upper int64) {
- this.written.sequence = upper
-}
diff --git a/writer_arm.go b/writer_arm.go
deleted file mode 100644
index a2a3d76..0000000
--- a/writer_arm.go
+++ /dev/null
@@ -1,5 +0,0 @@
-package disruptor
-
-func (this *Writer) Commit(lower, upper int64) {
- this.written.Store(upper)
-}
diff --git a/writer_test.go b/writer_test.go
deleted file mode 100644
index 46cfeb3..0000000
--- a/writer_test.go
+++ /dev/null
@@ -1,44 +0,0 @@
-package disruptor
-
-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()
-
- 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)
- }
-}
-
-func BenchmarkWriterNextWrapPoint(b *testing.B) {
- read := NewCursor()
- written := NewCursor()
-
- writer := NewWriter(written, read, 1024)
- iterations := int64(b.N)
- b.ReportAllocs()
- b.ResetTimer()
-
- read.Store(MaxSequenceValue)
- for i := int64(0); i < iterations; i++ {
- writer.Reserve(1)
- }
-}
