Unbounded FlatMap with slow consumer

[akarnokd]

The unbounded flatMap has terrible performance if a consumer, such as observeOn processes slower.

The main reasons:

• copy-on-write subscriber tracking, which will create N arrays of ever increasing size with O(n * n) complexity and then it takes O(n * n) to remove them one-by-one
• the cleanup loop runs over all n sources in case requested is zero to evict empty and completed sources, adding another O(n * n) complexity.

For example, this test runs in about 30s:

range(1, 100_000).flatMap(v -> range(1, 10)).subscribe(new TestSubscriber<>(0L));

(then it takes 12s to drain by requesting 96 elements in a loop, see PublisherFlatMapTest.slowDrain().)

The cleanup can be short circuited because if one finds a non-empty source, there is no need to continue removing any potential finished and empty sources after that, the request-based drain will take care of those entries eventually.

The tracking overhead can be reduced via a free slot queue and power-of-2 allocations, but with some cost because of synchronized and no compaction as of now.

The benchmark results (i7 4790, Windows 7 x64, Java 8u72):

Where COW is the original copy-on-write tracking, FL is a free-list based solution which already shows some promise; COW-2 is COW where the cleanup is short-circuited and FL-2 is FL with the cleanup short circuited as well.

FL-2 seems to be promising with a bunch of +/- 5% loss/gain, probably due to noise.

I'm looking into non-synchronized options with the FL solution.

[akarnokd]

The FL-2 seems to work well.