blob: 3d4cfc8e0fd1ffd83b1b8dfd809d01194cb153f8 [file] [log] [blame]
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) {
for shift, mask := this.shift, this.mask; lower <= upper; lower++ {
this.committed[lower&mask] = int32(lower >> shift)
}
}