Skip to content

4.x: Streamable - optimize several operator internals #8238

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
akarnokd merged 2 commits into from
Jul 17, 2026
Merged

4.x: Streamable - optimize several operator internals #8238

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
0477e31
4.x: Streamable - optimize several operator internals
akarnokd Jul 17, 2026
93b84e1
Fix mistakes in concat(Iterable)
akarnokd Jul 17, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
149 changes: 149 additions & 0 deletions src/jmh/java/io/reactivex/rxjava4/streamable/StreamableConcatIterablePerf.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,149 @@
/*
* 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.streamable;

import java.util.List;
import java.util.concurrent.TimeUnit;

import org.openjdk.jmh.annotations.*;

import io.reactivex.rxjava4.core.Streamable;

///
/// The concat(Iterable) seems to be one of the high allocators in the Scrabble benchmark
/// because of the continuation probably?
///
/// i9 275HX, 32GB LPDDR5 6400MT CL52, Windows 25H2, JDK 26.0.1
///
/// # 0. Baseline
/// ```
/// Benchmark (times) Mode Cnt Score Error Units
/// StreamableConcatIterablePerf.benchmark 1 thrpt 5 3474744,064 ┬▒ 92593,235 ops/s
/// StreamableConcatIterablePerf.benchmark 10 thrpt 5 1640843,357 ┬▒ 30500,201 ops/s
/// StreamableConcatIterablePerf.benchmark 100 thrpt 5 236067,804 ┬▒ 1324,984 ops/s
/// StreamableConcatIterablePerf.benchmark 1000 thrpt 5 24348,807 ┬▒ 494,837 ops/s
/// StreamableConcatIterablePerf.benchmark 10000 thrpt 5 2469,689 ┬▒ 30,614 ops/s
/// StreamableConcatIterablePerf.benchmark 100000 thrpt 5 241,203 ┬▒ 5,330 ops/s
/// StreamableConcatIterablePerf.benchmark 1000000 thrpt 5 24,330 ┬▒ 0,886 ops/s
/// ```
///
/// # 1. Reduce allocation in whenComplete
///
/// No practical effect
///
/// ```
/// Benchmark (times) Mode Cnt Score Error Units
/// StreamableConcatIterablePerf.benchmark 1 thrpt 5 3517630,344 ┬▒ 203989,280 ops/s
/// StreamableConcatIterablePerf.benchmark 10 thrpt 5 1633567,521 ┬▒ 40147,286 ops/s
/// StreamableConcatIterablePerf.benchmark 100 thrpt 5 236815,630 ┬▒ 6185,836 ops/s
/// StreamableConcatIterablePerf.benchmark 1000 thrpt 5 23939,185 ┬▒ 319,632 ops/s
/// StreamableConcatIterablePerf.benchmark 10000 thrpt 5 2452,654 ┬▒ 98,012 ops/s
/// StreamableConcatIterablePerf.benchmark 100000 thrpt 5 239,677 ┬▒ 6,460 ops/s
/// StreamableConcatIterablePerf.benchmark 1000000 thrpt 5 24,930 ┬▒ 1,884 ops/s
/// ```
///
/// # 2. avoid calling `whenComplete`
///
/// +35% performance vs 1 for longer sequences
///
/// ```
/// Benchmark (times) Mode Cnt Score Error Units
/// StreamableConcatIterablePerf.benchmark 1 thrpt 5 3785671,506 ┬▒ 124882,806 ops/s
/// StreamableConcatIterablePerf.benchmark 10 thrpt 5 2028842,852 ┬▒ 76338,957 ops/s
/// StreamableConcatIterablePerf.benchmark 100 thrpt 5 300832,578 ┬▒ 9916,135 ops/s
/// StreamableConcatIterablePerf.benchmark 1000 thrpt 5 33103,046 ┬▒ 856,542 ops/s
/// StreamableConcatIterablePerf.benchmark 10000 thrpt 5 3272,894 ┬▒ 129,658 ops/s
/// StreamableConcatIterablePerf.benchmark 100000 thrpt 5 335,497 ┬▒ 4,678 ops/s
/// StreamableConcatIterablePerf.benchmark 1000000 thrpt 5 33,685 ┬▒ 0,565 ops/s
/// ```
///
/// # 3. avoid calling decrementAndGet every time on a synchronous/reentrant usage
///
/// +6% for the 1 case but, -5% performance regression vs optimization 2
/// ```
/// Benchmark (times) Mode Cnt Score Error Units
/// StreamableConcatIterablePerf.benchmark 1 thrpt 5 4021165,636 ┬▒ 93456,776 ops/s
/// StreamableConcatIterablePerf.benchmark 10 thrpt 5 1990327,224 ┬▒ 24421,179 ops/s
/// StreamableConcatIterablePerf.benchmark 100 thrpt 5 301872,212 ┬▒ 2880,776 ops/s
/// StreamableConcatIterablePerf.benchmark 1000 thrpt 5 30909,533 ┬▒ 620,143 ops/s
/// StreamableConcatIterablePerf.benchmark 10000 thrpt 5 3174,424 ┬▒ 37,618 ops/s
/// StreamableConcatIterablePerf.benchmark 100000 thrpt 5 316,643 ┬▒ 4,920 ops/s
/// StreamableConcatIterablePerf.benchmark 1000000 thrpt 5 31,970 ┬▒ 0,521 ops/s
/// ```
///
/// # 4. restore the decrementAndGet use
///
/// +/- 1% vs optimization 3
///
/// ```
/// Benchmark (times) Mode Cnt Score Error Units
/// StreamableConcatIterablePerf.benchmark 1 thrpt 5 4034863,551 ┬▒ 147716,809 ops/s
/// StreamableConcatIterablePerf.benchmark 10 thrpt 5 1951987,827 ┬▒ 29883,362 ops/s
/// StreamableConcatIterablePerf.benchmark 100 thrpt 5 302013,530 ┬▒ 4251,994 ops/s
/// StreamableConcatIterablePerf.benchmark 1000 thrpt 5 33260,723 ┬▒ 755,076 ops/s
/// StreamableConcatIterablePerf.benchmark 10000 thrpt 5 3329,779 ┬▒ 81,860 ops/s
/// StreamableConcatIterablePerf.benchmark 100000 thrpt 5 333,358 ┬▒ 8,474 ops/s
/// StreamableConcatIterablePerf.benchmark 1000000 thrpt 5 32,943 ┬▒ 1,187 ops/s
/// ```
/// # 5. fall through to getting the first item from the next source when it is picked, save a drain call
///
/// +5% for short sequences, +22% for short sequences
///
/// ```
/// Benchmark (times) Mode Cnt Score Error Units
/// StreamableConcatIterablePerf.benchmark 1 thrpt 5 4252027,061 ┬▒ 65983,230 ops/s
/// StreamableConcatIterablePerf.benchmark 10 thrpt 5 2125263,936 ┬▒ 46999,223 ops/s
/// StreamableConcatIterablePerf.benchmark 100 thrpt 5 304414,571 ┬▒ 3778,558 ops/s
/// StreamableConcatIterablePerf.benchmark 1000 thrpt 5 32123,342 ┬▒ 915,151 ops/s
/// StreamableConcatIterablePerf.benchmark 10000 thrpt 5 3328,728 ┬▒ 38,444 ops/s
/// StreamableConcatIterablePerf.benchmark 100000 thrpt 5 331,795 ┬▒ 4,418 ops/s
/// StreamableConcatIterablePerf.benchmark 1000000 thrpt 5 33,268 ┬▒ 0,899 ops/s
/// ```
///
/// # 6. synchronous-biased next()
///
/// +77% short sequences, +762% for long sequences
///
/// ```
/// Benchmark (times) Mode Cnt Score Error Units
/// StreamableConcatIterablePerf.benchmark 1 thrpt 5 6172450,906 ┬▒ 66554,843 ops/s
/// StreamableConcatIterablePerf.benchmark 10 thrpt 5 5367812,331 ┬▒ 115663,523 ops/s
/// StreamableConcatIterablePerf.benchmark 100 thrpt 5 1820204,546 ┬▒ 143049,591 ops/s
/// StreamableConcatIterablePerf.benchmark 1000 thrpt 5 218017,883 ┬▒ 49050,865 ops/s
/// StreamableConcatIterablePerf.benchmark 10000 thrpt 5 17406,789 ┬▒ 1079,512 ops/s
/// StreamableConcatIterablePerf.benchmark 100000 thrpt 5 2103,477 ┬▒ 360,090 ops/s
/// StreamableConcatIterablePerf.benchmark 1000000 thrpt 5 207,888 ┬▒ 22,014 ops/s
/// ```
@BenchmarkMode(Mode.Throughput)
@Warmup(iterations = 5, time = 1, timeUnit = TimeUnit.SECONDS)
@Measurement(iterations = 5, time = 1, timeUnit = TimeUnit.SECONDS)
@OutputTimeUnit(TimeUnit.SECONDS)
@Fork(value = 1)
@State(Scope.Thread)
public class StreamableConcatIterablePerf {
@Param({ "1", "10", "100", "1000", "10000", "100000", "1000000" })
public int times;

Streamable<Integer> result;

@Setup
public void setup() {
result = Streamable.concat(List.of(Streamable.range(1, times), Streamable.range(times + 1, times)));
}

@Benchmark
public Object benchmark() {
return result.blockingLast();
}
}
91 changes: 91 additions & 0 deletions src/jmh/java/io/reactivex/rxjava4/streamable/StreamableLastAsSinglePerf.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,91 @@
/*
* 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.streamable;

import java.util.concurrent.TimeUnit;

import org.openjdk.jmh.annotations.*;

import io.reactivex.rxjava4.core.*;

///
/// The map is one of the most used operator in the ecosystem so it must be fast and
/// it must support operator fusion across itself.
///
/// i9 275HX, 32GB LPDDR5 6400MT CL52, Windows 25H2, JDK 26.0.1
///
/// # 0. Baseline
/// ```
/// Benchmark (times) Mode Cnt Score Error Units
/// StreamableLastAsSinglePerf.benchmark 1 thrpt 5 14165950,142 ┬▒ 277135,092 ops/s
/// StreamableLastAsSinglePerf.benchmark 10 thrpt 5 5437536,431 ┬▒ 89291,499 ops/s
/// StreamableLastAsSinglePerf.benchmark 100 thrpt 5 688500,242 ┬▒ 9755,649 ops/s
/// StreamableLastAsSinglePerf.benchmark 1000 thrpt 5 67446,091 ┬▒ 4208,462 ops/s
/// StreamableLastAsSinglePerf.benchmark 10000 thrpt 5 6880,385 ┬▒ 223,058 ops/s
/// StreamableLastAsSinglePerf.benchmark 100000 thrpt 5 690,980 ┬▒ 29,543 ops/s
/// StreamableLastAsSinglePerf.benchmark 1000000 thrpt 5 66,935 ┬▒ 1,086 ops/s
/// ```
///
/// # 1. avoid whenComplete
///
/// +17% for times 1, +50% for a million
///
/// ```
/// Benchmark (times) Mode Cnt Score Error Units
/// StreamableLastAsSinglePerf.benchmark 1 thrpt 5 16684817,192 ┬▒ 337671,603 ops/s
/// StreamableLastAsSinglePerf.benchmark 10 thrpt 5 7713698,470 ┬▒ 72192,494 ops/s
/// StreamableLastAsSinglePerf.benchmark 100 thrpt 5 992063,105 ┬▒ 31518,477 ops/s
/// StreamableLastAsSinglePerf.benchmark 1000 thrpt 5 105383,988 ┬▒ 4062,065 ops/s
/// StreamableLastAsSinglePerf.benchmark 10000 thrpt 5 10103,558 ┬▒ 494,420 ops/s
/// StreamableLastAsSinglePerf.benchmark 100000 thrpt 5 1022,181 ┬▒ 38,701 ops/s
/// StreamableLastAsSinglePerf.benchmark 1000000 thrpt 5 100,744 ┬▒ 1,390 ops/s
/// ```
///
/// # 2. batch wip accounting
///
/// +6.8% for times 1 vs optimization 1. +23.3% for times million vs #0
/// +25.8% for times 1 vs baseline, +82% for times million vs baseline
///
/// ```
/// Benchmark (times) Mode Cnt Score Error Units
/// StreamableLastAsSinglePerf.benchmark 1 thrpt 5 17823989,674 ┬▒ 243372,183 ops/s
/// StreamableLastAsSinglePerf.benchmark 10 thrpt 5 9319936,731 ┬▒ 157326,848 ops/s
/// StreamableLastAsSinglePerf.benchmark 100 thrpt 5 1308207,680 ┬▒ 14246,778 ops/s
/// StreamableLastAsSinglePerf.benchmark 1000 thrpt 5 138722,075 ┬▒ 29515,013 ops/s
/// StreamableLastAsSinglePerf.benchmark 10000 thrpt 5 11938,692 ┬▒ 354,381 ops/s
/// StreamableLastAsSinglePerf.benchmark 100000 thrpt 5 1408,117 ┬▒ 134,934 ops/s
/// StreamableLastAsSinglePerf.benchmark 1000000 thrpt 5 123,326 ┬▒ 29,358 ops/s
/// ```
@BenchmarkMode(Mode.Throughput)
@Warmup(iterations = 5, time = 1, timeUnit = TimeUnit.SECONDS)
@Measurement(iterations = 5, time = 1, timeUnit = TimeUnit.SECONDS)
@OutputTimeUnit(TimeUnit.SECONDS)
@Fork(value = 1)
@State(Scope.Thread)
public class StreamableLastAsSinglePerf {
@Param({ "1", "10", "100", "1000", "10000", "100000", "1000000" })
public int times;

Single<Integer> result;

@Setup
public void setup() {
result = Streamable.range(1, times).lastOrError();
}

@Benchmark
public Object benchmark() {
return result.blockingGet();
}
}
Loading