Kotlinx.coroutines: Redundant 'async' call may be reduced to 'kotlinx.coroutines.withContext'

[ScottPierce]

I have the following code:

DATABASE_CONTEXT.async(block = block).await()

IntelliJ gives me a warning and wants to simplify it to use withContext instead of async.

These two things aren't exactly equivalent though, right? i.e. if I cancel the coroutine, it will also stop the coroutine code from executing if I user withContext, but if I use DATABASE_CONTEXT.async(block = block).await(), that has a new coroutine hierarchy, and that will ensure that the code inside the block runs to completion, right?

Those are two very important distinctions, and I like to differentiate between the use-cases. The fact that the IDE is suggesting these to me as equivalent seems like an error to me.

[circusmagnus]

I think, You should define your DATABASE_CONTEX as NonCancellable (with appropriate dispatcher) and use withContext(DATABASE_CONTEX).

If you just need non cancellability, then just use withContext(NON_CANCELLABLE).

The code you posted, is misleading. Async invocation suggests, that there is some concurrency in this code, but there is none.

[ScottPierce]

Didn't know about that. So you are saying

private val DATABASE_CONTEXT: CoroutineScope = GlobalScope +
        Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors()).asCoroutineDispatcher() +
        NonCancellable

is what I want, and that with the DATABASE_CONTEXT defined like this, it's safe to use withContext?

[circusmagnus]

I would leave GlobalScope out.

I do not know your use case fully, but I think something along this lines should work:

private val DB_DISPATCHER = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors()).asCoroutineDispatcher()
private val DB_CONTEXT: CoroutineContext = DB_DISPATCHER + NonCancellable

private val DB_SCOPE: CoroutineScope = CoroutineScope(DB_CONTEXT)

suspend fun demo1(): Data = withContext(DB_CONTEXT) {
    db.getData()
}

suspend fun demo2(data: Data) = DB_SCOPE.launch {
    db.saveData(data)
}

Using such a DB_SCOPE however might be dangerous, as you will not be able to cancel it ever (screen exit, application exit, etc)

You may consider using a top level scope with SuperVisorJob (instead of NonCancellable) - and use you non-cancellable DB_CONTEXT just for critical pieces of code, which really need to be non-cancellable. Like:

val OridinaryScope = Dispatchers.Deafault + SupervisorJob

suspend fun demo2(data: Data) = OrdinaryScope.launch {
    bla...
   bla...
    withContext(DB_CONTEXT) { db.saveData(data) }

    bla...
}

That way your whole coroutine can be cancelled if needed, will not totally fail if error is encountered (thanks to SupervisorJob) but piece of code dealing with database will not get cancelled.

[elizarov]

@ScottPierce Indeed, they are not equivalent. Yes, usually async { ... }.await() is an error anyway. Can you, please, elaborate what is exactly your use-case for writing code like that?

[ScottPierce]

@elizarov My motivation came purely from the ability to create a separate coroutine hierarchy that wouldn't be cancelled, and would always complete after started, if its owning coroutine scope is cancelled.

I think one of the biggest problems with the kotlinx.coroutines library at the moment is the current cancellation behavior, which is enabled by default. It can lead to unexpected, difficult to find, difficult to debug race conditions, especially whenever any long-running state is involved.

[elizarov]

@ScottPierce Can you, please, add a little bit more substance to your use-case. Like specific examples of what kinds of tasks you want this "separate coroutine hierarchy that wouldn't be cancelled" for?

[ScottPierce]

Sure. I certainly don't consider myself a coroutine expert, but potentially any time you have long-lived state shared between multiple coroutines, and that state is set in at least two locations with a suspension point between them, there is a potential for race conditions if that coroutine were to be cancelled during that suspension point. I think this makes the current coroutine cancellation model a bit dangerous to people who don't know to watch for something like this.

I understand you probably want some real life examples. Take for example that we wanted to keep track of the current active number of requests. We could do something like this:

  protected suspend fun <T> withApiContext(
    block: suspend CoroutineScope.() -> T
  ): T {
    return withContext(apiScope.coroutineContext) {
        runningRequests++
        block()
        runningRequests--
    }
  }

Code running in playground

Notice how in the above playground that the running requests variable quickly gets out of sync. I can get the code to succeed, but I have to know and be aware that this code can be cancelled out from under me, and use a try / finally block

We encountered similar issues to this when coding our data layer in our Android app. Our Data layer is a MPP enabled layer that is responsible for retrieving all data in our app. We have a cache in this layer. Our cache keeps track of what cache-keys have actively running requests, and if a cache-key has an active request, then further requests to that cache-key just join to the current request and wait for that response, instead of firing off another server request. We either have to ensure that the UI cancelling it's coroutine context properly removes it's running request from the map of running requests using a finally block, or we have to ensure that the entry point to the data layer starts a new coroutine using a different parent context entirely so it isn't part of the cancellation.

We ended up doing both. The data layer described above has many other pieces of state, and when we started to see some weird issues we didn't think were possible, we theorized it could be caused by cancellation issues. We learned how coroutine cancellations work, were a little surprised, and then stopped using withContext, instead using something like this:

protected suspend fun <T> withDataContext(
  block: suspend CoroutineScope.() -> T
): T {
  try {
    val result = dataScope.async {
      block()
    }
    return result.await()
  } catch (e: Exception) {
    logger.i("Api Exception:", e)

    val apiException = e.toApiException()

    ...

    throw apiException
  }
}

We haven't seen any weird cancelation issues since.

Now whenever I work with coroutines, I create logical layer boundaries that I ensure have different parent contexts, so I don't accidentally have unexpected cancellation races.

[elizarov]

Would wrapping this kind of non-divisible (atomic?) block of code into withContext(NonCancellable) help you? Like this:

withContext(NonCancellable) {
    runningRequests++
    block()
    runningRequests--
}

[ScottPierce]

From an efficiency standpoint, yes. Now that I've learned about NonCancellable being able to be added into CoroutineContext, we can add it into our data layer scope:

val dataScope: CoroutineScope = GlobalScope +
          Platform.createDataCoroutineDispatcher() +
          CacheCoroutineContextElement(get()) +
          CoroutineExceptionHandler { coroutineContext, throwable ->
            // This should catch exceptions from launch coroutines where a result isn't expected.
            // Not having this would cause those exceptions (Network and HttpExceptions included) to crash the app.
            logger.i("Unhandled Exception:", t = throwable)
          } +
          NonCancellable

This should allow us to start using withContext instead of async / await.

I will say that the cancellation behavior has probably been one of the more surprising parts of my team switching to coroutines a year+ back. So much of coroutines are so intuitive that when you have to change the way you think about common problems because of things like the cancellation behavior, it can be fairly unexpected.

Proper layered architecture, and manually maintaining those layer boundaries has been essential so we don't have to worry about problems like this. I very much look forward to the compiler plugin API being exposed experimentally (hopefully in 1.4) so that the maintaining layer boundaries can be automated without adding significantly to compile times. 😃

[elizarov]

@ScottPierce I think the missing piece is the good out-of-the-box support for an actor pattern that we've delayed to roll out due to our work on flow. See discussion #87. Actor is an entity with it own scope, so the code running inside an actor does not get affected by cancelation of other subsystems that are interacting with it.

[ScottPierce]

I'm not sure Actors are a great solution. I'm not sure I'd want to introduce them to my team given that they only allow a single request at a time, and can deadlock. Heck, in the sample app you created for one of your kotlin-conf talks, didn't it have deadlocks?

[elizarov]

@ScottPierce How do you coordinately runn multiple requests at the same time without getting into data races? Can you share some larger piece of code that shows your current architecture?

[ScottPierce]

@elizarov To be clear, the ticket I originally filed was for server code, but what I'm about to describe is for my mobile teams Android application.

Here is a very high level overview of our architecture. We've been writing it using MPP for the past 1.5+ years, but only compiling it to Android / JVM. The View code is the only platform specific code, and most everything else is common, which is roughly 60-75% of the code.

How do you coordinately run multiple requests at the same time without getting into data races

Each high level module operates under a specified Dispatcher. i.e. the UI layer operates under the Main, and the Data layer operates under it's own custom single-threaded Dispatcher that I show the creation of in an above post. Since it's all running under a single thread, it's super easy to reason about, and we've had no issues, aside from the fact that we have to maintain the edge of the data layer manually.

What I mean by that, is that we have to follow the rule that all public methods in the data layer are required to:

1. Have the suspend keyword.
2. Immediately transfer to the data CoroutineContext with the withDataContext method I showed above.

i.e.

  /** Authenticate with email / password. */
  suspend fun authEmail(
    email: String,
    password: String
  ): AuthResult = withDataContext {
    ...
  }

My goal is to create a Kotlin compiler plugin that manages these module / thread boundaries for us automatically by generating a wrapping class.

The key difference between this and an actor is that it allows multiple requests at a single time. If you draw out the flow of data from a high level, it'd look very similar to a web worker or an isolate, only without requiring serialization / deserialization.

Obviously this threading model can be limited, and doesn't handle all use cases, but it's very effective for stateful application code, which is the vast majority of what mobile devs write. We don't need anything else because any long-running thread blocking operations are all just network requests for us, and ktor owns the parallelization of that. If we had a need for more parallel tasks, we'd consider using something similar to actors possibly, but they'd likely have certain restrictions, such as requiring multiple instances, being stateless, and confined to the very edge of the application to minimize having to worry about headaches like deadlocks.

[elizarov]

Thanks a lot! It is also good to know that this architecture works well with withContext. As for "wrapping plugin" -- we indeed periodically run into cases where this or that kind of wrapping might be needed and some day this kind of "auto-wrapping" might become a separate language feature.

[ScottPierce]

Now that I know about NonCancellable, it seems like it'd be fine to close this. While they aren't equivalent, proper usage of contexts can lead to them being equivalent from my perspective.

[Zordid]

I have exactly the same issue, maybe somebody could again help??
I do NEED async { jmsconsumer.receive() }.await() to call my external JMS (MQ) receive command and still be able to cancel from the outer scope.
When I apply the fix IntelliJ suggests, merely saying withContext { jmsconsumer.receive() }, it works, but if the job hierarchy gets cancelled, this one is stuck forever.

Generally speaking: my problem is the bridging between old libraries that do asynchronous IO activity, just like JmsConsumer.receive() does and my clean hierarchy of coroutines in Kotlin.

I simply don't see how I call one from the other and still adhere to the overall principles of the cleaner Kotlin coroutines... :-(

[qwwdfsad]

When I apply the fix IntelliJ suggests, merely saying withContext { jmsconsumer.receive() }, it works, but if the job hierarchy gets cancelled, this one is stuck forever.

It gets stuck anyway, but in case of async you just don't wait for it on the outer scope. So this behaviour is rather uncovered a potential leak in your code.
If JMS supports interruption, I would recommend using runInterruptible to support cooperative cancellation of JMS