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.

1 Like
5

I’m actually legitimately surprised by the result of your test. Unless you’re running it on a custom dispatcher, it should just run on the main thread, where only one coroutine is active at a time.

As far as I understand, the answer to thread safety with coroutines is it depends on the dispatcher. If you just use runBlocking, then your coroutines are all running on the same thread, and only one coroutine can be active at any one time on that thread. So there’s no real concurrency. However, if you use a thread pool dispatcher, then each thread can be running one coroutine, and concurrency becomes a problem.

Like ebrowne72 said, the best solution would be to return the number and sum them at the end. Here’s an example using a Flow.

listOfStuff.asFlow()
    .flatMapMerge {
        flowOf(suspendingFunctionThatReturnsNumber(it.number))
    }
    .sum()

This code might not work perfectly, since I don’t know exactly what your source code is, but flatMapMerge will collect the returned Flows concurrently, so it’s basically doing async for the lambda inside flatMapMerge.

6

There’s an option that’s a lot closer to your code:

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

If you really want to do concurrent modification like that, use atomics
Either using AtomicInt directly:

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

or using your own Thing wrapped in an AtomicReference:

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

The OP’s code doesn’t specify what scope the function is called from. If it was called from a scope with no dispatcher set, or the dispatcher was Default, then it would behave the way they said it did.

1 Like
8

I actually did specify in my code that shows that it is not thread safe. It’s called from a suspend main function which means it uses the Default dispatcher.

9

Sure, but ALL suspending code is ultimately called from inside a runBlocking, or attached to a scope. How is your main() function called? runBlocking { main() }? Or do you mean that suspend fun main() is the entry point to the program? I’d be very surprised if Kotlin allows you to use a suspending function as the entry point… I gotta test that.

10

Ok, so apparently you can have suspend fun main() and just run it. That’s new, I learned something new today.

I did a bit of testing, and found very interesting behaviour differences between a suspending main method and using runBlocking. Consider the following code:

suspend fun main() {
    var total = 0

    coroutineScope {
        (1..1000).map {
            launch { total++ }
        }
    }

    println(total)
}

When running this code, I get varied results, anywhere from 980 - 1000. However, if I change the coroutineScope to runBlocking, I get consistent results every time.

suspend fun main() {
    var total = 0

    runBlocking {
        (1..1000).map {
            launch { total++ }
        }
    }

    println(total)
}

With this code, I ALWAYS get 1000, exactly. I haven’t once seen it print out anything other than 1000, after running it 10-20 times.

I tried checking what the CoroutineDispatcher was inside the suspending main function, but there is no dispatcher; it looks like the coroutine context has absolutely nothing in it, so I have no idea how it’s being executed, or what the difference is. I also checked the thread name, and it’s just running on main.

So… if you do want consistent, predictable execution on a single thread, wrap your code in runBlocking instead of using suspend fun main(). But this is pretty academic; the proper solution is to write your code in a thread-safe, concurrency-safe manner. :slight_smile:

1 Like
11

Did some more testing and figured it out. ebrowne72 and benjishults were right. When you use runBlocking, the dispatcher is set to BlockingEventLoop. If you have no dispatcher, it uses Dispatchers.Default, which has multiple threads. I’ve learned a few interesting things today.

3 Likes
12

One can also include thread name in execution, so it is immediately obvious when it is a multi-threaded access I suppose.

suspend fun main() {
    var total = 0

    coroutineScope {
        (1..1000).map {
            launch {
                total++
                println("${Thread.currentThread().name}")
            }
        }
    }

    println(total)
}

The code above output will start with something like

DefaultDispatcher-worker-14 @coroutine#14
DefaultDispatcher-worker-20 @coroutine#20
DefaultDispatcher-worker-17 @coroutine#17
DefaultDispatcher-worker-4 @coroutine#4
...

So it’s pretty clear that it is not safe.

1 Like
13

I was printing the thread name outside the loop, since my experience with coroutines is that they always run on the thread that the outer code is running on. But apparently, there is an exception when no dispatcher is set, which I think is only possible when you use suspend fun main().

Once I did print the thread name inside the launch, that’s when I started seeing “DefaultDispatcher-worker-1”, and went digging.

1 Like
14

Not the case for launch, async, and any kind of withContext IIUC. They’ll stay on the same or the provided dispatcher, but that might mean they switch to a different thread. Otherwise, how would parallelism even work?! You’d have to be switching dispatchers all the time

15

True, you’re right. I should have said the outer context.

16

as an general rule, if you see the below happening, in any language always look twice .

In this case (at least for me) the big issue is starting anything async/threads inside a function without calling that function as a unit itself… I hope I wrote that clear :slight_smile:

image
image1196×556 50.6 KB