[Coroutines] resumeWithException not working as expected with custom continuation

Support
1

Hello all. I’ve done my best to discover what is wrong, but to no avail. This problem is based on some work I’d done earlier and am convinced worked as expected. However, I am not able to reproduce the exception condition reliably, having tried back as far as kotlinx 0.14.1.

The following code demonstrates what I believe is either the issue or a hole in my understanding.

package coroutines

import kotlinx.coroutines.experimental.CoroutineDispatcher
import kotlinx.coroutines.experimental.asCoroutineDispatcher  
import kotlinx.coroutines.experimental.async
import kotlinx.coroutines.experimental.runBlocking
import java.util.concurrent.CountDownLatch
import java.util.concurrent.Executors
import java.util.concurrent.TimeUnit
import kotlin.coroutines.experimental.AbstractCoroutineContextElement
import kotlin.coroutines.experimental.Continuation
import kotlin.coroutines.experimental.ContinuationInterceptor
import kotlin.coroutines.experimental.CoroutineContext

val dispatchExecutor: java.util.concurrent.ExecutorService = Executors.newFixedThreadPool(2)

// the thread local I want to maintain
val threadInt = ThreadLocal<Int>()
val latch = CountDownLatch(1)
fun main(args: Array<String>) {

    val result = runBlocking {
        val dispatcher: CoroutineDispatcher = dispatchExecutor.asCoroutineDispatcher()
        val intContext = InterceptingContext(dispatcher, latch)

        async(intContext) {
            println("Before exception")
            throw Exception("Whoops")
            println("After exception")
            3 //arbitrary value to return from Deferred
        }.await()

    }
    latch.await(5, TimeUnit.SECONDS)
    println("Done $result")
}

    class InterceptingContext(private val delegateInterceptor: ContinuationInterceptor,
                      private val latch: CountDownLatch)
    : AbstractCoroutineContextElement(ContinuationInterceptor), ContinuationInterceptor {

    override fun <T> interceptContinuation(continuation: Continuation<T>): Continuation<T> {

        val wrapped = WrappedContinuation(continuation,
                                      latch,
                                      { println("In pre-block") },
                                      { println("In post-block") })

        return delegateInterceptor.interceptContinuation(wrapped)
    }
}

class WrappedContinuation<in T>(
    private val continuation: Continuation<T>,
    private val latch: CountDownLatch,
    private val preblock: () -> Unit,
    private val postblock: () -> Unit)
: Continuation<T> {

    override val context: CoroutineContext
        get() = continuation.context

    override fun resume(value: T) {
        println("${Thread.currentThread().name} In resume")
        preblock()
        try {
            continuation.resume(value)
        } finally {
            postblock()
        }
    }

    override fun resumeWithException(exception: Throwable) {
        println("${Thread.currentThread().name} In resumeWithException")
        preblock()
        try {
            continuation.resumeWithException(exception)
        } finally {
            postblock()
            latch.countDown()
        }
    }

}

The point of the above code is to enable arbitrary execution of code before and after any resumption of a continuation. I’ve made two assumptions:

  1. My WrappedContinuation instance will intercept and delegate “resume(…)” (It does)
  2. if any exception occurs during execution of resume, the WrappedContinuation instance will intercept and delegate “resumeWIthException(…)” (it does not)

Am I missing something? I’ve assumed symmetry in this case, and my endeavour to debug through the kotlinx code has not left me any wiser, though I’ve seen plenty of invocations of resumeWithException as a result of the exception…just not on my wrapped continuation!

output:

pool-1-thread-1 @coroutine#2 In resume
In pre-block
Before exception
In post-block
Exception in thread "main" java.lang.Exception: Whoops
at coroutines.TestExecutor2Kt$main$result$1$1.doResume(TestExecutor2.kt:29)
at kotlin.coroutines.experimental.jvm.internal.CoroutineImpl.resume(CoroutineImpl.kt:54)
at coroutines.WrappedContinuation.resume(TestExecutor2.kt:68)
at kotlinx.coroutines.experimental.DispatchedContinuation$resume$1.run(CoroutineDispatcher.kt:152)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)

Any insight would be gratefully received.

2

Can you, please, explain what is not working? E.g. what results from running this code you’ve expected and what are the actual results you are getting?

I’m running this code and I’m not seeing any anomalies. The coroutine that you have in this code has a single continuation – from the start of the coroutine to its completion with the throw (it does not have suspensions so the initial continuation is all it has), so I’m seeing one interception occurring, printing:

In pre-block          // before continuation (starting coroutine)
Before exception  // execution code in the coroutine
In post-block        // after continuation (completing coroutine)
3

In my enthusiasm for using coroutines, I made some assumptions about how and when resumeWithException(…) would be called.

Assumption 1: If an exception is thrown in the coroutine after “onResume”, the same coroutine context would have resumeWithException(…) raised as an exception “callback” of sorts.

– the above appears to be wrong, given the test code already provided in this post. I think I also understand why my assumption was wrong.

Assumption 2: If coroutine B suspends on coroutine A, and coroutine A throws an Exception, then coroutine B would have “resumeWithException” raised.

so I tried a different test, based on the one above.

package coroutines

import kotlinx.coroutines.experimental.CoroutineDispatcher
import kotlinx.coroutines.experimental.Deferred  
import kotlinx.coroutines.experimental.asCoroutineDispatcher  
import kotlinx.coroutines.experimental.async
import kotlinx.coroutines.experimental.delay
import kotlinx.coroutines.experimental.runBlocking
import java.util.concurrent.Executors
import java.util.concurrent.RejectedExecutionException
import kotlin.coroutines.experimental.AbstractCoroutineContextElement
import kotlin.coroutines.experimental.Continuation
import kotlin.coroutines.experimental.ContinuationInterceptor
import kotlin.coroutines.experimental.CoroutineContext

//just an coroutines.getExecutor with two threads and a blocking backing queue
val dispatchExecutor1: java.util.concurrent.ExecutorService = Executors.newFixedThreadPool(2)
val dispatchExecutor2: java.util.concurrent.ExecutorService = Executors.newFixedThreadPool(2)

// the thread local I want to maintain
fun main(args: Array<String>) {
    val dispatcher1: CoroutineDispatcher = dispatchExecutor1.asCoroutineDispatcher()
    val dispatcher2: CoroutineDispatcher = dispatchExecutor2.asCoroutineDispatcher()

    val deferred1: Deferred<Int> = dispatch(dispatcher1) {
        println("Before exception")
//        delay(5000)
        throw Exception("Whoops")
    }

    val deferred2: Deferred<Int> = dispatch(dispatcher2) {
        val result1 = deferred1.await()
        result1 + 1
    }

    runBlocking {
        val finalResult = deferred2.await()
        println("Done $finalResult")
    }

    dispatchExecutor1.shutdownNow()
    dispatchExecutor2.shutdownNow()
}

class InterceptingContext(private val delegateInterceptor: ContinuationInterceptor)
    : AbstractCoroutineContextElement(ContinuationInterceptor), ContinuationInterceptor {

    override fun <T> interceptContinuation(continuation: Continuation<T>): Continuation<T> {

        val wrapped = WrappedContinuation(continuation,
                                      { println("In pre-block ${Thread.currentThread().name}") },
                                      { println("In post-block ${Thread.currentThread().name}") })

        return delegateInterceptor.interceptContinuation(wrapped)
    }
}

class WrappedContinuation<in T>(
        private val continuation: Continuation<T>,
        private val preblock: () -> Unit,
        private val postblock: () -> Unit)
    : Continuation<T> {

    override val context: CoroutineContext
        get() = continuation.context

    override fun resume(value: T) {
        println("${Thread.currentThread().name} In resume")
        preblock()
        try {
            continuation.resume(value)

        } finally {
            postblock()
        }
    }

    override fun resumeWithException(exception: Throwable) {
        println("${Thread.currentThread().name} In resumeWithException")
        preblock()
        try {
            continuation.resumeWithException(exception)
        } finally {
            postblock()
        }
    }

}

private fun <R> dispatch(dispatcher: CoroutineDispatcher,
                     block: suspend () -> R): Deferred<R> {

    return try {
        async(InterceptingContext(dispatcher)) {
            block()
        }

    } catch (ree: RejectedExecutionException) {
        ree.printStackTrace()
        throw ree
    }
}

Main difference is that there are now two coroutines instead of one, and the second waits on the first. And here’s where I am confused again. If I uncomment the “delay(5000)” line in coroutine 1, then I do, in fact get resumeWithException raised in coroutine 2.

pool-1-thread-1 @coroutine#1 In resume
pool-2-thread-1 @coroutine#2 In resume
In pre-block pool-2-thread-1 @coroutine#2
In pre-block pool-1-thread-1 @coroutine#1
Before exception
In post-block pool-2-thread-1 @coroutine#2
In post-block pool-1-thread-1 @coroutine#1
pool-1-thread-2 @coroutine#1 In resume
In pre-block pool-1-thread-2 @coroutine#1
In post-block pool-1-thread-2 @coroutine#1
pool-2-thread-2 @coroutine#2 In resumeWithException
In pre-block pool-2-thread-2 @coroutine#2
In post-block pool-2-thread-2 @coroutine#2
Exception in thread "main" java.lang.Exception: Whoops
at coroutines.TestExecutor2Kt$main$deferred1$1.doResume(TestExecutor2.kt:28)
at kotlin.coroutines.experimental.jvm.internal.CoroutineImpl.resume(CoroutineImpl.kt:54)
at coroutines.WrappedContinuation.resume(TestExecutor2.kt:71)
at kotlinx.coroutines.experimental.DispatchedContinuation$resume$1.run(CoroutineDispatcher.kt:152)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)

But if I leave the delay commented out, then resumeWithException never gets called…the exception just percolates out of the main method. This implies to me some timing issue in getting the resumeWithException invocation, in that coroutine1 failed before coroutine2 was even waiting for it, and thus the exception was “lost”.

My biggest hope here is that there is clear best practice on capturing exceptions in pipelines of coroutines. The only solution I can see to get exception logging closer to the exception detection is to wrap the call to “block()” in a try catch, and log there.

4

Without delay, the exception is thrown by the deferred1 coroutine so fast, that the deferred2 coroutine does not have chance to suspend. It just continues execution with exception. However, with delay in the deferred1, the second coroutine suspends, and then resumes with exception.

5

OK, I understand now. Thank you for taking the time to answer.

6

@elizarov when can we expect coroutines to be finalized? I am really excited to play with them.

7

They will be finalized when we receive enough feedback to have a confidence in all the details of their design. Meanwhile, we’ve released them in Kotlin 1.1, so that you can start using them. See here for details Can "experimental" Kotlin coroutines be used in production? - Stack Overflow

8

A post was split to a new topic: Functions with generic return typ