Performance issues after upgrading from 1.0.4 -> 1.0.8

[neerajrj]

Hello,

We are seeing high levels of lock contention after our upgrade to RxJava 108

Apologize for not having a unit test to reproduce this we have a fairly complex system and we are having trouble figuring out which areas to dig deeper to find a reproducible case.

This is a paste from a JMC view. As far as we know nothing should be getting unsubscribed in our application.
We would appreciate if anyone can shed some light on what kind of behavior would trigger the stack below.

Stack Trace Sample Count    Percentage(%)
java.util.concurrent.locks.LockSupport.unpark(Thread)   1,113   76.233
 java.util.concurrent.locks.AbstractQueuedSynchronizer.unparkSuccessor(AbstractQueuedSynchronizer$Node) 1,113   76.233
      java.util.concurrent.locks.AbstractQueuedSynchronizer.release(int)    1,113   76.233
         java.util.concurrent.locks.ReentrantLock.unlock()  1,113   76.233
            rx.internal.util.SubscriptionList.remove(Subscription)  1,113   76.233
               rx.internal.schedulers.ScheduledAction$Remover2.unsubscribe()    1,113   76.233
                  rx.internal.util.SubscriptionList.unsubscribeFromAll(Collection)  1,113   76.233
                     rx.internal.util.SubscriptionList.unsubscribe()    1,113   76.233
                        rx.internal.schedulers.ScheduledAction.unsubscribe()    1,113   76.233
                           rx.internal.schedulers.ScheduledAction.run() 1,113   76.233
                              java.util.concurrent.Executors$RunnableAdapter.call() 1,113   76.233
                                 java.util.concurrent.FutureTask.run()  1,113   76.233
                                    java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor$ScheduledFutureTask)    1,113   76.233
                                       java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run()   1,113   76.233
                                          java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor$Worker)  1,113   76.233
                                             java.util.concurrent.ThreadPoolExecutor$Worker.run()   1,113   76.233
                                                java.lang.Thread.run()  1,113   76.233

[akarnokd]

I guess you are scheduling a lot of work on the computation scheduler. We changed the tracking of tasks from using synchronized to j.u.c.Lock because it gives better throughput according to our JMH benchmarks. It appears task addition takes longer while holding the lock and the unsubscribe part spins and parks; most of the time, spin should be enough. Do you measure performance degradation?

[neerajrj]

We do not have formal performance benchmarks on our jobs. We autoscale our cluster based on resource utilization and we saw our cluster size go up by about 30% - 100% depending on the workload.
To give you a bit more context we have built a reactive stream processing system that basically reads data items as they flow in and processes them on a computation threadpool using observeOn. In terms of scale the jobs seeing the most impact process <= 10k messages/sec of size a few kBs each

[akarnokd]

Sounds like your data rate reaches a critical frequency where the submission of new values in observeOn overlaps its drain and thus causes extra contention. The change from 1.0.4 to 1.0.8 consists of two parts: lock in SubscriptionList and the use of SubscriptionList instead of CompositeSubscription for non-timed tasks inside the computation scheduler. What is the java version you are running and can you name the virtualization environment?

[neerajrj]

Hey David,

We are on Java 8 and we running in a Mesos container inside an AWS instance
(m3-2xl series)

Thanks
Neeraj

On Mon, Apr 6, 2015 at 11:56 AM, David Karnok notifications@github.com
wrote:

Sounds like your data rate reaches a critical frequency where the
submission of new values in observeOn overlaps its drain and thus causes
extra contention. The change from 1.0.4 to 1.0.8 consists of two parts:
lock in SubscriptionList and the use of SubscriptionList instead of
CompositeSubscription for non-timed tasks inside the computation scheduler.
What is the java version you are running and can you name the
virtualization environment?

—
Reply to this email directly or view it on GitHub
#2857 (comment).

[akarnokd]

Thanks. Not sure what's the cause, but you could try shifting the contention by using batching before the observeOn and unbatch after it;

source.batch(4).observeOn(Scheduler.computation()).concatMap(v -> Observable.from(v))

[davidmoten]

That would be buffer rather than batch, right? I was going to suggest the
same thing, sounds like a fast round robin on a scheduler.

Thanks. Not sure what's the cause, but you could try shifting the
contention by using batching before the observeOn and unbatch after it;

source.batch(4).observeOn(Scheduler.computation()).concatMap(v ->
Observable.from(v))

—
Reply to this email directly or view it on GitHub
#2857 (comment).

[akarnokd]

@davidmoten Sure.

[neerajrj]

Hey David,
Batching is a good suggestion however that would be a breaking change

On Mon, Apr 6, 2015 at 1:02 PM, David Karnok notifications@github.com
wrote:

@davidmoten https://github.com/davidmoten Sure.

—
Reply to this email directly or view it on GitHub
#2857 (comment).

[akarnokd]

This is why I suggested using concatMap to flatten the batches again after the observeOn so your subsequent computation chain doesn't need to change.

[neerajrj]

Oops hit the send button too soon.

As I was saying adding a batch or buffer would change our public API from T
to a List.

You mentioned earlier you saw significant throughput improvement in the JMH
tests can you provide more details, I am curious why the JMH tests do not
reflect the performance we see in production.
Were there any other benefits (did it resolve the issue of the NPE in merge
as well ?)

Thanks
Neeraj

On Mon, Apr 6, 2015 at 1:04 PM, Neeraj Joshi njoshi@netflix.com wrote:

Hey David,
Batching is a good suggestion however that would be a breaking change

On Mon, Apr 6, 2015 at 1:02 PM, David Karnok notifications@github.com
wrote:

@davidmoten https://github.com/davidmoten Sure.

—
Reply to this email directly or view it on GitHub
#2857 (comment).

[neerajrj]

Ah ok, I see now, let me try out your suggestion. Thanks!

[akarnokd]

There are 2 PRs that did perf enhancements:

#2603, #2773

Benchmark                                   (size)     1.0.4      PR 2603    PR 2773
r.s.ComputationSchedulerPerf.observeOn           1  104110.926  115707.286  113905.358
r.s.ComputationSchedulerPerf.observeOn        1000    3212.434   13020.027   28618.423
r.s.ComputationSchedulerPerf.observeOn     1000000       9.508      16.559      32.166