How to convert Java blocking function into cancellable suspend function?

Question

Kotlin suspend functions should be nonblocking by convention (1). Often we have old Java code which relies on java Thread interruption mechanism, which we cannot (don't want to) modif (2):

public void doSomething(String arg) {
    for (int i = 0; i < 100_000; i++) {
        heavyCrunch(arg, i);
        if (Thread.interrupted()) {
            // We've been interrupted: no more crunching.
            return;
        }
    }
}

What is the best way to adapt this code for usage in coroutines?

Version A: is unacceptable because it will run the code on the caller thread. So it will violate the "suspending functions do not block the caller thread" convention:

suspend fun doSomething(param: String) = delegate.performBlockingCode(param)

Version B: is better because it would run the blocking function in background thread, thus it wouldn't block the callers thread (except if by chance the caller uses the same thread from Dispatchers.Default threads pool). But coroutines job cancelling wouldn't interrupt performBlockingCode() which relies on thread interruption.

suspend fun doSomething(param: String) = withContext(Dispatchers.Default) {
    delegate.performBlockingCode(param)
}

Version C: is currently the only way which I see to make it working. The idea is to convert blocking function into nonblocking with Java mechanisms and later use suspendCancellableCoroutine (3) for converting asynchronous method into suspend function:

private ExecutorService executor = Executors.newSingleThreadExecutor();

public Future doSomethingAsync(String arg) {
    return executor.submit(() -> {
        doSomething(arg);
    });
}

suspend fun doSomething(param: String) = suspendCancellableCoroutine<Any> { cont ->
    try {
        val future = delegate.doSomethingAsync(param)
    } catch (e: InterruptedException) {
        throw CancellationException()
    }
    cont.invokeOnCancellation { future.cancel(true) }
}

As commented below, above code won't work properly, because continuation.resumeWith() is not called

Version D: uses CompletableFuture: which provides a way to register callback for when completable completes: thenAccept

private ExecutorService executor = Executors.newSingleThreadExecutor();

public CompletableFuture doSomethingAsync(String arg) {
    return CompletableFuture.runAsync(() -> doSomething(arg), executor);
}

suspend fun doSomething(param: String) = suspendCancellableCoroutine<Any> { cont ->
    try {
        val completableFuture = delegate.doSomethingAsync(param)
        completableFuture.thenAccept { cont.resumeWith(Result.success(it)) }
        cont.invokeOnCancellation { completableFuture.cancel(true) }
    } catch (e: InterruptedException) {
        throw CancellationException()
    }
}

Do you know any better way for that?

Your 3rd approach is broken, it never completes on its own. You'll find completion a challenge because all you get from a Future is the blocking get() or join(). You need a CompletableFuture: supplyAsync(fn, executor). — Marko Topolnik
The way the Kotlin library supports this is by providing CompletableFuture.await(). Alas, it is hardcoded to call future.cancel(false). But I think you could simply write your own extension differently. Writing an extension on CompletableFuture is more composable with other code, could relieve you from writing a suspendable wrapper every time. Keep in mind that interrupting coroutine-carrying threads is in general a dangerous practice and could cause the interrupt signal to be received by an unintended recipient. — Marko Topolnik
I have edited description, added "Version D". Is this is how you implement it? (I haven't tested if the code run). Also I have found this answer: stackoverflow.com/a/58402108/2075875 That should be a working solution with Futures as well. (The Android which we target doesn't have CompletableFuture) — Malachiasz
Yes, that looks like a good solution. In the absence of CompletableFuture I'd go with that. — Marko Topolnik

Maxim Kochetkov Maxim Kochetkov · Accepted Answer · 2020-11-27T19:32:33

You may wrap blocking code via suspend fun kotlinx.coroutines.runInterruptible

It suppressed compile warning and blocking code will throw InterruptedException on cancellation

val job = launch {
  runInterruptible {
    Thread.sleep(500)
  }
}

job.cancelAndJoin() // Cause will be 'java.lang.InterruptedException'

Tested on org.jetbrains.kotlinx:kotlinx-coroutines-core-jvm:1.4.2

How to convert Java blocking function into cancellable suspend function?

1 Answers