Netflix/RxJava — pull request #1403 by benjchristensen (2014-07-03 07:29:20+00:00)
Title: Backpressure-8 – Merge & ObserveOn

This is a functioning but non-performant implementation of backpressure. Of the major unbounded queue operators in 0.19, this implements `observeOn` and `merge` but not yet `zip`. 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 implements Observer, 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 -
```