FluxPublishOn to detach itself after further requests

[MikkelHJuul]

The FluxPublishOnSubscriber#runAsync will never escape its infinite loop if the call to onNext take longer time than the period between events from upstream.

This makes it so that the thread will never stop looping. And it has the unfortunate effect that; for a Scheduler of pool size N. If M (greater than N) upstream fluxes #publishOn that scheduler, only N of those fluxes will be subscribed and worked on.

While upstream supply is available I would prefer even distribution across all fluxes.

I suggest adding a flag to the publishOn that would allow control that the loop escape after requesting, or performing the request in the scheduler itself. This would free up the thread to pick up other tasks from the taskqueue.

[chemicL]

Hey, @MikkelHJuul! Thanks for the report. I can see how this can become an issue, however the proposed solution can have a negative impact on performance (request upon rescheduling) or could lead to regressions (escape the loop after requesting). The publishOn operator is a very frequently used one and it has been stable for a long time. Introducing a change like this might not be ideal. The work-stealing mechanisms in reactor are far from being fair, that's a valid point. However, if you run into issues of stream starvation, perhaps it's worth considering a dedicated Scheduler instead of changing the internals?

Let me just enlist the options (perhaps there are other?):

• issue request to upstream by rescheduling the request Runnable onto the Scheduler
  • pro: fair ordering of task execution by the Scheduler - no starvation
  • con: significant performance drop by having to go through the scheduling
• break from the loop after (or rather prior to?) issuing the upstream request
  • pro: no starvation, fair ordering
  • con: in reality similar to the above, since requesting outside of the loop would require releasing the exclusive access marker and allowing the upstream to immediately deliver a signal, which in turn would mean rescheduling -> which leads to performance degradation

With the above comments and my current lack of idea about a better solution, do you think the proposed workaround is good enough for such busy scenarios and we could somehow enrich our documentation to hint that very busy Fluxes should use a dedicated Scheduler if they're offloading work? Would you be willing to add a note in our reference documentation about this?