Make JMH tests a top level gradle module

- Move JMH into dedicated submodule "benchmarks"
- Add @akarnokd scrabble benchmarks

Signed-off-by: Stephane Maldini <smaldini@pivotal.io>

benchmarks/build.gradle

@@ -0,0 +1,35 @@
+apply plugin: "com.github.johnrengelman.shadow"
+apply plugin: "me.champeau.gradle.jmh"
+
+
+dependencies{
+  compile project(':reactor-core')
+
+  compile "org.openjdk.jmh:jmh-core:1.21"
+  compile "org.openjdk.jmh:jmh-generator-annprocess:1.21"
+  compile "io.projectreactor.addons:reactor-extra:3.3.0.BUILD-SNAPSHOT"
+
+}
+
+
+// Safer to use the following to run benchmarks, otherwise you may get unexpected errors:
+// ./gradlew --no-daemon cleanJmhJar jmh -Pjmh="ShakespearePlaysScrabble"
+
+jmh {
+  jmhVersion = '1.21'
+  duplicateClassesStrategy DuplicatesStrategy.INCLUDE
+  failOnError = true
+  resultFormat = 'CSV'
+  if (project.hasProperty('jmh')) {
+	include = [".*" + project.jmh + ".*"]
+  }
+//  includeTests = false
+}
+
+//Do not use the default naming convention to avoid CI scan
+jmhJar {
+  baseName 'benchmarks'
+  classifier = null
+  version = null
+  destinationDir = file("$rootDir")
+}

reactor-core/src/jmh/java/reactor/AssemblyTraceBenchmark.java → benchmarks/src/jmh/java/reactor/AssemblyTraceBenchmark.java

@@ -33,6 +33,7 @@
 
 /**
  * @author Sergei Egorov
+ * @author Stephane Maldini
  */
 @BenchmarkMode({Mode.AverageTime})
 @Warmup(iterations = 5, time = 2, timeUnit = TimeUnit.SECONDS)
@@ -44,25 +45,33 @@ public class AssemblyTraceBenchmark {
 
 	static final String JAVA_8 = "/Library/Java/JavaVirtualMachines/jdk1.8.0_192.jdk/Contents/Home/bin/java";
 
-	static final String JAVA_11 = "/Library/Java/JavaVirtualMachines/openjdk-11.0.1.jdk/Contents/Home/bin/java";
+	static final String JAVA_11 = "/Library/Java/JavaVirtualMachines/openjdk-11.0.2.jdk/Contents/Home/bin/java";
 
 	@Param({"10", "40", "80"})
 	int stackSize;
 
 	@Benchmark
 	@Fork(
-			jvm = JAVA_8,
 			jvmArgsAppend = {"-XX:-OmitStackTraceInFastThrow", "-Dreactor.trace.operatorStacktrace=true"}
 	)
-	public void withTracingOnJDK8(Blackhole bh) {
+	public void withTracing(Blackhole bh) {
 		stack(stackSize, bh);
 	}
 
-	@Benchmark
-	@Fork(
-			jvm = JAVA_11,
-			jvmArgsAppend = {"-XX:-OmitStackTraceInFastThrow", "-Dreactor.trace.operatorStacktrace=true"}
-	)
+//	@Benchmark
+//	@Fork(
+//			jvm = JAVA_8,
+//			jvmArgsAppend = {"-XX:-OmitStackTraceInFastThrow", "-Dreactor.trace.operatorStacktrace=true"}
+//	)
+//	public void withTracingOnJDK8(Blackhole bh) {
+//		stack(stackSize, bh);
+//	}
+
+//	@Benchmark
+//	@Fork(
+//			jvm = JAVA_11,
+//			jvmArgsAppend = {"-XX:-OmitStackTraceInFastThrow", "-Dreactor.trace.operatorStacktrace=true"}
+//	)
 	public void withTracingOnJDK11(Blackhole bh) {
 		stack(stackSize, bh);
 	}

benchmarks/src/jmh/java/reactor/CheckpointBenchmark.java

@@ -0,0 +1,69 @@
+/*
+ * Copyright (c) 2011-2017 Pivotal Software Inc, All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package reactor;
+
+import java.util.concurrent.TimeUnit;
+
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Warmup;
+import reactor.core.publisher.Flux;
+
+/**
+ * @author Simon Baslé
+ */
+public class CheckpointBenchmark {
+
+	Flux<String> findAllUserByName(Flux<String> source) {
+		return source.map(s -> { throw new IllegalStateException("boom"); })
+		             .map(s -> s + "-user");
+	}
+
+	@Benchmark()
+	@OutputTimeUnit(TimeUnit.MILLISECONDS)
+	@Warmup(iterations = 5, time = 1)
+	@Measurement(iterations = 5, time = 1)
+	@Fork(1)
+	@BenchmarkMode({Mode.Throughput, Mode.SampleTime})
+	public void withFullCheckpoint() {
+		this.findAllUserByName(Flux.just("pedro", "simon", "stephane"))
+		    .transform(f -> f.filter(s -> s.startsWith("s")))
+		    .transform(f -> f.elapsed())
+		    .checkpoint("checkpoint description", true)
+		    .subscribe(System.out::println, t -> {
+		    });
+	}
+
+	@Benchmark
+	@OutputTimeUnit(TimeUnit.MILLISECONDS)
+	@Warmup(iterations = 5, time = 1)
+	@Measurement(iterations = 5, time = 1)
+	@Fork(1)
+	@BenchmarkMode({Mode.Throughput, Mode.SampleTime})
+	public void withLightCheckpoint() {
+		this.findAllUserByName(Flux.just("pedro", "simon", "stephane"))
+		    .transform(f -> f.filter(s -> s.startsWith("s")))
+		    .transform(f -> f.elapsed())
+		    .checkpoint("light checkpoint identifier")
+		    .subscribe(System.out::println, t -> {
+		    });
+	}
+}

benchmarks/src/jmh/java/reactor/core/scrabble/FluxCharSequence.java

@@ -0,0 +1,153 @@
+package reactor.core.scrabble;
+import java.util.concurrent.atomic.AtomicLongFieldUpdater;
+
+import org.reactivestreams.Subscriber;
+import reactor.core.CoreSubscriber;
+import reactor.core.Fuseable;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Operators;
+
+/**
+ * Streams the characters of a string.
+ */
+final class FluxCharSequence extends Flux<Integer> implements Fuseable {
+
+	final CharSequence string;
+
+	FluxCharSequence(CharSequence string) {
+		this.string = string;
+	}
+
+	@Override
+	public void subscribe(CoreSubscriber<? super Integer> actual) {
+		actual.onSubscribe(new CharSequenceSubscription(actual, string));
+	}
+
+	static final class CharSequenceSubscription implements Fuseable.QueueSubscription<Integer> {
+
+		final CoreSubscriber<? super Integer> actual;
+
+		final CharSequence string;
+
+		final int end;
+
+		int index;
+
+		volatile boolean cancelled;
+
+		volatile     long                                             requested;
+		CharSequenceSubscription(CoreSubscriber<? super Integer> actual, CharSequence string) {
+			this.actual = actual;
+			this.string = string;
+			this.end = string.length();
+		}
+
+		@Override
+		public void cancel() {
+			cancelled = true;
+		}
+
+		@Override
+		public void clear() {
+			index = end;
+		}
+
+		@Override
+		public boolean isEmpty() {
+			return index != end;
+		}
+
+		@Override
+		public Integer poll() {
+			int i = index;
+			if (i != end) {
+				index = i + 1;
+				return (int) string.charAt(i);
+			}
+			return null;
+		}
+
+		@Override
+		public void request(long n) {
+			if (Operators.validate(n)) {
+				if (Operators.addCap(REQUESTED, this, n) == 0) {
+					if (n == Long.MAX_VALUE) {
+						fastPath();
+					}
+					else {
+						slowPath(n);
+					}
+				}
+			}
+		}
+
+		@Override
+		public int requestFusion(int requestedMode) {
+			return requestedMode & Fuseable.SYNC;
+		}
+
+		@Override
+		public int size() {
+			return end - index;
+		}
+
+		void fastPath() {
+			int e = end;
+			CharSequence s = string;
+			Subscriber<? super Integer> a = actual;
+
+			for (int i = index; i != e; i++) {
+				if (cancelled) {
+					return;
+				}
+
+				a.onNext((int) s.charAt(i));
+			}
+
+			if (!cancelled) {
+				a.onComplete();
+			}
+		}
+
+		void slowPath(long r) {
+			long e = 0L;
+			int i = index;
+			int f = end;
+			CharSequence s = string;
+			Subscriber<? super Integer> a = actual;
+
+			for (; ; ) {
+
+				while (e != r && i != f) {
+					if (cancelled) {
+						return;
+					}
+
+					a.onNext((int) s.charAt(i));
+
+					i++;
+					e++;
+				}
+
+				if (i == f) {
+					if (!cancelled) {
+						a.onComplete();
+					}
+					return;
+				}
+
+				r = requested;
+				if (e == r) {
+					index = i;
+					r = REQUESTED.addAndGet(this, -e);
+					if (r == 0L) {
+						break;
+					}
+				}
+			}
+		}
+		static final AtomicLongFieldUpdater<CharSequenceSubscription> REQUESTED =
+				AtomicLongFieldUpdater.newUpdater(CharSequenceSubscription.class, "requested");
+	}
+
+}