0.19 - Potential Leak in CompositeSubscription with Executor?

[daschl]

Hey folks,

doing performance tests with 0.19 did bring up something interesting - check this out on a 4gig heap:

96% of my heap are taken up by the CompositeSubscription.

I'm running a workload like this, that is fetching docs out of the server (but here for example without any docs, so they are not found - to not mess with object decoding in the benchmark.)

        List<GetRequest> keys = new ArrayList<GetRequest>(1024);
        for (int i = 0; i < 1024; i++) {
            keys.add(new GetRequest("key" + i, bucket));
        }

        while(true) {
            final CountDownLatch latch = new CountDownLatch(1);
            Observable.from(keys).flatMap(new Func1<GetRequest, Observable<?>>() {
                @Override
                public Observable<GetResponse> call(GetRequest s) {
                    return cluster.send(s);
                }
            }).toList().doOnCompleted(new Action0() {
                @Override
                public void call() {
                    latch.countDown();
                }
            }).subscribe();
            try {
                latch.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

Looking into the CompositeSubscription the dump makes sense since it uses a HashSet which is internally backed by a HashMap.

Another hint might be that they seem to be hanging around in the executor?

Not sure if there is a leak somewhere (I'm assuming, since 95% are reachable in the heap dump!), but it is causing unbearable pressure on the test system. In case you wonder about the workload, its around 100k ops/s until it goes nuts since GC takes up all time.

[daschl]

pinging @akarnokd and @benjchristensen. let me know if you need more info from that heap dump, it's quite large. That thing prevents me from testing the new 0.19 perf improvements since it dominates it all :)

[daschl]

Hmm it could be me scheduling too much stuff, but the hanging around so long doesn't make sense to me. I'll do further investigations and then reopen if I have news.

[daschl]

Okay, reopened this because I think I'm either misusing it completely or there is indeed something wrong with freeing stuff.

Let me give you some context: I'm writing those requests into a netty channel, but if channel.isWritable() returns false (indicating a overload on the writer side), I push it back into a response RingBuffer.

Now in my handler there I basically say "since I got a request in there, I need to reschedule it so that it will be retried, but lets do it with a backoff since doing it right now might now have any value."

So what I'm doing is:

        Schedulers.computation().createWorker().schedule(new Action0() {
            @Override
            public void call() {
                cluster.send(request);
            }
        }, 10, TimeUnit.MILLISECONDS);

Which means after 10 MS push the request again into the request RingBuffer so it will go into the request path again.

I think this is where all those objects come from, but shouldn't they be cleaned up at some point? Or do I need to do something special to indicate they can be freed? The send method returns immediately.

Any idea on how to approach this differently if I'm misusing the system?

[akarnokd]

You create a worker for each reschedule action which creates a lot of subscriptions to track stuff. If you don't do any other computation there, you could reuse an existing worker, maybe from Schedulers.newThread() so it doesn't disrupt the computation threads that much.

[daschl]

@akarnokd thanks, how do you mean? If I change the code to Schedulers.newThread().createWorker() I immediately run out of memory since it cant create more threads.

[akarnokd]

Cache the worker returned from Schedulers.newThread().createWorker() and use that.

[daschl]

@akarnokd I get more or less the same heap dump, too much subscriptions and hash sets created. Maybe I need to use my own HashedWheelTimer or so to reschedule those ops? Looks like the overhead to track this stuff in RxJava is too much?

[daschl]

@akarnokd but on the other hand, why are they sticking around for so long? If you look at my test case, I wait until the 1024 return so they went through correctly (which means they should be removed from the worker already?) They are still keeping around

[akarnokd]

I need a heapdump for further investigation. Could you create a smaller program I could run on my own accompanied by a dump?

[daschl]

@akarnokd I'll try to come up with something self-contained but it might take a while. I'll ping you once I have it.

[akarnokd]

Okay.

One drawback with this HashSet-based composite is that the HashSet grows and never shrinks. When I was researching #1145, many sources indicated that there is no good way to establish a shrink policy on java Collection classes.

However, we could introduce a simple policy in which the HashSet is recreated with default capacity if the composite reaches 0 size. Unfortunately, HashMap doesn't expose the underlying table's capacity so we could avoid constant recreation in case the size changes back and forth between 0 and some small value.

[benjchristensen]

That code also is never cleaning up the Worker that is created.

Here is a simple example to show how the Worker needs to be registered for cleanup: https://github.com/Netflix/RxJava/blob/master/rxjava-core/src/main/java/rx/internal/operators/OperatorRepeat.java#L76