Backpressure-8 – Merge & ObserveOn

This is a functioning implementation of backpressure. Of the major unbounded queue operators in 0.19, this implements `observeOn` and `merge` but not yet `zip`. Compared with the previous PRs (backpressure-7 and earlier) this brings performance to levels that are possible to be merged into master.

### New Signatures

```java
public interface Producer {
    public void request(int n);
}
```

```java
public abstract class Subscriber<T> implements Observer<T>, Subscription {

	/**
	 * Optionally called from `Observable.OnSubscribe`.
	 */
	public final void setProducer(Producer producer);
	
	/**
	 * Allow implementing classes to capture, decorate or override the `Producer` without affecting the `Subscriber` logic.
	 */
	protected Producer onSetProducer(Producer producer);
	
	public final void request(int n);
}
```


### Operator Changes

`OnSubscribeFromIterable` and `OnSubscribeRange` are both updated to support `Producer` and will only emit as many `onNext` as requested. The implementations are a little more complex than needed as they optimize for cases where backpressure is not requested and fallback to the "firehose" model of 0.19 and earlier. This allows for higher throughput in case where backpressure is not needed.

`parallel` had to be rewritten with a different approach to support backpressure. The `filter`, `take`, and `skip` operators were modified to support backpressure and propagate request modifications based on each of their logic. `takeUntil` needed to be rewritten because `NotificationLite` can't be allowed to leak across operators. The `subscribeOn` operator was modified to correctly reschedule work on the given `Scheduler` when a `request(n)` occurs after the `Producer` has let itself shut down. 


### Design Questions

##### 1) Producer

Is the name and signature of `Producer.request(n)` correct?

##### 2) Subscriber.setProducer

Is this the correct signature for an `Observable.OnSubscribe` to register the `Producer`?

##### 3) Requesting <0 for Infinite (no backpressure)

To be backwards compatible, and support performance optimizations when backpressure is not needed, `Producer.request(n)` can be invoked with a negative number. This means "no backpressure" and the `Producer` can then emit data in a tight-loop without further limitations.

Is this acceptable and the best solution?

##### 4) onBackpressure*

Are the `onBackpressureBuffer` and `onBackpressureDrop` operators well-named, clear in their usage and implemented correctly? Should we implemented others such as `onBackpressureUnsubscribe` and `onBackpressureBuffer(int limitUntilDropping)`?

##### 5) Unit Tests and Use Cases

Are there major use cases that are missing from `BackpressureTests`? 

What problems are there that are not yet resolved or handled by the design?


### Performance

```
-------------- ../gradlew benchmarks '-Pjmh=-f 1 -wi 5 -i 5 -r 2 .*PerfBaseline.observable.*'

------- 0.19

Benchmark                                       (size)   Mode   Samples        Score  Score error    Units
r.PerfBaseline.iterableViaForLoopConsumption         1  thrpt         5 250139466.600  6920354.087    ops/s
r.PerfBaseline.iterableViaForLoopConsumption      1000  thrpt         5    182500.174     5244.888    ops/s
r.PerfBaseline.iterableViaForLoopConsumption   1000000  thrpt         5       170.911        3.823    ops/s
r.PerfBaseline.iterableViaHasNextConsumption         1  thrpt         5 320148600.957 11599243.208    ops/s
r.PerfBaseline.iterableViaHasNextConsumption      1000  thrpt         5    297314.203     9485.727    ops/s
r.PerfBaseline.iterableViaHasNextConsumption   1000000  thrpt         5       269.004        4.151    ops/s
-----------------------------------------------------------------------------------------------------------
r.PerfBaseline.observableConsumption                 1  thrpt         5  29298639.717   277675.587    ops/s
r.PerfBaseline.observableConsumption              1000  thrpt         5    243968.954    10311.720    ops/s
r.PerfBaseline.observableConsumption           1000000  thrpt         5       252.916        3.420    ops/s
r.PerfBaseline.observableConsumptionUnsafe           1  thrpt         5 199935597.149 31884367.299    ops/s
r.PerfBaseline.observableConsumptionUnsafe        1000  thrpt         5    311704.723    39346.976    ops/s
r.PerfBaseline.observableConsumptionUnsafe     1000000  thrpt         5       307.188       58.163    ops/s
r.PerfBaseline.observableViaRange                    1  thrpt         5  23843851.260   324794.935    ops/s
r.PerfBaseline.observableViaRange                 1000  thrpt         5    193208.479     3741.146    ops/s
r.PerfBaseline.observableViaRange              1000000  thrpt         5       188.261       12.172    ops/s
r.PerfBaseline.observableViaRangeUnsafe              1  thrpt         5  45762792.585  1379450.701    ops/s
r.PerfBaseline.observableViaRangeUnsafe           1000  thrpt         5    240280.347     3312.756    ops/s
r.PerfBaseline.observableViaRangeUnsafe        1000000  thrpt         5       234.143        1.163    ops/s



------- backpressure-8

Benchmark                                       (size)   Mode   Samples        Score  Score error    Units
r.PerfBaseline.iterableViaForLoopConsumption         1  thrpt         5 250601807.892 12009864.080    ops/s
r.PerfBaseline.iterableViaForLoopConsumption      1000  thrpt         5    184859.869     2400.585    ops/s
r.PerfBaseline.iterableViaForLoopConsumption   1000000  thrpt         5       172.127        4.979    ops/s
r.PerfBaseline.iterableViaHasNextConsumption         1  thrpt         5 322464817.267  8117858.197    ops/s
r.PerfBaseline.iterableViaHasNextConsumption      1000  thrpt         5    294868.674     7349.485    ops/s
r.PerfBaseline.iterableViaHasNextConsumption   1000000  thrpt         5       266.602        5.682    ops/s
-----------------------------------------------------------------------------------------------------------
r.PerfBaseline.observableConsumption                 1  thrpt         5  33622221.238   963042.530    ops/s +
r.PerfBaseline.observableConsumption              1000  thrpt         5    244097.704     4564.965    ops/s =
r.PerfBaseline.observableConsumption           1000000  thrpt         5       256.587        5.404    ops/s =
r.PerfBaseline.observableConsumptionUnsafe           1  thrpt         5 206453075.285  4940138.228    ops/s =
r.PerfBaseline.observableConsumptionUnsafe        1000  thrpt         5    320284.165     6034.968    ops/s =
r.PerfBaseline.observableConsumptionUnsafe     1000000  thrpt         5       314.259        7.683    ops/s =
r.PerfBaseline.observableViaRange                    1  thrpt         5  37486671.210   887127.291    ops/s ++
r.PerfBaseline.observableViaRange                 1000  thrpt         5    172643.588     7439.349    ops/s -
r.PerfBaseline.observableViaRange              1000000  thrpt         5       173.171        2.794    ops/s -
r.PerfBaseline.observableViaRangeUnsafe              1  thrpt         5 236013059.425  7822030.114    ops/s +++
r.PerfBaseline.observableViaRangeUnsafe           1000  thrpt         5    211461.722    13717.277    ops/s -
r.PerfBaseline.observableViaRangeUnsafe        1000000  thrpt         5       202.761       12.915    ops/s -



-------------- ../gradlew benchmarks '-Pjmh=-f 1 -wi 5 -i 5 -r 2 .*rx.operators.OperatorMapPerf.*'

------- 0.19

Benchmark                                 (size)   Mode   Samples        Score  Score error    Units
r.o.OperatorMapPerf.mapPassThru                1  thrpt         5 17853002.462   556655.574    ops/s 
r.o.OperatorMapPerf.mapPassThru             1000  thrpt         5   150669.958    16170.385    ops/s
r.o.OperatorMapPerf.mapPassThru          1000000  thrpt         5      162.473        3.961    ops/s
r.o.OperatorMapPerf.mapPassThruViaLift         1  thrpt         5 18947641.742   644348.550    ops/s
r.o.OperatorMapPerf.mapPassThruViaLift      1000  thrpt         5   160047.002    20086.976    ops/s
r.o.OperatorMapPerf.mapPassThruViaLift   1000000  thrpt         5      162.310        6.204    ops/s


------- backpressure-8

Benchmark                                 (size)   Mode   Samples        Score  Score error    Units
r.o.OperatorMapPerf.mapPassThru                1  thrpt         5 24092718.114   586172.216    ops/s ++
r.o.OperatorMapPerf.mapPassThru             1000  thrpt         5   149191.341     3154.653    ops/s -
r.o.OperatorMapPerf.mapPassThru          1000000  thrpt         5      150.013        3.124    ops/s -
r.o.OperatorMapPerf.mapPassThruViaLift         1  thrpt         5 26144219.306   368383.801    ops/s ++
r.o.OperatorMapPerf.mapPassThruViaLift      1000  thrpt         5   150500.298     2074.296    ops/s - 
r.o.OperatorMapPerf.mapPassThruViaLift   1000000  thrpt         5      150.896        1.945    ops/s -





-------------- ../gradlew benchmarks '-Pjmh=-f 1 -wi 5 -i 5 -r 2 .*rx.operators.OperatorObserveOnPerf.*'

------- 0.19

Benchmark                                         (size)   Mode   Samples        Score  Score error    Units
r.o.OperatorObserveOnPerf.observeOnComputation         1  thrpt         5   110481.792      991.073    ops/s
r.o.OperatorObserveOnPerf.observeOnComputation      1000  thrpt         5     8386.714      103.294    ops/s
r.o.OperatorObserveOnPerf.observeOnComputation   1000000  thrpt         5        8.568        4.330    ops/s
r.o.OperatorObserveOnPerf.observeOnImmediate           1  thrpt         5 13144808.808   852878.900    ops/s
r.o.OperatorObserveOnPerf.observeOnImmediate        1000  thrpt         5   191230.661     4066.226    ops/s
r.o.OperatorObserveOnPerf.observeOnImmediate     1000000  thrpt         5      186.919        6.075    ops/s
r.o.OperatorObserveOnPerf.observeOnNewThread           1  thrpt         5    15972.029     1955.679    ops/s
r.o.OperatorObserveOnPerf.observeOnNewThread        1000  thrpt         5     6606.316      104.860    ops/s
r.o.OperatorObserveOnPerf.observeOnNewThread     1000000  thrpt         5       10.683        2.060    ops/s


------- backpressure-8

Benchmark                                         (size)   Mode   Samples        Score  Score error    Units
r.o.OperatorObserveOnPerf.observeOnComputation         1  thrpt         5    93371.544    12564.954    ops/s
r.o.OperatorObserveOnPerf.observeOnComputation      1000  thrpt         5     5209.287      161.070    ops/s
r.o.OperatorObserveOnPerf.observeOnComputation   1000000  thrpt         5       10.070        5.387    ops/s
r.o.OperatorObserveOnPerf.observeOnImmediate           1  thrpt         5 15855354.194    74122.090    ops/s
r.o.OperatorObserveOnPerf.observeOnImmediate        1000  thrpt         5   173351.676     5924.790    ops/s
r.o.OperatorObserveOnPerf.observeOnImmediate     1000000  thrpt         5      170.760        4.379    ops/s
r.o.OperatorObserveOnPerf.observeOnNewThread           1  thrpt         5    15524.785     1373.757    ops/s
r.o.OperatorObserveOnPerf.observeOnNewThread        1000  thrpt         5     4878.633       54.696    ops/s
r.o.OperatorObserveOnPerf.observeOnNewThread     1000000  thrpt         5       15.285        0.564    ops/s






-------------- ../gradlew benchmarks '-Pjmh=-f 1 -wi 5 -i 5 -r 2 .*rx.operators.OperatorMergePerf.*'

------- 0.19


Benchmark                                          (size)   Mode   Samples        Score  Score error    Units
r.o.OperatorMergePerf.merge1SyncStreamOfN               1  thrpt         5  3047431.417    33607.539    ops/s
r.o.OperatorMergePerf.merge1SyncStreamOfN            1000  thrpt         5    70708.027     2468.027    ops/s
r.o.OperatorMergePerf.merge1SyncStreamOfN         1000000  thrpt         5       70.950        3.242    ops/s
r.o.OperatorMergePerf.mergeNAsyncStreamsOfN             1  thrpt         5   106426.955     3921.743    ops/s
r.o.OperatorMergePerf.mergeNAsyncStreamsOfN          1000  thrpt         5        7.645        0.409    ops/s
r.o.OperatorMergePerf.mergeNSyncStreamsOf1              1  thrpt         5  2924643.167    90602.452    ops/s
r.o.OperatorMergePerf.mergeNSyncStreamsOf1            100  thrpt         5    65041.997     4387.032    ops/s
r.o.OperatorMergePerf.mergeNSyncStreamsOf1           1000  thrpt         5     6727.579      140.162    ops/s
r.o.OperatorMergePerf.mergeNSyncStreamsOfN              1  thrpt         5  2923542.700    55565.638    ops/s
r.o.OperatorMergePerf.mergeNSyncStreamsOfN           1000  thrpt         5       71.998        1.992    ops/s
r.o.OperatorMergePerf.mergeTwoAsyncStreamsOfN           1  thrpt         5    80321.863    19568.569    ops/s
r.o.OperatorMergePerf.mergeTwoAsyncStreamsOfN        1000  thrpt         5     3832.958      236.712    ops/s
r.o.OperatorMergePerf.oneStreamOfNthatMergesIn1         1  thrpt         5  2927868.656   517769.910    ops/s
r.o.OperatorMergePerf.oneStreamOfNthatMergesIn1      1000  thrpt         5     6799.933      180.123    ops/s
r.o.OperatorMergePerf.oneStreamOfNthatMergesIn1   1000000  thrpt         5        6.408        0.206    ops/s



------- backpressure-8

Benchmark                                          (size)   Mode   Samples        Score  Score error    Units
r.o.OperatorMergePerf.merge1SyncStreamOfN               1  thrpt         5  5475300.198   156741.334    ops/s
r.o.OperatorMergePerf.merge1SyncStreamOfN            1000  thrpt         5    68932.278     1311.023    ops/s
r.o.OperatorMergePerf.merge1SyncStreamOfN         1000000  thrpt         5       64.405        0.611    ops/s
r.o.OperatorMergePerf.mergeNAsyncStreamsOfN             1  thrpt         5   102736.219     2948.102    ops/s
r.o.OperatorMergePerf.mergeNAsyncStreamsOfN          1000  thrpt         5        5.893        0.278    ops/s xxxxx
r.o.OperatorMergePerf.mergeNSyncStreamsOf1              1  thrpt         5  4941243.788   109965.809    ops/s
r.o.OperatorMergePerf.mergeNSyncStreamsOf1            100  thrpt         5   368061.743    11456.008    ops/s
r.o.OperatorMergePerf.mergeNSyncStreamsOf1           1000  thrpt         5    38941.345     1974.002    ops/s
r.o.OperatorMergePerf.mergeNSyncStreamsOfN              1  thrpt         5  6070127.935    58505.579    ops/s
r.o.OperatorMergePerf.mergeNSyncStreamsOfN           1000  thrpt         5       57.002        1.708    ops/s
r.o.OperatorMergePerf.mergeTwoAsyncStreamsOfN           1  thrpt         5    77911.340     2413.596    ops/s
r.o.OperatorMergePerf.mergeTwoAsyncStreamsOfN        1000  thrpt         5     4258.650      116.595    ops/s
r.o.OperatorMergePerf.oneStreamOfNthatMergesIn1         1  thrpt         5  5429881.489    26728.872    ops/s
r.o.OperatorMergePerf.oneStreamOfNthatMergesIn1      1000  thrpt         5    28932.279      572.963    ops/s
r.o.OperatorMergePerf.oneStreamOfNthatMergesIn1   1000000  thrpt         5       30.147        1.253    ops/s




-------------- ../gradlew benchmarks '-Pjmh=-f 1 -wi 5 -i 5 -r 2 .*rx.operators.OperatorFlatMapPerf.*'

------- 0.19

Benchmark                                          (size)   Mode   Samples        Score  Score error    Units
r.o.OperatorFlatMapPerf.flatMapIntPassthruAsync         1  thrpt         5   352212.163    24668.380    ops/s
r.o.OperatorFlatMapPerf.flatMapIntPassthruAsync      1000  thrpt         5      350.930        8.635    ops/s
r.o.OperatorFlatMapPerf.flatMapIntPassthruAsync   1000000  thrpt         5        0.352        0.010    ops/s
r.o.OperatorFlatMapPerf.flatMapIntPassthruSync          1  thrpt         5  3323839.174    38716.562    ops/s
r.o.OperatorFlatMapPerf.flatMapIntPassthruSync       1000  thrpt         5     6456.484     1262.577    ops/s
r.o.OperatorFlatMapPerf.flatMapIntPassthruSync    1000000  thrpt         5        6.337        0.149    ops/s
r.o.OperatorFlatMapPerf.flatMapTwoNestedSync            1  thrpt         5  2117425.124    45292.911    ops/s
r.o.OperatorFlatMapPerf.flatMapTwoNestedSync         1000  thrpt         5    35047.747      637.443    ops/s
r.o.OperatorFlatMapPerf.flatMapTwoNestedSync      1000000  thrpt         5       35.408        1.304    ops/s



------- backpressure-8

Benchmark                                          (size)   Mode   Samples        Score  Score error    Units
r.o.OperatorFlatMapPerf.flatMapIntPassthruAsync         1  thrpt         5   307489.156    24595.370    ops/s -
r.o.OperatorFlatMapPerf.flatMapIntPassthruAsync      1000  thrpt         5      315.122       79.497    ops/s -
r.o.OperatorFlatMapPerf.flatMapIntPassthruAsync   1000000  thrpt         5      305.585       33.345    ops/s +++ (why is this better?)
r.o.OperatorFlatMapPerf.flatMapIntPassthruSync          1  thrpt         5  5932020.586   123609.972    ops/s ++
r.o.OperatorFlatMapPerf.flatMapIntPassthruSync       1000  thrpt         5    26402.347     4096.510    ops/s +++
r.o.OperatorFlatMapPerf.flatMapIntPassthruSync    1000000  thrpt         5       29.338        5.235    ops/s +++
r.o.OperatorFlatMapPerf.flatMapTwoNestedSync            1  thrpt         5  3843959.776   204899.691    ops/s ++
r.o.OperatorFlatMapPerf.flatMapTwoNestedSync         1000  thrpt         5    29316.657      446.403    ops/s ++
r.o.OperatorFlatMapPerf.flatMapTwoNestedSync      1000000  thrpt         5       32.767        0.754    ops/s -
```

We could put in a safeguard, it should only happen once.