Kotlin coroutines progress counter

Question

I'm making thousands of HTTP requests using async/await and would like to have a progress indicator. I've added one in a naive way, but noticed that the counter value never reaches the total when all requests are done. So I've created a simple test and, sure enough, it doesn't work as expected:

fun main(args: Array<String>) {
    var i = 0
    val range = (1..100000)
    range.map {
        launch {
            ++i
        }
    }
    println("$i ${range.count()}")
}

The output is something like this, where the first number always changes:

98800 100000

I'm probably missing some important detail about concurrency/synchronization in JVM/Kotlin, but don't know where to start. Any tips?

UPDATE: I ended up using channels as Marko suggested:

/**
 * Asynchronously fetches stats for all symbols and sends a total number of requests
 * to the `counter` channel each time a request completes. For example:
 *
 *     val counterActor = actor<Int>(UI) {
 *         var counter = 0
 *         for (total in channel) {
 *             progressLabel.text = "${++counter} / $total"
 *         }
 *     }
 */
suspend fun getAssetStatsWithProgress(counter: SendChannel<Int>): Map<String, AssetStats> {
    val symbolMap = getSymbols()?.let { it.map { it.symbol to it }.toMap() } ?: emptyMap()
    val total = symbolMap.size
    return symbolMap.map { async { getAssetStats(it.key) } }
        .mapNotNull { it.await().also { counter.send(total) } }
        .map { it.symbol to it }
        .toMap()
}

Answer 1

The explanation what exactly makes your wrong approach fail is secondary: the primary thing is fixing the approach.

Instead of async-await or launch, for this communication pattern you should instead have an actor to which all the HTTP jobs send their status. This will automatically handle all your concurrency issues.

Here's some sample code, taken from the link you provided in the comment and adapted to your use case. Instead of some third party asking it for the counter value and updating the GUI with it, the actor runs in the UI context and updates the GUI itself:

import kotlinx.coroutines.experimental.*
import kotlinx.coroutines.experimental.channels.*
import kotlin.system.*
import kotlin.coroutines.experimental.*

object IncCounter

fun counterActor() = actor<IncCounter>(UI) {
    var counter = 0
    for (msg in channel) {
        updateView(++counter)
    }
}

fun main(args: Array<String>) = runBlocking {
    val counter = counterActor()
    massiveRun(CommonPool) {
        counter.send(IncCounter)
    }
    counter.close()
    println("View state: $viewState")
}


// Everything below is mock code that supports the example
// code above:

val UI = newSingleThreadContext("UI")

fun updateView(newVal: Int) {
    viewState = newVal
}

var viewState = 0

suspend fun massiveRun(context: CoroutineContext, action: suspend () -> Unit) {
    val numCoroutines = 1000
    val repeatActionCount = 1000
    val time = measureTimeMillis {
        val jobs = List(numCoroutines) {
            launch(context) {
                repeat(repeatActionCount) { action() }
            }
        }
        jobs.forEach { it.join() }
    }
    println("Completed ${numCoroutines * repeatActionCount} actions in $time ms")
}

Running it prints

Completed 1000000 actions in 2189 ms
View state: 1000000