Complete removal of buffers, miscounting BaseProcessor fix

Signed-off-by: Daniel Kec <[email protected]>

Author: danielkec

common/reactive/src/main/java/io/helidon/common/reactive/BaseProcessor.java

@@ -31,32 +31,47 @@
  * @param <T> subscribed type (input)
  * @param <U> published type (output)
  */
-public abstract class BaseProcessor<T, U> implements Processor<T, U>, Subscription {
+public abstract class BaseProcessor<T, U> implements Processor<T, U>, Subscription, StrictProcessor {
 
     private Subscription subscription;
     private final SingleSubscriberHolder<U> subscriber;
     private final RequestedCounter requested;
+    private final RequestedCounter ableToSubmit;
+    private final ReentrantLock subscriptionLock = new ReentrantLock();
     private final AtomicBoolean ready;
     private final AtomicBoolean subscribed;
-    private ReentrantLock publisherSequentialLock = new ReentrantLock();
     private volatile boolean done;
     private Throwable error;
+    private boolean strictMode = DEFAULT_STRICT_MODE;
+
 
     /**
      * Generic processor used for the implementation of {@link Multi} and {@link Single}.
      */
     protected BaseProcessor() {
-        requested = new RequestedCounter();
+        requested = new RequestedCounter(strictMode);
+        ableToSubmit = new RequestedCounter(strictMode);
         ready = new AtomicBoolean();
         subscribed = new AtomicBoolean();
         subscriber = new SingleSubscriberHolder<>();
     }
 
+    @Override
+    public BaseProcessor<T, U> strictMode(boolean strictMode) {
+        this.strictMode = strictMode;
+        return this;
+    }
+
     @Override
     public void request(long n) {
-        StreamValidationUtils.checkRequestParam(n, this::failAndCancel);
-        requested.increment(n, this::failAndCancel);
-        tryRequest(subscription);
+        ableToSubmit.increment(n, this::failAndCancel);
+        subscriptionLock(() -> {
+            if (subscription != null && !subscriber.isClosed()) {
+                subscription.request(n);
+            } else {
+                requested.increment(n, this::failAndCancel);
+            }
+        });
         if (done) {
             tryComplete();
         }
@@ -74,39 +89,27 @@ public void cancel() {
 
     @Override
     public void onSubscribe(Subscription s) {
-        try {
-            // https://github.com/reactive-streams/reactive-streams-jvm#1.3
-            publisherSequentialLock.lock();
-            // https://github.com/reactive-streams/reactive-streams-jvm#2.13
-            Objects.requireNonNull(s);
+        Objects.requireNonNull(s);
+        subscriptionLock(() -> {
             if (subscription == null) {
                 this.subscription = s;
                 tryRequest(s);
             } else {
-                // https://github.com/reactive-streams/reactive-streams-jvm#2.5
                 s.cancel();
             }
-        } finally {
-            publisherSequentialLock.unlock();
-        }
+        });
     }
 
     @Override
     public void onNext(T item) {
+        if (done) {
+            throw new IllegalStateException("Subscriber is closed!");
+        }
+        Objects.requireNonNull(item);
         try {
-            publisherSequentialLock.lock();
-            if (done) {
-                throw new IllegalStateException("Subscriber is closed!");
-            }
-            // https://github.com/reactive-streams/reactive-streams-jvm#2.13
-            Objects.requireNonNull(item);
-            try {
-                hookOnNext(item);
-            } catch (Throwable ex) {
-                failAndCancel(ex);
-            }
-        } finally {
-            publisherSequentialLock.unlock();
+            hookOnNext(item);
+        } catch (Throwable ex) {
+            failAndCancel(ex);
         }
     }
 
@@ -116,7 +119,6 @@ public void onNext(T item) {
      * @param ex Exception to be reported downstream
      */
     protected void fail(Throwable ex) {
-        // https://github.com/reactive-streams/reactive-streams-jvm#2.13
         Objects.requireNonNull(ex);
         done = true;
         if (error == null) {
@@ -147,19 +149,20 @@ public void onComplete() {
     }
 
     @Override
-    public void subscribe(Subscriber<? super U> s) {
-        try {
-            publisherSequentialLock.lock();
+    public void subscribe(Subscriber<? super U> sub) {
+        subscriptionLock(() -> {
+            var s = sub;
+            if (strictMode) {
+                s = SequentialSubscriber.create(sub);
+            }
             if (subscriber.register(s)) {
                 ready.set(true);
                 s.onSubscribe(this);
                 if (done) {
                     tryComplete();
                 }
             }
-        } finally {
-            publisherSequentialLock.unlock();
-        }
+        });
     }
 
     /**
@@ -172,31 +175,13 @@ protected Optional<Subscription> getSubscription() {
         return Optional.ofNullable(subscription);
     }
 
-    /**
-     * Processor's {@link SingleSubscriberHolder}.
-     *
-     * @return {@link SingleSubscriberHolder}
-     */
-    protected SingleSubscriberHolder<U> getSubscriber() {
-        return subscriber;
-    }
-
-    /**
-     * Returns {@link RequestedCounter} with information about requested vs. submitted items.
-     *
-     * @return {@link RequestedCounter}
-     */
-    protected RequestedCounter getRequestedCounter() {
-        return requested;
-    }
-
     /**
      * Submit an item to the subscriber.
      *
      * @param item item to be submitted
      */
     protected void submit(U item) {
-        if (requested.tryDecrement()) {
+        if (ableToSubmit.tryDecrement()) {
             try {
                 subscriber.get().onNext(item);
             } catch (InterruptedException ex) {
@@ -257,10 +242,10 @@ protected void hookOnCancel(Flow.Subscription subscription) {
      */
     protected final void doSubscribe(Publisher<U> delegate) {
         if (subscribed.compareAndSet(false, true)) {
-            delegate.subscribe(new Subscriber<U>() {
+            delegate.subscribe(new Subscriber<>() {
                 @Override
                 public void onSubscribe(Subscription subscription) {
-                    tryRequest(subscription);
+                    tryRequest(ableToSubmit, subscription);
                 }
 
                 @Override
@@ -311,11 +296,37 @@ protected void tryComplete() {
      * @param subscription {@link Flow.Subscription} to make a request from
      */
     protected void tryRequest(Subscription subscription) {
-        if (subscription != null && !subscriber.isClosed()) {
-            long n = requested.get();
-            if (n > 0) {
-                subscription.request(n);
+        tryRequest(requested, subscription);
+    }
+
+    private void tryRequest(RequestedCounter counter, Subscription subscription) {
+        if (subscription == null || subscriber.isClosed()) {
+            return;
+        }
+
+        long n;
+        try {
+            counter.lock();
+            n = counter.get();
+            if (n < 1) {
+                return;
             }
+            subscription.request(n);
+        } finally {
+            counter.unlock();
+        }
+    }
+
+    private void subscriptionLock(Runnable guardedBlock) {
+        if (!strictMode) {
+            guardedBlock.run();
+            return;
+        }
+        try {
+            subscriptionLock.lock();
+            guardedBlock.run();
+        } finally {
+            subscriptionLock.unlock();
         }
     }
 }

common/reactive/src/main/java/io/helidon/common/reactive/MultiFilterProcessor.java

@@ -48,7 +48,7 @@ protected void hookOnNext(T item) {
         if (predicate.test(item)) {
             submit(item);
         } else {
-            getRequestedCounter().increment(1, this::fail);
+            tryRequest(getSubscription().get());
         }
     }
 }

common/reactive/src/main/java/io/helidon/common/reactive/MultiFlatMapProcessor.java

@@ -30,17 +30,18 @@
  * @param <T> input item type
  * @param <X> output item type
  */
-public class MultiFlatMapProcessor<T, X> implements Flow.Processor<T, X>, Multi<X> {
+public class MultiFlatMapProcessor<T, X> implements Flow.Processor<T, X>, Multi<X>, StrictProcessor {
 
     private Function<T, Flow.Publisher<X>> mapper;
     private RequestedCounter requestCounter = new RequestedCounter();
-    private SequentialSubscriber<? super X> subscriber;
+    private Flow.Subscriber<? super X> subscriber;
     private Flow.Subscription subscription;
     private volatile Flow.Subscription innerSubscription;
     private volatile Flow.Publisher<X> innerPublisher;
-    private volatile boolean upstreamsCompleted;
+    private AtomicBoolean upstreamsCompleted = new AtomicBoolean(false);
     private Optional<Throwable> error = Optional.empty();
     private ReentrantLock stateLock = new ReentrantLock();
+    private boolean strictMode = BaseProcessor.DEFAULT_STRICT_MODE;
 
 
     private MultiFlatMapProcessor() {
@@ -76,10 +77,16 @@ public static <T, U> MultiFlatMapProcessor<T, U> fromPublisherMapper(Function<T,
         return flatMapProcessor;
     }
 
+    @Override
+    public MultiFlatMapProcessor<T, X> strictMode(boolean strictMode) {
+        this.strictMode = strictMode;
+        return this;
+    }
+
     private class FlatMapSubscription implements Flow.Subscription {
         @Override
         public void request(long n) {
-            stateLock(() -> requestCounter.increment(n, MultiFlatMapProcessor.this::onError));
+            requestCounter.increment(n, MultiFlatMapProcessor.this::onError);
             if (Objects.nonNull(innerSubscription)) {
                 innerSubscription.request(n);
             } else {
@@ -96,8 +103,8 @@ public void cancel() {
 
     @Override
     public void subscribe(Flow.Subscriber<? super X> subscriber) {
+        this.subscriber = SequentialSubscriber.create(subscriber);
         stateLock(() -> {
-            this.subscriber = SequentialSubscriber.create(subscriber);
             if (Objects.nonNull(this.subscription)) {
                 this.subscriber.onSubscribe(new FlatMapSubscription());
             }
@@ -141,8 +148,8 @@ public void onError(Throwable t) {
 
     @Override
     public void onComplete() {
+        upstreamsCompleted.set(true);
         stateLock(() -> {
-            upstreamsCompleted = true;
             if (requestCounter.get() == 0 || Objects.isNull(innerPublisher)) {
                 //Have to wait for all Publishers to be finished
                 subscriber.onComplete();
@@ -166,11 +173,9 @@ public void onSubscribe(Flow.Subscription subscription) {
         @Override
         public void onNext(X item) {
             Objects.requireNonNull(item);
-            stateLock(() -> {
-                if (requestCounter.tryDecrement()) {
-                    subscriber.onNext(item);
-                }
-            });
+            if (requestCounter.tryDecrement()) {
+                subscriber.onNext(item);
+            }
         }
 
         @Override
@@ -182,22 +187,31 @@ public void onError(Throwable throwable) {
 
         @Override
         public void onComplete() {
+            if (upstreamsCompleted.get()) {
+                subscriber.onComplete();
+            }
             stateLock(() -> {
                 innerPublisher = null;
-                if (upstreamsCompleted) {
-                    subscriber.onComplete();
-                }
+            });
+            try {
+                requestCounter.lock();
                 if (requestCounter.get() > 0) {
                     subscription.request(1);
                 }
-            });
+            } finally {
+                requestCounter.unlock();
+            }
         }
     }
 
     /**
      * Protect critical sections when working with states of innerPublisher.
      */
     private void stateLock(Runnable runnable) {
+        if (!strictMode) {
+            runnable.run();
+            return;
+        }
         try {
             stateLock.lock();
             runnable.run();