Concat array enhancement

[danielkec]

• More effective cancellation handling
• Better handling of negative requests

JMH test results:

Benchmark                                     Mode  Cnt        Score        Error  Units
ConcatArrayComparison.reqMaxNewCA            thrpt   10   742789.861 ±  25826.167  ops/s
ConcatArrayComparison.reqMaxOldCA            thrpt   10   716121.440 ±  28539.035  ops/s
ConcatArrayComparison.variableRequestsNewCA  thrpt   10  6538445.756 ± 629438.611  ops/s
ConcatArrayComparison.variableRequestsOldCA  thrpt   10  5267535.840 ± 650941.922  ops/s

[akarnokd]

The code looks quite unconventional to me and the state transitions are somewhat hard to follow. In general, concat can break when there is a switchover from source to source at the same time the downstream requests thus the outstanding request amount is not properly forwarded to them.

Here is a test that passes on master with the SubscriptionArbiter but fails on this implementation because some request(1)s are not handed over properly:

    static void race(Runnable r1, Runnable r2, ExecutorService exec) {

        AtomicInteger sync = new AtomicInteger(2);
        CountDownLatch cdl = new CountDownLatch(1);

        exec.submit(() -> {
            if (sync.decrementAndGet() != 0) {
                while (sync.get() != 0) ;
            }

            try {
                r1.run();
            } finally {
                cdl.countDown();
            }
        });

        if (sync.decrementAndGet() != 0) {
            while (sync.get() != 0) ;
        }
        r2.run();

        try {
            Assert.assertTrue(cdl.await(5, TimeUnit.SECONDS));
        } catch (InterruptedException ex) {
            Assert.fail("", ex);
        }
    }


    @Test
    public void switchOverRequestRace() {
        var exec = Executors.newSingleThreadExecutor();
        try {
            for (int i = 0; i < 10_000; i++) {
                System.out.printf("Round %d%n", i + 1);
                TestSubscriber<Object> ts = new TestSubscriber<>(0L);
                AtomicReference<Flow.Subscriber<? super Integer>> sref = new AtomicReference<>();

                Multi.concatArray(new Flow.Publisher<Integer>() {

                    @Override
                    public void subscribe(Flow.Subscriber<? super Integer> subscriber) {
                        subscriber.onSubscribe(EmptySubscription.INSTANCE);
                        sref.set(subscriber);
                    }
                }, Multi.range(1, 100)).subscribe(ts);

                race(() -> {
                    sref.get().onComplete();
                }, () -> {
                    var s = ts.getSubcription();
                    for (int j = 0; j < 100; j++) {
                        s.request(1);
                    }
                }, exec);

                ts.awaitDone(5, TimeUnit.SECONDS);
                ts.assertItemCount(100);
            }
        } finally {
            exec.shutdown();
        }
    }

[olotenko]

Thanks.

The problem with the original code is the lack of atomicity in state transitions around the same place, which can also be shown to lead to problems (not conformant to the specification). Consequently, an attempt to capture all state transitions in a single atomic counter.

[olotenko]

Two things need doing:
long oldProduced = produced; on the line before change of REQUESTED, then use oldProduced to the end of this method.

This addresses the case raised by @akarnok - concurrent request results in concurrent onNext, which updates produced.

and

REQUESTED.setVolatile(this, p0) to replace setOpaque - opaque doesn't guarantee ordering with other stores.

[olotenko]

The code looks quite unconventional to me and the state transitions are somewhat hard to follow. In general, concat can break when there is a switchover from source to source at the same time the downstream requests thus the outstanding request amount is not properly forwarded to them.

Here is a test that passes on master with the SubscriptionArbiter but fails on this implementation because some request(1)s are not handed over properly:

I even remember considering the race condition this test shows.

It really only shows Multi.range does not conform to the spec - it is able to deliver onNext during onSubscribe. The spec requires these signals to be delivered serially, and there is a TCK capturing this condition (which fails for several Publishers).

https://github.com/reactive-streams/reactive-streams-jvm/blob/006202eb877c54cf64b258a8a78328e4fb38cab6/tck/src/main/java/org/reactivestreams/tck/PublisherVerification.java#L244 - this is the test that Multi.range will fail. It looks specifically for the occurrences of onNext executed concurrently with onSubscribe.

[akarnokd]

103 tests for concurrent access violation which range does not do. Also §2.1 and §3.2 cover reentrance and §3.3 covers recursion which range conforms as well.

Of course, you can ensure no request is delivered until the subscriber.onSubscribe invocation has returned via the deferred subscription approach. However, that incurs extra allocation and full-barrier atomics thus I'd expect no clear performance benefit over the current implementation.

[olotenko]

That is not the proof of the absence of the race in range. The proof of the existence of the race is the existence of this failing condition, which can be shown to occur because onNext is delivered concurrently with onSubscribe by range.

Even without such a test it is easy to see from the code of the range that it does not deliver signals in a total order for all the signals.

We have a prototype that works around this issue with spec conformance of the range, @danielkec is preparing a merge.

[olotenko]

Reading this again, strictly speaking this fence is not needed, because we have stronger fences between the stores targeted by this fence. But no need to change the code as the cost of this fence on the target platforms is zero.

[olotenko]

Appears to have only cosmetic changes compared to the internal proposal.