Reactor-core: Map and flatMap operator chaining is not stack-safe

[pakoito]

Expected behavior

Chaining large numbers of map and flatMap operators is stack-safe

Actual behavior

StackOverflow on MonoMapFuseable.subscribe

Steps to reproduce

Mono<Integer> initial = Mono.just(0);

for (int i = 0; i < 5000; i++) {
    int currentI = i;
    initial = initial.map(previous -> currentI);
}

assert initial.block() == 5000;

Yields a StackOverflow error:

java.lang.StackOverflowError
    at reactor.core.publisher.MonoMapFuseable.subscribe(MonoMapFuseable.java)
    at reactor.core.publisher.MonoMapFuseable.subscribe(MonoMapFuseable.java:59)
    at reactor.core.publisher.MonoMapFuseable.subscribe(MonoMapFuseable.java:59)
    at reactor.core.publisher.MonoMapFuseable.subscribe(MonoMapFuseable.java:59)
    at reactor.core.publisher.MonoMapFuseable.subscribe(MonoMapFuseable.java:59)
    at reactor.core.publisher.MonoMapFuseable.subscribe(MonoMapFuseable.java:59)
    at reactor.core.publisher.MonoMapFuseable.subscribe(MonoMapFuseable.java:59)
        ...

Reactor Core version

3.1.8.RELEASE

JVM version (e.g. java -version)

1.7 with Kotlin in the project, 1.8 in the system

[smaldini]

It could be a trade-off that could affect day-to-day Flux/Mono use (I think its around 10-30 operators for a given use case/request): we could always check via instanceof the current operator to see if its the same type and conflate the function via "Function.andThen".
That would be assembly-time fusion and it would create a single map/flatMap as we do when we chain multiple concatWith, mergeWith and zipWith.

Note that if we generalize this pattern its something to keep in mind as well for a future generation of reactor architecture.

[pakoito]

Another potential optimization is to immediately apply these operators if the origin is just, just updating the value internally. Same for error, up until the first error handling operator.

[smaldini]

It might be too optimized in that case because there is still a user expectation of deferring the call until subscribe. In fact its only the error/empty case we could really optimize because they don't need to invoke at all the function. Also was thinking about flatMap, that might be a challenge to optimize because its not as simple as Function.andThen there is an asynchronous boundary (the returned Publisher that is merged back).

[pakoito]

Ooooh impurities :D Fair, fair.

[simonbasle]

1761 would fix this, I believe

[nomisRev]

👏 👏 👏 Excited to see this change! @simonbasle thanks for the update.

Arrow has a good set of tests to verify the stack-safety if you're interested to collaborate.

We have a bunch of other complicated tests if you're interested to include them in your need-investigation setup. I am working on making the test suite independent from Arrow since I am looking for some help/feedback with complex custom operators using Reactor/RxJava. So Reactor even add them in some community/integration suite if you're interested in that.

[simonbasle]

the PR above has now been merged, but it doesn't entirely fix this issue. it eliminates subscribe, but when chaining a huge amount of operators there is still request calls and onNext calls and onComplete calls... ie. the original snippet probably still overflows due to requests. It will be way harder to eliminate these.

[pakoito]

IIRC, and @nomisRev can help me here, what we've done in Arrow is trampoline every N operators to reset the stack.

[simonbasle]

@pakoito @nomisRev any help would be appreciated but I think implementing a trampoline function would involve a major architectural change to Reactor, which I don't think we're ready for. I you have any practical idea, proof of concept would be interesting 👍

[nomisRev]

Hey @simonbasle,

I created a small example that shows how the original code above can be made stack safe by simply interleaving a trampoline function on the desired stackDepth.

All the interesting code is found above the main function. The code to do the actual trampolining I just copied from Arrow and can be found here.

https://gist.github.com/nomisRev/7e857956cc79e00810a4bf3da55fea1b

I hope that helps! If there is anything else I can help I'd be glad to do so.

[simonbasle]

@pakoito @nomisRev can you chime in #2147 ? I'm afraid the implicit thread-hopping for stack reset might indeed be an issue in the long term, though...

[simonbasle]

after spiking on it for a few days, I'm starting to have the uneasy feeling that this will introduce way more unpredictable issues down the road than it fixes...

[pakoito]

It's okay, it was just a suggestion about what we found when trying to implement general recursion when building wrappers in https://github.com/arrow-kt/arrow. I believe we found an alternate solution or made so the tests were not as deep with the recursion, and it's been going okay.

EDIT: actually we made it blocking huh. Not awesome, just serviceable.

https://github.com/arrow-kt/arrow-fx/blob/deca849a1f5b551e9a4b92b67a4f577887d54fca/arrow-fx-reactor/src/main/kotlin/arrow/fx/reactor/FluxK.kt#L288-L295

https://github.com/arrow-kt/arrow-fx/blob/deca849a1f5b551e9a4b92b67a4f577887d54fca/arrow-fx-reactor/src/main/kotlin/arrow/fx/reactor/MonoK.kt#L292-L298

[simonbasle]

I'm going to close this one unfortunately, as the potential for side effects, subtle bugs down the road and general maintainability hit outweights the benefits in my opinion. Keeping this in a corner of our head in case there's a heavy rearchitecturing of Reactor in the future, though.

[pakoito]

Thank you for spending time on it @simonbasle !