Added significant logging to drive out shared writer race condition.
diff --git a/example/example_consumer.go b/example/example_consumer.go
index ac65ebf..a84eee5 100644
--- a/example/example_consumer.go
+++ b/example/example_consumer.go
@@ -2,7 +2,6 @@
import (
"fmt"
- "time"
"github.com/smartystreets/go-disruptor"
)
@@ -13,31 +12,26 @@
}
}
func consume1(reader *disruptor.Reader) {
- started := time.Now()
+ // started := time.Now()
- debug := make([]int64, 5)
- debug = debug[0:]
-
+ fmt.Printf("\t\t\t\t\t[CONSUMER] Starting consumer...\n")
for {
sequence, remaining := reader.Receive()
if remaining >= 0 {
- for remaining >= 0 {
+ fmt.Printf("\t\t\t\t\t[CONSUMER] Received messages starting at sequence %d, with %d messages remaining\n", sequence, remaining)
- if sequence%ReportingFrequency == 0 {
- finished := time.Now()
- fmt.Println(sequence, finished.Sub(started))
- started = time.Now()
- }
+ for remaining >= 0 {
+ // if sequence%ReportingFrequency == 0 {
+ // finished := time.Now()
+ // fmt.Println(sequence, finished.Sub(started))
+ // started = time.Now()
+ // }
message := ringBuffer[sequence&RingMask]
+ fmt.Printf("\t\t\t\t\t[CONSUMER] Consuming sequence %d. Message Payload: %d\n", sequence, message)
if sequence != message {
- for i := sequence - 5; i <= sequence; i++ {
- debug = append(debug, ringBuffer[i&RingMask])
- }
-
- alert := fmt.Sprintf("Race Condition::Sequence: %d, Message %d\n", sequence, message)
+ alert := fmt.Sprintf("--------------\n\t\t\t\t\t[CONSUMER] ***Race Condition***::Sequence: %d, Message %d\n", sequence, message)
fmt.Println(alert)
- fmt.Println("Partial Ring Buffer Snapshot:", debug)
panic(alert)
}
@@ -46,8 +40,16 @@
remaining--
sequence++
}
+
+ fmt.Println("\t\t\t\t\t[CONSUMER] All messages consumed, committing up to sequence ", sequence-1)
reader.Commit(sequence - 1)
} else {
+ if remaining == disruptor.Gating {
+ fmt.Println("\t\t\t\t\t[CONSUMER] Consumer gating at sequence", sequence)
+ } else if remaining == disruptor.Idling {
+ fmt.Println("\t\t\t\t\t[CONSUMER] Consumer idling at sequence", sequence)
+ }
+ //time.Sleep(time.Millisecond * 10)
}
}
}
diff --git a/example/example_producer.go b/example/example_producer.go
index 94b3095..9888d4f 100644
--- a/example/example_producer.go
+++ b/example/example_producer.go
@@ -1,18 +1,24 @@
package main
-import "github.com/smartystreets/go-disruptor"
+import (
+ "fmt"
-func publish(writer *disruptor.SharedWriter) {
+ "github.com/smartystreets/go-disruptor"
+)
+
+func publish(id int, writer *disruptor.SharedWriter) {
+
+ fmt.Printf("[PRODUCER %d] Starting producer...\n", id)
+
for {
- sequence := writer.Reserve(ItemsToPublish)
-
- if sequence != disruptor.Gating {
- for lower := sequence - ItemsToPublish; lower < sequence; {
- lower++
- ringBuffer[(lower)&RingMask] = lower
- }
-
+ if sequence := writer.Reserve(id, ItemsToPublish); sequence != disruptor.Gating {
+ fmt.Printf("[PRODUCER %d] Writing %d to slot %d\n", id, sequence, sequence)
+ ringBuffer[sequence&RingMask] = sequence
+ fmt.Printf("[PRODUCER %d] Committing sequence %d\n", id, sequence)
writer.Commit(sequence)
+ } else {
+ // fmt.Println("[PRODUCER] Gating")
+ //time.Sleep(time.Millisecond)
}
}
}
diff --git a/example/main.go b/example/main.go
index c687b8b..81f813c 100644
--- a/example/main.go
+++ b/example/main.go
@@ -10,9 +10,9 @@
MaxConsumersPerGroup = 1
MaxConsumerGroups = 1
MaxProducers = 2
- ItemsToPublish = 1
- ReportingFrequency = 1000000 * 10 // 1 million * N
- RingSize = 1024 * 16
+ ItemsToPublish = 2
+ ReportingFrequency = 10000 //1000000 * 10 // 1 million * N
+ RingSize = 2
RingMask = RingSize - 1
)
@@ -30,10 +30,10 @@
}
func startProducers(writer *disruptor.SharedWriter) {
for i := 0; i < MaxProducers-1; i++ {
- go publish(writer)
+ go publish(i, writer)
}
- publish(writer)
+ publish(MaxProducers-1, writer)
}
func startConsumerGroups(upstream disruptor.Barrier, writer *disruptor.Cursor) disruptor.Barrier {
diff --git a/reader.go b/reader.go
index 3570f00..50c6ceb 100644
--- a/reader.go
+++ b/reader.go
@@ -1,5 +1,7 @@
package disruptor
+import "fmt"
+
const (
Gating = -2
Idling = -3
@@ -24,12 +26,16 @@
func (this *Reader) Receive() (int64, int64) {
next := this.readerCursor.Load() + 1
ready := this.upstreamBarrier.Load()
+ fmt.Printf("\t\t\t\t\t[READER] Next: %d, Ready: %d\n", next, ready)
if next <= ready {
+ fmt.Printf("\t\t\t\t\t[READER] Next Sequence: %d, Remaining: %d\n", next, ready-next)
return next, ready - next
- } else if next <= this.writerCursor.Load() {
+ } else if gate := this.writerCursor.Load(); next <= gate {
+ fmt.Println("\t\t\t\t\t[READER] Gating at sequence:", gate)
return next, Gating
} else {
+ fmt.Println("\t\t\t\t\t[READER] Gating")
return next, Idling
}
}
diff --git a/shared_writer.go b/shared_writer.go
index e235a60..e60b241 100644
--- a/shared_writer.go
+++ b/shared_writer.go
@@ -1,6 +1,9 @@
package disruptor
-import "sync/atomic"
+import (
+ "fmt"
+ "sync/atomic"
+)
type SharedWriter struct {
capacity int64
@@ -24,23 +27,32 @@
}
}
-func (this *SharedWriter) Reserve(count int64) int64 {
+func (this *SharedWriter) Reserve(id int, count int64) int64 {
for {
previous := this.reservation.Load()
next := previous + count
wrap := next - this.capacity
+ fmt.Printf("[WRITER %d] Previous: %d, Next: %d, Wrap: %d\n", id, previous, next, wrap)
+
if wrap > this.gate {
+ fmt.Printf("[WRITER %d] Previous gate: %d\n", id, this.gate)
min := this.upstream.Load()
if wrap > min {
+ fmt.Printf("[WRITER %d] New gate (need to wait more): %d\n", id, min)
return Gating
}
+ fmt.Printf("[WRITER %d] New gate (accepted): %d\n", id, min)
this.gate = min // doesn't matter which write wins, BUT will most likely need to be a Cursor
}
+ fmt.Printf("[WRITER %d] Updating reservation. Previous: %d, Next: %d\n", id, previous, next)
if atomic.CompareAndSwapInt64(&this.reservation.value, previous, next) {
+ fmt.Printf("[WRITER %d] CAS accepted\n", id)
return next
+ } else {
+ fmt.Printf("[WRITER] CAS failed, retrying\n", id)
}
}
}
diff --git a/shared_writer_barrier.go b/shared_writer_barrier.go
index b9499d9..5d913c3 100644
--- a/shared_writer_barrier.go
+++ b/shared_writer_barrier.go
@@ -1,6 +1,9 @@
package disruptor
-import "math"
+import (
+ "fmt"
+ "math"
+)
type SharedWriterBarrier struct {
committed []int32
@@ -30,11 +33,14 @@
}
func (this *SharedWriterBarrier) Load() int64 {
+
for sequence := this.reservation.Load(); sequence >= 0; sequence-- {
if this.committed[sequence&this.mask] == int32(sequence>>this.shift) {
+ fmt.Printf("[SHARED-WRITER-BARRIER] Barrier Sequence: %d\n", sequence)
return sequence
}
}
+ fmt.Printf("[SHARED-WRITER-BARRIER] Barrier Sequence: -1\n")
return InitialSequenceValue
}
