How to do parallel flatMap in Kotlin?

user_1743253 0
Viewed 3130

I need to do parallel flat map. Let's say I have this code:

val coll: List<Set<Int>> = ...
coll.flatMap{set -> setOf(set, set + 1)}

I need something like this:

coll.pFlatMap{set -> setOf(set, set + 1)} // parallel execution
kotlinx.coroutines kotlin coroutine
3 Answers

Kotlin doesn’t provide any threading out of the box. But you can use kotlinx.coroutines to do something like this:

val coll: List<Set<Int>> = ...
val result = coll
 .map {set -> 
    // Run each task in own coroutine,
    // you can limit concurrency using custom coroutine dispatcher
    async { doSomethingWithSet(set) } 
 }
 .flatMap { deferred -> 
    // Await results and use flatMap
    deferred.await() // You can handle errors here
 }
user_4204120

Alternatively, you can do it without coroutines:

fun <T, R> Collection<T>.pFlatMap(transform: (T) -> Collection<R>): List<R> =
    parallelStream().flatMap { transform(it).stream() }.toList()

This solution requires Kotlin on JDK 8 or higher.

You can also make it more general (analogous to Kotlin's flatMap):

fun <T, R> Iterable<T>.pFlatMap(transform: (T) -> Iterable<R>): List<R> =
    toList()
        .parallelStream()
        .flatMap { transform(it).toList().stream() }
        .toList()
user_4438360

You need to add Coroutine scope first (runBlocking), and apply deferred execution to your function (async):

val coll: List<Set<Int>> = listOf()
val x = runBlocking {
    coll.map{ set ->
        async {
            setOf(set, set + 1)
        }
    }.awaitAll().flatten()
}

x ....
user_9354080
Related