Skip to main content

CompletableFuture can't be interrupted

I wrote a lot about InterruptedException and interrupting threads already. In short if you call Future.cancel() not inly given Future will terminate pending get(), but also it will try to interrupt underlying thread. This is a pretty important feature that enables better thread pool utilization. I also wrote to always prefer CompletableFuture over standard Future. It turns out the more powerful younger brother of Future doesn't handle cancel() so elegantly. Consider the following task, which we'll use later throughout the tests:

class InterruptibleTask implements Runnable {

    private final CountDownLatch started = new CountDownLatch(1)
    private final CountDownLatch interrupted = new CountDownLatch(1)

    @Override
    void run() {
        started.countDown()
        try {
            Thread.sleep(10_000)
        } catch (InterruptedException ignored) {
            interrupted.countDown()
        }
    }

    void blockUntilStarted() {
        started.await()
    }

    void blockUntilInterrupted() {
        assert interrupted.await(1, TimeUnit.SECONDS)
    }

}
Client threads can examine InterruptibleTask to see whether it has started or was interrupted. First let's see how InterruptibleTask reacts to cancel() from outside:

def "Future is cancelled without exception"() {
    given:
        def task = new InterruptibleTask()
        def future = myThreadPool.submit(task)
        task.blockUntilStarted()
    and:
        future.cancel(true)
    when:
        future.get()
    then:
        thrown(CancellationException)
}

def "CompletableFuture is cancelled via CancellationException"() {
    given:
        def task = new InterruptibleTask()
        def future = CompletableFuture.supplyAsync({task.run()} as Supplier, myThreadPool)
        task.blockUntilStarted()
    and:
        future.cancel(true)
    when:
        future.get()
    then:
        thrown(CancellationException)
}
So far so good. Clearly both Future and CompletableFuture work pretty much the same way - retrieving result after it was canceled throws CancellationException. But what about thread in myThreadPool? I thought it will be interrupted and thus recycled by the pool, how wrong was I!

def "should cancel Future"() {
    given:
        def task = new InterruptibleTask()
        def future = myThreadPool.submit(task)
        task.blockUntilStarted()
    when:
        future.cancel(true)
    then:
        task.blockUntilInterrupted()
}

@Ignore("Fails with CompletableFuture")
def "should cancel CompletableFuture"() {
    given:
        def task = new InterruptibleTask()
        def future = CompletableFuture.supplyAsync({task.run()} as Supplier, myThreadPool)
        task.blockUntilStarted()
    when:
        future.cancel(true)
    then:
        task.blockUntilInterrupted()
}
First test submits ordinary Runnable to ExecutorService and waits until it's started. Later we cancel Future and wait until InterruptedException is observed. blockUntilInterrupted() will return when underlying thread is interrupted. Second test, however, fails. CompletableFuture.cancel() will never interrupt underlying thread, so despite Future looking as if it was cancelled, backing thread is still running and no InterruptedException is thrown from sleep(). Bug or a feature? It's documented, so unfortunately a feature:

Parameters: mayInterruptIfRunning - this value has no effect in this implementation because interrupts are not used to control processing.

RTFM, you say, but why CompletableFuture works this way? First let's examine how "old" Future implementations differ from CompletableFuture. FutureTask, returned from ExecutorService.submit() has the following cancel() implementation (I removed Unsafe with similar non-thread safe Java code, so treat it as pseudo code only):

public boolean cancel(boolean mayInterruptIfRunning) {
    if (state != NEW)
        return false;
    state = mayInterruptIfRunning ? INTERRUPTING : CANCELLED;
    try {
        if (mayInterruptIfRunning) {
            try {
                Thread t = runner;
                if (t != null)
                    t.interrupt();
            } finally { // final state
                state = INTERRUPTED;
            }
        }
    } finally {
        finishCompletion();
    }
    return true;
}
FutureTask has a state variable that follows this state diagram:



In case of cancel() we can either enter CANCELLED state or go to INTERRUPTED through INTERRUPTING. The core part is where we take runner thread (if exists, i.e. if task is currently being executed) and we try to interrupt it. This branch takes care of eager and forced interruption of already running thread. In the end we must notify all threads blocked on Future.get() in finishCompletion() (irrelevant here). So it's pretty obvious how old Future cancels already running tasks. What about CompletableFuture? Pseudo-code of cancel():

public boolean cancel(boolean mayInterruptIfRunning) {
    boolean cancelled = false;
    if (result == null) {
        result = new AltResult(new CancellationException());
        cancelled = true;
    }
    postComplete();
    return cancelled || isCancelled();
}
Quite disappointing, we barely set result to CancellationException, ignoring mayInterruptIfRunning flag. postComplete() has a similar role to finishCompletion() - notifies all pending callbacks registered on that future. Its implementation is rather unpleasant (using non-blocking Treiber stack) but it definitely doesn't interrupt any underlying thread.

Reasons and implications

Limited cancel() in case of CompletableFuture is not a bug, but a design decision. CompletableFuture is not inherently bound to any thread, while Future almost always represents background task. It's perfectly fine to create CompletableFuture from scratch (new CompletableFuture<>()) where there is simply no underlying thread to cancel. Still I can't help the feeling that majority of CompletableFutures will have an associated task and background thread. In that case malfunctioning cancel() is a potential problem. I no longer advice blindly replacing Future with CompletableFuture as it might change the behavior of applications relying on cancel(). This means CompletableFuture intentionally breaks Liskov substitution principle - and this is a serious implication to consider.

Comments

  1. Is there a way to get around this limitation? I'd like to keep using CompletableFuture, but still be able to interrupt the underlying thread.

    ReplyDelete
  2. Kenneth,

    First, you need an alternative implementation of the CompletionStage that acts as RunnableFuture. Second, you have to make your own wrapper/implementation of Executor-s that uses this RunnableFuture.

    Please take a look at my attempt:
    https://github.com/vsilaev/java-async-await/tree/master/com.farata.lang.async.examples/src/main/java/com/farata/concurrent

    It relies on the following library as a basis -- https://github.com/lukas-krecan/completion-stage. The library provides basic CompletionStage impl. Afterwards implementing CompletableTask (CompletionStage+RunnableFuture) was a breeze. You may see for yourself -- CompletableTaskImpl.java + J8Executors.java

    ReplyDelete
  3. This limitation is not limited to `CompletableFuture` - a `Future` returned by `Executors.newWorkStealingPool` will ignore the boolean flag in the cancel method also.

    The word 'may' is there for a reason: you need to know where it's executing to know whether it applies. (At least the CompletableFuture Javadoc is clear on this, once I, too, RTFM ;-). It doesn't break the contract, as the possibility is documented in 'Future.cancel' also)

    Setting an interrupt applies to the entire Thread, not just the code inside the task, so that's where the mismatch comes from - in some cases it is not a safe or even possible to do.

    But inconvenient, this is. I'm also on the hunt for a good solution...

    ReplyDelete
  4. I've just released a small library that provides an implementation of CompletionStage for long-running blocking tasks, with true cancellation support - including thread interruption. The code and documentation are available here -- https://github.com/vsilaev/tascalate-concurrent. Any feedback or suggestions are welcome.

    ReplyDelete

Post a Comment