+str Add flatmapConcat with parallelism.

[He-Pin]

References #31958
was: #31602

I ran the tests after changing the default flatMapConcat to FlattenConcat; the tests passed, too.

[info] FlatMapConcatBenchmark.completedFuture         thrpt    5    832464.979 ± 1045741.316  ops/s
[info] FlatMapConcatBenchmark.completedFutureP1       thrpt    5  10560065.545 ±  406623.774  ops/s
[info] FlatMapConcatBenchmark.internalSingleSource    thrpt    5  17202143.700 ±  735428.818  ops/s
[info] FlatMapConcatBenchmark.internalSingleSourceP1  thrpt    5  16402884.417 ±  481353.061  ops/s
[info] FlatMapConcatBenchmark.mapBaseline             thrpt    5  37437826.208 ±  590107.276  ops/s
[info] FlatMapConcatBenchmark.normalFuture            thrpt    5    315802.300 ±   23575.498  ops/s
[info] FlatMapConcatBenchmark.normalFutureP1          thrpt    5    339282.443 ±    6550.085  ops/s
[info] FlatMapConcatBenchmark.oneElementList          thrpt    5   1019333.244 ±   50691.164  ops/s
[info] FlatMapConcatBenchmark.oneElementListP1        thrpt    5  10686004.485 ±  430848.272  ops/s
[info] FlatMapConcatBenchmark.sourceDotSingle         thrpt    5   7971587.554 ± 1193544.525  ops/s
[info] FlatMapConcatBenchmark.sourceDotSingleP1       thrpt    5  11181956.601 ±  124978.076  ops/s


jmh:run -i 5 -wi 5 -f1 -t1 akka.stream.FlatMapConcatBenchmark 

[He-Pin]

before:

[info] Benchmark                                                         Mode  Cnt         Score        Error   Units
[info] FlatMapConcatBenchmark.internalSingleSource                      thrpt   10   9111519.387 锟斤拷 183996.447   ops/s
[info] FlatMapConcatBenchmark.mapBaseline                               thrpt   10  20336793.367 锟斤拷 887065.039   ops/s
[info] FlatMapConcatBenchmark.oneElementList                            thrpt   10    371981.248 锟斤拷   3765.397   ops/s
[info] FlatMapConcatBenchmark.sourceDotSingle                           thrpt   10   4578547.437 锟斤拷 479604.201   ops/s

after:

info] Benchmark                                                         Mode  Cnt          Score            Error   Units
[info] FlatMapConcatBenchmark.internalSingleSource                      thrpt   10    9027659.060 锟斤拷     136614.084   ops/s
[info] FlatMapConcatBenchmark.mapBaseline                               thrpt   10   20947850.178 锟斤拷    1039800.021   ops/s
[info] FlatMapConcatBenchmark.oneElementList                            thrpt   10    4451591.431 锟斤拷      55815.355   ops/s
[info] FlatMapConcatBenchmark.sourceDotSingle                           thrpt   10    4547444.612 锟斤拷     136653.128   ops/s

[GreyPlane]

I wonder if it is possible to write a Buffer.peek(n: Int) which peek from current to n, and by that, is it easy to impl Merge via this, just change inflightSources.peek() eq self to check through the range from 0 to breadth - 1.

[He-Pin]

@GreyPlane How about comment in line, thanks

[gael-ft]

Thank you for the implementation.
This will allow us to have more than 1 source pulled at a time keeping elements ordered from downstream perspective.

[He-Pin]

Seems should add another parameter eager too. When true will complete / fail the stream when upstream fail.

[gael-ft]

Seems should add another parameter eager too. When true will complete / fail the stream when upstream fail.

Not obvious to me.
Goal flatMapConcat is to keep elements ordered.
If order is not a concern, you could just use merge (which has the concept of eager).

So I can't see a scenario where you would like to continue after one of the upstream sources cancelled when using this operator.

That being said, if you feel that could be useful, let's go 😉

[He-Pin]

@gael-ft that's true. but I saw your PR and you can return Source.failed(...) sometime. and for the eager I mean eagerError.
If when materilization, the upstream is Source.failed would I failed the stream ASAP when the parallesm > 1?

And for most case, I think we should make the eagerError = false, even there is an inflight Source.failed(...).

I was wantted to optimize for Source.future and Source.failed, but in which way the Source can be already failed. and there are Flux.concatMapDelayError/flatMapSequentialDelayError, I think delaying error should be the default behavior as the operator only parallelism materialization but not emitting.

[gael-ft]

@He-Pin I might be a bit too much use case centric (i.e. S3 object upload where we don't want any missing part).

If when materilization, the upstream is Source.failed would I failed the stream ASAP when the parallesm > 1?

So this question can probably be a valid question in some cases.

But regarding the default value of eager I think it should be intentional to continue processing after any error occured.
As with Flux, using flatMapSequentialDelayError is intentional vs flatMapSequential.

In my understanding of flatMapConcat it guarantees that the element you process will be in order.
If for some reason you can afford skipping element N+1 and process directly N+2 that should be intentional.

Once again, just my opinion, but as long as this is parameter, users can use it as they want.

[He-Pin]

@gael-ft Yes, let's just keep it simple for now. And I saw the FailedSource which will failStage when preStartI().
And the FutureSource only trigger the failStage after be pulled. And in your code you are using the FailedSource which will cause the stream fails very soon, So I think I need adjust my code for that too.

[He-Pin]

@gael-ft @johanandren @patriknw I think it's ready now, and happy new year.

[He-Pin]

check the ordering

[johanandren]

Sorry for not reviewing this yet, busy day, we'll get there eventually so please be patient.

[He-Pin]

@johanandren take easy, $Work first :)