Netflix/RxJava — pull request #1382 by benjchristensen (2014-06-24 03:49:41+00:00)
Title: Backpressure-7 - 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`. Several attempts at performance improvements last week failed so I have rolled them back and left this with naive data structures while restarting the performance work.

I want to kick off the design and functionality review while continuing to work on performance.

### 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);
}
```

### `Subscriber.request` vs `Producer.request`

This is possibly confusing and I think we may want to eliminate `Subscriber.request`. It would make some operators like `take` more complicated, but for the rare time it is needed, I think it's more clear that you must "capture" the `Producer` and call `Producer.request`.

### 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) Subscriber.request

This seems confusing and should probably be removed, despite it making some use cases such as `take` and `filter` more complicated.

##### 4) 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?

##### 5) Producer.BUFFER_SIZE

Currently there is a public constant, `Producer.BUFFER_SIZE` that represents the system-wide size of buffers used for backpressure. This is exposed so that an operator implementation can optimize to skip the `setProducer` path when it knows it will emit less data that the BUFFER_SIZE.

This does have a performance benefit, but it comes at the cost of exposing implementation details. It also means all buffers MUST use the same buffer size, or at least not be any smaller.

Is this something we want? Is the performance benefit worth the trade-offs and complexity?

##### 6) Other operators like filter?

Are there other operators that filter out `onNext` like `filter` does that we must modify to call `request(1)` each time it filters something out? Is there a better way of handling this so operators do not need to manually handle this?

##### 7) 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)`?

##### 8) 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?

##### 9) observeOn

Is the design of `observeOn` as efficient and correct as it can be?

##### 10) merge

The `merge` operator currently has a naive algorithm for handling the queue of buffers. We need to design a better one.

There are some unit tests marked as ignored for `merge` in the `BackpressureTests` class. Should those work? What should be the behavior of `merge` when merging all synchronous, all asynchronous, or mixed sync/async Observables?

### Performance

Performance is not good right now. The biggest problem is `merge`, and that's critical since it's key to `flatMap` which is a core operator.

Performance can be tested via the JMH performance tests. This is where I'm currently focusing.