Commenting out diagnostics--race condition identified.
diff --git a/example/example_consumer.go b/example/example_consumer.go
index a84eee5..4c9b55f 100644
--- a/example/example_consumer.go
+++ b/example/example_consumer.go
@@ -2,6 +2,7 @@
import (
"fmt"
+ "time"
"github.com/smartystreets/go-disruptor"
)
@@ -12,25 +13,26 @@
}
}
func consume1(reader *disruptor.Reader) {
- // started := time.Now()
+ started := time.Now()
- fmt.Printf("\t\t\t\t\t[CONSUMER] Starting consumer...\n")
+ // fmt.Printf("\t\t\t\t\t[CONSUMER] Starting consumer...\n")
for {
sequence, remaining := reader.Receive()
if remaining >= 0 {
- fmt.Printf("\t\t\t\t\t[CONSUMER] Received messages starting at sequence %d, with %d messages remaining\n", sequence, remaining)
+ // fmt.Printf("\t\t\t\t\t[CONSUMER] Received messages starting at sequence %d, with %d messages remaining\n", sequence, remaining)
for remaining >= 0 {
- // if sequence%ReportingFrequency == 0 {
- // finished := time.Now()
- // fmt.Println(sequence, finished.Sub(started))
- // started = time.Now()
- // }
+ 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)
+ // fmt.Printf("\t\t\t\t\t[CONSUMER] Consuming sequence %d. Message Payload: %d\n", sequence, message)
if sequence != message {
- alert := fmt.Sprintf("--------------\n\t\t\t\t\t[CONSUMER] ***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)
+ alert := fmt.Sprintf("***Race Condition***::Sequence: %d, Message %d\n", sequence, message)
fmt.Println(alert)
panic(alert)
}
@@ -41,15 +43,15 @@
sequence++
}
- fmt.Println("\t\t\t\t\t[CONSUMER] All messages consumed, committing up to sequence ", sequence-1)
+ // 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)
+ // } 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 9888d4f..3607547 100644
--- a/example/example_producer.go
+++ b/example/example_producer.go
@@ -1,24 +1,20 @@
package main
-import (
- "fmt"
-
- "github.com/smartystreets/go-disruptor"
-)
+import "github.com/smartystreets/go-disruptor"
func publish(id int, writer *disruptor.SharedWriter) {
- fmt.Printf("[PRODUCER %d] Starting producer...\n", id)
+ // fmt.Printf("[PRODUCER %d] Starting producer...\n", id)
for {
if sequence := writer.Reserve(id, ItemsToPublish); sequence != disruptor.Gating {
- fmt.Printf("[PRODUCER %d] Writing %d to slot %d\n", id, sequence, sequence)
+ // 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)
+ // fmt.Printf("[PRODUCER %d] Committing sequence %d\n", id, sequence)
writer.Commit(sequence)
- } else {
- // fmt.Println("[PRODUCER] Gating")
- //time.Sleep(time.Millisecond)
+ // } else {
+ // // fmt.Println("[PRODUCER] Gating")
+ // //time.Sleep(time.Millisecond)
}
}
}
diff --git a/example/main.go b/example/main.go
index 81f813c..7af8dea 100644
--- a/example/main.go
+++ b/example/main.go
@@ -10,7 +10,7 @@
MaxConsumersPerGroup = 1
MaxConsumerGroups = 1
MaxProducers = 2
- ItemsToPublish = 2
+ ItemsToPublish = 1
ReportingFrequency = 10000 //1000000 * 10 // 1 million * N
RingSize = 2
RingMask = RingSize - 1
diff --git a/reader.go b/reader.go
index 50c6ceb..006bfa4 100644
--- a/reader.go
+++ b/reader.go
@@ -1,7 +1,5 @@
package disruptor
-import "fmt"
-
const (
Gating = -2
Idling = -3
@@ -26,16 +24,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)
+ // 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)
+ // fmt.Printf("\t\t\t\t\t[READER] Next Sequence: %d, Remaining: %d\n", next, ready-next)
return next, ready - next
} else if gate := this.writerCursor.Load(); next <= gate {
- fmt.Println("\t\t\t\t\t[READER] Gating at sequence:", 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")
+ // fmt.Println("\t\t\t\t\t[READER] Gating")
return next, Idling
}
}
diff --git a/shared_writer.go b/shared_writer.go
index e60b241..0ffb9dc 100644
--- a/shared_writer.go
+++ b/shared_writer.go
@@ -1,9 +1,6 @@
package disruptor
-import (
- "fmt"
- "sync/atomic"
-)
+import "sync/atomic"
type SharedWriter struct {
capacity int64
@@ -33,26 +30,26 @@
next := previous + count
wrap := next - this.capacity
- fmt.Printf("[WRITER %d] Previous: %d, Next: %d, Wrap: %d\n", id, previous, next, wrap)
+ // 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)
+ // 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)
+ // fmt.Printf("[WRITER %d] New gate (waiting for consumers): %d\n", id, min)
return Gating
}
- fmt.Printf("[WRITER %d] New gate (accepted): %d\n", id, min)
+ // fmt.Printf("[WRITER %d] New gate found: %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)
+ // 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)
+ // fmt.Printf("[WRITER %d] Reservation updated\n", id)
return next
- } else {
- fmt.Printf("[WRITER] CAS failed, retrying\n", id)
+ // } else {
+ // fmt.Printf("[WRITER %d] Reservation rejected, retrying\n", id)
}
}
}
diff --git a/shared_writer_barrier.go b/shared_writer_barrier.go
index 5d913c3..d77910a 100644
--- a/shared_writer_barrier.go
+++ b/shared_writer_barrier.go
@@ -1,9 +1,6 @@
package disruptor
-import (
- "fmt"
- "math"
-)
+import "math"
type SharedWriterBarrier struct {
committed []int32
@@ -36,11 +33,11 @@
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)
+ // fmt.Printf("\t\t\t\t\t[SHARED-WRITER-BARRIER] Barrier Sequence: %d\n", sequence)
return sequence
}
}
- fmt.Printf("[SHARED-WRITER-BARRIER] Barrier Sequence: -1\n")
+ // fmt.Printf("\t\t\t\t\t[SHARED-WRITER-BARRIER] Barrier Sequence: -1\n")
return InitialSequenceValue
}
