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.

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

Wrong builder. Use

channelFlow

for stuff like this.

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

offer

or

sendBlocking

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 { this@channelFlow.send(transform(item)) }
        }
}

Outer launch is not needed

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))
                }
            }
        }
    }
}

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