Blocking buffer until experiment

[akarnokd]

This is a solution to the time gap problem for #844.

I've added an subscribeOn overload where the user can explicitly request a buffering behavior. In addition, SubscribeOn checks the type of the Observable and enters buffering mode for GroupedObservable and PublishSubject. I think these code options should be mutually exclusive:

1. either we only check for Observable type, but then new kinds of observables or hidden observables won't work,
2. or ask the programmer in the documentation/tutorial to explicitly request buffering in certain operator compositions.

I personally favor option 2).

A drawback is that this blocking subscribeOn deadlocks on pools with a single thread. We can, of course, check for Trampoline, Test and Immediate schedulers, but not schedulers created via Schedulers.executor, or the computation scheduler on a single-core machine.

[cloudbees-pull-request-builder]

RxJava-pull-requests #792 FAILURE
Looks like there's a problem with this pull request

[cloudbees-pull-request-builder]

RxJava-pull-requests #793 SUCCESS
This pull request looks good

[akarnokd]

Test testRepeatTakeWithSubscribeOn passed locally.

I guess there is a race issue with repeat() as it schedules a new repeat after take unsubscribes. I guess adding a child.isUnsubscribed test before L85 should do the trick.

[benjchristensen]

Reviewing code ... considering the drawbacks, what do you think is worse, possibility of (deterministic?) deadlock with this solution? or possibility of non-deterministic data-loss when using subscribeOn on hot Observables?

Is the deadlock deterministic (it would always happen in dev so it gets found) or could it happen if a Scheduler becomes saturated, or the buffer size is higher than available threads?

[akarnokd]

Non-deterministic data loss is definitely worse.

Deadlock due to the computation scheduler being single threaded is worrying, but might affect other concurrent operators as well regardless. I think the documentation could mention that if pushback or blocking behavior is expected, one should use NewThread or IO scheduler for the unblocking operation.

[benjchristensen]

Non-deterministic data loss is definitely worse.

I agree, so let's continue down this path :-) I'll review through your code in a bit.

[benjchristensen]

First pass through reading this code it seems good, and mature enough to handle the different scenarios we could throw against it. I'm going to spend some more time playing but nothing right now suggests that this should not be the path we take.

[benjchristensen]

This overload should help mitigate issues when subscribing to a PublishSubject (and derivatives such as GroupedObservable in operator groupBy) and events fired between the original and actual subscriptions are lost.

That doesn't seem to need this overload since we special-case those two Observable instances even if bufferSize is not passed in and we say false for dontLoseEvents.

[benjchristensen]

Ah, I see ... the automatic has an unbounded buffer so it never blocks. This overload allows for blocking as well.

Are you suggesting subscribeOn by itself never use the blocking form but any operator implementation that uses subscribeOn would choose to do so?

In that case, should subscribeOn do any buffering by default or only on demand?

[akarnokd]

My suggestion is that subscribeOn(Scheduler) never blocks with any source and events may be lost, and subscribeOn(Scheduler, int) does not lose events and may block depending on the buffer size.

[benjchristensen]

I added some things on top of this at #869.

[benjchristensen]

Work on this is picked up in #869