4.x: Streamable optimizations 2026.07.21 (#8254)

* 4.x: Streamable optimizations & fusion expansions

* blockingX to accept StreamerCancellation for external management

* Fix style and validator reported errors

Author: akarnokd

src/jmh/java/io/reactivex/rxjava4/streamable/StreamableSkipPerf.java

@@ -13,11 +13,13 @@
 
 package io.reactivex.rxjava4.streamable;
 
+import java.util.List;
 import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
 
 import org.openjdk.jmh.annotations.*;
 
-import io.reactivex.rxjava4.core.*;
+import io.reactivex.rxjava4.core.Streamable;
 
 ///
 /// The map is one of the most used operator in the ecosystem so it must be fast and
@@ -66,6 +68,48 @@
 /// StreamableSkipPerf.benchmark   100000  thrpt    5        2168,925 ┬▒        32,591  ops/s
 /// StreamableSkipPerf.benchmark  1000000  thrpt    5         214,055 ┬▒         5,629  ops/s
 /// ```
+///
+/// # 3. sync bias via Claude Fable atomics
+///
+/// Small regression on 1, but +200% on million vs 2., almost +410% vs baseline
+///
+/// ```
+/// Benchmark                     (times)   Mode  Cnt           Score           Error  Units
+/// StreamableSkipPerf.benchmark        1  thrpt    5  1392748873,097 ┬▒ 240564597,648  ops/s
+/// StreamableSkipPerf.benchmark       10  thrpt    5    50478482,677 ┬▒   2503440,823  ops/s
+/// StreamableSkipPerf.benchmark      100  thrpt    5     8260512,961 ┬▒    832256,445  ops/s
+/// StreamableSkipPerf.benchmark     1000  thrpt    5      617903,033 ┬▒    127135,231  ops/s
+/// StreamableSkipPerf.benchmark    10000  thrpt    5       65116,345 ┬▒      8425,364  ops/s
+/// StreamableSkipPerf.benchmark   100000  thrpt    5        6298,539 ┬▒       739,686  ops/s
+/// StreamableSkipPerf.benchmark  1000000  thrpt    5         634,945 ┬▒        99,134  ops/s
+/// ```
+///
+/// # 4. indexable/enumerable/deferredenumerable
+///
+/// ```
+/// Benchmark                      (times)   Mode  Cnt           Score           Error  Units
+/// StreamableSkipPerf.benchmark         1  thrpt    5  1371065918,333 ┬▒ 114154178,555  ops/s
+/// StreamableSkipPerf.benchmark        10  thrpt    5    48387844,296 ┬▒   1998160,368  ops/s
+/// StreamableSkipPerf.benchmark       100  thrpt    5     8344391,259 ┬▒    251723,213  ops/s
+/// StreamableSkipPerf.benchmark      1000  thrpt    5      700332,413 ┬▒     28667,777  ops/s
+/// StreamableSkipPerf.benchmark     10000  thrpt    5       68645,229 ┬▒      3123,103  ops/s
+/// StreamableSkipPerf.benchmark    100000  thrpt    5        6047,152 ┬▒       725,594  ops/s
+/// StreamableSkipPerf.benchmark   1000000  thrpt    5         649,134 ┬▒        36,349  ops/s
+/// StreamableSkipPerf.enumerable        1  thrpt    5    71117553,552 ┬▒   6738602,495  ops/s
+/// StreamableSkipPerf.enumerable       10  thrpt    5    43426136,653 ┬▒   2744303,836  ops/s
+/// StreamableSkipPerf.enumerable      100  thrpt    5     2819347,969 ┬▒     20437,839  ops/s
+/// StreamableSkipPerf.enumerable     1000  thrpt    5      298000,822 ┬▒     28754,714  ops/s
+/// StreamableSkipPerf.enumerable    10000  thrpt    5       28839,499 ┬▒      3096,840  ops/s
+/// StreamableSkipPerf.enumerable   100000  thrpt    5        2657,147 ┬▒        23,023  ops/s
+/// StreamableSkipPerf.enumerable  1000000  thrpt    5         187,770 ┬▒        42,082  ops/s
+/// StreamableSkipPerf.indexed           1  thrpt    5    75403131,814 ┬▒   4676965,390  ops/s
+/// StreamableSkipPerf.indexed          10  thrpt    5    42730509,975 ┬▒   6004178,786  ops/s
+/// StreamableSkipPerf.indexed         100  thrpt    5     3561284,547 ┬▒    381609,301  ops/s
+/// StreamableSkipPerf.indexed        1000  thrpt    5      346781,462 ┬▒     28909,008  ops/s
+/// StreamableSkipPerf.indexed       10000  thrpt    5       54254,177 ┬▒      7440,753  ops/s
+/// StreamableSkipPerf.indexed      100000  thrpt    5        5153,256 ┬▒       310,895  ops/s
+/// StreamableSkipPerf.indexed     1000000  thrpt    5         259,912 ┬▒        76,746  ops/s
+/// ```
 @BenchmarkMode(Mode.Throughput)
 @Warmup(iterations = 5, time = 1, timeUnit = TimeUnit.SECONDS)
 @Measurement(iterations = 5, time = 1, timeUnit = TimeUnit.SECONDS)
@@ -77,14 +121,28 @@ public class StreamableSkipPerf {
     public int times;
 
     Streamable<Integer> result;
+    Streamable<List<Integer>> resultIndexed;
+    Streamable<List<Integer>> resultEnumerable;
 
     @Setup
     public void setup() {
         result = Streamable.range(1, times).skip(times / 2);
+        resultIndexed = Streamable.range(1, times).skip(times / 2).collect(Collectors.toList());
+        resultEnumerable = Streamable.range(1, times).filter(_ -> true).skip(times / 2).collect(Collectors.toList());
     }
 
     @Benchmark
     public Object benchmark() {
         return result.blockingLast();
     }
+
+    @Benchmark
+    public Object indexed() {
+        return resultIndexed.blockingLast();
+    }
+
+    @Benchmark
+    public Object enumerable() {
+        return resultEnumerable.blockingLast();
+    }
 }

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

@@ -623,6 +623,21 @@ static <T> Streamable<List<T>> zip(Iterable<? extends Streamable<? extends T>> s
     default T blockingFirst() {
         return StreamableBlocking.blockingFirst(this);
     }
+    /**
+     * Blocks the current thread until this {@code Streamable} produces one item, which is then returned.
+     * @param cancellation the external cancellation resource to pass into the chain
+     * @return the first item of this {@code Streamable}
+     * @throws NoSuchElementException if the this {@code Streamable} is empty
+     * @throws CancellationException if this {@code Streamable} failed with a checked exception
+     * @throws RuntimeException if this {@code Streamable} failed with an unchecked exception
+     * @throws NullPointerException if {@code cancellation} is {@code null}
+     */
+    @CheckReturnValue
+    @NonNull
+    default T blockingFirst(StreamerCancellation cancellation) {
+        Objects.requireNonNull(cancellation, "cancellation is null");
+        return StreamableBlocking.blockingFirst(this, cancellation);
+    }
 
     /**
      * Blocks the current thread until this {@code Streamable} produces all of its items
@@ -638,6 +653,23 @@ default T blockingLast() {
         return StreamableBlocking.blockingLast(this);
     }
 
+    /**
+     * Blocks the current thread until this {@code Streamable} produces all of its items
+     * and the very last is then returned.
+     * @param cancellation the external cancellation resource to pass into the chain
+     * @return the very last item of this {@code Streamable}
+     * @throws NoSuchElementException if the this {@code Streamable} is empty
+     * @throws CancellationException if this {@code Streamable} failed with a checked exception
+     * @throws RuntimeException if this {@code Streamable} failed with an unchecked exception
+     * @throws NullPointerException if {@code cancellation} is {@code null}
+     */
+    @CheckReturnValue
+    @NonNull
+    default T blockingLast(StreamerCancellation cancellation) {
+        Objects.requireNonNull(cancellation, "cancellation is null");
+        return StreamableBlocking.blockingLast(this, cancellation);
+    }
+
     /**
      * Collects all upstream values via the use of a {@link Collector} configuration
      * and emits its resulting value as a single item of the returned {@code Streamable}.

src/main/java/io/reactivex/rxjava4/disposables/DisposableStreamerCancellation.java

@@ -13,6 +13,8 @@
 
 package io.reactivex.rxjava4.disposables;
 
+import io.reactivex.rxjava4.internal.disposables.NeverDisposableStreamerCancellation;
+
 /**
  * Represents the full, disposable cancellation interface for {@code Streamer}
  * operations.
@@ -23,4 +25,13 @@
  */
 public interface DisposableStreamerCancellation extends StreamerCancellation, Disposable {
 
+    /**
+     * Returns a constant instance which does nothing, cannot be disposed and
+     * accepts any incoming Disposable without registering it or handling it in any form,
+     * because this {@code never} instance cannot be disposed to begin with.
+     * @return the shared constant no-op instance
+     */
+    static DisposableStreamerCancellation never() {
+        return NeverDisposableStreamerCancellation.INSTANCE;
+    }
 }

src/main/java/io/reactivex/rxjava4/internal/disposables/NeverDisposableStreamerCancellation.java

@@ -0,0 +1,54 @@
+/*
+ * Copyright (c) 2016-present, RxJava Contributors.
+ *
+ * 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
+ *
+ * http://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 io.reactivex.rxjava4.internal.disposables;
+
+import io.reactivex.rxjava4.annotations.NonNull;
+import io.reactivex.rxjava4.disposables.*;
+
+/// A [DisposableStreamerCancellation] handler that does nothing.
+public enum NeverDisposableStreamerCancellation implements DisposableStreamerCancellation {
+    INSTANCE
+    ;
+
+    @Override
+    public boolean isDisposed() {
+        return false; // always because it has no state
+    }
+
+    @Override
+    public boolean add(@NonNull Disposable d) {
+        return true; // always succeeds to avoid infinite retry loops
+    }
+
+    @Override
+    public boolean remove(@NonNull Disposable d) {
+        return true; // always succeeds
+    }
+
+    @Override
+    public boolean delete(@NonNull Disposable d) {
+        return true; // always succeeds
+    }
+
+    @Override
+    public @NonNull DisposableStreamerCancellation derive() {
+        return this;
+    }
+
+    @Override
+    public void dispose() {
+        // deliberately no-op
+    }
+
+}

src/main/java/io/reactivex/rxjava4/internal/operators/streamable/IsEnumerableStreamable.java

@@ -0,0 +1,26 @@
+/*
+ * Copyright (c) 2016-present, RxJava Contributors.
+ *
+ * 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
+ *
+ * http://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 io.reactivex.rxjava4.internal.operators.streamable;
+
+import io.reactivex.rxjava4.core.*;
+import io.reactivex.rxjava4.operators.EnumerableSource;
+
+/// Marker interface to indicate a [Streamable] source will produce
+/// an [EnumerableSource]-enabled [Streamer] and thus enables
+/// optimizations and operator fusion during assembly time.
+/// @param <T> the element type of the `Streamable`
+/// @since 4.0.0
+public interface IsEnumerableStreamable<T> extends Streamable<T> {
+
+}

src/main/java/io/reactivex/rxjava4/internal/operators/streamable/IsIndexableStreamable.java

@@ -0,0 +1,26 @@
+/*
+ * Copyright (c) 2016-present, RxJava Contributors.
+ *
+ * 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
+ *
+ * http://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 io.reactivex.rxjava4.internal.operators.streamable;
+
+import io.reactivex.rxjava4.core.*;
+import io.reactivex.rxjava4.operators.IndexableSource;
+
+/// Marker interface to indicate a [Streamable] source will produce
+/// an [IndexableSource]-enabled [Streamer] and thus enables
+/// optimizations and operator fusion during assembly time.
+/// @param <T> the element type of the `Streamable`
+/// @since 4.0.0
+public interface IsIndexableStreamable<T> extends Streamable<T> {
+
+}

src/main/java/io/reactivex/rxjava4/internal/operators/streamable/IsSynchronousStreamable.java

@@ -0,0 +1,26 @@
+/*
+ * Copyright (c) 2016-present, RxJava Contributors.
+ *
+ * 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
+ *
+ * http://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 io.reactivex.rxjava4.internal.operators.streamable;
+
+import io.reactivex.rxjava4.core.*;
+import io.reactivex.rxjava4.operators.*;
+
+/// Marker interface to indicate a [Streamable] source will produce
+/// both an [IndexableSource] and an [EnumerableSource] capable
+/// [Streamer].
+/// @param <T> the element type of the `Streamable`
+/// @since 4.0.0
+public interface IsSynchronousStreamable<T> extends IsIndexableStreamable<T>, IsEnumerableStreamable<T> {
+
+}

src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableBlocking.java

@@ -18,7 +18,7 @@
 
 import io.reactivex.rxjava4.annotations.*;
 import io.reactivex.rxjava4.core.Streamable;
-import io.reactivex.rxjava4.disposables.CompositeDisposable;
+import io.reactivex.rxjava4.disposables.*;
 import io.reactivex.rxjava4.exceptions.Exceptions;
 import io.reactivex.rxjava4.internal.util.ExceptionHelper;
 
@@ -36,7 +36,23 @@ public record StreamableBlocking() {
     @CheckReturnValue
     @NonNull
     public static <T> T blockingFirst(Streamable<T> source) {
-        var streamer = source.stream(new CompositeDisposable());
+        return blockingFirst(source, new CompositeDisposable());
+    }
+
+    /**
+     * Consumes the first item and finishes the {@link Streamable},
+     * throwing {@link NoSuchElementException} if the source is empty.
+     * @param <T> the element type
+     * @param source the source {@code Streamable}
+     * @param cancellation the external cancellation manager
+     * @return the first item
+     * @throws RuntimeException if the source signals an unchecked exception
+     * @throws CompletionException if the source signals a checked exception
+     */
+    @CheckReturnValue
+    @NonNull
+    public static <T> T blockingFirst(Streamable<T> source, StreamerCancellation cancellation) {
+        var streamer = source.stream(cancellation);
         Throwable nextException = null;
         Throwable finishException = null;
         T result = null;
@@ -74,7 +90,21 @@ public static <T> T blockingFirst(Streamable<T> source) {
      * @throws CompletionException if the source signals a checked exception
      */
     public static <T> T blockingLast(Streamable<T> source) {
-        var streamer = source.stream(new CompositeDisposable());
+        return blockingLast(source, new CompositeDisposable());
+    }
+
+    /**
+     * Consumes all upstream items and returns the very last or throws
+     * a {@link NoSuchElementException}.
+     * @param <T> the element type
+     * @param source the source sequence
+     * @param cancellation the external cancellation manager
+     * @return the very last value
+     * @throws RuntimeException if the source signals an unchecked exception
+     * @throws CompletionException if the source signals a checked exception
+     */
+    public static <T> T blockingLast(Streamable<T> source, StreamerCancellation cancellation) {
+        var streamer = source.stream(cancellation);
         Throwable nextException = null;
         Throwable finishException = null;
         T result = null;

src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableEmpty.java

@@ -21,7 +21,7 @@
 import io.reactivex.rxjava4.disposables.StreamerCancellation;
 import io.reactivex.rxjava4.operators.*;
 
-public enum StreamableEmpty implements Streamable<Object> {
+public enum StreamableEmpty implements IsSynchronousStreamable<Object> {
 
     INSTANCE;
 

src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableJust.java

@@ -20,7 +20,7 @@
 import io.reactivex.rxjava4.disposables.StreamerCancellation;
 import io.reactivex.rxjava4.operators.*;
 
-public record StreamableJust<T>(@NonNull T item) implements Streamable<T> {
+public record StreamableJust<T>(@NonNull T item) implements IsSynchronousStreamable<T> {
 
     @Override
     public @NonNull Streamer<@NonNull T> stream(@NonNull StreamerCancellation cancellation) {