updates versions and fixes memleak in UnboundedProcessor

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) {}
+}