Netflix/RxJava — pull request #962 by benjchristensen (2014-03-13 18:01:05+00:00)
Title: Migrate from SynchronizedObserver to SerializedObserver

This pull request replaces use of `SynchronizedObserver` with `SerializedObserver`.

### Why?

#### 1) Deadlocks

Holding a lock while emitting notifications (`onNext`, `onCompleted`, `onError`) allows for deadlocks if the event results in a cycle back to the parent.

While testing RxJava 0.17.0 in Netflix production we ran into one of these. The vulnerability has existed all along but we finally hit it.

This issue has also been reported before such as: http://twistedoakstudios.com/blog/Post8424_deadlocks-in-practice-dont-hold-locks-while-notifying

#### 2) Blocking Threads

The use of `synchronized` can block threads. If it's used in areas such as modifying a data structure this can be okay. When emitting a notification however it is a problem as the result of an `onNext` can take a non-deterministically long time to complete. This means any other thread trying to emit will be blocked.

If the source threads are event loops (such as Vert.x or Netty) this will block the event loops.

For example, if two network calls off two Netty event loops are being merged (such as via `flatMap`) and one of them does further slow processing that causes `onNext` to be slow, it will block the other `onNext` which blocks the event loop and prevents any further IO on that thread. This is a significant problem for system scale and breaks the promise of Rx being a non-blocking, reactive library.

### Solution

The `synchronize`, `SynchronizedObserver` and `SynchronizedSubscriber` operator and classes have been deprecated. They are replaced by `serialize`, `SerializedObserver` and `SerializedSubscriber`.

The `SerializedObserver` still ensures only a single thread can emit `onNext`, `onCompleted`, or `onError` at a time but does not hold a lock while doing so. Instead of blocking threads it will accept the incoming events into a buffer. Thus, it becomes an asynchronous operator.

The `merge` operator (which impacts `flatMap`) now uses `SerializedObserver`, along with any other place in RxJava that needed synchronization.

### Implementation

3 implementatations were written and tested:

- [queue and counter](https://github.com/benjchristensen/RxJava/blob/serialize-implementations/rxjava-core/src/main/java/rx/observers/SerializedObserverViaQueueAndCounter.java)
- [queue and lock](https://github.com/benjchristensen/RxJava/blob/serialize-implementations/rxjava-core/src/main/java/rx/observers/SerializedObserverViaQueueAndLock.java)
- [atomic state machine](https://github.com/benjchristensen/RxJava/blob/serialize-implementations/rxjava-core/src/main/java/rx/observers/SerializedObserverViaStateMachine.java)

Performance testing [revealed](https://github.com/benjchristensen/RxJava/blob/serialize-implementations/rxjava-core/src/perf/java/rx/operators/OperatorSerializePerformance.java):

```java
/**
* 1 streams emitting in a tight loop. Testing for single-threaded overhead.
*
* -> blocking synchronization (SynchronizedObserver)
*
* Run: 10 - 58,186,310 ops/sec
* Run: 11 - 60,592,037 ops/sec
* Run: 12 - 58,099,263 ops/sec
* Run: 13 - 59,034,765 ops/sec
* Run: 14 - 58,231,548 ops/sec
*
* -> state machine technique (SerializedObserverViaStateMachine)
*
* Run: 10 - 34,668,810 ops/sec
* Run: 11 - 32,874,312 ops/sec
* Run: 12 - 33,389,339 ops/sec
* Run: 13 - 35,269,946 ops/sec
* Run: 14 - 34,165,013 ops/sec
*
* -> using queue and counter technique (SerializedObserverViaQueueAndCounter)
*
* Run: 10 - 19,548,387 ops/sec
* Run: 11 - 19,471,069 ops/sec
* Run: 12 - 19,480,112 ops/sec
* Run: 13 - 18,720,550 ops/sec
* Run: 14 - 19,070,383 ops/sec
*
* -> using queue and lock technique (SerializedObserverViaQueueAndLock)
*
* Run: 10 - 51,295,152 ops/sec
* Run: 11 - 50,317,937 ops/sec
* Run: 12 - 51,126,331 ops/sec
* Run: 13 - 52,418,291 ops/sec
* Run: 14 - 51,694,710 ops/sec
*/

/**
* 2 streams emitting in tight loops so very high contention.
*
* -> blocking synchronization (SynchronizedObserver)
*
* Run: 10 - 8,361,252 ops/sec
* Run: 11 - 7,184,728 ops/sec
* Run: 12 - 8,249,685 ops/sec
* Run: 13 - 6,831,595 ops/sec
* Run: 14 - 8,003,358 ops/sec
*
* (faster because it allows each thread to be "single threaded" while blocking the other)
*
* -> state machine technique (SerializedObserverViaStateMachine)
*
* Run: 10 - 4,060,062 ops/sec
* Run: 11 - 3,561,131 ops/sec
* Run: 12 - 3,721,387 ops/sec
* Run: 13 - 3,693,909 ops/sec
* Run: 14 - 3,516,324 ops/sec
*
* -> using queue and counter technique (SerializedObserverViaQueueAndCounter)
*
* Run: 10 - 4,300,229 ops/sec
* Run: 11 - 4,395,995 ops/sec
* Run: 12 - 4,551,550 ops/sec
* Run: 13 - 4,443,235 ops/sec
* Run: 14 - 4,158,475 ops/sec
*
* -> using queue and lock technique (SerializedObserverViaQueueAndLock)
*
* Run: 10 - 6,369,781 ops/sec
* Run: 11 - 6,933,872 ops/sec
* Run: 12 - 5,652,535 ops/sec
* Run: 13 - 5,503,716 ops/sec
* Run: 14 - 6,219,264 ops/sec
*/

/**
* 2 streams emitting once a millisecond. Slow emission so little to no contention.
*
* -> blocking synchronization (SynchronizedObserver)
*
* Run: 10 - 1,996 ops/sec
* Run: 11 - 1,996 ops/sec
* Run: 12 - 1,995 ops/sec
* Run: 13 - 1,997 ops/sec
* Run: 14 - 1,996 ops/sec
*
* -> state machine technique (SerializedObserverViaStateMachine)
*
* Run: 10 - 1,996 ops/sec
* Run: 11 - 1,996 ops/sec
* Run: 12 - 1,996 ops/sec
* Run: 13 - 1,996 ops/sec
* Run: 14 - 1,996 ops/sec
*
* -> using queue and counter technique (SerializedObserverViaQueueAndCounter)
*
* Run: 10 - 1,996 ops/sec
* Run: 11 - 1,996 ops/sec
* Run: 12 - 1,996 ops/sec
* Run: 13 - 1,996 ops/sec
* Run: 14 - 1,995 ops/sec
*
* -> using queue and lock technique (SerializedObserverViaQueueAndLock)
*
* Run: 10 - 1,996 ops/sec
* Run: 11 - 1,996 ops/sec
* Run: 12 - 1,997 ops/sec
* Run: 13 - 1,996 ops/sec
* Run: 14 - 1,995 ops/sec
*/
```

The state machine solution was tested in production but caused performance problems, most likely due to the immense object allocation it needs to do.

The elegant "queue and counter" solution does not perform well enough in the non-contended case.

The "queue and lock" model performs well in the non-contended case and under contention, despite not being a very elegant solution and requiring the use of mutex locks for the state changes (but it does not hold the locks during notification).

### Considerations

This does allow unbounded buffer growth, the same as `observeOn` and `zip` instead of blocking the producer threads.

### Conclusion

The implementation in this pull request can and likely will be improved over time. The other implementations are purposefully being shown to allow others to provide further insight on how to do this better.

This change is important to ensure RxJava is non-blocking and our canary testing of this change in the Netflix production environment suggests this change is both performant and functional.