fixes with UnboundedProcessor

Signed-off-by: Oleh Dokuka <[email protected]>
Signed-off-by: Oleh Dokuka <[email protected]>
Signed-off-by: OlegDokuka <[email protected]>

Author: OlegDokuka

build.gradle

@@ -33,7 +33,7 @@ subprojects {
     apply plugin: 'com.github.sherter.google-java-format'
     apply plugin: 'com.github.vlsi.gradle-extensions'
 
-    ext['reactor-bom.version'] = '2020.0.25-SNAPSHOT'
+    ext['reactor-bom.version'] = '2020.0.25'
     ext['logback.version'] = '1.2.10'
     ext['netty-bom.version'] = '4.1.85.Final'
     ext['netty-boringssl.version'] = '2.0.54.Final'

gradle/wrapper/gradle-wrapper.properties

@@ -1,5 +1,5 @@
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-7.3-bin.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-7.5.1-bin.zip
 zipStoreBase=GRADLE_USER_HOME
 zipStorePath=wrapper/dists

rsocket-core/src/jcstress/java/io/rsocket/internal/UnboundedProcessorStressTest.java

@@ -895,6 +895,70 @@ public void arbiter(LLL_Result r) {
     }
   }
 
+  @JCStressTest
+  @Outcome(
+      id = {
+        "0, 1, 0, 5",
+        "1, 1, 0, 5",
+        "2, 1, 0, 5",
+        "3, 1, 0, 5",
+        "4, 1, 0, 5",
+        "5, 1, 0, 5",
+      },
+      expect = Expect.ACCEPTABLE,
+      desc = "onComplete()")
+  @State
+  public static class Smoke33StressTest extends UnboundedProcessorStressTest {
+
+    final StressSubscriber<ByteBuf> stressSubscriber =
+        new StressSubscriber<>(Long.MAX_VALUE, Fuseable.NONE);
+    final ByteBuf byteBuf1 = UnpooledByteBufAllocator.DEFAULT.buffer().writeByte(1);
+    final ByteBuf byteBuf2 = UnpooledByteBufAllocator.DEFAULT.buffer().writeByte(2);
+    final ByteBuf byteBuf3 = UnpooledByteBufAllocator.DEFAULT.buffer().writeByte(3);
+    final ByteBuf byteBuf4 = UnpooledByteBufAllocator.DEFAULT.buffer().writeByte(4);
+    final ByteBuf byteBuf5 = UnpooledByteBufAllocator.DEFAULT.buffer().writeByte(5);
+
+    {
+      unboundedProcessor.subscribe(stressSubscriber);
+    }
+
+    @Actor
+    public void next1() {
+      unboundedProcessor.tryEmitNormal(byteBuf1);
+      unboundedProcessor.tryEmitPrioritized(byteBuf2);
+    }
+
+    @Actor
+    public void next2() {
+      unboundedProcessor.tryEmitPrioritized(byteBuf3);
+      unboundedProcessor.tryEmitNormal(byteBuf4);
+    }
+
+    @Actor
+    public void complete() {
+      unboundedProcessor.tryEmitFinal(byteBuf5);
+    }
+
+    @Arbiter
+    public void arbiter(LLLL_Result r) {
+      r.r1 = stressSubscriber.onNextCalls;
+      r.r2 =
+          stressSubscriber.onCompleteCalls
+              + stressSubscriber.onErrorCalls * 2
+              + stressSubscriber.droppedErrors.size() * 3;
+
+      r.r4 = stressSubscriber.values.get(stressSubscriber.values.size() - 1).readByte();
+      stressSubscriber.values.forEach(ByteBuf::release);
+
+      r.r3 =
+          byteBuf1.refCnt()
+              + byteBuf2.refCnt()
+              + byteBuf3.refCnt()
+              + byteBuf4.refCnt()
+              + byteBuf5.refCnt();
+    }
+  }
+
   @JCStressTest
   @Outcome(
       id = {

rsocket-core/src/main/java/io/rsocket/internal/BaseDuplexConnection.java

@@ -24,16 +24,22 @@
 public abstract class BaseDuplexConnection implements DuplexConnection {
 
   protected final Sinks.Empty<Void> onClose = Sinks.empty();
-  protected final UnboundedProcessor sender = new UnboundedProcessor(onClose::tryEmitEmpty);
+  protected final UnboundedProcessor sender =
+      new UnboundedProcessor(
+          () -> {
+            onClose.tryEmitEmpty();
+            dispose();
+          },
+          (__) -> {});
 
   public BaseDuplexConnection() {}
 
   @Override
   public void sendFrame(int streamId, ByteBuf frame) {
     if (streamId == 0) {
-      sender.onNextPrioritized(frame);
+      sender.tryEmitPrioritized(frame);
     } else {
-      sender.onNext(frame);
+      sender.tryEmitNormal(frame);
     }
   }
 

rsocket-core/src/main/java/io/rsocket/internal/UnboundedProcessor.java

@@ -24,9 +24,8 @@
 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
 import java.util.concurrent.atomic.AtomicLongFieldUpdater;
 import java.util.function.Consumer;
-import java.util.function.Function;
 import java.util.stream.Stream;
-import org.reactivestreams.Suwipbscription;
+import org.reactivestreams.Subscription;
 import reactor.core.CoreSubscriber;
 import reactor.core.Disposable;
 import reactor.core.Exceptions;
@@ -93,10 +92,12 @@ public final class UnboundedProcessor extends Flux<ByteBuf>
   static final AtomicLongFieldUpdater<UnboundedProcessor> REQUESTED =
       AtomicLongFieldUpdater.newUpdater(UnboundedProcessor.class, "requested");
 
+  ByteBuf last;
+
   boolean outputFused;
 
   public UnboundedProcessor() {
-    this(() -> {});
+    this(() -> {}, __ -> {});
   }
 
   public UnboundedProcessor(Runnable onFinalizedHook, Consumer<ByteBuf> onValueDelivered) {
@@ -123,79 +124,122 @@ public Object scanUnsafe(Attr key) {
     return null;
   }
 
-  public boolean tryEmitNext
-
-  @Deprecated
-  public void onNextPrioritized(ByteBuf t) {
+  public boolean tryEmitPrioritized(ByteBuf t) {
     if (this.done || this.cancelled) {
       release(t);
-      return;
+      return false;
     }
 
     if (!this.priorityQueue.offer(t)) {
       onError(Operators.onOperatorError(null, Exceptions.failWithOverflow(), t, currentContext()));
       release(t);
-      return;
+      return false;
     }
 
     final long previousState = markValueAdded(this);
     if (isFinalized(previousState)) {
       this.clearSafely();
-      return;
+      return false;
     }
 
     if (isSubscriberReady(previousState)) {
       if (this.outputFused) {
         // fast path for fusion
         this.actual.onNext(null);
-        return;
+        return true;
       }
 
       if (isWorkInProgress(previousState)) {
-        return;
+        return true;
       }
 
       if (hasRequest(previousState)) {
         drainRegular(previousState);
       }
     }
+    return true;
   }
 
-  @Override
-  @Deprecated
-  public void onNext(ByteBuf t) {
+  public boolean tryEmitNormal(ByteBuf t) {
     if (this.done || this.cancelled) {
       release(t);
-      return;
+      return false;
     }
 
     if (!this.queue.offer(t)) {
       onError(Operators.onOperatorError(null, Exceptions.failWithOverflow(), t, currentContext()));
       release(t);
-      return;
+      return false;
     }
 
     final long previousState = markValueAdded(this);
     if (isFinalized(previousState)) {
       this.clearSafely();
-      return;
+      return false;
     }
 
     if (isSubscriberReady(previousState)) {
       if (this.outputFused) {
         // fast path for fusion
         this.actual.onNext(null);
-        return;
+        return true;
       }
 
       if (isWorkInProgress(previousState)) {
-        return;
+        return true;
       }
 
       if (hasRequest(previousState)) {
         drainRegular(previousState);
       }
     }
+
+    return true;
+  }
+
+  public boolean tryEmitFinal(ByteBuf t) {
+    if (this.done || this.cancelled) {
+      release(t);
+      return false;
+    }
+
+    this.done = true;
+    this.last = t;
+
+    final long previousState = markValueAddedAndTerminated(this);
+    if (isFinalized(previousState)) {
+      this.clearSafely();
+      return false;
+    }
+
+    if (isSubscriberReady(previousState)) {
+      if (this.outputFused) {
+        // fast path for fusion
+        this.actual.onNext(null);
+        return true;
+      }
+
+      if (isWorkInProgress(previousState)) {
+        return true;
+      }
+
+      if (hasRequest(previousState)) {
+        drainRegular(previousState);
+      }
+    }
+
+    return true;
+  }
+
+  @Deprecated
+  public void onNextPrioritized(ByteBuf t) {
+    tryEmitPrioritized(t);
+  }
+
+  @Override
+  @Deprecated
+  public void onNext(ByteBuf t) {
+    tryEmitNormal(t);
   }
 
   @Override
@@ -371,6 +415,11 @@ boolean checkTerminated(boolean done, boolean empty, CoreSubscriber<? super Byte
     }
 
     if (done && empty) {
+      final ByteBuf last = this.last;
+      if (last != null) {
+        this.last = null;
+        a.onNext(last);
+      }
       clearAndFinalize(this);
       Throwable e = this.error;
       if (e != null) {
@@ -523,6 +572,7 @@ public void cancel() {
   }
 
   @Override
+  @Deprecated
   public void dispose() {
     this.cancelled = true;
 
@@ -606,6 +656,12 @@ void clearUnsafely() {
     final Queue<ByteBuf> queue = this.queue;
     final Queue<ByteBuf> priorityQueue = this.priorityQueue;
 
+    final ByteBuf last = this.last;
+
+    if (last != null) {
+      release(last);
+    }
+
     ByteBuf byteBuf;
     while ((byteBuf = queue.poll()) != null) {
       release(byteBuf);
@@ -753,6 +809,36 @@ static long markValueAdded(UnboundedProcessor instance) {
     }
   }
 
+  /**
+   * Sets {@link #FLAG_HAS_VALUE} flag if it was not set before and if flags {@link
+   * #FLAG_FINALIZED}, {@link #FLAG_CANCELLED}, {@link #FLAG_DISPOSED} are unset. Also, this method
+   * increments number of work in progress (WIP) if {@link #FLAG_HAS_REQUEST} is set
+   *
+   * @return previous state
+   */
+  static long markValueAddedAndTerminated(UnboundedProcessor instance) {
+    for (; ; ) {
+      final long state = instance.state;
+
+      if (isFinalized(state)) {
+        return state;
+      }
+
+      long nextState = state;
+      if (isWorkInProgress(state)) {
+        nextState = addWork(state);
+      } else if (isSubscriberReady(state) && !instance.outputFused) {
+        if (hasRequest(state)) {
+          nextState = addWork(state);
+        }
+      }
+
+      if (STATE.compareAndSet(instance, state, nextState | FLAG_HAS_VALUE | FLAG_TERMINATED)) {
+        return state;
+      }
+    }
+  }
+
   /**
    * Sets {@link #FLAG_TERMINATED} flag if it was not set before and if flags {@link
    * #FLAG_FINALIZED}, {@link #FLAG_CANCELLED}, {@link #FLAG_DISPOSED} are unset. Also, this method

rsocket-transport-local/src/main/java/io/rsocket/transport/local/LocalClientTransport.java

@@ -79,8 +79,10 @@ public Mono<DuplexConnection> connect() {
 
           Sinks.One<Object> inSink = Sinks.one();
           Sinks.One<Object> outSink = Sinks.one();
-          UnboundedProcessor in = new UnboundedProcessor(() -> inSink.tryEmitValue(inSink));
-          UnboundedProcessor out = new UnboundedProcessor(() -> outSink.tryEmitValue(outSink));
+          UnboundedProcessor in =
+              new UnboundedProcessor(() -> inSink.tryEmitValue(inSink), (__) -> {});
+          UnboundedProcessor out =
+              new UnboundedProcessor(() -> outSink.tryEmitValue(outSink), (__) -> {});
 
           Mono<Void> onClose = inSink.asMono().zipWith(outSink.asMono()).then();
 

rsocket-transport-local/src/main/java/io/rsocket/transport/local/LocalDuplexConnection.java

@@ -89,17 +89,16 @@ public Flux<ByteBuf> receive() {
   @Override
   public void sendFrame(int streamId, ByteBuf frame) {
     if (streamId == 0) {
-      out.onNextPrioritized(frame);
+      out.tryEmitPrioritized(frame);
     } else {
-      out.onNext(frame);
+      out.tryEmitNormal(frame);
     }
   }
 
   @Override
   public void sendErrorAndClose(RSocketErrorException e) {
     final ByteBuf errorFrame = ErrorFrameCodec.encode(allocator, 0, e);
-    out.onNext(errorFrame);
-    dispose();
+    out.tryEmitFinal(errorFrame);
   }
 
   @Override