4.x cleanup: spinwait, grammar, minor syntax simplifications

build.gradle

@@ -180,14 +180,15 @@ jmh {
 def isCI = System.getenv("CI") != null
 def testLoggingConfig = ["skipped", "failed"]
 def parallelism = Runtime.runtime.availableProcessors()
+def heap = "1400m"
 
 if (!isCI) {
     testLoggingConfig = ["failed"]
     parallelism = parallelism.intdiv(2) ?: 1
 }
 
 test {
-    maxHeapSize = "1200m"
+    maxHeapSize = heap
     maxParallelForks = parallelism
     timeout = Duration.ofMinutes(30)
 
@@ -206,7 +207,7 @@ tasks.register('testNG', Test) {
     // Tell it to use TestNG explicitly
     useTestNG()
 
-    maxHeapSize = "1200m"
+    maxHeapSize = heap
     maxParallelForks = parallelism
     timeout = Duration.ofMinutes(30)
 

src/jmh/java/io/reactivex/rxjava4/core/OperatorFlatMapPerf.java

@@ -49,7 +49,7 @@ public void flatMapIntPassthruAsync(Input input) throws InterruptedException {
         PerfSubscriber latchedObserver = input.newLatchedObserver();
         input.flowable.flatMap((Function<Integer, Publisher<Integer>>) i -> Flowable.just(i).subscribeOn(Schedulers.computation())).subscribe(latchedObserver);
         if (input.size == 1) {
-            while (latchedObserver.latch.getCount() != 0) { }
+            while (latchedObserver.latch.getCount() != 0) { Thread.onSpinWait(); }
         } else {
             latchedObserver.latch.await();
         }

src/jmh/java/io/reactivex/rxjava4/core/OperatorMergePerf.java

@@ -35,7 +35,7 @@ public void oneStreamOfNthatMergesIn1(final InputMillion input) throws Interrupt
         Flowable.merge(os).subscribe(o);
 
         if (input.size == 1) {
-            while (o.latch.getCount() != 0) { }
+            while (o.latch.getCount() != 0) { Thread.onSpinWait(); }
         } else {
             o.latch.await();
         }
@@ -49,7 +49,7 @@ public void merge1SyncStreamOfN(final InputMillion input) throws InterruptedExce
         Flowable.merge(os).subscribe(o);
 
         if (input.size == 1) {
-            while (o.latch.getCount() != 0) { }
+            while (o.latch.getCount() != 0) { Thread.onSpinWait(); }
         } else {
             o.latch.await();
         }
@@ -61,7 +61,7 @@ public void mergeNSyncStreamsOfN(final InputThousand input) throws InterruptedEx
         PerfSubscriber o = input.newLatchedObserver();
         Flowable.merge(os).subscribe(o);
         if (input.size == 1) {
-            while (o.latch.getCount() != 0) { }
+            while (o.latch.getCount() != 0) { Thread.onSpinWait(); }
         } else {
             o.latch.await();
         }
@@ -73,7 +73,7 @@ public void mergeNAsyncStreamsOfN(final InputThousand input) throws InterruptedE
         PerfSubscriber o = input.newLatchedObserver();
         Flowable.merge(os).subscribe(o);
         if (input.size == 1) {
-            while (o.latch.getCount() != 0) { }
+            while (o.latch.getCount() != 0) { Thread.onSpinWait(); }
         } else {
             o.latch.await();
         }
@@ -85,7 +85,7 @@ public void mergeTwoAsyncStreamsOfN(final InputThousand input) throws Interrupte
         Flowable<Integer> ob = Flowable.range(0, input.size).subscribeOn(Schedulers.computation());
         Flowable.merge(ob, ob).subscribe(o);
         if (input.size == 1) {
-            while (o.latch.getCount() != 0) { }
+            while (o.latch.getCount() != 0) { Thread.onSpinWait(); }
         } else {
             o.latch.await();
         }
@@ -96,7 +96,7 @@ public void mergeNSyncStreamsOf1(final InputForMergeN input) throws InterruptedE
         PerfSubscriber o = input.newLatchedObserver();
         Flowable.merge(input.observables).subscribe(o);
         if (input.size == 1) {
-            while (o.latch.getCount() != 0) { }
+            while (o.latch.getCount() != 0) { Thread.onSpinWait(); }
         } else {
             o.latch.await();
         }

src/jmh/java/io/reactivex/rxjava4/core/PerfAsyncConsumer.java

@@ -73,7 +73,7 @@ public void onComplete() {
      */
     public PerfAsyncConsumer await(int count) {
         if (count <= 1000) {
-            while (getCount() != 0) { }
+            while (getCount() != 0) { Thread.onSpinWait(); }
         } else {
             try {
                 await();

src/jmh/java/io/reactivex/rxjava4/core/RangePerf.java

@@ -63,7 +63,7 @@ public void rangeAsync(Blackhole bh) throws Exception {
         rangeAsync.subscribe(lo);
 
         if (times == 1) {
-            while (lo.latch.getCount() != 0) { }
+            while (lo.latch.getCount() != 0) { Thread.onSpinWait(); }
         } else {
             lo.latch.await();
         }
@@ -76,7 +76,7 @@ public void rangePipeline(Blackhole bh) throws Exception {
         rangeAsyncPipeline.subscribe(lo);
 
         if (times == 1) {
-            while (lo.latch.getCount() != 0) { }
+            while (lo.latch.getCount() != 0) { Thread.onSpinWait(); }
         } else {
             lo.latch.await();
         }

src/jmh/java/io/reactivex/rxjava4/core/TakeUntilPerf.java

@@ -48,13 +48,13 @@ public void setup() {
 
         flowable = Flowable.range(1, 1000 * 1000).takeUntil(Flowable.fromCallable((Callable<Object>) () -> {
             int c = count;
-            while (items < c) { }
+            while (items < c) { Thread.onSpinWait(); }
             return 1;
         }).subscribeOn(Schedulers.single()));
 
         observable = Observable.range(1, 1000 * 1000).takeUntil(Observable.fromCallable((Callable<Object>) () -> {
             int c = count;
-            while (items < c) { }
+            while (items < c) { Thread.onSpinWait(); }
             return 1;
         }).subscribeOn(Schedulers.single()));
     }
@@ -65,7 +65,7 @@ public void flowable() {
 
         flowable.subscribe(this, Functions.emptyConsumer(), cdl::countDown);
 
-        while (cdl.getCount() != 0) { }
+        while (cdl.getCount() != 0) { Thread.onSpinWait(); }
     }
 
     @Benchmark
@@ -74,6 +74,6 @@ public void observable() {
 
         observable.subscribe(this, Functions.emptyConsumer(), cdl::countDown);
 
-        while (cdl.getCount() != 0) { }
+        while (cdl.getCount() != 0) { Thread.onSpinWait(); }
     }
 }

src/main/java/io/reactivex/rxjava4/core/Flowable.java

@@ -14526,7 +14526,7 @@ public final Flowable<T> retryUntil(@NonNull BooleanSupplier stop) {
      * <img width="640" height="430" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/retryWhen.f.v3.png" alt="">
      * <p>
      * Example:
-     *
+     * <p>
      * This retries 3 times, each time incrementing the number of seconds it waits.
      *
      * <pre><code>

src/main/java/io/reactivex/rxjava4/core/FlowableSubscriber.java

@@ -35,7 +35,7 @@ public interface FlowableSubscriber<@NonNull T> extends Subscriber<T> {
      * calling {@link Subscription#request(long)}. In practice this means
      * no initialization should happen after the {@code request()} call and
      * additional behavior is thread safe in respect to {@code onNext}.
-     *
+     * <p>
      * {@inheritDoc}
      */
     @Override

src/main/java/io/reactivex/rxjava4/core/Maybe.java

@@ -4358,7 +4358,7 @@ public final Maybe<T> retryUntil(@NonNull BooleanSupplier stop) {
      * <img width="640" height="405" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Maybe.retryWhen.png" alt="">
      * <p>
      * Example:
-     *
+     * <p>
      * This retries 3 times, each time incrementing the number of seconds it waits.
      *
      * <pre><code>
@@ -5391,7 +5391,7 @@ public final Maybe<T> unsubscribeOn(@NonNull Scheduler scheduler) {
     /**
      * Waits until this and the other {@link MaybeSource} signal a success value then applies the given {@link BiFunction}
      * to those values and emits the {@code BiFunction}'s resulting value to downstream.
-     *
+     * <p>
      * <img width="640" height="451" src="https://raw.github.com/wiki/ReactiveX/RxJava/images/rx-operators/Maybe.zipWith.png" alt="">
      *
      * <p>If either this or the other {@code MaybeSource} is empty or signals an error, the resulting {@code Maybe} will

src/main/java/io/reactivex/rxjava4/exceptions/CompositeException.java

@@ -21,14 +21,14 @@
 /**
  * Represents an exception that is a composite of one or more other exceptions. A {@code CompositeException}
  * does not modify the structure of any exception it wraps, but at print-time it iterates through the list of
- * Throwables contained in the composite in order to print them all.
- *
+ * {@link Throwable}s contained in the composite in order to print them all.
+ * <p>
  * Its invariant is to contain an immutable, ordered (by insertion order), unique list of non-composite
  * exceptions. You can retrieve individual exceptions in this list with {@link #getExceptions()}.
- *
+ * <p>
  * The {@link #printStackTrace()} implementation handles the StackTrace in a customized way instead of using
  * {@code getCause()} so that it can avoid circular references.
- *
+ * <p>
  * If you invoke {@link #getCause()}, it will lazily create the causal chain but will stop if it finds any
  * Throwable in the chain that it has already seen.
  */
@@ -42,9 +42,9 @@ public final class CompositeException extends RuntimeException {
     private Throwable cause;
 
     /**
-     * Constructs a CompositeException with the given array of Throwables as the
+     * Constructs a CompositeException with the given array of {@link Throwable}s as the
      * list of suppressed exceptions.
-     * @param exceptions the Throwables to have as initially suppressed exceptions
+     * @param exceptions the {@code Throwable}s to have as initially suppressed exceptions
      *
      * @throws IllegalArgumentException if <code>exceptions</code> is empty.
      */
@@ -54,24 +54,20 @@ public CompositeException(@NonNull Throwable... exceptions) {
     }
 
     /**
-     * Constructs a CompositeException with the given array of Throwables as the
+     * Constructs a CompositeException with the given sequence of {@link Throwable}s as the
      * list of suppressed exceptions.
-     * @param errors the Throwables to have as initially suppressed exceptions
+     * @param errors the {@code Throwable}s to have as initially suppressed exceptions
      *
      * @throws IllegalArgumentException if <code>errors</code> is empty.
      */
     public CompositeException(@NonNull Iterable<? extends Throwable> errors) {
         Set<Throwable> deDupedExceptions = new LinkedHashSet<>();
         if (errors != null) {
             for (Throwable ex : errors) {
-                if (ex instanceof CompositeException) {
-                    deDupedExceptions.addAll(((CompositeException) ex).getExceptions());
+                if (ex instanceof CompositeException ce) {
+                    deDupedExceptions.addAll(ce.getExceptions());
                 } else
-                if (ex != null) {
-                    deDupedExceptions.add(ex);
-                } else {
-                    deDupedExceptions.add(new NullPointerException("Throwable was null!"));
-                }
+                    deDupedExceptions.add(Objects.requireNonNullElseGet(ex, () -> new NullPointerException("Throwable was null!")));
             }
         } else {
             deDupedExceptions.add(new NullPointerException("errors was null"));
@@ -104,7 +100,7 @@ public String getMessage() {
     @NonNull
     public synchronized Throwable getCause() { // NOPMD
         if (cause == null) {
-            String separator = System.getProperty("line.separator");
+            String separator = System.lineSeparator();
             if (exceptions.size() > 1) {
                 Map<Throwable, Boolean> seenCauses = new IdentityHashMap<>();
 
@@ -164,9 +160,9 @@ public synchronized Throwable getCause() { // NOPMD
     }
 
     /**
-     * All of the following {@code printStackTrace} functionality is derived from JDK {@link Throwable}
+     * All the following {@code printStackTrace} functionality is derived from JDK {@link Throwable}
      * {@code printStackTrace}. In particular, the {@code PrintStreamOrWriter} abstraction is copied wholesale.
-     *
+     * <p>
      * Changes from the official JDK implementation:<ul>
      * <li>no infinite loop detection</li>
      * <li>smaller critical section holding {@link PrintStream} lock</li>

src/main/java/io/reactivex/rxjava4/exceptions/Exceptions.java

@@ -55,9 +55,9 @@ public static RuntimeException propagate(@NonNull Throwable t) {
      * <li>{@code LinkageError}</li>
      * </ul>
      * This can be useful if you are writing an operator that calls user-supplied code, and you want to
-     * notify subscribers of errors encountered in that code by calling their {@code onError} methods, but only
-     * if the errors are not so catastrophic that such a call would be futile, in which case you simply want to
-     * rethrow the error.
+     * notify subscribers of errors encountered in that code by calling their {@code onError} methods.
+     * But only if the errors are not so catastrophic that such a call would be futile, in which case
+     * you simply want to rethrow the error.
      *
      * @param t
      *         the {@code Throwable} to test and perhaps throw

src/main/java/io/reactivex/rxjava4/exceptions/OnErrorNotImplementedException.java

@@ -31,7 +31,7 @@ public final class OnErrorNotImplementedException extends RuntimeException {
 
     /**
      * Customizes the {@code Throwable} with a custom message and wraps it before it
-     * is signalled to the {@code RxJavaPlugins.onError()} handler as {@code OnErrorNotImplementedException}.
+     * is signaled to the {@code RxJavaPlugins.onError()} handler as {@code OnErrorNotImplementedException}.
      *
      * @param message
      *          the message to assign to the {@code Throwable} to signal
@@ -44,7 +44,7 @@ public OnErrorNotImplementedException(String message, @NonNull Throwable e) {
 
     /**
      * Wraps the {@code Throwable} before it
-     * is signalled to the {@code RxJavaPlugins.onError()}
+     * is signaled to the {@code RxJavaPlugins.onError()}
      * handler as {@code OnErrorNotImplementedException}.
      *
      * @param e

src/main/java/io/reactivex/rxjava4/flowables/ConnectableFlowable.java

@@ -44,8 +44,8 @@
  * {@code reset()} in this case.
  * <p>
  * Note that although {@link #connect()} and {@link #reset()} are safe to call from multiple threads, it is recommended
- * a dedicated thread or business logic manages the connection or resetting of a {@code ConnectableFlowable} so that
- * there is no unwanted signal loss due to early {@code connect()} or {@code reset()} calls while {@code Subscriber}s are
+ * a dedicated thread or business logic manages the connection or resetting of a {@code ConnectableFlowable}.
+ * This ensures there is no unwanted signal loss due to early {@code connect()} or {@code reset()} calls while {@code Subscriber}s are
  * still being subscribed to this {@code ConnectableFlowable} to receive signals from the get-go.
  * <p>
  * @see <a href="https://github.com/ReactiveX/RxJava/wiki/Connectable-Observable-Operators">RxJava Wiki: Connectable Observable Operators</a>
@@ -279,7 +279,7 @@ public final Flowable<T> refCount(int subscriberCount, long timeout, @NonNull Ti
      * during the lifetime of the returned {@code Flowable}. If this {@code ConnectableFlowable}
      * terminates, the connection is never renewed, no matter how {@code Subscriber}s come
      * and go. Use {@link #refCount()} to renew a connection or dispose an active
-     * connection when all {@code Subscriber}s have cancelled their {@link Subscription}s.
+     * connection when all {@code Subscriber}s have canceled their {@link Subscription}s.
      * <p>
      * This overload does not allow disconnecting the connection established via
      * {@link #connect(Consumer)}. Use the {@link #autoConnect(int, Consumer)} overload
@@ -314,7 +314,7 @@ public Flowable<T> autoConnect() {
      * during the lifetime of the returned {@code Flowable}. If this {@code ConnectableFlowable}
      * terminates, the connection is never renewed, no matter how {@code Subscriber}s come
      * and go. Use {@link #refCount()} to renew a connection or dispose an active
-     * connection when all {@code Subscriber}s have cancelled their {@link Subscription}s.
+     * connection when all {@code Subscriber}s have canceled their {@link Subscription}s.
      * <p>
      * This overload does not allow disconnecting the connection established via
      * {@link #connect(Consumer)}. Use the {@link #autoConnect(int, Consumer)} overload
@@ -352,7 +352,7 @@ public Flowable<T> autoConnect(int numberOfSubscribers) {
      * during the lifetime of the returned {@code Flowable}. If this {@code ConnectableFlowable}
      * terminates, the connection is never renewed, no matter how {@code Subscriber}s come
      * and go. Use {@link #refCount()} to renew a connection or dispose an active
-     * connection when all {@code Subscriber}s have cancelled their {@link Subscription}s.
+     * connection when all {@code Subscriber}s have canceled their {@link Subscription}s.
      * <dl>
      *  <dt><b>Backpressure:</b></dt>
      *  <dd>The operator itself doesn't interfere with backpressure which is determined by

src/main/java/io/reactivex/rxjava4/internal/operators/maybe/MaybeMergeArray.java

@@ -388,7 +388,7 @@ public boolean isEmpty() {
 
         @Override
         public void clear() {
-            while (poll() != null && !isEmpty()) { }
+            while (poll() != null && !isEmpty()) { Thread.onSpinWait(); }
         }
 
         @Override

src/main/java/io/reactivex/rxjava4/internal/queue/MpscLinkedQueue.java

@@ -98,7 +98,7 @@ public T poll() {
         }
         else if (currConsumerNode != lvProducerNode()) {
             // spin, we are no longer wait free
-            while ((nextNode = currConsumerNode.lvNext()) == null) { } // NOPMD
+            while ((nextNode = currConsumerNode.lvNext()) == null) { Thread.onSpinWait(); } // NOPMD
             // got the next node...
 
             // we have to null out the value because we are going to hang on to the node
@@ -120,7 +120,7 @@ public boolean offer(T v1, T v2) {
 
     @Override
     public void clear() {
-        while (poll() != null && !isEmpty()) { } // NOPMD
+        while (poll() != null && !isEmpty()) { Thread.onSpinWait(); } // NOPMD
     }
     LinkedQueueNode<T> lvProducerNode() {
         return producerNode.get();

src/main/java/io/reactivex/rxjava4/internal/schedulers/BlockingCurrentThreadScheduler.java

@@ -259,7 +259,7 @@ public void run() throws Exception {
                     }
                 }
             } finally {
-                while (get() == INTERRUPTING) { }
+                while (get() == INTERRUPTING) { Thread.onSpinWait(); }
 
                 if (get() == INTERRUPTED) {
                     Thread.interrupted();
@@ -385,7 +385,7 @@ public void run() throws Exception {
                         }
                     }
                 } finally {
-                    while (get() == INTERRUPTING) { }
+                    while (get() == INTERRUPTING) { Thread.onSpinWait(); }
 
                     if (get() == INTERRUPTED) {
                         Thread.interrupted();

src/main/java/io/reactivex/rxjava4/internal/subscriptions/EmptySubscription.java

@@ -87,7 +87,8 @@ public void clear() {
 
     @Override
     public int requestFusion(int mode) {
-        return mode & ASYNC; // accept async mode: an onComplete or onError will be signalled after anyway
+        // accept async mode: an onComplete or onError will be signaled after anyway
+        return mode & ASYNC;
     }
 
     @Override

src/main/java/io/reactivex/rxjava4/internal/util/OpenHashSet.java

@@ -171,4 +171,4 @@ public Object[] keys() {
     public int size() {
         return size;
     }
-}
+}