RxAwait missing await(context: CoroutineContext)

[dmytrodanylyk]

The Problem

Let's say we have API which expose Single.

private fun loadData() = Single.fromCallable {
    ...
}

I want to execute loadData on background thread and then switch to ui thread. To do so I have to write write following code, which looks a bit weird (with run it looks better):

fun main() = launch(UI) {

    val result = async(CommonPool) { loadData().await() }.await() // background thread

    print(result) // ui thread
}

The Solutions

Add following function to RxAwait:

public suspend fun <T> SingleSource<T>.await(context: CoroutineContext) 
        = async(context) { await() }.await()

So the result code would look like this:

fun main() = launch(UI) {

    val result = loadData().await(CommonPool) // background thread

    print(result) // ui thread
}

Any thoughts on this?

[elizarov]

What's wrong with Rx subscribeOn operator, which is specifically designed from this purpose?

val result = loadData().subscribeOn(Schedulers.IO).await()

As a matter of style, it might make sense to include subscribeOn(Schedulers.IO) into the definition of loadData so it is always done in background thread.

Actually, the key question here is why do you need Rx here at all? You can just replace:

private fun loadData() = Single.fromCallable {
    ...
}

with

private suspend fun loadData() = run(CommonPool) {
    ...
}

And it will be much more convenient to use. No need to await or anything like that. Just do:

fun main() = launch(UI) {
    val result = loadData() // background thread due to the way loadData is defined
    print(result) // ui thread
}

[dmytrodanylyk]

Actually, the key question here is why do you need Rx here at all?

I don't need it, that's what 3rd party library expose.

What's wrong with Rx subscribeOn operator, which is specifically designed from this purpose?

I don't want to mix up RxJava Schedulers and CoroutineContext, would prefer to control everything via CoroutineContext.

[dmytrodanylyk]

No need to await or anything like that.

Actually I try to avoid run and use async because I am not able to catch both coroutine CancellationException and exception thrown inside function wrapped via run coroutine.

Example

private fun loadData() = launch(uiContext) {
    view.showLoading() // ui thread
    try {
        // dataProvider.loadUser() throws Exception after 5 sec
        val result = run(bgContext) { dataProvider.loadUser() } 

        view.showData(result) // ui thread
    } catch (e: RuntimeException) {
        e.printStackTrace()
    }
}

// run
val job = loadData()
job.cancel()

Now, loadData return Job. If you cancel job CancellationException will be caught inside try-catch block. Next dataProvider.loadUser() completes and throws Exception which crash the app.

Following code with async works fine, because exception is stored inside Deferred.

private fun loadData() = launch(uiContext) {
    view.showLoading() // ui thread

    try {
        // dataProvider.loadUser() throws Exception after 5 sec
        val task = async(bgContext) { dataProvider.loadUser() }
        val result = task.await() 

        view.showData(result) // ui thread
    } catch (e: RuntimeException) {
        e.printStackTrace()
    }
}

[elizarov]

The exception is stored in async and you loose it. So, if there is anything genuinely wrong with loadUser, you'll never know it.

The problem with cancellation using run most likely stems from the way that bgContext is defined.
I assume, it is defined something like this:

val bgContext = newFixedThreadPoolContext(n, "bg")

However, unfortunately, bgContext is a combination of dispatcher and job, which is preventing the cancellation to propagate. It will get fixed (see #149). Meanwhile, you can can redefine bgContext like this:

val bgContext = newFixedThreadPoolContext(n, "bg")[ContinuationInterceptor]!!

or like this:

val bgContext = Executors.newScheduledThreadPool(n).asCoroutineDispatcher()

After that, you can define your own awaitIO extension function like this:

suspend fun <T> Single<T>.awaitIO() = run(bgContext) { await() }

With this extension in hand, you can simply use dataProvider.loadUser().awaitIO() anywhere you need it.

[dmytrodanylyk]

The exception is stored in async and you loose it. So, if there is anything genuinely wrong with loadUser, you'll never know it.

True, but if I cancel this job I don't care about result, I just want to make sure it doesn't crash my app.

The problem with cancellation using run most likely stems from the way that bgContext is defined.

Actually its CommonPool. Here are some more tests with run.

Test 1

// bgContext = newFixedThreadPoolContext(n, "bg")
// run (bgContext)
// loadUser wait 5 sec and throws IllegalStateException
val job = loadData()
job.cancel()

App doesn't crash, and when loadUser is completed CancellationException is caught by try-catch block.

Test 2

// bgContext = newFixedThreadPoolContext(n, "bg")[ContinuationInterceptor]!!
// run (bgContext)
// loadUser wait 5 sec and throws IllegalStateException
val job = loadData()
job.cancel()

CancellationException is caught by try-catch block, and when loadUser is completed app crash with IllegalStateException.

Test 3

// bgContext = Executors.newScheduledThreadPool(n).asCoroutineDispatcher()
// run (bgContext)
// loadUser wait 5 sec and throws IllegalStateException
val job = loadData()
job.cancel()

Same as Test 2

[elizarov]

Sorry, but I cannot reproduce your problem. Can you provide some kind of self-contained unit-test that reproduces it?

As a side note, if loadUser throws IllegalStateException then it crashes your app by design. It has nothing to do with cancellation. If an uncaught exception is thrown, app crashes. The same with NullPointerException, etc. You don't want to ignore them. Those exceptions indicate that their is a bug in your code, by default, Android apps crash to let their developers know of the bug.

[dmytrodanylyk]

As a side note, if loadUser throws IllegalStateException then it crashes your app by design. It has nothing to do with cancellation. If an uncaught exception is thrown, app crashes.

Actually IllegalStateException is caught by try-catch block (2), if a I do not cancel job (1). If I do cancel job (1), try-catch block catch CancellationException and then it's unable to catch IllegalStateException and app crash.

private fun loadData() = launch(UI) { // (1)
    view.showLoading() // ui thread

    try {
        val result = run(CommonPool) { thisCodeThrowsException()}
        view.showData(result) // ui thread
    } catch (e: RuntimeException) { // (2)
        e.printStackTrace()
    }
}

Working on self-contained unit-test.

[elizarov]

I might be starting to understand what problem you are running into. Does your app crash because of UnexpectedCoroutineException: Unexpected exception while cancellation is in progress?

[dmytrodanylyk]

No, here is tack trace.

10-19 21:11:55.688 E: FATAL EXCEPTION: ForkJoinPool.commonPool-worker-1
                      Process: com.dmytrodanylyk, PID: 4795
                      java.lang.IllegalArgumentException: Ooops you are unlucky
                          at com.dmytrodanylyk.exception.trycatch.DataProvider.mayThrowException(MainActivity.kt:52)
                          at com.dmytrodanylyk.exception.trycatch.DataProvider.loadData(MainActivity.kt:46)
                          at com.dmytrodanylyk.exception.trycatch.MainPresenter$loadData$1$result$1.doResume(MainActivity.kt:90)
                          at kotlin.coroutines.experimental.jvm.internal.CoroutineImpl.resume(CoroutineImpl.kt:54)
                          at kotlinx.coroutines.experimental.DispatchTask.run(CoroutineDispatcher.kt:123)
                          at java.util.concurrent.ForkJoinTask$RunnableExecuteAction.exec(ForkJoinTask.java:1383)
                          at java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:256)
                          at java.util.concurrent.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1123)
                          at java.util.concurrent.ForkJoinPool.scan(ForkJoinPool.java:1961)
                          at java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1909)
                          at java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:128)

I think I finally understood the problem, please correct me if I am wrong. Consider following code:

private fun loadData() = launch(UI) { // parent coroutines
    try {
        val result = run(CommonPool) { loadData() } // loadData throws exception
    } catch (e: RuntimeException) {
        e.printStackTrace()
    }
}

1. I am launching 2 coroutines here, parent with launch and child with run.
2. Child coroutine which execute loadData may through exception.
3. If I cancel parent coroutine try-catch block (which is defined in parent coroutine) will never catch exception thrown in child coroutine because parent coroutines has been cancelled. That's why app crash after child coroutine complete it's work.

Is my assumptions correct?

Interesting things are happening when I replace CommonPool with newFixedThreadPoolContext(2, "bg").
When I cancel parent coroutines, I don't see cancellation stack trace immediately. Instead, JobCancellationException: Job was cancelled normally stack trace appears when function loadData is completed.

I will try to prepare some unit tests tomorrow.

[dmytrodanylyk]

Here is source code of Android Instrumented Test, which crash with: FATAL EXCEPTION: ForkJoinPool.commonPool-worker-1

@RunWith(AndroidJUnit4::class)
class ExampleInstrumentedTest {

    private val latch = CountDownLatch(1)

    @Test
    fun myTest() {
        val job = loadData()
        SystemClock.sleep(100) // wait before cancelling job
        job.cancel()

        latch.await()
    }

    private fun loadData() = launch(UI) {
        try {
            run(CommonPool) { loadUser() } // throws exception
        } catch (e: IllegalArgumentException) {
            e.printStackTrace()
            latch.countDown()
        }
    }

    private fun loadUser() {
        SystemClock.sleep(1000)
        throw IllegalArgumentException("Ooops you are unlucky")
    }
}

[dmytrodanylyk]

This source code of Unit Test, it runs forever because try-catch block never catch exception. In log you should see ForkJoinPool.commonPool-worker-1" java.lang.IllegalArgumentException: Ooops you are unlucky exception.

class MyTest {

    val uiContext = Unconfined
    val latch = CountDownLatch(1)

    @Test
    fun myTest() {
        val job = loadData()
        Thread.sleep(100) // wait before cancelling job
        job.cancel()
        latch.await()
    }

    private fun loadData() = launch(uiContext) {
        try {
            run(CommonPool) { mayThrowException() } // throws exception
        } catch (e: IllegalArgumentException) {
            e.printStackTrace()
            latch.countDown()
        }
    }

    private fun mayThrowException() {
        Thread.sleep(1000)
        throw IllegalArgumentException("Ooops you are unlucky")
    }

}

[dmytrodanylyk]

FYI: if you replace CommonPool with newSingleThreadContext("bg") in above tests everything works perfectly fine, exception is caught in try-catch block and test finishes successfully.

[elizarov]

Thanks a lot. I'll get this fixed soon.