blob: e0820d54ed2b083abb27688a89b9c60cba786fd3 [file] [log] [blame]
package disruptor
import "sync/atomic"
type SharedWriter struct {
written *Cursor
upstream Barrier
capacity int64
gate *Cursor
mask int64
shift uint8
committed []int32
}
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,
}
}
func (this *SharedWriter) Reserve(count int64) int64 {
for {
previous := this.written.Load()
upper := previous + count
for upper-this.capacity > this.gate.Load() {
this.gate.Store(this.upstream.Read(0))
}
if atomic.CompareAndSwapInt64(&this.written.sequence, previous, upper) {
return upper
}
}
}
func (this *SharedWriter) Commit(lower, upper int64) {
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--
}
}
}