So I was thinking about parallel mapping. So basic...

# coroutines

[Stephan Schroeder]

So I was thinking about parallel mapping. So basically having an extension function like this:

fun <T, R> Iterable<T>.pmap(
    exec: ExecutorService,
    transform: (T) -> R
): List<R>

which uses an instance of ExecutorService to do all the transformations in parallel. An implementation of this is given e.g. here: https://stackoverflow.com/questions/34697828/parallel-operations-on-kotlin-collections/35638609#35638609
The problem is that I want to continue to work on the Rs as soon as they’re done, so List<R> is the wrong collection class (because a list will not be returned until all transformations are done). First I was thinking Sequence<R> would be the anwser for this. But now I think it isn’t. Firstly I could even make it work because I can’t simply modify to code to return a Sequence<R):

= sequence{
    ....
    for (item in this) {
        exec.submit { yield transform(item) }
    }
    ....
}

is actually a compile-time error because I can’t call yield from a “sub-lambda”). But also a Sequence is supposed to be lazy, so you don’t start computing the next value until it’s requested. So what is the correct collection-like class??
Is this what coroutines’ Flow is for? I guess I could also use a Java Stream, but that just seams so unidiomatic.

UPDATE:
So I tried the flow-builder but it runs into the same problem as with Sequence:

fun <T, R> pFlowMap(
    items: Iterable<T>,
    exec: ExecutorService,
    transform: (T) -> R): Flow<R> = flow {

    for (item in items) {
        exec.submit { emit(transform(item)) }
    }

    exec.shutdown()
    exec.awaitTermination(1, TimeUnit.DAYS)
}

The emit is flagged with “Suspension functions can be called only within coroutine body”. So I tried wrapping that into a `runBlocking`:

runBlocking {
        for (item in items) {
            exec.submit { emit(transform(item)) }
        }
    }

but that dosn’t change a thing.

[Dennis Schröder]

Long story short: Look at Flow (https://kotlinlang.org/docs/reference/coroutines/flow.html)

[Dominaezzz]

Wrong builder. Use channelFlow for stuff like this.

[Stephan Schroeder]

channelFlow with send runs into the same problem of send being flagged with “Suspension functions can be called only within coroutine body”.
Playground link: https://pl.kotl.in/rKJJDKpDP

[Dominaezzz]

offer or sendBlocking

[Dominaezzz]

https://pl.kotl.in/FWbUVMUGh

[Dominaezzz]

parallelMap should work for you.

[Stephan Schroeder]

looks good 😃
Now if i dont want to care about the creation of the ExecutorService myself, i guess this is the way to go? The inner launches will be executed on the ThreadPool of the outer launch!?

= channelFlow {
     launch(newFixedThreadPoolContext(numberOfThreads, "tp1")) {
        items.forEach {item ->
            launch { [e-mail](transform(item)) }
        }
}

[Stephan Schroeder]

yes, that works 😃 (I added a 5s delay in the inner launch, and it definitely took only 5s and not 15!)

[gildor]

Outer launch is not needed

[gildor]

Also careful with creating own pools, if you so not close them it cause leak of threads

[gildor]

Probably IO or Default dispatcher is enough, but depends on case, but IO and Default share threads

[Stephan Schroeder]

I’m writting a chess program, I want control over my pools. I guess I’ll just have to handle it properly then 😎
The latest version looks more like this:

inline fun <T, R> Iterable<T>.parallelTransform(
    numberOfThreads: Int,
    crossinline transform: (T) -> R
): Flow<R> = channelFlow {

    val items: Iterable<T> = this@parallelTransform
    val channelFlowScope: ProducerScope<R> = this@channelFlow

    Executors.newFixedThreadPool(numberOfThreads).asCoroutineDispatcher().use { dispatcher ->
        launch( dispatcher ) {
            items.forEach { item ->
                launch {
                    channelFlowScope.send(transform(item))
                }
            }
        }
    }
}

[gildor]

This is not very efficient, on every launch of this function you create a thread pool, it creates threads and then you shutdown everything.
Also it's hard to control resources, what if someone launch too many of those parallel transform?
If you do not want to use standard dispatchers (which also provide API to configure them) I would recommend to create own and reuse it across you app