Make some Mono sources and aggregators lazier

benchmarks/src/main/java/reactor/core/publisher/MonoAllBenchmark.java

@@ -0,0 +1,57 @@
+/*
+ * Copyright (c) 2022 VMware Inc. or its affiliates, 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.core.publisher;
+
+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.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.Warmup;
+
+@BenchmarkMode({Mode.AverageTime})
+@Warmup(iterations = 5, time = 5, timeUnit = TimeUnit.SECONDS)
+@Measurement(iterations = 5, time = 5, timeUnit = TimeUnit.SECONDS)
+@Fork(value = 1)
+@OutputTimeUnit(TimeUnit.NANOSECONDS)
+@State(Scope.Benchmark)
+public class MonoAllBenchmark {
+
+	@Param({"0", "10", "1000", "100000"})
+	int rangeSize;
+
+	public static void main(String[] args) throws Exception {
+		reactor.core.scrabble.ShakespearePlaysScrabbleOpt
+				s = new reactor.core.scrabble.ShakespearePlaysScrabbleOpt();
+		s.init();
+		System.out.println(s.measureThroughput());
+	}
+
+	@SuppressWarnings("unused")
+	@Benchmark
+	public void measureThroughput() {
+		Flux.range(0, rangeSize)
+			.all(i -> i < Integer.MAX_VALUE)
+			.block();
+	}
+}

benchmarks/src/main/java/reactor/core/publisher/MonoCallableBenchmark.java

@@ -0,0 +1,57 @@
+/*
+ * Copyright (c) 2022 VMware Inc. or its affiliates, 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.core.publisher;
+
+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.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.Warmup;
+
+@BenchmarkMode({Mode.AverageTime})
+@Warmup(iterations = 5, time = 5, timeUnit = TimeUnit.SECONDS)
+@Measurement(iterations = 5, time = 5, timeUnit = TimeUnit.SECONDS)
+@Fork(value = 1)
+@OutputTimeUnit(TimeUnit.NANOSECONDS)
+@State(Scope.Benchmark)
+public class MonoCallableBenchmark {
+
+	@Param({"10", "1000", "100000"})
+	int rangeSize;
+
+	public static void main(String[] args) throws Exception {
+		reactor.core.scrabble.ShakespearePlaysScrabbleOpt
+				s = new reactor.core.scrabble.ShakespearePlaysScrabbleOpt();
+		s.init();
+		System.out.println(s.measureThroughput());
+	}
+
+	@SuppressWarnings("unused")
+	@Benchmark
+	public void measureThroughput() {
+		Flux.range(0, rangeSize)
+			.all(i -> i < Integer.MAX_VALUE)
+			.block();
+	}
+}

reactor-core/src/main/java/reactor/core/publisher/FluxCallable.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2016-2021 VMware Inc. or its affiliates, All Rights Reserved.
+ * Copyright (c) 2016-2022 VMware Inc. or its affiliates, 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.
@@ -37,21 +37,7 @@ final class FluxCallable<T> extends Flux<T> implements Callable<T>, Fuseable, So
 
 	@Override
 	public void subscribe(CoreSubscriber<? super T> actual) {
-		Operators.MonoSubscriber<T, T> wrapper = new Operators.MonoSubscriber<>(actual);
-		actual.onSubscribe(wrapper);
-
-		try {
-			T v = callable.call();
-			if (v == null) {
-				wrapper.onComplete();
-			}
-			else {
-				wrapper.complete(v);
-			}
-		}
-		catch (Throwable ex) {
-			actual.onError(Operators.onOperatorError(ex, actual.currentContext()));
-		}
+		actual.onSubscribe(new MonoCallable.MonoCallableSubscription<>(actual, callable));
 	}
 
 	@Override

reactor-core/src/main/java/reactor/core/publisher/FluxDefaultIfEmpty.java

@@ -17,12 +17,9 @@
 package reactor.core.publisher;
 
 import java.util.Objects;
-import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
 import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
 
-import org.reactivestreams.Subscription;
 import reactor.core.CoreSubscriber;
-import reactor.core.Fuseable;
 import reactor.util.annotation.Nullable;
 
 /**
@@ -52,15 +49,9 @@ public Object scanUnsafe(Attr key) {
 	}
 
 	static final class DefaultIfEmptySubscriber<T>
-			implements InnerOperator<T, T>,
-			           Fuseable, //for constants only
-			           Fuseable.QueueSubscription<T> {
+			extends Operators.BaseFluxToMonoOperator<T, T> {
 
-		final CoreSubscriber<? super T> actual;
-
-		Subscription s;
-
-		boolean hasRequest;
+		boolean done;
 
 		boolean hasValue;
 
@@ -69,29 +60,17 @@ static final class DefaultIfEmptySubscriber<T>
 		static final AtomicReferenceFieldUpdater<DefaultIfEmptySubscriber, Object> FALLBACK_VALUE =
 				AtomicReferenceFieldUpdater.newUpdater(DefaultIfEmptySubscriber.class, Object.class, "fallbackValue");
 
-		volatile int state;
-		@SuppressWarnings("rawtypes")
-		static final AtomicIntegerFieldUpdater<DefaultIfEmptySubscriber> STATE =
-				AtomicIntegerFieldUpdater.newUpdater(DefaultIfEmptySubscriber.class, "state");
-
 		DefaultIfEmptySubscriber(CoreSubscriber<? super T> actual, T fallbackValue) {
-			this.actual = actual;
+			super(actual);
 			FALLBACK_VALUE.lazySet(this, fallbackValue);
 		}
 
-		@Override
-		public CoreSubscriber<? super T> actual() {
-			return this.actual;
-		}
-
 		@Override
 		@Nullable
 		public Object scanUnsafe(Attr key) {
-			if (key == Attr.PARENT) return s;
-			if (key == Attr.PREFETCH) return 0;
-			if (key == Attr.RUN_STYLE) return Attr.RunStyle.SYNC;
+			if (key == Attr.TERMINATED) return done;
 
-			return InnerOperator.super.scanUnsafe(key);
+			return super.scanUnsafe(key);
 		}
 
 		@Override
@@ -121,22 +100,14 @@ public void request(long n) {
 
 		@Override
 		public void cancel() {
-			s.cancel();
+			super.cancel();
+
 			final T fallbackValue = this.fallbackValue;
 			if (fallbackValue != null && FALLBACK_VALUE.compareAndSet(this, fallbackValue, null)) {
 				Operators.onDiscard(fallbackValue, actual.currentContext());
 			}
 		}
 
-		@Override
-		public void onSubscribe(Subscription s) {
-			if (Operators.validate(this.s, s)) {
-				this.s = s;
-
-				actual.onSubscribe(this);
-			}
-		}
-
 		@Override
 		public void onNext(T t) {
 			if (!hasValue) {
@@ -153,30 +124,14 @@ public void onNext(T t) {
 
 		@Override
 		public void onComplete() {
-			if (!hasValue) {
-				if (hasRequest) {
-					final T fallbackValue = this.fallbackValue;
-					if (fallbackValue != null && FALLBACK_VALUE.compareAndSet(this,
-							fallbackValue,
-							null)) {
-						actual.onNext(fallbackValue);
-						actual.onComplete();
-					}
-					return;
-				}
+			if (done) {
+				return;
+			}
 
-				final int state = this.state;
-				if (state == 0 && STATE.compareAndSet(this, 0, 2)) {
-					return;
-				}
+			done = true;
 
-				final T fallbackValue = this.fallbackValue;
-				if (fallbackValue != null && FALLBACK_VALUE.compareAndSet(this,
-						fallbackValue,
-						null)) {
-					actual.onNext(fallbackValue);
-					actual.onComplete();
-				}
+			if (!hasValue) {
+				completeWhenEmpty();
 
 				return;
 			}
@@ -186,6 +141,11 @@ public void onComplete() {
 
 		@Override
 		public void onError(Throwable t) {
+			if (done) {
+				return;
+			}
+
+			done = true;
 			if (!hasValue) {
 				final T fallbackValue = this.fallbackValue;
 				if (fallbackValue != null && FALLBACK_VALUE.compareAndSet(this, fallbackValue, null)) {
@@ -197,28 +157,12 @@ public void onError(Throwable t) {
 		}
 
 		@Override
-		public int requestFusion(int requestedMode) {
-			return Fuseable.NONE; // prevent fusion because of the upstream
-		}
-
-		@Override
-		public T poll() {
+		T resolveValue() {
+			final T fallbackValue = this.fallbackValue;
+			if (fallbackValue != null && FALLBACK_VALUE.compareAndSet(this, fallbackValue, null)) {
+				return fallbackValue;
+			}
 			return null;
 		}
-
-		@Override
-		public int size() {
-			return 0;
-		}
-
-		@Override
-		public boolean isEmpty() {
-			return false;
-		}
-
-		@Override
-		public void clear() {
-
-		}
 	}
 }