Modifying a var within an async coroutine

Support
1

I’ve inherited some code and I’m wondering about how thread-safe it is.

What’s bugging me is that the code makes an assignment to a var from within code that could be executed on any number of threads.

Here’s the idea:

data class Stuff(val number: Int)
data class Thing(val accumulation: Int)

suspend fun foo(): Int {
    var count: Thing = Thing(0)
    val listOfStuff = listOf<Stuff>(/* a bunch of Stuffs here */)
    coroutineScope {
        listOfStuff.map { s ->
            async {
                count = Thing(count.accumulation + s.number)
            }
        }
        .awaitAll()
    }
    return count.accumulation
}

My question is, is the assignment to count thread safe?

Put another way: If those coroutines are run in parallel on different threads (and there’s nothing here preventing that happening) is there a possibility that the value of count will be wrong in the end?

Obviously, this could be fixed by making count an AtomicReference but I don’t want to go to that trouble if I don’t have to.

Ideally, if the answer is “yes, it is safe” please also provide a link to some docs or a spec I can read that will confirm that because I’m interested to learn how they do that. (It seems to me that it should NOT be safe.)

2

I just wrote a test and it was pretty easy to show that this is NOT thread safe.

I wonder how much code like this is out there.

I ran this code:

data class Stuff(val number: Int)
data class Thing(val accumulation: Int)

suspend fun foo(): Int {
    var count: Thing = Thing(0)
    val listOfStuff = buildList {
        repeat(10_000) { add(Stuff(1)) }
    }
    coroutineScope {
        listOfStuff.map { s ->
            async {
                count = Thing(count.accumulation + s.number)
            }
        }
            .awaitAll()
    }
    println(count.accumulation)
    return count.accumulation
}

suspend fun main() {
    foo()
}

and it output 9929 instead of 10_000.

3

I don’t know coroutines, but if they run on multiple threads, then I strongly suspect that it’s NOT thread-safe. (I can’t quote you a reference, though.)

I think one failure mode is that when one thread updates count, another thread could see the updated value but NOT the initialised Thing instance it refers to. (There are probably additional failure modes too.)

Although that access could be made safe with an AtomicReference (or possibly a @Volatile annotation or some other technique), I think that would still leave a race condition where two threads could read the count.accumulation, and then both could create and assign Thing instances from it — one of their updates would be lost.

In this simple case, I think a much better approach would be for each coroutine to simply yield its value independently, and then the surrounding code to sum all the values. (Though, as I said, I don’t know coroutines so I can’t write the code.) Of course, your more complex real-world case may not allow that approach.

4

If the async can return the count, then something like this would be better.

val count = coroutineScope {
    listOfStuff.map { s ->
        async {
           s.number
        }
    }
    .fold(0) { count, deferred ->
        count + deferred.await()
    }
}

If not then you could have each coroutine output to a Channel, and receive all the counts in one receiver that updates the Thing.