Another question It is probably somewhere in the documentati

Question

Another question It is probably somewhere in the documentation but is the `Flow` map operation parallelizable

gildor · Answer

You can use flatMap for this <https kotlin github io kotlinx coroutines kotlinx coroutines core kotlinx coroutines flow flat map merge html>

gildor · Answer

map is sequential

altavir · Answer

for 1 gt 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 · Answer

Something like this ``` fun lt T R gt Flow lt T gt mapParallel scope CoroutineScope transform suspend T gt R = map scope async transform it map it await ```

altavir · Answer

Probably won t work since intermediate operations are lazy

gildor · Answer

gt I will need to create single element flows There are extension functions for functional types including suspend which create flow from lambda function

altavir · Answer

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 · Answer

If all you need is to map you can do ``` flowWith Dispatchers Default BG flatMapMerge flowOf slowOperation it ```

altavir · Answer

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

altavir · Answer

< elizarov> Your solution does not work I ve just placed in my test and it evaluates sequentially not in parallel

altavir · Answer

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 · Answer

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 · Answer

The working function is here the test is here

elizarov · Answer

Yes you are right Transforms are computed sequentially

sdeleuze · Answer

Don t you think there is a missing operator between sequential `map` and `flatMapMerge Concat` which forces to use artificially a single element `Flow` to achieve parrallel execution Processing each element of a `Flow` with slow operations remote http call for each element is very common

sdeleuze · Answer

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 · Answer

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

altavir · Answer

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 · Answer

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 · Answer

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 · Answer

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 · Answer

I guess the related issue where we should continue the discussion is <https github com Kotlin kotlinx coroutines issues 1147>