Window operators now support backpressure in the inner observable.

[akarnokd]

Repost of #3050.

[abersnaze]

Do we need to maintain both BufferUntilSubscriber V1 and V2 at the same time?

[akarnokd]

BufferUntilSubscriber does some black backpressure magic inside GroupBy which I wasn't able to figure out and the V2 doesn't pass unit tests with it.

[stealthcode]

Could you explain why this is necessary? This could only happen if you are the emitting thread and no other thread has tried to emit while you are emitting and the only changes in this if block seem unrelated.

[akarnokd]

If no other thread tried to do something, we quit the emission loop. This means other actions can now enter the emission loop. This is part of the standard emission-loop approach with an optimization for unbounded consumers.

[stealthcode]

Yes I see that. This condition is tracked above on line 230 using the emitting field. I am unclear on why we have a missed field. It seems like missed could be a local variable or deleted entirely.

[akarnokd]

The method may be entered from multiple threads and only one can enter the emission loop. The others need to signal their attempt which can't happen if missed is a local variable.

[stealthcode]

What you are saying makes perfect sense and I can see that from the code. I am asking why does this if statement guard lines 280-284? The fact that another thread did not try to emit while your thread was emitting doesn't seem relevant for these lines. It seems like this block should always run.

[akarnokd]

That if block ends in return which shouldn't run if there was missed activity in which case the the loop has to try and emit more values if possible, again. That missed may come from a new value becoming available or from more requests.

[stealthcode]

I see. You are saying "if there have not been additional requests while emitting then we should return". At first this seemed like a race condition however I see that the onNext would just hop threads to the other thread in that case. Do you think it would give us better performance and possible read easier by taking out the synchronized block and the missed field and replacing it with a check after each for loop iteration if (get() > 0)?

[akarnokd]

No. This form of caughtUp optimiziation requires an atomic block which wouldn't work with the queue-drain approach. Since this represents the requested amount, having a non-zero requested doesn't mean there is values available in the queue, therefore it would just loop indefinitely. The performance is relative to the use pattern. Mostly single-threaded use benefits from synchronized because the JIT can eliminate it.

[stealthcode]

Okay. Thanks for the explanation.

[stealthcode]

It seems like BufferUntilSubscriber is being used as a special case buffer for OperatorGroupBy while BufferUntilSubscriberV2 is custom for the windowing operators. If they cannot be reconciled to work for both cases then could you please move them into the operators as static nested classes?

Also since groupBy's back pressure is being patched maybe it'd be a good idea to revisit this pull request with those changes to try to combine the two BufferUntilSubscribers.

[akarnokd]

These are internal classes and can live in their own files. The groupBy in #3428 no longer uses it and once merged, this BufferUntilSubscriberV2 can be renamed; or better yet, promoted to a standard API UnicastSubject.

I don't think we should be so eager to combine things; such combinations end up in monstrous classes which are then sources of misunderstanding and complexity.

[stealthcode]

BufferUntilSubscriber was a proposed as a solution to the time gap problem in group by so because BufferUntilSubscriberV2 doesn't work to solve this problem for group by leads me to question it's naming. Is it built to handle the same problem or a different problem? Are you proposing to rename v2 to UnicastSubject?

[stealthcode]

I don't think we should be so eager to combine things; such combinations end up in monstrous classes which are then sources of misunderstanding and complexity.

I'm proposing to hide away internal functionality from the public API.

[akarnokd]

Neither of these are public API since they reside in the internal package. Hiding them further away seems to be unnecessary. In addition, BufferUntilSubscriberV2 is effectively turned into the official UnicastSubject in 2.x.

Historically, BufferUntilSubscriber started out to solve the time-gap problem for the non-backpressured groupBy and window. When the backpressure was introduced, groupBy had to support backpressure but BufferUntilSubscriber wasn't enhanced. Instead, BufferUntilSubscriber is used as a middle man and the subscription process is hijacked to inject a Producer that attempts to coordinate requests. Apparently, this didn't cover all request pattern hence the original bug.

Due to this very specific hijacking, adding backpressure to BufferUntilSubscriber directly didn't work out as it conflicted with the request coordination in a way I couldn't resolve. Therefore, I decided to have a separate class that does backpressure but doesn't have to be involved in request coordination for window (which I believe can't be established with a reasonable strategy).

[stealthcode]

All I'm asking for is a rename from v2 in 1.x

[akarnokd]

Maybe I missed it but what's the name you'd like?

[akarnokd]

I've renamed the class to UnicastSubject and combined the State with the OnSubscribeBUS class.

[akarnokd]

I'm going to sleep now (it's midnight here) so if you only have concerns about naming, location and visibility, I suggest merging this PR then posting a separate PR with your changes (and merge it if you can get somebody else to like it in the meantime).

[stealthcode]

@akarnokd would you mind rebasing this?

[akarnokd]

Done.

[stealthcode]

👍