Removed diagnostics.
diff --git a/example/example_consumer.go b/example/example_consumer.go
index 0579616..e60ed5a 100644
--- a/example/example_consumer.go
+++ b/example/example_consumer.go
@@ -15,12 +15,9 @@
func consume1(reader *disruptor.Reader) {
started := time.Now()
- // 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)
-
for remaining >= 0 {
if sequence%ReportingFrequency == 0 {
finished := time.Now()
@@ -29,29 +26,19 @@
}
message := ringBuffer[sequence&RingMask]
- // 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("***Race Condition***::Sequence: %d, Message %d\n", sequence, message)
fmt.Println(alert)
panic(alert)
}
- //ringBuffer[sequence&RingMask] = sequence % 2
+ ringBuffer[sequence&RingMask] = sequence % 2
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 472bbab..ec32ab7 100644
--- a/example/example_producer.go
+++ b/example/example_producer.go
@@ -1,27 +1,15 @@
package main
-import (
- "fmt"
- "time"
+import "github.com/smartystreets/go-disruptor"
- "github.com/smartystreets/go-disruptor"
-)
-
-func publish(id int, writer *disruptor.SharedWriter) {
-
- fmt.Printf("[PRODUCER %d] Starting producer...\n", id)
+func publish(writer *disruptor.SharedWriter) {
for {
// TODO: return lower, upper instead? or some kind of struct "Reservation"
// upon which commit can be invoked?
- if sequence := writer.Reserve(id, ItemsToPublish); sequence != disruptor.Gating {
- // fmt.Printf("[PRODUCER %d] Writing %d to slot %d\n", id, sequence, sequence)
+ if sequence := writer.Reserve(ItemsToPublish); sequence != disruptor.Gating {
ringBuffer[sequence&RingMask] = sequence
- // fmt.Printf("[PRODUCER %d] Committing from sequence %d\n", id, sequence)
writer.Commit(sequence, sequence+ItemsToPublish-1)
- } else {
- // // fmt.Printf("[PRODUCER %d] Gating\n", id)
- time.Sleep(time.Millisecond * 10)
}
}
}
diff --git a/example/main.go b/example/main.go
index bcbfb0a..c687b8b 100644
--- a/example/main.go
+++ b/example/main.go
@@ -30,10 +30,10 @@
}
func startProducers(writer *disruptor.SharedWriter) {
for i := 0; i < MaxProducers-1; i++ {
- go publish(i, writer)
+ go publish(writer)
}
- publish(MaxProducers-1, writer)
+ publish(writer)
}
func startConsumerGroups(upstream disruptor.Barrier, writer *disruptor.Cursor) disruptor.Barrier {
diff --git a/reader.go b/reader.go
index 1ea5ec6..7c7a278 100644
--- a/reader.go
+++ b/reader.go
@@ -24,16 +24,12 @@
func (this *Reader) Receive() (int64, int64) {
next := this.readerCursor.Load() + 1
ready := this.upstreamBarrier.LoadBarrier(next)
- // 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 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 0f909e6..6550f22 100644
--- a/shared_writer.go
+++ b/shared_writer.go
@@ -24,39 +24,28 @@
}
}
-func (this *SharedWriter) Reserve(id int, count int64) int64 {
+func (this *SharedWriter) Reserve(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.LoadBarrier(0)
if wrap > min {
- // fmt.Printf("[WRITER %d] New gate (waiting for consumers): %d\n", id, min)
return Gating
}
- // 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)
if atomic.CompareAndSwapInt64(&this.reservation.value, previous, next) {
- // fmt.Printf("[WRITER %d] Reservation updated\n", id)
return next
- // } else {
- // fmt.Printf("[WRITER %d] Reservation rejected, retrying\n", id)
}
}
}
func (this *SharedWriter) Commit(lower, upper int64) {
- // fmt.Printf("[PRODUCER] Committing Reservation. Lower: %d, Upper: %d\n", lower, upper)
-
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
index fb9c280..203334e 100644
--- a/shared_writer_barrier.go
+++ b/shared_writer_barrier.go
@@ -32,16 +32,11 @@
func (this *SharedWriterBarrier) LoadBarrier(lower int64) int64 {
shift, mask := this.shift, this.mask
upper := this.reservation.Load()
- // fmt.Printf("\t\t\t\t\t[BARRIER] Next (lower): %d, Reservation (upper): %d\n", lower, upper)
- // fmt.Println("\t\t\t\t\t[BARRIER] Committed:", this.committed)
for ; lower <= upper; lower++ {
- // fmt.Printf("\t\t\t\t\t[BARRIER] Inside Loop. Index: %d, Value: %d\n", sequence&mask, sequence>>shift)
if this.committed[lower&mask] != int32(lower>>shift) {
- // fmt.Println("\t\t\t\t\t[BARRIER] Upstream Barrier:", sequence-1, this.committed)
return lower - 1
}
}
- // fmt.Println("\t\t\t\t\t[BARRIER] Upstream Barrier (default):", lower)
return upper
}
