How to properly shutdown disruptor?

[Curtis Stanford]

Hi, just wondering what's the proper way to shutdown a disruptor ring and the executor service behind it? I tried calling executor.shutdown but it times out. Calling halt on the consumers doesn't finish what's currently in the ring. My application won't exit with the executor service running. Is there some way to finish what's in the ring but shutdown the threads, etc. afterwards?

Thanks

Curtis

[Olivier Deheurles]

Hi Curtis,

Calling halt on a consumer will finish processing current entry (or current entries if you're in the middle of a batch). This call is asynchronous (ie. the call can return before the consumer is properly stoped).

If you want to stop the consumer(s) when all pending entries have been processed:
1) stop publishing ;)
2) check consumers have processed the latest sequence number
3) Halt consumers when they have reached this sequence number

If you use Adrian's DisruptorWizard DSL you will see there is a halt method on the DisruptorWizard (but this does not guarantee all pending entries have been processed).

@LMAX guys: should the halt method on the wizard and on the batch consumer take the sequence number after which you release the consumer thread?

Olivier

[Curtis Stanford]

Thanks Olivier. It would be nice to have an API for that.

[Martin Thompson]

I've tried to keep the API similar to other threading frameworks. You
are right about the steps above. I like the idea that halt() just
stops the consumer once it has finished the current cycle in a very
similar way to Thread.interrupt() works. I often have a method that
checks if a backlog of items is still being processed then call halt.
Remember that if you interrupt a thread you still have to wait on it
finishing by calling join() or similar.

1.
<stop producing>

2.
while (hasBackLog()) // see below
{
// busy spin
}

3.
for (Consumer consumer: consumers)
{
consumer.halt();
}

<can restart threads at this point if we want>

4.
executor.shutdown():

/**
* Check if any consumers have work to do before they catch up with
the producers.
*/
public boolean hasBacklog()
{
final long cursor = ringBuffer.getCursor();

for (Consumer consumer : consumers)
{
if (cursor != consumer.getSequence())
{
return true;
}
}

return false;

[Curtis Stanford]

Awesome, thanks!

[Curtis Stanford]

I've added a method to the disruptor method to implement this:

public void haltWhenRingEmpty()

{

while (ringHasBacklog()) {

}

halt();

}

private boolean ringHasBacklog()

{

long cursor = ringBuffer.getCursor();

for (ConsumerInfo consumerInfo : consumerRepository)

{

if (cursor != consumerInfo.getConsumer().getSequence()) return true;

}

return false;

}