Serialization Behavior

[benjchristensen]

Opening this issue to capture, document and discuss how serialization (serialize(), merge(), flatMap(), mergeMap()) behaves and is implemented.

Prior to 0.17.1 all serialization was done via blocking synchronization. This is how Rx.Net does it so we adopted the same pattern. This however breaks the model of Rx which is supposed to be non-blocking and be usable in non-blocking environments (think NIO event loops like Netty, Vert.x and Node.js-style apps). Blocking a thread while merging can significantly impact throughput in an application based on event-loops.

The migration to doing "serialization" instead of "synchronization" came with trade-offs.

Back Pressure

To be non-blocking means it must becomes async and allow threads to deliver their notifications, queueing if necessary and return. This can result in buffer-bloat and typical back-pressure problems. Solutions to this are being explored internally and with other teams/companies and will result in changes in the future.

Concurrency

One way of solving the problem without blocking is similar to observeOn and everything gets dropped into a queue and another thread pulls off the queue.

This however means that we are injecting additional concurrency in places it is not expected and generally not desired.

The current implementation does not do this. It uses the threads that are pushing events through and "steals" a single thread at a time to push through whatever is in the queue and then itself and then return to do its own work.

Thread Starvation

Stealing threads opens up the possibility of thread starvation. If a thread loops continually to drain the queue and the queue always is getting filled by other threads it will never be released to do its own work. This would means that the events intended for it to deliver would never be delivered as it is always busy delivering events on behalf of other threads.

Delayed Delivery

To prevent thread starvation the current implementation only allows draining the queue once. This can be increased to multiple iterations, but at some point it stops draining and returns and allows another thread to "win" and start draining.

During the time gap between the draining thread finishing and and a new thread taking over there may be a delay where events stay in the queue. This can delay delivery of events.

In a fast moving stream this is not a problem as another thread immediately takes over. In an intermittent stream however this can possibly mean long, non-determistic delays.

Possible Improvements

There are a few ways to improve this situation without reverting back to blocking.

Metrics could be kept to know if a stream is fast moving and thus thread-starvation is an issue and draining should be handed off to another thread. If starvation is not an issue then the queue could be fully drained before returning. This is still not perfect and would still risk one of the two occurring, but could probably solve most cases. The difficult is doing so without significantly impacting normal performance.

Another option is conditionally scheduling delivery onto another Scheduler when starvation is happening. This would allow most cases to be done by stealing threads, but flip to an observeOn style model if contention and/or starvation is happening.

Next steps

If the current functionality is breaking your use cases, you may want to stay on 0.17.0 while working with us to improve. Most use cases have shown to work fine with 0.17.1 behavior and the non-blocking and deadlock-free characteristics are necessary.

I welcome discussion, unit tests, pull requests and assistance on this.

Existing performance tests (that need work):

• https://github.com/Netflix/RxJava/blob/master/rxjava-core/src/perf/java/rx/operators/OperatorSerializePerf.java
• https://github.com/Netflix/RxJava/blob/master/rxjava-core/src/perf/java/rx/archive/operators/OperatorSerializePerformance.java

Existing unit tests:

• https://github.com/Netflix/RxJava/blob/master/rxjava-core/src/test/java/rx/operators/OperatorSerializeTest.java
• https://github.com/Netflix/RxJava/blob/master/rxjava-core/src/test/java/rx/observers/SerializedObserverTest.java

[Strilanc]

I recommend the following:

• Steal the current thread, as currently happens, but continue draining even if work is queued while you're draining.
• If some limit is exceeded, either in number of work items or time duration, spin off a thread or other async-execution-thingy to handle draining the queue from then on.

This guarantees items can't sit indefinitely in the queue, and will not live lock of callers indefinitely.

[benjchristensen]

spin off a thread or other async-execution-thingy to handle draining the queue from then on.

So it's clear what this means, the Rx contract is to never inject concurrency unless a Scheduler overload is offered that controls what Scheduler is used.

This means every flatMap, mergeMap, merge and mergeDelayError method needs an overload with a Scheduler. This is around 30 methods.

[Strilanc]

The livelock is only an issue when there's already concurrency (watch out for re-entrancy though), since other threads have to be adding data... but I suppose if every one of them was in some context the async-thingy we spun off might violate that.

What about some sort of hand-off? After the limit is exceeded, the current consumer sets a flag that others check when they're producing. The first producer that comes along is then forced to block so the consumer can hand-off to them after it finishes the next job. So we double-pay the latency of a job, once on the current consumer and once on blocking the to-be consumer, but gain not blocking anything indefinitely.

[benjchristensen]

The first producer that comes along is then forced to block so the consumer can hand-off to them after it finishes the next job.

The problem I see with this is that we have no idea how quickly the onNext will complete and thus can block for non-deterministic periods of time. On event-loops that's a real problem. We could argue that if anyone is doing something like that they are "doing the wrong thing" but I'm not sure we can.

[benjchristensen]

The backpressure (#1000) solution will affect the design of this. In particular it will mean the buffer becomes bounded and may naturally prevent thread starvation as the fast producers will park themselves. The addition of backpressure machinery may or may not necessitate the additional concurrent consumer.

[benjchristensen]

I suggest changing the current implementation to tradeoff for risking starvation rather than delaying delivery. The starvation problem seems less likely to occur and a symptom of other problems (lack of backpressure?).

Or should we leave this as is until we get the backpressure stuff in place?

I'm not all that comfortable yet adding more threads to deal with serializing other threads. It feels wrong and makes for a confusing API (particularly on flatMap) since a Scheduler would exist but not be used for real work, just sometimes pushing notifications.

[headinthebox]

suggest changing the current implementation to tradeoff for risking starvation

I would vote for that as a temporary fix until we have the back pressure stuff in place.

[Strilanc]

Sounds acceptable to me.

[davidmoten]

And me. Interesting discussion.

[benjchristensen]

Please review changes on this pull request: #999

In particular this commit: benjchristensen@5b317ad

[petermd]

(moving comment from pull-request)

apologies for joining this a little late, but had a couple of observations/questions

• in theory this wasn't a problem for Vert.x because as long as all the events that trigger the callbacks come from the Vert.x event loop those calls are already serialized (so there is no contention)
• it does imply that all code using RxJava in Vert.x does not use any other threads/schedulers that might cause it to block or switch threads. in the Vert.x model the design requires you to delegate slow/blocking operations to separate dedicated worker-verticles that use their own thread-pool (and communication with those threads is via an async event bus mechanism)
• this means that there is no "slow operation" that can happen via onNext, each processing pipeline should complete quickly (otherwise the event loop will starve other Verticles tied to the same Thread)

From what I understand of the current solution, this means that in Vert.x the short-circuit where no queue is created will always be executed and there will be minimal impact.

It also means that Vert.x is relying on not introducing Concurrency without an explicitly provided Scheduler (aside: I assume naively this never happens?) as the user would also need a matching observeOn() with a customer Scheduler to ensure the thread isolation is preserved.