`Sinks.many().replay().latest()` holds two values in cache

[MikkelHJuul]

The caching used for Sinks.many().replay().latest() holds both head and tail in the cache, although only one is available for replay (as it should). This means that the extra reference holds unnecessary memory, and holds onto a reference that cannot be reclaimed.

Expected Behavior

The cache holds only a single reference in memory.

Actual Behavior

The https://github.com/reactor/reactor-core/blob/main/reactor-core/src/main/java/reactor/core/publisher/FluxReplay.java#L774 SizeBoundReplayBuffer holds two items in cache.

Steps to Reproduce

package reactor.core.publisher;

@Test
void reproCase() {
     var replayBuffer = new FluxReplay.SizeBoundReplayBuffer<Integer>(1);
     replayBuffer.add(11);
     replayBuffer.add(22);
     assert replayBuffer.head != replayBuffer.tail;
     assert replayBuffer.head.value != replayBuffer.tail.value;
}

Possible Solution

package reactor.core.publisher;

@Test
void fixCase() {
     var replayBuffer = new FluxReplay.SizeBoundReplayBuffer<Integer>(0);
     replayBuffer.add(11);
     replayBuffer.add(22);
     assert replayBuffer.head == replayBuffer.tail;
     assert replayBuffer.head.value == replayBuffer.tail.value;
     assert replayBuffer.head.value == 22;
}

Your Environment

doesn't bloody matter

• Reactor version(s) used: 3.5.0

[MikkelHJuul]

I will just add that the possible solution stated here fixes the head != tail issue but the buffer will not emit its value, it looks like this is a "simple" off-by-one error. (edit: limit 2 yields 3 items referenced)

Consider making a singleton implementation of the buffer. And moving to using the Java's AtomicReferenceArray in stead of the linked list (Node extends AtomicReference<Node>). This should improve memory and performance.

[MikkelHJuul]

My least effort implementations of a SingletonReplayBuffer and an AtomiReferenceArray-based implementation

There is bound to be some issues with these, but they both passed a naive test:

    public static <T> Sinks.Many<T> replayLatestSink() {
        var sink = new SinkManyReplayProcessor<>(new AtomicArraySizeBoundReplayBuffer<T>(1));
        return sink; // and wrapped: new SinkManySerialized<>(sink, sink);
    }
    public static <T> Sinks.Many<T> replayLatestSinkSingleton() {
        var sink = new SinkManyReplayProcessor<>(new SingletonReplayBuffer<T>());
        return  sink; // and wrapped: new SinkManySerialized<>(sink, sink);
    }

    @Test
    void testFunctionalityOfSinksManyReplayLatest() {
        Sinks.Many<Integer> sink  = replayLatestSink(); // or replayLatestSingleton();
        sink.tryEmitNext(12);
        var firstResult = sink.asFlux().blockFirst();
        assert firstResult != null;
        assert firstResult == 12;
        var secondResult = sink.asFlux().blockFirst();
        assert firstResult.equals(secondResult);

        sink.tryEmitNext(21);
        firstResult = sink.asFlux().blockFirst();
        assert firstResult != null;
        assert firstResult == 21;
        secondResult = sink.asFlux().blockFirst();
        assert firstResult.equals(secondResult);
    }

sorry it's not in a PR form, I do this on my work PC and I cannot clone from github (private PC not set up for Java, nor IntelliJ)

    static final class SingletonReplayBuffer<T> implements FluxReplay.ReplayBuffer<T> {
        final AtomicReference<T> val = new AtomicReference<>(null);

        final int indexUpdateLimit = Operators.unboundedOrLimit(1);
        volatile boolean done;
        Throwable error;
        @Override
        public void add(T value) {
            val.set(value);
        }

        @Override
        public void onError(Throwable ex) {
            error = ex;
            done = true;
        }

        @Override
        public Throwable getError() {
            return error;
        }

        @Override
        public void onComplete() {
            done = true;
        }

        @Override
        public void replay(FluxReplay.ReplaySubscription<T> rs) {
            if (!rs.enter()) {
                return;
            }

            if (rs.fusionMode() == FluxReplay.NONE) {
                replayNormal(rs);
            }
            else {
                replayFused(rs);
            }
        }

        private void replayNormal(FluxReplay.ReplaySubscription<T> rs) {
            final Subscriber<? super T> a = rs.actual();

            int missed = 1;
            long r = rs.requested(); // can this be 0?
            long e = 0L;

            for (; ; ) {

                @SuppressWarnings("unchecked") T curr = (T) rs.node();
                T t = val.get();
                if (curr == null) {
                    curr = t;
                    rs.node(curr);
                }
                boolean d = done;
                boolean empty = t == null;

                if (d && empty) {
                    rs.node(null);
                    Throwable ex = error;
                    if (ex != null) {
                        a.onError(ex);
                    }
                    else {
                        a.onComplete();
                    }
                    return;
                }

                if (empty) {
                    continue;
                }

                if (rs.isCancelled()) {
                    rs.node(null);
                    return;
                }


                a.onNext(t);
                e++;

                if (r != Long.MAX_VALUE) {
                    rs.produced(1);
                }

                rs.node(t);

                missed = rs.leave(missed);
                if (missed == 0 || e == r) {
                    break;
                }
            }
        }

        private void replayFused(FluxReplay.ReplaySubscription<T> rs) {
            // I have no idea...
        }

        @Override
        public boolean isDone() {
            return done;
        }

        @Override
        public T poll(FluxReplay.ReplaySubscription<T> rs) {
            @SuppressWarnings("unchecked") T node = (T) rs.node();
            T next = val.get();
            if (node == null) {
                rs.node(next);
            }

//            if ((1) % indexUpdateLimit == 0) {
//                rs.requestMore(1);
//            } // this should be rethought for this implementation

            return next;
        }

        @Override
        public void clear(FluxReplay.ReplaySubscription<T> rs) {
            rs.node(null);
        }

        @Override
        public boolean isEmpty(FluxReplay.ReplaySubscription<T> rs) {
            @SuppressWarnings("unchecked") T node = (T) rs.node();
            if (node == null) {
                rs.node(val.get());
                return false;
                // doesn't this mean that the subscription is not up-to-date?
                // then the subscription can progress, why add the node, then?
            }
            return node == val.get();
        }

        @Override
        public int size(FluxReplay.ReplaySubscription<T> rs) {
            @SuppressWarnings("unchecked") T node = (T) rs.node();
            T t = val.get();
            if (node == null) {
                rs.node(t);
            }
            return sizeOf(t);
        }

        @Override
        public int size() {
            return  sizeOf(val.get());
        }

        private int sizeOf(@Nullable T t) {
            return t != null ? 1 : 0;
        }

        @Override
        public int capacity() {
            return 1;
        }

        @Override
        public boolean isExpired() {
            return false;
        }
    }

    static final class AtomicArraySizeBoundReplayBuffer<T> implements FluxReplay.ReplayBuffer<T> {

        final AtomicInteger head = new AtomicInteger();
        final int indexUpdateLimit;
        volatile boolean done;
        Throwable error;

        final AtomicReferenceArray<T> buffer;

        AtomicArraySizeBoundReplayBuffer(int limit) {
            if (limit < 0) {
                throw new IllegalArgumentException("Limit cannot be negative");
            }
            this.indexUpdateLimit = Operators.unboundedOrLimit(limit);

            this.buffer = new AtomicReferenceArray<>(limit);
        }

        @Override
        public boolean isExpired() {
            return false;
        }

        @Override
        public int capacity() {
            return buffer.length();
        }

        @Override
        public void add(T value) {
            buffer.set(
                    head.accumulateAndGet(capacity(), (curr, cap) -> {
                        if (curr + 1 == cap) return 0;
                        return curr + 1;
                    }),
                    value
            );
        }

        @Override
        public void onError(Throwable ex) {
            error = ex;
            done = true;
        }

        @Override
        public void onComplete() {
            done = true;
        }

        void replayNormal(FluxReplay.ReplaySubscription<T> rs) {
            final Subscriber<? super T> a = rs.actual();

            int missed = 1;

            for (; ; ) {

                long r = rs.requested();
                long e = 0L;

                Integer curr = (Integer) rs.node();
                if (curr == null) {
                    curr = head.get();
                    rs.node(curr);
                }
                int start = curr;

                while (e != r) {
                    if (rs.isCancelled()) {
                        rs.node(null);
                        return;
                    }

                    boolean d = done;
                    T next = buffer.get(curr);
                    boolean empty = next == null;

                    if (d && empty) { // if start is not 0 here that is a clear bug!
                        rs.node(null);
                        Throwable ex = error;
                        if (ex != null) {
                            a.onError(ex);
                        }
                        else {
                            a.onComplete();
                        }
                        return;
                    }

                    if (empty) {
                        break;
                    }

                    a.onNext(next);

                    e++;
                    if (curr == capacity() && start != 0) {
                        curr = 0;
                    } else {
                        curr++;
                    }

                    if (curr == start) break;

                    if ((curr + 1) % indexUpdateLimit == 0) {
                        rs.requestMore(curr + 1);
                    }
                }

                if (e != 0L) {
                    if (r != Long.MAX_VALUE) {
                        rs.produced(e);
                    }
                }

                rs.node(curr);

                missed = rs.leave(missed);
                if (missed == 0) {
                    break;
                }
            }
        }

        void replayFused(FluxReplay.ReplaySubscription<T> rs) {
            int missed = 1;

            final Subscriber<? super T> a = rs.actual();

            for (; ; ) {

                if (rs.isCancelled()) {
                    rs.node(null);
                    return;
                }

                boolean d = done;

                a.onNext(null);

                if (d) {
                    Throwable ex = error;
                    if (ex != null) {
                        a.onError(ex);
                    }
                    else {
                        a.onComplete();
                    }
                    return;
                }

                missed = rs.leave(missed);
                if (missed == 0) {
                    break;
                }
            }
        }

        @Override
        public void replay(FluxReplay.ReplaySubscription<T> rs) {
            if (!rs.enter()) {
                return;
            }

            if (rs.fusionMode() == FluxReplay.NONE) {
                replayNormal(rs);
            }
            else {
                replayFused(rs);
            }
        }

        @Override
        @Nullable
        public Throwable getError() {
            return error;
        }

        @Override
        public boolean isDone() {
            return done;
        }

        @Override
        @Nullable
        public T poll(FluxReplay.ReplaySubscription<T> rs) {
            Integer node = (Integer) rs.node();
            if (node == null) {
                node = head.get();
                rs.node(node);
            }

            T next = buffer.get(node);
            if (next == null) {
                return null;
            }
            rs.node(node+1);

            if ((node + 1) % indexUpdateLimit == 0) {
                rs.requestMore(node + 1);
            }

            return next;
        }

        @Override
        public void clear(FluxReplay.ReplaySubscription<T> rs) {
            //clear array?
            rs.node(null);
        }

        @Override
        public boolean isEmpty(FluxReplay.ReplaySubscription<T> rs) {
            Integer node = (Integer) rs.node();
            if (node == null) {
                node = head.get();
                rs.node(node);
            }
            return buffer.get(0) == null;
        }

        @Override
        public int size(FluxReplay.ReplaySubscription<T> rs) {
            Integer node = (Integer) rs.node();
            if (node == null) {
                node = head.get();
            }
            return sizeFrom(node);
        }

        @Override
        public int size() {
            return sizeFrom(head.get());
        }

        private int sizeFrom(int position) {
            if (
                    position+1 >= capacity()
                            || buffer.get(position+1) != null
            ) return capacity();
            return position;
        }
    }

There is bound to be some bugs here... but anyway the implementations are pretty simple. I didn't really understand how to use the rs.node() nor requestMore for the singleton implementation;

[MikkelHJuul]

fixed a bug in replay:

       void replayNormal(FluxReplay.ReplaySubscription<T> rs) {
            final Subscriber<? super T> a = rs.actual();

            int missed = 1;

            for (; ; ) {

                long r = rs.requested();
                long e = 0L;

                Integer curr = (Integer) rs.node();
                if (curr == null) {
                    curr = head.get();
                    rs.node(curr);
                }
                int last = curr;
                curr = nextOrReset(curr);

                while (e != r) {
                    if (rs.isCancelled()) {
                        rs.node(null);
                        return;
                    }

                    boolean d = done;
                    T next = buffer.get(curr);
                    boolean empty = next == null;

                    if (d && empty) { // if start is not 0 here that is a clear bug!
                        rs.node(null);
                        Throwable ex = error;
                        if (ex != null) {
                            a.onError(ex);
                        }
                        else {
                            a.onComplete();
                        }
                        return;
                    }

                    if (empty) {
                        break;
                    }

                    a.onNext(next);

                    e++;
                    if (curr == last) break;
                    curr = nextOrReset(curr);


                    if ((curr + 1) % indexUpdateLimit == 0) {
                        rs.requestMore(curr + 1);
                    }
                }

                if (e != 0L) {
                    if (r != Long.MAX_VALUE) {
                        rs.produced(e);
                    }
                }

                rs.node(curr);

                missed = rs.leave(missed);
                if (missed == 0) {
                    break;
                }
            }
        }

        private Integer nextOrReset(Integer curr) {
            if (curr + 1 == capacity()) {
                return 0;
            }
            return curr + 1;
        }

But there is some timing issue/race condition in the code... Replay is called twice, producing double responses. but stepping through it works... and some times I got 11 and 13 outputs in a buffer of size 10 (always in correct order), by adding a breakpoint... it may be easier to reimplement than to go with this implementation then :(

[OlegDokuka]

Hi, @MikkelHJuul!

Can you reproduce that issue via black-box test as well? Something like a test that demonstrates the operator fails on its own can be useful. Consider JCStress Test, etc

Thanks,
Oleh

[MikkelHJuul]

Hi, @MikkelHJuul!

Can you reproduce that issue via black-box test as well? Something like a test that demonstrates the operator fails on its own can be useful. Consider JCStress Test, etc

Thanks,

Oleh

In the implementation? Or the original issue? Because the original issue is not really something that is testable via the API, but I can tell give you some jmap histogram showing the 2:1 correlation. Again, I cannot really be of much help because I cannot share actual code with you guys because I don't have a dev PC where it is permitted.

[OlegDokuka]

@MikkelHJuul I'm not sure the problem is mitigatable since in all possible cases the previous value is referenced and being held longer than needed. This is "by design" of the current replay storage and I can only guess that this is the best effort, the most efficient implementation. I'm afraid we will not fix it

actually, RxJava has the same issue, so I wonder what @akarnokd thinks about it

[OlegDokuka]

@MikkelHJuul can you tell us how critical is this issue for you?

[MikkelHJuul]

@OlegDokuka - Well... our use of this is kinda spurious as well.. holding all cache objects in a Sinks.many().replay().latest() in stead of only the data that is subscribed downstream. also eagerly caching too much. but it is the single largest shallow memory usage apart from String and byte[] in our current use.

for a possible fix of the above add eager tests for equality:

                Integer curr = (Integer) rs.node();
                if (curr == null) {
                    curr = head.get();
                    rs.node(curr);
                }

replaced by

                Integer curr = (Integer) rs.node();
                Integer hd = head.get();
                if (hd.equals(curr)) {
                    break;
                }
                if (curr == null) {
                    curr = hd;
                    rs.node(curr);
                }

This I am mostly afraid of a subscriber that dissallows #enter until it is incidentally at the same head again making it miss the limited offset.. so the Integer that marks the ringbuffer head should increment, and mark the modulated position in stead.

and for the Singleton; add a simple

                 if (curr == t) {
                    break;
                }

add the similar point for the Singletons implementation

It seemed to fix all the issues, but you will have to test yourself.

I'm not entirely sure that I bite into the we save one dangling AtomicReference too many "by design". I understand that the implementation works wrt. your tests, that is not the same as it being the most correct implementation.

(we had some actual memory leaks (growing) which is also probably the only reason that I found this at all)

I may try to get a developer environment running on my personal PC so I can make an actual PR if you are not interested in taking over.

[OlegDokuka]

Hi, @MikkelHJuul!

Thank you for the open response with deeper explanation of the problem.

First of all, any contribution is appreciated, and it is not equal to 'we dont wanna mess with that' 😅

What i mean is that the atomic replay store is singular case, while the rest of scenarios (e.g. cache(x)) are still having the same issue.

That means although your case is covered the others arent.

From, what you say it sounds you care alot about performance, so I start wondering whether you can replace sink API with something different to unblock you for now. 🧐

Nevertheless, I want to do a bit deeper research first to see if we can solve the general scenario as well, and weight your proposal as an good alternative. 🤓

That say, dont hesitate to send your PR while we going to spawn own research in parallel 🙂