0

I have a queue with multiple producers - one consumer. The consumer runs periodically and drains the queue fully (no messages left after). A graceful algorithm should run the consumer and wait for it with timeout or just wait if the consumer is already running.

Currently we have smth like this:

void stop(boolean graceful) {
        if (graceful && !checkAndStopDirectly()) {
            executor.shutdown();
            try {
                if (!executor.awaitTermination(shutdownWaitInterval, shutdownWaitIntervalUnit)) {
                    log.warn("...");
                }
            } catch (InterruptedException e) {
                log.error("...", e);
            }
        } else {
            executor.shutdownNow();
        }

private boolean checkAndStopDirectly() {
    ExecutorService shutdownExecutor = Executors.newSingleThreadExecutor();
    try {
        return shutdownExecutor.submit(new Callable<Boolean>(){
            @Override
            public Boolean call() throws Exception {
                if (isAlreadyRan.compareAndSet(false, true)) {
                    try {
                        runnableDrainTask.run();
                    } finally {
                        isAlreadyRan.set(false);
                    }
                    return true;
                } else {
                    return false;
                }
            }
        }).get(shutdownWaitInterval, shutdownWaitIntervalUnit);

Does anybody see more elegant way to do it? e.g. I'm looking a way w/o using additional AtomicBoolean (isAlreadyRan) or double waiting logic with time intervals as object fields etc. btw, the poison pill pattern comes to my mind...

1
  • Can you be clearer with your requirements? what should wait? A client to the stop() call? what should it wait for, the queue to drain? Do you want a call to stop() to block until the thing is drained, no matter what?
    – Toby
    Sep 6, 2011 at 13:36

1 Answer 1

1

Are you talking here about a graceful shutdown of your application should do the following?

  1. wait for the queue to drain, or
  2. timeout if the drain is taking too long

I might need to understand how you drain the queue but if you're able to do that in such a way as that it is interruptible, you can timeout on the get of the Future and attempt a shutdownNow (interrupting if not fully drained) regardless; how does this look to you?

ExecutorService pool = Executors.newSingleThreadExecutor();

public void stop() {
    try {
        pool.submit(new DrainTask()).get(100, MILLISECONDS);
    } catch (TimeoutException e) {
        // nada, the timeout indicates the queue hasn't drained yet
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    } catch (Exception e) {
        // nada, defer to finally
    } finally {
        pool.shutdownNow();
    }
}

private class DrainTask implements Callable<Void> {
    @Override
    public Void call() throws Exception {
        while (!Thread.currentThread().isInterrupted())
            ; // drain away
        return null;
    }
}

Is the compareAndSet part used to guard against multiple concurrent calls to stop? I think I'd favor a general lock or use synchronized instead. However, in the event of a conflict, an ExecutionException would be thrown and repeating calls to shutdownNow are fine.

It kind of relies on the DrainTask being able to stop what's its doing in response to a call to interrupt (as shutdownNow will attempt to call interrupt on any currently running threads).

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.