Another question. It is probably somewhere in the ...

# coroutines

[altavir]

Another question. It is probably somewhere in the documentation, but is the Flow map operation parallelizable?

[gildor]

You can use flatMap* for this
https://kotlin.github.io/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/flat-map-merge.html

[gildor]

map is sequential

[altavir]

for 1->1 map I will need to create single-element flows, which is not quite convenient, but thanks, I will look into the source. I was thinking about using regular map to generate Deferred and then await them in next map

[altavir]

Something like this:

fun <T,R> Flow<T>.mapParallel(scope: CoroutineScope, transform: suspend (T)->R) = map{
    scope.async { transform(it) }
}.map{
    it.await()
}

[altavir]

Probably won't work since intermediate operations are lazy...

[gildor]

I will need to create single-element flows

There are extension functions for functional types (including suspend) which create flow from lambda/function

[altavir]

There still is a problem of structured concurrency and dispatcher. flatMapMerge does not provide control over scope and dispatcher, so it could give a work around at best. I think this is rather common task so it should have dedicated solution.

[elizarov]

If all you need is to map, you can do:

.flowWith(Dispatchers.Default) { // BG!
    flatMapMerge {
        flowOf(slowOperation(it))
    }
}

[altavir]

OK, I think I understand. But it is not obvious. I think it still needs a separate functions.

[altavir]

@elizarov Your solution does not work. I've just placed in my test and it evaluates sequentially, not in parallel.

[altavir]

I think it is because dispatcher does not actually change and there is a little clause in flowWith that skips channel generation in case context is the same

[altavir]

Nope, even if I start with IO or Unconfined, it still does not work. In order to get parallel execution in coroutines, one need to first generate a list of async, store them and then await them. In my solution channel plays the role of interediate storage, but I do not think any solution with flowOn would generate this intermediate state because, channel there is consumed immediately.

[altavir]

The working function is here: https://github.com/mipt-npm/kmath/blob/68ccc0b3fcb698a6c87c5fd26fd8e32033ff5da8/kmath-coroutines/src/commonMain/kotlin/scientifik/kmath/streaming/BufferFlow.kt#L70 , the test is here: https://github.com/mipt-npm/kmath/blob/68ccc0b3fcb698a6c87c5fd26fd8e32033ff5da8/kmath-coroutines/src/jvmTest/kotlin/scientifik/kmath/streaming/BufferFlowTest.kt#L15

[elizarov]

Yes, you are right. Transforms are computed sequentially.

[sdeleuze]

Also discouraging people to use flatMap in the doc seems misleading to me, I think most people want parrallel execution when calling suspending function to transform element of a Flow

[sdeleuze]

Or maybe map could do parrallel transformation when used with suspending functions?

[altavir]

I managed to make it work here: https://github.com/mipt-npm/kmath/blob/dev/kmath-coroutines/src/commonMain/kotlin/scientifik/kmath/CoroutinesExtra.kt

[altavir]

The idea is first to transform flow into flow of lazy value (I can't use lazy deferred due to structured concurrency leak). Then I trigger start for required number of parallel lazy computations when I do collect (new collect method requires explicit number of started tasks).

[gildor]

I agree with Sebastien, map is sequential in Rx just because it doesn't support async operation there by definition, but with Coroutines it's possible to make map parallel, even do this optional (add concurrency param with default value 1), so it will use fast path by default (current implementation) and optional easy to use concurrency

[sdeleuze]

I like the idea of a map optional parameter, and I think I would go as far as using same concurency value that flatMap. Coroutines are cheap, Flow is designed to provide a powerful map operator where people will expect parrallel execution for suspending functions.

[sdeleuze]

I guess the related issue where we should continue the discussion is https://github.com/Kotlin/kotlinx.coroutines/issues/1147