Stateful Goroutines in Kotlin

Our example demonstrates how to manage state using coroutines and channels in Kotlin. This approach aligns with Kotlin’s ideas of sharing memory by communicating and having each piece of data owned by exactly one coroutine.

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
import java.util.concurrent.atomic.AtomicLong
import kotlin.random.Random

data class ReadOp(val key: Int, val response: CompletableDeferred<Int>)
data class WriteOp(val key: Int, val value: Int, val response: CompletableDeferred<Boolean>)

suspend fun main() = coroutineScope {
    val readOps = AtomicLong(0)
    val writeOps = AtomicLong(0)

    val reads = Channel<ReadOp>()
    val writes = Channel<WriteOp>()

    // Coroutine that owns the state
    launch {
        val state = mutableMapOf<Int, Int>()
        while (true) {
            select<Unit> {
                reads.onReceive { read ->
                    read.response.complete(state[read.key] ?: 0)
                }
                writes.onReceive { write ->
                    state[write.key] = write.value
                    write.response.complete(true)
                }
            }
        }
    }

    // Launch 100 coroutines to perform reads
    repeat(100) {
        launch {
            while (true) {
                val read = ReadOp(Random.nextInt(5), CompletableDeferred())
                reads.send(read)
                read.response.await()
                readOps.incrementAndGet()
                delay(1)
            }
        }
    }

    // Launch 10 coroutines to perform writes
    repeat(10) {
        launch {
            while (true) {
                val write = WriteOp(Random.nextInt(5), Random.nextInt(100), CompletableDeferred())
                writes.send(write)
                write.response.await()
                writeOps.incrementAndGet()
                delay(1)
            }
        }
    }

    // Let the coroutines work for a second
    delay(1000)

    // Capture and report the op counts
    println("readOps: ${readOps.get()}")
    println("writeOps: ${writeOps.get()}")
}

In this Kotlin version:

We use data class for ReadOp and WriteOp instead of structs.
We use CompletableDeferred for the response channels, which allows us to await the response.
The main function is marked as suspend and wrapped in coroutineScope to allow the use of coroutines.
We use Kotlin’s Channel instead of Go’s channels.
The state-owning goroutine is replaced with a coroutine launched using launch.
We use Kotlin’s select function to handle multiple channels.
Instead of goroutines, we launch multiple coroutines using launch.
We use AtomicLong for atomic operations instead of the atomic package in Go.
time.Sleep is replaced with delay.

Running our program shows that the coroutine-based state management example completes about 80,000 total operations.

$ kotlinc -cp .:kotlinx-coroutines-core-1.5.2.jar StatefulCoroutines.kt -include-runtime -d StatefulCoroutines.jar
$ java -jar StatefulCoroutines.jar
readOps: 71708
writeOps: 7177

For this particular case, the coroutine-based approach in Kotlin is quite similar to the goroutine-based approach in Go. It might be useful in certain cases, especially where you have other channels involved or when managing multiple such mutexes would be error-prone. You should use whichever approach feels most natural, especially with respect to understanding the correctness of your program.

Mutexes in Kotlin

Sorting in Kotlin

Learn X By Example

Title here

Stateful Goroutines in Kotlin