Parallel flow processing

[altavir]

It makes sense that Flow and all map operations are sequential by default, but it makes sense to allow some heavy operations to be performed in parallel on explicitly provided dispatcher.

I've implemented the feature based on existing flowOn implementation:

@FlowPreview
fun <T, R> Flow<T>.mapParallel(scope: CoroutineScope, bufferSize: Int = 16, transform: suspend (T) -> R) =
    flow {
        val currentContext = coroutineContext.minusKey(Job) // Jobs are ignored

        coroutineScope {
            val channel = produce(currentContext, capacity = bufferSize) {
                collect { value ->
                    send(scope.async { transform(value) })
                }
            }

            (channel as Job).invokeOnCompletion { if (it is CancellationException && it.cause == null) cancel() }
            for (element in channel) {
                emit(element.await())
            }
//
//            val producer = channel as Job
//            if (producer.isCancelled) {
//                producer.join()
//                throw producer.getCancellationException()
//            }
        }
    }

It seems to be working fine, but I can't use internal kotlinx,coroutines methods. I think that this use-case require a separate function in the library.

[elizarov]

We do plan a set of separate functions for that and to close #172 with them. However, the key challenge here is to introduce a composable set of concurrency operators as opposed to specific operators like concurrentMap, which is hard to scale to other operators (concurrentFilter, concurrentFold, etc). Tentatively, it is going to look like this:

flow.concurrent(n) { 
    // inside here we can have a sequence of flow operations,
    // each doing its own thing concurrently
    // this: Flow<T> here similarly to `flowWith` operator
    filter { ... }.map { ... }.filter { ... } // -> Flow<R>
} // -> Flow<Flow<R>> 

Alternatively you can call concurrent without a lambda. This way you get Flow<Flow<T>> which you can map manually (so concurrent with lambda is just integrated with map):

// different way to write the same code as above
flow.concurrent(n).map { 
    // it: Flow<T> here
    it.filter { ... }.map { ... }.filter { ... } // -> Flow<R>
} // -> Flow<Flow<R>> 

Note, that the result is going to be Flow<Flow<R>>. Right now the only useful thing you can do on it is flattenMerge() to get Flow<R>. But what if you want to reduce/fold the result somehow -- that will require a sequence of invocations and will not parallelize properly. You can repeat reduction inside and outside, but that does not look pretty. A classical concurrent map-reduce pipeline would be written, for example, like this:

flow.concurrent(n) { // this: Flow<Input>
    map(transform) // -> Flow<Mapped>
   .reduce(reducer) // -> R
} // -> Flow<R>
.reduce(reducer) // we need to reduce again to get -> R 

So that part of design for concurrent terminal operators is still TBD with map-reduce being the primary use-case.

The similar design problem we have with terminal concurrent collect:

flow.concurrent(n) {
    collect { ... } // collect in `n` coroutines concurrently 
}.collect {} // but must write an empty collect here to activate it all.

[altavir]

I see the idea, but parallel is not necessary concurrent. On the contrary, in my example, there is no concurrency. The key feature needed for parallel map is an ability to provide a place-holder ahead of time so when you actually call this place-holder, you trigger calculation, do not wait for result and move further. Now I understand that you use inner Flow in your examples as a placeholder. In my example it is eager Deferred.

Maybe we can modify your structure by replacing inner Flow by lazy Deferred? it would remove a lot of confusion and would allow more fine grained control of results. Then parallelMap will transform Flow<T> into Flow<Deferred<R>> and we can add extensions for that type or something like that. I will try to think about it later.

As for initial example, I work a lot with parallel data processing and map/reduce workflows. And basically the only operation that could be parallelized is map (safe for group-based reduce which could be represented as parallel map + non-parallel reduce). So it covers most of the use-cases.

[elizarov]

@altavir Actually, it is the converse with the respect to the traditional distinction between concurrency and parallelism (it is quite a recent distinction, < 20 years old, but quite established by now).

Concurrent is not necessarily parallel. A mapParallel operator is concurrent, because different calls to transform happen concurrently with each other. Which means, for example, that if you update any kind of shared mutable state from inside of transform you'd run into a data race.

However, even if you take concurrent operator, it does not necessarily mean that it is going to actually run different calls to transform in parallel to each other. For example, if you run it on JS you'll get no parallelism whatsoever (since there is only a single thread), yet you can still have a lot of concurrency even on JS.

On JVM you can run your version mapParallel in a single-threaded dispatcher and while it provides concurrency between different calls to transform, there will no parallelism in a single-threaded dispatcher. You can get parallelism on JVM, though, by plugging a multi-threaded dispatcher into the context.

[altavir]

I won't argue with you about the terminology, especially remembering that significant fraction of my knowledge of it comes from your articles. Still, it does not change the fact that all we need for parallel processing is a way to start multiple computations ahead of time. I am trying to implement it via transforming a Flow into Flow of Deferred and batch starting those Deferred on collect. The code looks good from API point of view, but for some reason does not work right now. This function blocks on channel production stage and I am trying to figure out why.

[altavir]

The problem was in the conflict of lazy deferred with structured concurrency. I've fixed it by replacing deferred by my imitation. The working code is available here.

In theory, it should be possible to make concurrent collect without calling async in advance. One just have to create a number of placeholders (for example CompletableDeferred) and then fill and pop them to collector as soon as they come. I will think about it a little more. Maybe channel + switch or one channel for placeholders and the second for results.

[elizarov]

Let's hold off a discussion on implementation strategy a bit. There are lots of ways to implement it. What are you trying to achieve in the end? You've started this thread with a proposal for mapParallel, but what is your end goal? What are you trying to build?

[altavir]

The particular case is a streaming mathematical processor with ability to parallel computations. In the version before the Flow I had Producers - generators, which could lazily create elements of some type (suspend receive) and Consumers working similar to actors (with suspend send). And I had Processors which implemented both interfaces. Also I had a mechanism to lock a Consumer on Producer forcing this specific Producer to send all its elements to Consumer (and no one else).

The typical application of such mechanism is the analysis of time series, when you have a very long block of numbers and you want to continuously run it through some chain of mathematical operations (like moving window, Fourier transform etc) and drop result somewhere without loading the whole block in the memory. Some of those operations could be parallelized. For example, one could split the incoming data into fixed chunks and then apply FFT to each chunk in parallel. Basically it is the same old map-reduce technology. And very similar to Java Stream processing (I would use Java Streams, but I want to implement it on MPP). Flow seems to be good replacement for my initial construct since next Flow always consumes previous one and the Flow does not start generating until final consumer is called. The internal sequential processing also probably allows to avoid context switching performance losses.

[qwwdfsad]

Actually, concurrentMap can be expressed via flatMapMerge without using any private API:

private fun <T, R> Flow<T>.concurrentMap(dispatcher: CoroutineDispatcher, concurrencyLevel: Int, bufferSize: Int, transform: suspend (T) -> R): Flow<R> {
    return flatMapMerge(concurrencyLevel, bufferSize) { value ->
        flow { emit(transform(value)) }
    }.flowOn(dispatcher)
}

[salamanders]

I think I'm debating the same need over on https://discuss.kotlinlang.org/t/must-go-fast-sequences-flows-bounded-channels/12494 - but with less knowledge of the various options than this thread, so I'm going to sit quietly and watch here instead. :)

Why I think my need is the same use case:

1. Have a source. (In my case a video file that produces BufferedImages). The source is fast enough to not be the bottleneck if it is given a dedicated thread.
2. Need to do a lot of math, and the math can be done in parallel. (distilling chunks of images down to a single frame)
3. But too many in parallel "in flight" is bad (doesn't make sense to be grinding on more than I have cores, and I could run out of memory)
4. The destination (encoding to a MP4) could slow things down, so again, don't have too many in-flight. (which brings up back-pressure)
5. Even if we do some work in parallel, the order matters.
6. Desire to have more steps as it gets more interesting.

And I have too many options and not enough knowledge if there is one that is "use this until we make the Kotlin canonical lib"

• Java Streams
• Kotlin Sequences
• Kotlin Flows
• Java Capacity-Bounded Channels (that block)
• Kotlin Capacity-Bounded Channels (that suspend)
• (any of the above) containing Deferred

[altavir]

@qwwdfsad Seems to be working. I looked into the source and it does the trick with two channels I was intending to do. I've changed signature a bit: fun <T, R> Flow<T>.map(concurrencyLevel: Int, dispatcher: CoroutineDispatcher = Dispatchers.Default, bufferSize: Int = concurrencyLevel, transform: suspend (T) -> R . Now it could be called instead of regular map replacing map by map(4).

I think it mostly solves my case. Few additional comments:

• I think that async/collect syntax in case we need multiple subsequent heavy operations.
• The bufferSize purpose is not clear from documentation. Back-pressure should start to work when the process is slow and limit the speed of incoming messages. In this case the speed is limited by suspension when incoming channel is full. This parameter does not seem to do anything.
• I have not tested everything yet, but it would be nice to have subsequent operations to be run on the same thread to minimize context switching. My solution does not seem to work this way.

[sdeleuze]

On Spring side, the use case is pretty common: you want to chain 2 remote webservices, the first will return a list of IDs and the second will allow to get the details associated with each IDs. For that you are typically using flatMap in Rx world and you will naturally convert it to map with Flow following the documentation which is IMO misleading:

Note that even though this operator looks very familiar, we discourage its usage in a regular application-specific flows. Most likely, suspending operation in map operator will be sufficient and linear transformations are much easier to reason about.

So developers will naturally use a sequential map thinking it handles naturally parallel processing (concurrency behavior is not specified in map operator documentation) since it supports suspending functions, but it will process each element sequentially which makes little sense to me.

I agree with @gildor when he says map is sequential in Rx (or Sequence) because it doesn't support async operation there by definition. Since map in Flow support suspending functions, it is critical to be explicit about concurrency in its documentation, and to provide parallel map capabilities that does not force developers to use single Flow element + flatMap*.

I would personally be in favor to support parallel processing by default in map because the suspending function parameter is a clear signal that may happen to me, and I see much more use cases for this behavior than sequential processing with suspending functions where use cases are unclear to me. @elizarov said he is not, point taken. But please at least update the documentation of map, flatMap* and provide a parallel map operator.

[elizarov]

I'm all for adding concurrent APIs, too, but in an explicit way. There's a lot to design, to think about, though. I've already presented one way to provide explicit concurrent API in this thread. The other would be something like this:

flow.concurrent().map { ... }.merge() 

This is both explicit about concurrency and about what happens with all those concurrently mapped things -- they are all merged at the end (the other possible termination is to reduce all those concurrent flows, which nicely gives you the classical concurrent map/reduce paradigm in the same API).

Actually, I have one really radical idea and this thread is good place to share it. I'm thinking that we should deprecate flatMapMerge because it is implicitly concurrent. The whole concurrency parameter to flatMapMerge is a design smell. The replacement for flatMapMerge should be:

flow.concurrent().flatMap { ... }.merge()
//             ^ optional concurrency limit goes here if needed

At the same time flatMapConcat should be deprecated, too. A simple flow.flatMap should perform a sequential flatMapConcat just like it happens on a plain sequential sequence.

[sdeleuze]

I like the simplification that leads to more consistency with only map and flatMap while providing both sequential and concurrent behavior, but I am puzzled by the requirement to add 2 additional steps for operations as simple as map or flatmap. What types would concurrent(), map { } and flatMap { } return?