backpressure story

[benlesh]

This issue on StackOverflow illustrates a problem with backpressure in current RxJS, I think

While I think this guy could solve his problem with an Iterator of Promise, I think it would be better if RxJS could support this with fewer object allocations.

cc/ @benjchristensen

[headinthebox]

My opinion about back pressure is that you either have a truly reactive system where the producer is fully in charge and the onus is on the consumer to keep up; or a truly interactive the consumer is fully in charge and the onus is on the producer to keep up. This is one of the cases where having different types of collections makes sense (just like we have different collections like lists, maps, sets, ... with different performance characteristics and different operations). @benjchristensen has a different opinion on that.

[benjchristensen]

Any system that has hidden unbounded buffers is a recipe for disaster. RxJava was rightfully dismissed outright by many because it would consume an entire stream into unbounded buffers in zip, merge and observeOn. Adding backpressure to RxJava was painful (to figure out and retrofit) but has been worth it.

Just because some use cases should be "completely reactive" (such as mouse events), it does not mean that signals are not needed for backpressure. Even with mouse events, if the consumer is slow, an unbounded buffer in observeOn, zip or merge should not be accumulating and the consumer silently falling behind and the application losing memory. It should be an error and blow up if the developer has not put in necessary flow control.

In other words, an application should not silently drop or do unbounded buffering data. It should always be explicit to do either of those things.

So even for the mouse event use case, the backpressure mechanism becomes a signal for flow control to kick in (conditionally drop, buffer, debounce, throttle, etc only when backpressure happens) or errors out and tells the developer to add flow control either conditionally (i.e. onBackpressure*) or permanently (temporal operators like debounce, sample, throttle, etc).

If the "reactive collection" doesn't have unbounded buffers, them I'm fine with it existing separately, but that means it can't have any async operators like merge, observeOn, or zip – and we know that those are necessary for application use cases. Since we need async operators, and I think we should never have unbounded buffering, that means the "reactive collection" should support backpressure signals.

The signals can be ignored. And infinite can be requested if backpressure is not wanted. But the API and semantics should support it so it composes through and the library can exist without silent unbounded buffering, memory leaks, or dropping of data.

[benjchristensen]

I suggest that the collaboration for Reactive Streams hardened the API, interaction model, and contracts quite well.

It would look something like this in Javascript:

interface Observable {
    subscribe(o: Observer<T>) : void; // or could return Subscription
    lift(operator): Observable<R>;
}

interface Subscription {
    unsubscribe(): void;
    isUnsubscribed(): boolean;
    request(n: number): void // optional if backpressure is added
}

interface Observer<T> {
   onSubscribe(s: Subscription): void;
   onNext(t: T): void;
   onError(e: any): void;
   onComplete(): void;
}

The consumer then can either invoke subscription.request(Number.MAX_VALUE) if no backpressure is needed (synchronous firehose) or request up the number of elements that can be enqueued: subscription.request(n). It keeps invoking request for more as it drains its internal queue and permits the producing Observable to keep emitting more.

If the Observable is hot, or can't otherwise be "paused", then it either blows up (i.e. BackpressuneException) or another operator composes in a flow control strategy (drop, buffer, throttle, sample, debounce, etc).

[headinthebox]

zip in particular was naively implemented, and should be a synchronization primitive like CyclicBarrier.

People have been using observeOn for ages in the UI space with no problems (https://msdn.microsoft.com/en-us/library/system.threading.synchronizationcontext.post(v=vs.110).aspx) or https://docs.oracle.com/javase/8/javafx/api/javafx/application/Platform.html, but if you do not want buffering you can use https://msdn.microsoft.com/en-us/library/system.threading.synchronizationcontext.send(v=vs.110).aspx.

If you are doing a merge of and infinite number of infinite streams that is as bas as doing unbound recursion, or allocating an unbounded number of objects. You make sure that this does not happen.

Also note that anytime you use say a http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/Semaphore.html or a http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/Executor.html you are using a hidden unbounded buffer.

Saying "unbounded buffer are verboten" is as fundamentalist as Haskell people saying all side-effects are bad, or the Akka folks saying everything is an actor with strict asynchronous boundaries, or Java insisting on checked exceptions. Programming is tastefully composing side effect, to write real stuff you have to break the rules, so you never say never, or always say always.

[benjchristensen]

Also note that anytime you use say a Semaphore or a Executor you are using a hidden unbounded buffer.

I know. That's why RxJava never holds a lock when emitting, and never uses synchronized or Semaphore to serialize multiple streams, but uses explicit, bounded, queues. This is why merge is an async operator that has buffers. It never blocks a thread, whereas Semaphores do – which is queueing.

Same issue with Executor as you mention. Again why we never expose Scheduler directly as a raw API in Observable sequences. The scheduling into the underlying unbounded queues of Executor are always governed by the backpressure of the Observable sequence so they are bounded.

if you do not want buffering you can use https://msdn.microsoft.com/en-us/library/system.threading.synchronizationcontext.send(v=vs.110).aspx.

Using synchronization, and call-stack blocking to avoid observeOn (or other buffers) is not an option when event-loops are involved, so they are not a solution. Event loops are almost always involved, particularly in Javascript. Thus, synchronous approaches to backpressure are not an option.

fundamentalist

This is not a fundamentalist approach. In fact, I'd argue that you are taking the fundamentalist approach claiming "reactive" requires unbounded buffers. I'm arguing against the "purity" of pure-push, and allowing the type to support moving between push and pull depending on the speed of the producer/consumer relationship. It is embracing the messiness of our applications and abstractions.

I do not say that "unbounded buffers are verboten", I say that hidden "unbounded buffers are verboten". If someone wants to say observable.onBackpressureBuffer().subscribe(o) and explicitly put an unbounded buffer in the mix, then so be it.

People have been using observeOn for ages in the UI space with no problems

It is not correct to say "no problems". People work around them because they haven't been given tools to do otherwise. Naively consume a Websocket stream for example using RxJS in a browser. I myself have had these problems. Ben Lesh linked to someone else having similar issues in another environment RxJS will be used: Node.

We can state that an Observable is ONLY reactive, without ability to support backpressure and flip into pull mode. But if that's the case, then the type is itself being fundamentalist and pushing people to find other solutions, which they will have to.

[staltz]

@headinthebox

the producer is fully in charge and the onus is on the consumer to keep up

@benjchristensen

In other words, an application should not silently drop or do unbounded buffering data. It should always be explicit to do either of those things.

I don't think you guys fundamentally disagree, but as you can see from the quotes, we just need to find a way how to determine consumer behavior when producer is too fast. Implicit ("hidden") has gotchas for developers. Explicit definition of lossy or lossless consumption should be how the consumer handles its onus.

[benlesh]

@benjchristensen could you give us a few use-cases for backpressure we'll run into at Netflix? (since that's part of the reason I'm proposing this) and I know that you an @jhusain disagree on the need for backpressure in RxJS.

[benjchristensen]

we just need to find a way how to determine consumer behavior when producer is too fast

Agreed, and I suggest that this is exactly what @headinthebox and I spent 9+ months with many in the RxJava community figuring out with "backpressure" which behaves as an upstream signal from the consumer to the producer. Hence my advocating for it here. It's something we have implemented and proven in production. It's not just a theory.

Here is more information on it:

• Docs: https://github.com/ReactiveX/RxJava/wiki/Backpressure
• The long road to making it happen: Backpressure RxJava#1000
• Presentation explaining it: https://speakerdeck.com/benjchristensen/reactive-streams-with-rx-at-javaone-2014?slide=67 and specifically starting at slide 89

[benlesh]

@benjchristensen can we get some use cases specifically to help make sure we're making good decisions?

I mean, it seems like one could do a something like this primitive example to deal with backpressure to some degree:

var requestor = new BehaviorSubject(true);

source.buffer(requestor)
  .flatMap(buffer => Observable.from(buffer)
    .finally(() => requestor.next(true)))
  .subscribe(x => console.log(x));

[benlesh]

I'm not for or against the idea (I created the issue, afterall), but I want it considered up front.

[benjchristensen]

@Blesh yes, I can do that a little later tonight

[benjchristensen]

@headinthebox going back to your point about having multiple collection types, I'd like to explore that idea with you. Let's say we had something like this:

• Observable – purely reactive, no backpressure
• Flowable (just for sake of a name) – reactive push/pull supporting backpressure

interface Flowable {
    subscribe(o: FlowObserver<T>) : void; // or could return Subscription
    lift(operator): Flowable<R>;
}

interface FlowableSubscription {
    unsubscribe(): void;
    isUnsubscribed(): boolean;
    request(n: number): void // optional if backpressure is added
}

interface FlowObserver<T> {
   onSubscribe(s: Subscription): void;
   onNext(t: T): void;
   onError(e: any): void;
   onComplete(): void;
}

Then Observable as the original and simpler:

interface Observable {
    subscribe(o: Observer<T>) : void; // or could return Subscription
    lift(operator): Observable<R>;
}

interface Subscription {
    unsubscribe(): void;
    isUnsubscribed(): boolean;
    add(s: Subscription): void
    remove(s: Subscription): void
}

interface Observer<T> extends Subscription {
   onNext(t: T): void;
   onError(e: any): void;
   onComplete(): void;
}

Then on Flowable we would NOT have temporal flow control operators like sample, throttle, debounce etc. On Observable we would have the temporal flow control.

A Flowable can easily convert into an Observable that ends up fire hosing. An Observable would convert into a Flowable with a strategy given to it such as drop, buffer, debounce, sample, etc.

Is this what you're thinking instead of having one type support both use cases?

[benjchristensen]

Use cases for backpressure:

• writing to async IO in a way that goes at the rate of writing (pauses when buffer is full)
  • this is applicable to writing to databases, event streams, websockets, etc
  • Node.js will start to be used like our Java apps so all similar use cases apply
• composing consumption and production over network boundaries
  • http://reactive-streams.io and http://reactive-socket.io
  • we will have these types of protocols over TCP, WebSockets, Aeron etc
  • we will have full duplex interaction of this nature between Java and Node.js, Node.js and browsers, Java and browsers
  • the network protocol will compose backpressure over it so we have application level consumption signals so the server-side doesn't fill network buffers (which can cause head-of-line blocking on the consumption side, do more work than necessary server-side, overwhelm memory, etc)
• composing hot and cold streams with operators like merge and zip
  • the cold stream will firehose everything into buffers inside merge and zip
  • particularly bad for infinite, lazy Observables like range
• file serving from Node.js?
  • this is probably a solved problem in Node.js, but if someone wants to use RxJS to read, manipulate, compose, transform before writing to IO, it needs backpressure unless it is a very small file
• distributed stream processing
  • it is not hard to see Node.js getting used in our Mantis stream processing system where they are infinite, async event streams across network boundaries that absolutely must apply backpressure
• just normal merge, zip, observeOn use cases while keeping memory usage low
  • for example, RxJava on Android has default bounded buffers kept very small since mobile devices have such limited memory
  • this applies to seemingly "small" sizes like a stream of 100s of items, even that may not want to be buffered, and by default RxJava on Android does not, it will only buffer up to 16 items, whereas server-side it defaults to 128. This means it "backpressures" and does not request more than 16 at a time.
  • considering the memory constraints on devices, this is likely the same concern for JS as it is on Android, though perhaps not at all a concern in a desktop browser
  • the expectation would be that request(n) semantics would apply to something like a websocket connection when streaming say 100s of video titles down, with no more than 16 outstanding at a time

This is a dumping of thoughts, some are more important than others, and all definitely don't apply to all environments (desktop browser, mobile browser, mobile app, Node.js server), but they are representative of the type of stuff we will deal with as we embrace more stream oriented programming and IO.

[benlesh]

Okay, it doesn't seem like the use cases you're mentioning are at all edge cases. In fact, some seem fairly common at least in server-land where you want to maximize data throughput.

So I think what we need to weigh here is:

1. VALUE: How this solves the above use cases better than what RxJS currently has
2. COST: How this will effect performance of the library. It's likely not zero cost (Although maybe it is if we have a different "Flowable" type)

I'm not too worried about cost in terms of "size" of the library, because there are plenty of ways to get around that.

[mattpodwysocki]

@Blesh people are pretty satisfied with the current backpressure mechanisms with pausableBuffered and controlled given that they're hard but they'e been fixed.