Concurrency Control with RingBuffer + SequenceGroup

[DarrenWang]

say I use a SequenceGroup as the gatingSequences as a RingBuffer, and
the SequenceGroup has some initial sequences:
seqGrp = SequenceGroup( sequence1, sequence2)
ringBuffer.setGatingSequences(seqGrp)

after some time, I want to add a new sequence to the SequenceGroup and
use the Util.getMinimumSequence(seqGrp) as the new sequence's initial
value.

The problem I wonder is , at the operation process：
seqGrp.add(new Sequence(Util.getMinimumSequence(seqGrp)))

Is there any possibility that the ringBuffer will wrap or the gating
sequence will proceed? if that's possible, then when I call
ringBuffer.get("the value of new Sequence added to seqGrp"), the value
of it will be the newest not the oldest, right?

[Alexei Valyaev]

[Michael Barker]

The problem I wonder is , at the operation process：
seqGrp.add(new Sequence(Util.getMinimumSequence(seqGrp)))

Is there any possibility that the ringBuffer will wrap or the gating
sequence will proceed? if that's possible, then when I call
ringBuffer.get("the value of new Sequence added to seqGrp"), the value
of it will be the newest not the oldest, right?

There is a chance yes.  I would recommend the following approach for adding a new value to the SequenceGroup:

Sequence newSequence = new Sequence(ringBuffer.getCursor());

seqGroup.add(newSequence);

newSequence.set(ringBuffer.getCursor()); // <-- This is the important bit.

Then start the event processor/event handler.

This will ensure that the disruptor will not wrap while the new Sequence is being added to the group.  You will always run the risk of wrapping the ring buffer if you try to start at the minimum sequence.  This is due to some subtleties around how the disruptor tracks the minimum sequence (I'll do longer post explaining this at a later date).  The key thing is to set it to a value that you know will be safe before starting the event handler or event processor.  The best way to guarantee this is to initialise the Sequence with a value that will be the same or less that value you intend to start with.  In this case I use the cursor value for both as it will only increase between calls and is the value least likely to cause the publishing thread(s) to stall.  You could initialise the sequence with -1, but if the publishing thread causes the buffer to get to a point where is needs to re-request the minimum sequence from the sequence group, before you have set sequence value, there is a higher chance that you could cause the publishing thread to block while waiting for space to become available.

I recognise that is not ideal from an API usability perspective.  However, the sequence group is a fairly new concept and is not one we're using heavily yet, you're on a road less travelled.  It was added for a project that has yet to be implemented, so we haven't ironed out all of the kinks.

Mike.

[DarrenWang]

Thanks Mike
In fact, I am trying to use disruptor's RingBuffer as a Message
destination, say Topic, and let multiple subscribers to keep their own
Sequences when they try to get their own pieces of messages, when new
subscribers come in, I would like to add them to same SequenceGroup of
the same Topic, that's why I want to add a new sequence to a
SequenceGroup at runtime :-)
I know seldom people use disruptor in this way, but it does fit my
need, I think, for low-latency, high throughput message passing,
disruptor can help a lot.

Rgds
Darren

[Michael Barker]

>    I know seldom people use disruptor in this way, but it does fit my
> need, I think, for low-latency, high throughput message passing,
> disruptor can help a lot.

I think it's something we should support well, we just haven't done so
yet. Requiring the slightly tricky set up I mentioned in the previous
post is not ideal. I thinking on how to improve the API for this
feature so that it is simpler and less error prone. I've got a couple
of use cases myself, I just haven't gotten around to implementing
them.

Mike.

[DarrenWang]

Great to hear that :-)
Hope u genius guys get it done soon ~