update versions and fixes memleak in UnboundedProcessor (#1106)

This PR updates dependencies and makes minor modifications to UnboundedProcessor due to repeating failures of UnboundedProcessorJCStreassTest, which started reproducing some unspotted issues.

Motivation:

UnboundedProcessor is a critical component in the RSocket-Java ecosystem and must work properly. After analysis of its internal state machine, it was spotted that sometimes:

The request may not be delivered due to natural concurrency
The terminal signal may not be delivered since it checks for demand which might be consumed already (due to natural concurrency)
The final value could be delivered violating reactive-streams spec
Modifications:

This PR adds a minimal set of changes, preserving old implementation but eliminating the mentioned bugs

---------

Signed-off-by: Oleh Dokuka <oleh.dokuka@icloud.com>

Author: OlegDokuka

build.gradle

@@ -33,10 +33,10 @@ subprojects {
     apply plugin: 'com.github.sherter.google-java-format'
     apply plugin: 'com.github.vlsi.gradle-extensions'
 
-    ext['reactor-bom.version'] = '2020.0.32'
-    ext['logback.version'] = '1.2.10'
-    ext['netty-bom.version'] = '4.1.93.Final'
-    ext['netty-boringssl.version'] = '2.0.61.Final'
+    ext['reactor-bom.version'] = '2020.0.39'
+    ext['logback.version'] = '1.3.14'
+    ext['netty-bom.version'] = '4.1.106.Final'
+    ext['netty-boringssl.version'] = '2.0.62.Final'
     ext['hdrhistogram.version'] = '2.1.12'
     ext['mockito.version'] = '4.11.0'
     ext['slf4j.version'] = '1.7.36'

rsocket-core/build.gradle

@@ -41,13 +41,14 @@ dependencies {
     testRuntimeOnly 'org.junit.jupiter:junit-jupiter-engine'
 
     jcstressImplementation(project(":rsocket-test"))
+    jcstressImplementation 'org.slf4j:slf4j-api'
     jcstressImplementation "ch.qos.logback:logback-classic"
     jcstressImplementation 'io.projectreactor:reactor-test'
 }
 
 jcstress {
-    mode = 'quick' //quick, default, tough
-    jcstressDependency = "org.openjdk.jcstress:jcstress-core:0.15"
+    mode = 'sanity' //sanity, quick, default, tough
+    jcstressDependency = "org.openjdk.jcstress:jcstress-core:0.16"
 }
 
 jar {

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

@@ -3,6 +3,9 @@
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.UnpooledByteBufAllocator;
 import io.rsocket.core.StressSubscriber;
+import io.rsocket.utils.FastLogger;
+import java.util.Arrays;
+import java.util.ConcurrentModificationException;
 import org.openjdk.jcstress.annotations.Actor;
 import org.openjdk.jcstress.annotations.Arbiter;
 import org.openjdk.jcstress.annotations.Expect;
@@ -14,14 +17,17 @@
 import org.openjdk.jcstress.infra.results.L_Result;
 import reactor.core.Fuseable;
 import reactor.core.publisher.Hooks;
+import reactor.util.Logger;
 
 public abstract class UnboundedProcessorStressTest {
 
   static {
     Hooks.onErrorDropped(t -> {});
   }
 
-  final UnboundedProcessor unboundedProcessor = new UnboundedProcessor();
+  final Logger logger = new FastLogger(getClass().getName());
+
+  final UnboundedProcessor unboundedProcessor = new UnboundedProcessor(logger);
 
   @JCStressTest
   @Outcome(
@@ -145,6 +151,8 @@ public void arbiter(LLL_Result r) {
       stressSubscriber.values.forEach(ByteBuf::release);
 
       r.r3 = byteBuf1.refCnt() + byteBuf2.refCnt() + byteBuf3.refCnt() + byteBuf4.refCnt();
+
+      checkOutcomes(this, r.toString(), logger);
     }
   }
 
@@ -270,6 +278,8 @@ public void arbiter(LLL_Result r) {
       stressSubscriber.values.forEach(ByteBuf::release);
 
       r.r3 = byteBuf1.refCnt() + byteBuf2.refCnt() + byteBuf3.refCnt() + byteBuf4.refCnt();
+
+      checkOutcomes(this, r.toString(), logger);
     }
   }
 
@@ -375,6 +385,8 @@ public void arbiter(LLL_Result r) {
       stressSubscriber.values.forEach(ByteBuf::release);
 
       r.r3 = byteBuf1.refCnt() + byteBuf2.refCnt() + byteBuf3.refCnt() + byteBuf4.refCnt();
+
+      checkOutcomes(this, r.toString(), logger);
     }
   }
 
@@ -476,6 +488,8 @@ public void arbiter(LLL_Result r) {
       stressSubscriber.values.forEach(ByteBuf::release);
 
       r.r3 = byteBuf1.refCnt() + byteBuf2.refCnt() + byteBuf3.refCnt() + byteBuf4.refCnt();
+
+      checkOutcomes(this, r.toString(), logger);
     }
   }
 
@@ -578,6 +592,8 @@ public void arbiter(LLL_Result r) {
       stressSubscriber.values.forEach(ByteBuf::release);
 
       r.r3 = byteBuf1.refCnt() + byteBuf2.refCnt() + byteBuf3.refCnt() + byteBuf4.refCnt();
+
+      checkOutcomes(this, r.toString(), logger);
     }
   }
 
@@ -701,6 +717,8 @@ public void arbiter(LLL_Result r) {
       stressSubscriber.values.forEach(ByteBuf::release);
 
       r.r3 = byteBuf1.refCnt() + byteBuf2.refCnt() + byteBuf3.refCnt() + byteBuf4.refCnt();
+
+      checkOutcomes(this, r.toString(), logger);
     }
   }
 
@@ -781,6 +799,8 @@ public void arbiter(LLL_Result r) {
       stressSubscriber.values.forEach(ByteBuf::release);
 
       r.r3 = byteBuf1.refCnt() + byteBuf2.refCnt() + byteBuf3.refCnt() + byteBuf4.refCnt();
+
+      checkOutcomes(this, r.toString(), logger);
     }
   }
 
@@ -837,9 +857,15 @@ public void arbiter(LLL_Result r) {
               + stressSubscriber.onErrorCalls * 2
               + stressSubscriber.droppedErrors.size() * 3;
 
+      if (stressSubscriber.concurrentOnNext || stressSubscriber.concurrentOnComplete) {
+        throw new ConcurrentModificationException("boo");
+      }
+
       stressSubscriber.values.forEach(ByteBuf::release);
 
       r.r3 = byteBuf1.refCnt() + byteBuf2.refCnt() + byteBuf3.refCnt() + byteBuf4.refCnt();
+
+      checkOutcomes(this, r.toString(), logger);
     }
   }
 
@@ -892,6 +918,8 @@ public void arbiter(LLL_Result r) {
       stressSubscriber.values.forEach(ByteBuf::release);
 
       r.r3 = byteBuf1.refCnt() + byteBuf2.refCnt() + byteBuf3.refCnt() + byteBuf4.refCnt();
+
+      checkOutcomes(this, r.toString(), logger);
     }
   }
 
@@ -1107,6 +1135,8 @@ public void arbiter(LLLL_Result r) {
       stressSubscriber.values.forEach(ByteBuf::release);
 
       r.r4 = byteBuf1.refCnt() + byteBuf2.refCnt() + byteBuf3.refCnt() + byteBuf4.refCnt();
+
+      checkOutcomes(this, r.toString(), logger);
     }
   }
 
@@ -1238,6 +1268,8 @@ public void arbiter(LLLL_Result r) {
       stressSubscriber.values.forEach(ByteBuf::release);
 
       r.r4 = byteBuf1.refCnt() + byteBuf2.refCnt() + byteBuf3.refCnt() + byteBuf4.refCnt();
+
+      checkOutcomes(this, r.toString(), logger);
     }
   }
 
@@ -1390,6 +1422,8 @@ public void arbiter(LLLL_Result r) {
       stressSubscriber.values.forEach(ByteBuf::release);
 
       r.r4 = byteBuf1.refCnt() + byteBuf2.refCnt() + byteBuf3.refCnt() + byteBuf4.refCnt();
+
+      checkOutcomes(this, r.toString(), logger);
     }
   }
 
@@ -1522,6 +1556,8 @@ public void arbiter(LLLL_Result r) {
       stressSubscriber.values.forEach(ByteBuf::release);
 
       r.r4 = byteBuf1.refCnt() + byteBuf2.refCnt() + byteBuf3.refCnt() + byteBuf4.refCnt();
+
+      checkOutcomes(this, r.toString(), logger);
     }
   }
 
@@ -1587,6 +1623,8 @@ public void arbiter(LLLL_Result r) {
       stressSubscriber.values.forEach(ByteBuf::release);
 
       r.r4 = byteBuf1.refCnt() + byteBuf2.refCnt() + byteBuf3.refCnt() + byteBuf4.refCnt();
+
+      checkOutcomes(this, r.toString(), logger);
     }
   }
 
@@ -1652,6 +1690,8 @@ public void arbiter(LLLL_Result r) {
       stressSubscriber.values.forEach(ByteBuf::release);
 
       r.r4 = byteBuf1.refCnt() + byteBuf2.refCnt() + byteBuf3.refCnt() + byteBuf4.refCnt();
+
+      checkOutcomes(this, r.toString(), logger);
     }
   }
 
@@ -1678,6 +1718,16 @@ public void subscribe2() {
     @Arbiter
     public void arbiter(L_Result r) {
       r.r1 = stressSubscriber1.onErrorCalls + stressSubscriber2.onErrorCalls;
+
+      checkOutcomes(this, r.toString(), logger);
+    }
+  }
+
+  static void checkOutcomes(Object instance, String result, Logger logger) {
+    if (Arrays.stream(instance.getClass().getDeclaredAnnotationsByType(Outcome.class))
+        .flatMap(o -> Arrays.stream(o.id()))
+        .noneMatch(s -> s.equalsIgnoreCase(result))) {
+      throw new RuntimeException(result + " " + logger);
     }
   }
 }

rsocket-core/src/jcstress/java/io/rsocket/utils/FastLogger.java

@@ -0,0 +1,137 @@
+package io.rsocket.utils;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.util.stream.Collectors;
+import reactor.util.Logger;
+
+/**
+ * Implementation of {@link Logger} which is based on the {@link ThreadLocal} based queue which
+ * collects all the events on the per-thread basis. </br> Such logger is designed to have all events
+ * stored during the stress-test run and then sorted and printed out once all the Threads completed
+ * execution (inside the {@link org.openjdk.jcstress.annotations.Arbiter} annotated method. </br>
+ * Note, this implementation only supports trace-level logs and ignores all others, it is intended
+ * to be used by {@link reactor.core.publisher.StateLogger}.
+ */
+public class FastLogger implements Logger {
+
+  final Map<Thread, List<String>> queues = new ConcurrentHashMap<>();
+
+  final ThreadLocal<List<String>> logsQueueLocal =
+      ThreadLocal.withInitial(
+          () -> {
+            final ArrayList<String> logs = new ArrayList<>(100);
+            queues.put(Thread.currentThread(), logs);
+            return logs;
+          });
+
+  private final String name;
+
+  public FastLogger(String name) {
+    this.name = name;
+  }
+
+  @Override
+  public String toString() {
+    return queues
+        .values()
+        .stream()
+        .flatMap(List::stream)
+        .sorted(
+            Comparator.comparingLong(
+                s -> {
+                  Pattern pattern = Pattern.compile("\\[(.*?)]");
+                  Matcher matcher = pattern.matcher(s);
+                  matcher.find();
+                  return Long.parseLong(matcher.group(1));
+                }))
+        .collect(Collectors.joining("\n"));
+  }
+
+  @Override
+  public String getName() {
+    return this.name;
+  }
+
+  @Override
+  public boolean isTraceEnabled() {
+    return true;
+  }
+
+  @Override
+  public void trace(String msg) {
+    logsQueueLocal.get().add(String.format("[%s] %s", System.nanoTime(), msg));
+  }
+
+  @Override
+  public void trace(String format, Object... arguments) {
+    trace(String.format(format, arguments));
+  }
+
+  @Override
+  public void trace(String msg, Throwable t) {
+    trace(String.format("%s, %s", msg, Arrays.toString(t.getStackTrace())));
+  }
+
+  @Override
+  public boolean isDebugEnabled() {
+    return false;
+  }
+
+  @Override
+  public void debug(String msg) {}
+
+  @Override
+  public void debug(String format, Object... arguments) {}
+
+  @Override
+  public void debug(String msg, Throwable t) {}
+
+  @Override
+  public boolean isInfoEnabled() {
+    return false;
+  }
+
+  @Override
+  public void info(String msg) {}
+
+  @Override
+  public void info(String format, Object... arguments) {}
+
+  @Override
+  public void info(String msg, Throwable t) {}
+
+  @Override
+  public boolean isWarnEnabled() {
+    return false;
+  }
+
+  @Override
+  public void warn(String msg) {}
+
+  @Override
+  public void warn(String format, Object... arguments) {}
+
+  @Override
+  public void warn(String msg, Throwable t) {}
+
+  @Override
+  public boolean isErrorEnabled() {
+    return false;
+  }
+
+  @Override
+  public void error(String msg) {}
+
+  @Override
+  public void error(String format, Object... arguments) {}
+
+  @Override
+  public void error(String msg, Throwable t) {}
+}