benjchristensen
commented
10 years ago This issue is related to #1204 where CompositeSubscription was found to be a major problem for object allocation and the implementation was changed.
Closed benjchristensen closed 10 years ago
benjchristensen
commented
10 years ago This issue is related to #1204 where CompositeSubscription was found to be a major problem for object allocation and the implementation was changed.
benjchristensen
commented
10 years ago If anyone wants to offer alternative implementations please ensure that 2 requirements are met (obviously beyond passing all functional unit tests):
1) It performs better in the JMH performance tests than what's currently in use (see above for how to execute them)
2) It does not cause excessive object allocation. (This is not as easy to identity in benchmarking. The best tool I know of right now is Java Flight Recorder. It can be fast in the JMH test but fail the object allocation requirement, like the CompositeSubscription state machine implementation.)
benjchristensen
commented
10 years ago Here is a resource for us to explore: https://github.com/JCTools/JCTools/tree/master/jctools-core/src/main/java/org/jctools
akarnokd
commented
10 years ago There is my hybrid composite in #1145. As for the hang, my guess is the lazySet that may overwrite a terminal state indicator, but it seems its performance isn't good enough anyway.
benjchristensen
commented
10 years ago A lot has been done and we now have rx.internal.util and rx.internal.until.unsafe for various optimized data structures.
Closing this out as this was a very generic issue and has served its purpose.
This is to document and explore blocking vs non-blocking solutions for a couple use cases, particularly the
SerializedObserverandCompositeSubscriptionwhich are hot code paths in most use of Rx.Thus far the
synchronizedblocking implementations have won and are currently being used. Despite usingsynchronized, the locks are not held while emitting notifications, only for mutating internal data structures.A key consideration is that object allocations must be kept low. The atomic state machine pattern has been attempted on both of these, and is elegant, but was a massive performance problem with
CompositeSubscriptiondue to some valid edge use cases withmergethat result in hundreds of subscriptions being added to the data structure viaCompositeSubscription.addand each time performing a state transition with object allocation.Following are details on the use cases, the current implementation, alternates that have been attempted, and performance results from JMH tests.
The intent of this is to document what has been attempted thus far and seek improvements from anyone who can provide better solutions.
SerializedObserver
Use Case:
Serialize
onNext/onError/onCompletedcalls from multiple threads to be sequential, but without synchronizing and blocking threads.Current: https://github.com/Netflix/RxJava/blob/master/rxjava-core/src/main/java/rx/observers/SerializedObserver.java
Discussions:
Alternate implementations:
Performance tests
Current Code (locks)
MPSC Queue
It completely hung once at this point:
queue and counter
state machine
CompositeSubscription/SubscriptionList
Use Case:
A
Subscriptionimplementation that allows the following:Subscriptionto be invoked when the parent is unsubscribedonNextin a tight loopisUnsubscribedto true and callunsubscribeon all subscriptions added viaaddSubscription, for example,mergeandflatMapwhich usesmergeThis means that the list of subscriptions is multiple writes, a single read at
unsubscribeat the end. The booleanisUnsubscribedis ready many times, modified once.Current CompositeSubscription: https://github.com/Netflix/RxJava/blob/master/rxjava-core/src/main/java/rx/subscriptions/CompositeSubscription.java Current ChainedSubscription: https://github.com/Netflix/RxJava/blob/master/rxjava-core/src/main/java/rx/subscriptions/ChainedSubscription.java |-> will be renamed
Discussions:
Alternate implementations:
add(Subscription)such asObservable.merge(hundreds-of-observables)Performance tests
Current Code (locks)
State Machine
MPSC Queue + AtomicInteger