In a pipeline can we skip event processors and just signal the slot is available ?

[basu76]

 * +----+    +-----+    +-----+    +-----+

 * | P1 |--->| EP1 |--->| EP2 |--->| EP3 |

 * +----+    +-----+    +-----+    +-----+

In a typical scenario like the one  above (which i took from one of the examples), if we determine in EP1, that the data is to be filtered and doesn't need to go all the way to EP3, can we signal that the slot be made available to the producer ?

thanks

Basanth

 

[ymo]

I had th exact same question here and was hoping someone would answer it.

What i do now is make an atomic boolean in the event class that all the processors check before doing anything. I am not sure if this is the best way tho !!!

[Sam Barker]

The other option would be to have EP1 publish events to be processed down stream to a second ring buffer and have the other processors consume from that. 

--
 
 

[ymo]

This would force a third or forth ringbuffer ! meaning you would need as many ringbuffers as there are EP which is not very apealing.

[Toby DiPasquale]

Why not just add a signal field to the event itself and have EP2 and
EP3 simply truncate processing when that field is set?

--
Toby DiPasquale

[Sam Barker]

Yes, it does but each ringbuffer is dealing with a subset of messages. Which does sound like what you want. Its no different to having a queue in front of each EP as you would do in a non disruptor model. 

 If you are gating each EventProccessor on the sequence of the previous one you would only need a primative boolean (single writer), but if your just letting them all read from the ringbuffer then you have a  race condition and are relying on EP1. 

The other option is are the downstream EP's really EP's or could they just be invoked when needed by the EP further up the chain?

--
 
 

[McKinley]

In the example you reference, OnePublisherToThreeProcessorPipelineThroughputTest, you might just check a condition set in EP1 to quickly return in the case of EP2 and EP3. By making it stateful to the event it seems to me you will not need an atomic variable (which would introduce contention). The SequenceBarriers will ensure that the event processors operate on the event in succession (stop me if I'm wrong).

// EP1 Handler

public void onEvent(Event event, long sequence, boolean endOfBatch) {

    processStep1(event);

    if(event.allDone)

        event.finalStep = 1;

}

// EP2 Handler

public void onEvent(Event event, long sequence, boolean endOfBatch) {

    if(event.finalStep < 2)

        return;

    processStep2(event);

}

This won't allow slots to race ahead back to availability. They will be fairly bounded in their batches from the start of EP1, but those batches will flow through the remaining event processors quickly, skipping events that do not need additional work.

McKinley

[Jason Jebanesan]

Thanks for the reply. This is what i'm currently doing, but when it comes to multiple consumers this becomes trickier. It would be a good to have feature if we can choose the sequence from the ringbuffer itself. anyway let me have a look at the source code. Thanks again for your help. 

Jason

[Jason Jebanesan]

I ran into the following problem,

Lets assume we have four worker handlers doing different tasks. Following is the sequence W1 --> W2 --> W3 --> W4

Each has one thread except for W2 which does alot of I/O calls so it has many threads (10), not all events need to be processed by W2 some are skipped. Now if all threads (10) in my W2 gets blocked on the IO, assuming the next event which is passed can be skipped the event cant be skipped because all my threads are in use doing IO, so now W3 has no events till atlest one of the W2 I/O job is completed for it to pick the next event and skip it.

Is there any workaround or solution for this? or have I just mist some feature which can do this? 

Thanks for the help

Jason

[Nikolay Tsankov]

Not sure if I get your case, but can't you split this in two, W1->W2 and W3->W4 and have W2 threads publish to the second ring buffer? You can then have a W1.5 handler that determines if the event is skippable by W2 and directly publishes to second ring buffer.

Hope that makes sense

--
 
 

[Jason Jebanesan]

Hi Nikolay

What im trying to do is, pass events on the following order W1 --> W2 --> W3 --> W4 but not all events need to be processed by every worker. My W2 worker is a I/O extensive code, so some events can skip it and just go to W3, but if all my threads in W2 is been used up on I/O process there will be no way to receive the event on W2 in the first place to skip it (W2 does not have free threads to pick from ringbuffer). 

If I talk in terms of queues, basically im trying to have two different queues one is a fast queue which can process messages fast and go to the next step next is a slow queue which takes it time to process (FIFO is not important) 

hope that make sense.

As you mentioned I can have two ringbuffers to do this, but im really trying to avoid that because i dont want another pooling of object and another overhead on copying details between objects to pass to the another ringbuffer.

[Sam Barker]

A couple of thoughts why have multiple threads in W2 at all? Have one thread in it and just have multiple of them reading off the same ring buffer and publishing to a second downstream ringbuffer (the cost of publication is going to be tiny compared to your IO. You can gate the w2 consumers on w1. So you would look more like:

       /-> W1     -\

RB1 --> W2a   --> RB2  --> W3

       |-> W2b    -|           \-> W4 

       \-> W2..n  -/

      

Though if W3 & W4 can do work without waiting for W2 you can just gate them on W1 and let W2n handle things in their own time as long as they can outpace the publishers.

Sam 

--