3.0.3 - Hangs when using combination of flatMap and publish

[sobersanta]

Hi.

After few days of desperate attempts to find the cause of my servers' spontaneous hangs, I managed to reproduce the issue, but still don't have a clear idea about the reason, but I think I have identified a possible direction.

All problems has started when I started using very promising operator groupBy with evicting map provider possibility:

<K, V> Flowable<GroupedFlowable<K, V>> groupBy(@NonNull Function<? super T, ? extends K> keySelector,
            @NonNull Function<? super T, ? extends V> valueSelector,
            boolean delayError, int bufferSize,
            @NonNull Function<? super Consumer<Object>, ? extends Map<K, Object>> evictingMapFactory)

First I realised that it doesn't work well with Caffeine cache, as it performs evictions on a separate thread pool, and with async eviction notification groupBy goes crazy.

After moving to Guava things got better but only to some extent.

The tests I want to demonstrate I believe reproduce the issue I have. Definitely chain of operators is way more complicated in the production code, but here with certain variant of .publish() operator it's reproducible as well.

Basically, I'm not sure if there is issue in the groupBy(...evictingMapFactory) or in the publish(Function...), or in both.

Also tests are demonstrating that eviction policy doesn't work very well when crossing async boundaries downstream.

Please take a look at the attached file, it contains few simple tests, I kindly ask to check carefully all parameters, as it's important.

FlowableOperators.java.zip

[akarnokd]

Hi and thanks for reporting this bug.

The problem is caused by the eviction double decrementing the group counter, once when the eviction completes the shared chain that cancels the group and once after the eviction.

I'll post a fix shortly.

[sobersanta]

Cool, looking forward for the fix!

[akarnokd]

/cc @davidmoten regarding the async eviction concerns.

[sobersanta]

Thanks for the prompt reply and the fix!
May I ask, when we can expect to have it to be released?
Maybe 3.0.4 soon? ;)

[akarnokd]

I suggest you verify it by testing with the current RxJava snapshot, just in case: https://github.com/ReactiveX/RxJava#snapshots

[sobersanta]

Sure, I will let you know about results.

[sobersanta]

Could you please comment on this?

    public static void main(String[] args) {

        final int groups = 20;

        // Not completed (Timed out), buffer is too small
        int groupByBufferSize = groups * 2;
        int flatMapMaxConcurrency = 2 * groups;
        boolean notifyOnExplicitEviction = false;
        test2(groupByBufferSize, flatMapMaxConcurrency, groups, notifyOnExplicitEviction);

        // Success, buffer is large enough
        groupByBufferSize = groups * 30;
        flatMapMaxConcurrency = 2 * groups;
        test2(groupByBufferSize, flatMapMaxConcurrency, groups, notifyOnExplicitEviction);

        // Timeout... explicit eviction notification makes difference
        groupByBufferSize = groups * 30;
        flatMapMaxConcurrency = 2 * groups;
        notifyOnExplicitEviction = true;
        test2(groupByBufferSize, flatMapMaxConcurrency, groups, notifyOnExplicitEviction);

        // MissingBackpressureException, flatMap concurrency == groups count is not enough, must be at least 2x
        groupByBufferSize = groups * 20;
        flatMapMaxConcurrency = groups;
        notifyOnExplicitEviction = false;
        test2(groupByBufferSize, flatMapMaxConcurrency, groups, notifyOnExplicitEviction);
    }

    public static void test2(int groupByBufferSize, int flatMapMaxConcurrency, int groups,
                             boolean notifyOnExplicitEviction) {
        try {
            TestSubscriber<Integer> test = Flowable
                .range(1, 500_000)
                .map(i -> i % groups)
                .groupBy(i -> i, i -> i, false, groupByBufferSize,
                         // set cap too high
                         sizeCap(groups * 100, notifyOnExplicitEviction))
                .flatMap(gf -> gf
                             .take(10, TimeUnit.MILLISECONDS)
                    , flatMapMaxConcurrency)
                .test()
                .awaitDone(5, TimeUnit.SECONDS)
                .assertComplete();
            System.out.println("Success");
        } catch (Throwable e) {
            e.printStackTrace();
        }
    }

    public static <T> Function<Consumer<Object>, Map<T, Object>> sizeCap(int maxCapacity, boolean notifyOnExplicit) {
        return itemEvictConsumer ->
            CacheBuilder
                .newBuilder()
                .maximumSize(maxCapacity)
                .removalListener(n -> {
                    if (n.getCause() != RemovalCause.EXPLICIT || notifyOnExplicit) {
                        try {
                            itemEvictConsumer.accept(n.getValue());
                        } catch (Throwable throwable) {
                            throw new RuntimeException(throwable);
                        }
                    }
                })
                .<T, Object>build().asMap();
    }

[sobersanta]

changing .take(10, MILLISECONDS) to .take(10) makes only 3d case failing with timeout

[akarnokd]

Is this with the snapshot?