Backpressure

[benjchristensen]

Thus far Rx has left backpressure solutions "an exercise for the user". It can be done via manual feedback loops and operators like throttle, sample, window, etc.

The drawback of this is that it requires someone understanding the implication of every operator and when it might be async (buffer/queue) and requires the effort to create a feedback loop and hook it up correctly. Some use cases will probably always require this type of manual effort, but it's worth exploring whether we can solve the general use cases within Rx.

Most Rx operators are synchronous in nature meaning the producing thread does not return until the work is completed. The map operator is a good example. It does computation work to transform from T -> R and nothing else.

There are some operators though that are async in nature and involve unbounded buffers. These include observeOn, zip and merge (as of 0.17.1 which means flatMap is affected). It also happens if an Observer is writing the data out to non-blocking IO (such as a Netty channel).

In all of these example, if the producer is fast, such as an in-memory Iterable, loading a file, or a firehose of events over a network, buffer-bloat (and eventually OutOfMemory) can easily occur.

After some experimentation and talking with various individuals, teams and companies about this topic, a possible solution is the use of co-routines from the producing Observable that a Subscriber can request data from in batches.

Prototype Code

A prototype of this is being experimented with at https://github.com/benjchristensen/RxBackpressure though it is not yet complete (as of this writing).

The general idea is that an Observable.OnSubscribe implementation could register a producer co-routine with the Subscriber and then only push data down when the Subscriber has said how much it can receive. It becomes a conditional push-model. The co-routine can then be "parked" by the producer by finishing work and returning (releasing the thread) and then "unparked" or resumed by the Subscriber when it wants more. In this way no threads are blocked and only the amount of data the Subscriber can handle is sent. The same model can work across threads or across the network.

If the Observable chain has no async operators then it will be executed with an "infinite" request and behave exactly as it does today without any parking.

If an async operator is in the chain then it will "request" a batch size equaling it's internal buffer (say 128, 512, 1024 etc).

A simple producer of an Iterable would look like this (ignoring debates over naming and API):

public final class OnSubscribeFromIterable<T> implements OnSubscribe<T> {

    final Iterable<? extends T> is;

    public OnSubscribeFromIterable(Iterable<? extends T> iterable) {
        this.is = iterable;
    }

    @Override
    public void call(final Subscriber<? super T> o) {
        // state is outside the co-routine below
        final Iterator<? extends T> iter = is.iterator();
        // define the co-routine
        Action1<Request> func = new Action1<Request>() {
            @Override
            public void call(final Request r) {
                if (!r.countDown()) {
                    return;
                }

                while (iter.hasNext()) {
                    final T value = iter.next();
                    o.onNext(value);

                    if (!r.countDown()) {
                        // we have delivered all that was requested
                        // so "park" be returning and releasing this thread
                        return;
                    }
                }
                o.onCompleted();
            }

        };

        // register co-routine with Subscriber
        o.setProducer(func);
    }
}

The observeOn operator is async and thus will request batches as needed. The relevant code looks like this:

private void pollQueue() {
    do {
        Object v = queue.poll();
        if (v != null) {
            if (v instanceof Sentinel) {
                if (v == NULL_SENTINEL) {
                    observer.onNext(null);
                } else if (v == COMPLETE_SENTINEL) {
                    observer.onCompleted();
                } else if (v instanceof ErrorSentinel) {
                    observer.onError(((ErrorSentinel) v).e);
                }
            } else {
                observer.onNext((T) v);
            }
        }
        requested--;
    } while (counter.decrementAndGet() > 0 && !observer.isUnsubscribed());
    if (requested == 0) {
        requested += SIZE;
        // request more (starting the co-routine up again)
        request(SIZE);
    }
})

The subscribeOn operator decorates the co-routine so scheduling is retained and is started on the desired thread each time rather than the consuming thread doing the work.

public void setProducer(final Action1<rx.Subscriber.Request> producer) {
    subscriber.setProducer(new Action1<Request>() {

        @Override
        public void call(final rx.Subscriber.Request r) {
            inner.schedule(new Action1<Inner>() {

                @Override
                public void call(Inner inner) {
                    producer.call(r);
                }

            });

        }

    });

}

The take and skip operators compose the batch sizes to adjust accordingly to what is being taken or skipped:

• https://github.com/benjchristensen/RxBackpressure/blob/master/rx-bp-prototype/src/main/java/rx/operators/OperatorTake.java#L93
• https://github.com/benjchristensen/RxBackpressure/blob/master/rx-bp-prototype/src/main/java/rx/operators/OperatorSkip.java#L51

The Subscriber manages the life-cycle of when the co-routine is run: https://github.com/benjchristensen/RxBackpressure/blob/master/rx-bp-prototype/src/main/java/rx/Subscriber.java#L97

For example, if the final Subscriber in the chain is hit and no async operator was involved it immediately invokes the co-routine with "-1" request size so it is infinite.

Example Use Cases

Use case are being written here: https://github.com/benjchristensen/RxBackpressure/tree/master/rx-bp-examples/src/test/java/rx/examples

Handling Observables Without Co-routines

Implementing an Observable that supports backpressure is more complicated and not all will support it. In those cases the async operators will onError stating that an Observable is not respecting the backpressure and suggest resolutions (linking to documentation probably).

The two routes at that time are: 1) fix the Observable to support backpressure, or 2) use one of several backpressure operators that will be added such as:

• whileParkedDrop
• whileParkedBuffer
• whileParkedBufferThenDrop(int maxCount)
• whileParkedBlock (when blocking is okay, such as on Quasar fibers, IO threads, etc)
• whileParkedUnsubscribe (such as on a hot stream like mouse events)

Outstanding Work

An earlier prototype (not public) had zip functioning but the current code does not. We know how it is done and this model works with it but have not yet spent the time to re-implement it. We need to.

The biggest outstanding item is making merge work (and thus flatMap). This is the major hurdle of any backpressure solution working.

We (myself, @abersnaze, @headinthebox and others) have whiteboarded it and believe we have a solution but my schedule has not allowed me to code it. I have held off on writing this post as I wanted to code it up first, but since I still haven't had time I wanted to get this information public instead of holding it up.

The planned design for merge is that each Observable being merged would have it's own buffer (of say 128 items) and then the consumer would request n items which would come in round-robin fashion from all of the merged buffers. Each individual Observable would exert it's own backpressure upstream. Slow Observables may never fill their buffer while fast ones will.

We will not attempt to limit the horizontal growth (number of Observables being merged) but will limit the vertical growth (size of buffer for each Observable).

The challenge with this is making sure performance and fairness are balanced.

Next Steps

1. Finish prototype implementatation of merge and zip
2. Finish examples across threads and netwwork.
3. Bikeshed and debate APIs and naming conventions
4. Prove or disprove performance/functionality tradeoff.
5. Test long enough to convince ourselves we're not going to cause deadlocks/livelocks/hangs somewhere (co-routines not running).

I look forward to your help making this happen.

[zsxwing]

Does the source Observable need to guarantee to call setProducer? If not, looks setProducer can not be triggered automatically.

[kirkshoop]

The main issues that I have run into while trying to implement back-pressure involving continuations in rxcpp are; synchronization and additional complexity everywhere (it is not localized). Combined these really pessimize performance.

I think that there is a way to localize the complexity behind three interfaces. Two already exist; scheduler and connectable_observable. The new interface is resumable_observable which just adds pause() and resume() to observable. Edit: resumable_observable is not required of course, it is just a renaming of connectable_observable. The new name for connect() is resume() and the new name for unsubscribe() is pause()

The usage depends on the type of source.

Hot Observables

The pause() implementations that are meaningful for Hot Observables are; unsubscribe or buffer.
These can be realized with some operators.

• An operator that takes an existing connectable_observable source for the buffering behavior and returns a resumable_observable. pause() calls unsubscribe() and resume() calls connect()
• Operators that take observable and directly implement a buffering behaviour. pause() switches into buffering mode and resume() plays the buffer out.

Cold Observables

The Hot back-pressure policies are available by applying publish() to turn the observable Hot, then use one of the Hot operators that return a resumable_observable.

The additional policy, which only applies to Cold observables, is to stop the production of the next value until resumed. This does require a continuation. Fortunately, Cold Observables already create a continuation per value. Cold Observables take a scheduler and call schedule in some form for each value. The continuation that is needed to support resumable_observable is passed to schedule already!

To exploit this a new regulator type wraps an existing scheduler and has methods get_scheduler(), pause() and resume(). The regulator scheduler keeps its own queue and schedules itself on the wrapped scheduler to pop the next item and run it.

A new operator, that takes the regulator and an observable, returns a resumable_observable. pause() will stop scheduling the regulator queue and resume() will start scheduling the regulator queue.

Join Operators

These would change to accept resumable_observables.

Cold Example

regulator<current_thread> r;
auto counter = range<int>(0, 100000000, 1, r.get_scheduler());
counter.skip(1).resume(r)
    .zip(counter.resume(r), [](int lhs, int rhs){return std::make_pair(lhs, rhs);})
    .subscribe([](std::pair<int, int> p){std::cout << p << std::endl;});

Hot Example

button_downs.resume_with_subscribe()
    .zip(button_ups.resume_with_subscribe(), 
        [](button_down lhs, button_up rhs){return std::make_pair(lhs, rhs);})
    .subscribe([](std::pair<button_down, button_up> p){std::cout << p << std::endl;});

[benjchristensen]

Does the source Observable need to guarantee to call setProducer?

No. It can continue to act just like right now. Supporting backpressure is optional. While discussing with @abersnaze and @headinthebox we contemplated using a minimum buffer size throughout the codebase (like 32, 64, or 128) so if an Observable is going to emit less than that it doesn't need to worry about back pressure at all.

If not, looks setProducer can not be triggered automatically.

What do you mean by this? If it's not set, there's nothing to trigger and the source Observable is responsible for emitting via onNext as normal.

[benjchristensen]

@kirkshoop Thank you for getting involved in the conversation!

I apologize that I don't fully understand the proposed implementation but I'll provide comments based on my interpretation.

We initially implemented the pause() solution and moved on from it after discussion with other teams (such as at Typesafe) who have used the pause() solution before and found issues with it. We instead inverted "pause/resume" to the batch continuation model.

The pause solution works very well within a single process where pausing can be fairly deterministic by sharing an atomic/volatile reference across thread boundaries. It does not work well across network boundaries though which is often the source of data requiring back pressure. It is far too non-deterministic to asynchronously request a source to "pause" and therefore still requires unbounded buffers or dropping data.

Also, anything that is modeled after ConnectableObservable is immediately not composable and requires awkward coding just like publish() and connect() requires.

Cold Observables take a scheduler and call schedule in some form for each value.

This is very inefficient and exactly what we have sought to avoid in RxJava, particularly as of version 0.17 where the Subscription is injected into the source so that a while loop can happen synchronously and still unsubscribe. Scheduling each notification is not something we do very often at all.

Also, this does not apply well to network requests, which are the source for most cold observables. Generally they are happening on a separate thread, or an event loop and the notifications are not rescheduled again on a separate thread. Nor do we use CurrentThreadScheduler which was significantly diminished in importance as of RxJava 0.17 (due to the inefficiency of it in normal cases). Thus, cold observables generally do not have a thread or event loop to "pause". Forcing them to do so to achieve back pressure would force everyone to pay a performance penalty even in the best case where back pressure is not needed.

synchronization and additional complexity everywhere (it is not localized). Combined these really pessimize performance.

What extra synchronization is required? The proposed solution (using continuations) does not change the threading model or introduce any extra use of schedulers/concurrency.

I am concerned with performance impact for cases that don't require back pressure, but thus far because of the batch request approach which then allows async onNext notifications, it does not seem to be a performance problem (still too early to perform all performance testing to know for sure though). The case where performance will be impacted are the very ones where buffer-bloat is a problem and it is purposefully slowing it down to the rate of consumption. The one area where performance is an interesting challenge (and will be with continuations as well) is non-blocking serialization: #998

[kirkshoop]

Okay I am caught up now. I was designing for a different system. :)

I think that we agree that we are happy with the existing Rx behavior for Hot Observables and that this back-pressure discussion is for Cold Observables.

It is far too non-deterministic to asynchronously request a source to "pause" and therefore still requires unbounded buffers or dropping data.

I disagree with this. The buffer can be bounded declaratively on the source operator. In fact, I argue that a declarative chunk-size and buffer-size, that is passed to Request in your design, provides two benefits.

1. The appropriate chunk-size is dependent on the source, not the subscriber or operators. Modifications of the request value in Take and Skip would not affect the throughput of the source because the Request will always schedule chunks to keep the local buffer full, so they are not needed.
2. Providing declarative sizes allows the Request to create overlapping calls for subsequent chunks (when the buffer has room for a chunk). I have a application where the latency for a request is 13sec but the data in that request only contains the data needed for the next 10sec. If the requests are not overlapped then the pipeline stalls horribly.

The buffer-size ensures that no data is dropped and that the size is bounded. A good minimum for the buffer-size is 1.5 * chunk-size.

I still think that pause() and resume() suffice as the control surface. They would not control the scheduling directly, they would synchronously control the buffer draining, which would indirectly trigger more chunks as the buffer emptied. The source should be paused synchronously in on_next when the declared input buffer for the subscriber is full. The source should be resumed when the buffer has space. This ensures that the buffers in the whole expression are bounded.

However, the buffer draining could be controlled with the current request(count) control surface as well. I still think it is important to change the Request impl to support overlapped chunk requests with declared sizes for the chunk and buffer.

I think that in addition to chunk-size and buffer-size, there should be two scheduler arguments. One for the overlapped chunk requests and the other for the buffer draining.

This is very inefficient and exactly what we have sought to avoid in RxJava

Yes, this is very inefficient. :) It added complexity to the scheduler in rxcpp to control the perf issue while still allowing current_thread to be fully supported. I have looked at the code for RxJava's approach now and I have a background mental task deciding whether to similarly limit support for current_thread in rxcpp. Thanks!

What extra synchronization is required?

The Request design does not have the same issues that I was running into with both the resumeWith + is_resumed and the 'allocateMarble + completeMarble' designs. Although the later would have had some interesting debugger applications.

[benjchristensen]

The source should be resumed when the buffer has space.

Agreed, but I don't understand how we would have resume() and request(count) in the same system. To me resume() and request(count) are competing implementations for the same thing – restarting work after it being paused.

It seems you are referring to having batch sizes when you say "the buffer can be bounded declaratively on the source operator" but then suggest we can work with just pause/resume and not request(count), so I'm not following.

What we're trying to achieve is that when 'resume` is invoked (whatever it is called) that is passes along how much space is in the buffer for the upstream to send.

Are you suggesting that only the IO boundary needs the request(count) functionality and everything operating internally in the same process then just use pause/resume to drain from the IO Observable buffer?

I think that in addition to chunk-size and buffer-size, there should be two scheduler arguments. One for the overlapped chunk requests and the other for the buffer draining.

We attempted this in an early prototype and it became ridiculously complex because the co-routine state now needed to be thread-safe and handle concurrent execution. I'm open to it as I agree it would be good, but it's not clear yet how to make it simple to implement and I haven't spent enough time to try and determine alternative implementations.

To understand what I mean, take a look at the OnSubscribeFromIterable example at the beginning of this issue, it has this line in it:

final Iterator<? extends T> iter = is.iterator();

If we allowed overlapped requests in this example the producer could no longer use a simple iterator.

The buffer can be bounded declaratively on the source operator.

Since I'm obviously not quite understanding, can you elaborate more on what you mean by this and how it interacts with resume/pause/request(count) please?

[kirkshoop]

I don't understand how we would have resume() and request(count) in the same system. To me resume() and request(count) are competing implementations for the same thing – restarting work after it being paused.

I am suggesting that a cold observable that supports back pressure has both a producer and a consumer inside.

The producer generates requests for a chunk-size of items and the consumer stores a buffer-size of items.

Subscriber::setProducer(void(Request, Subscriber){}) and int Request::get_chunk_size() are the surface of the producer.

pause(), resume() and optionally request(count) are the surface of the consumer. The internal consumer is exposed as the Observable that can be subscribed to.

resume, pause and request(count) all control the draining of the buffer in the source operator.

What we're trying to achieve is that when 'resume` is invoked (whatever it is called) that is passes along how much space is in the buffer for the upstream to send.

Conceptually request(count) is just resume() -> count * on_next -> pause().
Once the source has separated the producer from the consumer what benefit is achieved by passing the count up?

[benjchristensen]

Thanks for explaining. I need to think about this more and trace it through the composition of merge, zip, skip, take, observeOn, subscribeOn to understand. It won't happen quickly unfortunately as I'm wrapping up other things and traveling to London in a couple days for a conference.

My initial reaction though is that we shouldn't have both pause/resume and request(count).

[kirkshoop]

traveling to London in a couple days for a conference

Me too - will you be at React?

My initial reaction though is that we shouldn't have both pause / resume and request(count) .

I agree.

Complexity of source operator implementation

I was thinking more about the operator implementations in this design and it changed the design a bit.

range should not be overlapped or buffered. iterator should not be overlapped and should be buffered (I am thinking of a generator that takes time in getNext).

Thus, I think that a declarative max-concurrency parameter to the producer is also required. iterator and range would set it to 1 and neither would have to constrain the chunk or buffer sizes.

To simplify the implementation of the operators supporting overlapped calls:

Add a function Subscriber::setStateSelector<T>(T(T){}) and change Subscriber::setProducer<T>(void(T, Subscriber){})

The producer function calls

state = stateSelector(state); 
bufferStart = bufferNext;
bufferNext = bufferStart + chunk_size;
produce(state, new BufferSubscriber(buffer, bufferStart, bufferNext);

The producer will not overlap calls to stateSelector and the BufferSubscriber will control the synchronization of the buffer access for produced values.

The produce and stateSelector functions supplied by the source operator can choose to share the same state across overlapped calls and deal with the sync themselves or stateSelector can choose to return an independent state value so that produce does not synchronize at all.

[benjchristensen]

Me too - will you be at React?

Yes I will be. If you're there we'll definitely need to meet.

[alexandru]

IMHO, both pause/resume and request(count) is trying to shove a square peg into a round hole. I've been thinking about an alternative design inspired by alternative abstractions for handling streams:

  trait Observer[-T] {
    def onNext(elem: T): Future[Unit]
    def onError(ex: Throwable): Unit
    def onComplete(): Unit
  }

So basically Observer.onNext can tell the Observable about when it is ready for more by means of its return type, a Future. For the Observable itself, nothing much changes, i.e...

UPDATE: because I pulled this out of my arse, haven't thought about the implications - but the above doesn't solve the problem, still thinking about it.

[benjchristensen]

Hi @alexandru, thanks for getting involved.

In early prototypes I tried making 'onNext' return 'Future' and it became unwieldy very quickly. However, it still feels like it may be a valid approach. I too am considering this more.

As a side note, one practical concern is memory impact of allocating a future for every onNext.

[alexandru]

@benjchristensen you're right, it would be more heavy and I believe this would go against the design goals of Rx.NET / RxJava, as you'd have to make the whole abstraction to be entirely async. I personally prefer it that way, for the use-cases that are of concern to me, but it would make it impractical for certain use-cases for which Rx was designed for (the usual pick your poison applies).

I'm exploring it for a bit, asking for feedback on the scala-user mailing list, so if you can tolerate Scala code and my strong opinions, here are some details I've thought about: https://groups.google.com/forum/#!topic/scala-user/ckgrXz_4F_A