How do I run two coroutines concurrently but take the result of the first one that returns and cancel the other? Preferably a one liner.

I made a library (available on MavenCentral) for that. The implementation is not a one-liner, but usage is quite straightforward I think:

suspend fun testCoroutinesRacing() {
    val result = raceOf({
        delay(3)
        "slow"
    }, {
        delay(0)
        "fast"
    })
    assertEquals(expected = "fast", actual = result)
}

https://github.com/LouisCAD/Splitties/tree/main/modules/coroutines#racing-coroutines

there's a lot of discussion on https://github.com/Kotlin/kotlinx.coroutines/issues/424 as well

Finally opened an issue to ask for an API to race coroutines: https://github.com/Kotlin/kotlinx.coroutines/issues/2867

Here's my implementation.

@louiscad what if i want to provide a predicate to determine the “winner” ?

If

awaitCancellation()

doesn't suit your needs, then I guess you are looking for something different. If you come up with something, please share.

ohh

Related: https://kotlinlang.slack.com/archives/C1CFAFJSK/p1627721102019900

Arrow has race functions too

@Rob I guess. If prefer

raceOf

or

race

in regards to ensuring structured concurrency, but I'm biased.

Standard library already has

awaitAll()

. So it's matches their naming scheme. Thanks for the feedback.

It's not too hard with `Flow`:

merge(::method1.asFlow(), ::method2.asFlow()).first()

Little more awkward if you want the lambdas:

suspend fun testCoroutinesRacing() {
    val result = merge(suspend {
        delay(3)
        "slow"
    }.asFlow(), suspend {
        delay(0)
        "fast"
    }.asFlow()).first()
    assertEquals(expected = "fast", actual = result)
}

It's still not getting cancelled 🤦

Yes it is. first cancels the result of merge which cancels the still active "slow" flow.

I'm testing it, and it's not cancelling the slower coroutine.

fun main() {
    runBlocking {
        val x = merge<String>(suspend {
            try {
                delay(3)
                "slow"
            } catch (e: CancellationException) {
                println("Cancelled slow")
                throw e
            }
        }.asFlow(), suspend {
            delay(0)
            "fast"
        }.asFlow()).first()
        println(x)
    }
}

prints

Cancelled slow
fast

This doesn't cancel the first coroutine.

Sorry, I thought you were responding to my comments about using

merge

and

Flow

.

I finally figured out why it wasn't cancelling before. I was cancelling the job blocking on the deferred but not the deferred itself.

FYI, the code I linked way above can be copy pasted, it's self-contained in a file, it's not a one-liner, but it's not a thousand, nor a hundred of lines either.

It had some some things that made it not easily copy pasteable.

Seems like maybe you'd be better served with:

merge(::method1.asFlow().catch {}, ::method2.asFlow().catch{}).firstOrNull()

That is if you can work with suspend methods instead of

Deferred

. You seem to want to use

Deferred

but it's not clear to me if you actually need to.

I was following the precedent set by

awaitAll()

. I think

first()

works for my case. I do want to get an error if ALL the coroutines fail.

Keep in mind any solution that does not cancel your

Deferred

objects can be modified to just cancel everything once it has a value:

suspend fun <T> awaitFirstIgnoringErrors(vararg deferreds: Deferred<T>): T = deferreds
    .map { it::await.asFlow().catch {} }
    .merge()
    .first()
    .also { _ -> deferreds.forEach { it.cancel() }}

That didn't cancel the slow deferred in my test.

Try waiting longer than 3ms. If the coroutine manages to finish, it can't be cancelled.

Here is my test.

awaitFirstIgnoringErrors

is not throwing the exception.

runBlocking

is throwing the exception because a child failed. You have to wrap the code in

supervisorScope

.

Yeah. I read that unless you do

await()

on an

async

block exceptions can be lost. I thought I could catch

async

block errors, but they still cancel the scope when

await()

is called.

It has nothing to do with

await

. This will fail:

runBlocking { async { throw Exception() }}

Failed child jobs fail the parent. This is part of structured concurrency. The exception is if you use a

SupervisorJob

which does not mind if child jobs fail (in this case failed child exceptions are handled like top level coroutine exceptions).

I didn't know that

async

blocks cancelled their parent if they failed even if you didn't call

await

. I thought I read that

await

had to be called. Everything makes sense now 😄

GlobalScope works like SupervisorJob (technically it has no job at all). Methods like

Deferred.await

or

Channel.receive

(where you are asking for a result) will throw an exception ... since that was the result (it failed).

cancelling GlobalScope if a child failed would sort of amazing :D

I guess I thought `Deferred`s were like `Future`s, `Promise`s, `Single`s etc.

they are, it's just that failures also cancel everything up to the nearest supervisor (or global) scope. compare

runBlocking { supervisorScope { async { throw Exception() } } }

I'm doing it like this. It doesn't return anything, but it could probably be adapted to use `Deferred<T>`:

suspend inline fun awaitFirst(vararg funcs: suspend () -> Unit) {
  supervisorScope {
    select<Unit> {
      funcs.forEach { func -> launch { func() }.onJoin {} }
    }

    coroutineContext.cancelChildren()
  }
}