Skip to content

Commit 258caeb

Browse files
akarnokdakarnokd
authored
4.x: Streamable Operator optimizations 7/18 (#8239)
* 4.x: Streamable Operator optimizations 7/18 * Style fix
1 parent 8ed1fbc commit 258caeb

19 files changed

Lines changed: 975 additions & 107 deletions

‎src/jmh/java/io/reactivex/rxjava4/streamable/StreamableCollectPerf.java‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -102,6 +102,20 @@
102102
/// StreamableCollectPerf.benchmarkHidden 100000 thrpt 5 1161,570 ┬▒ 20,902 ops/s
103103
/// StreamableCollectPerf.benchmarkHidden 1000000 thrpt 5 116,332 ┬▒ 1,313 ops/s
104104
/// ```
105+
///
106+
/// ## 5 - Synchronous bias in the non-fused area, via Claude Fable
107+
///
108+
/// +198% for times 1 vs 4., +177% for times million vs 4
109+
///
110+
/// ```
111+
/// StreamableCollectPerf.benchmarkHidden 1 thrpt 5 47849604,612 ┬▒ 873982,158 ops/s
112+
/// StreamableCollectPerf.benchmarkHidden 10 thrpt 5 34146288,027 ┬▒ 395418,081 ops/s
113+
/// StreamableCollectPerf.benchmarkHidden 100 thrpt 5 3294354,815 ┬▒ 47802,050 ops/s
114+
/// StreamableCollectPerf.benchmarkHidden 1000 thrpt 5 329341,300 ┬▒ 6665,423 ops/s
115+
/// StreamableCollectPerf.benchmarkHidden 10000 thrpt 5 33513,984 ┬▒ 2196,936 ops/s
116+
/// StreamableCollectPerf.benchmarkHidden 100000 thrpt 5 3529,826 ┬▒ 71,479 ops/s
117+
/// StreamableCollectPerf.benchmarkHidden 1000000 thrpt 5 322,838 ┬▒ 5,698 ops/s
118+
/// ```
105119
@BenchmarkMode(Mode.Throughput)
106120
@Warmup(iterations = 5, time = 1, timeUnit = TimeUnit.SECONDS)
107121
@Measurement(iterations = 5, time = 1, timeUnit = TimeUnit.SECONDS)

‎src/jmh/java/io/reactivex/rxjava4/streamable/StreamableFilterPerf.java‎

Lines changed: 177 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,177 @@
1+
/*
2+
* Copyright (c) 2016-present, RxJava Contributors.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in
5+
* compliance with the License. You may obtain a copy of the License at
6+
*
7+
* http://www.apache.org/licenses/LICENSE-2.0
8+
*
9+
* Unless required by applicable law or agreed to in writing, software distributed under the License is
10+
* distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See
11+
* the License for the specific language governing permissions and limitations under the License.
12+
*/
13+
14+
package io.reactivex.rxjava4.streamable;
15+
16+
import java.util.*;
17+
import java.util.concurrent.TimeUnit;
18+
import java.util.stream.*;
19+
20+
import org.openjdk.jmh.annotations.*;
21+
22+
import io.reactivex.rxjava4.core.Streamable;
23+
24+
///
25+
/// The map is one of the most used operator in the ecosystem so it must be fast and
26+
/// it must support operator fusion across itself.
27+
///
28+
/// i9 275HX, 32GB LPDDR5 6400MT CL52, Windows 25H2, JDK 26.0.1
29+
///
30+
/// # 0. Baseline
31+
/// ```
32+
/// Benchmark (times) Mode Cnt Score Error Units
33+
/// StreamableFilterPerf.basic 1 thrpt 5 26170031,056 ┬▒ 100892,180 ops/s
34+
/// StreamableFilterPerf.basic 10 thrpt 5 4174832,259 ┬▒ 118915,456 ops/s
35+
/// StreamableFilterPerf.basic 100 thrpt 5 485563,469 ┬▒ 6977,640 ops/s
36+
/// StreamableFilterPerf.basic 1000 thrpt 5 52859,526 ┬▒ 392,085 ops/s
37+
/// StreamableFilterPerf.basic 10000 thrpt 5 4757,976 ┬▒ 67,715 ops/s
38+
/// StreamableFilterPerf.basic 100000 thrpt 5 475,093 ┬▒ 6,846 ops/s
39+
/// StreamableFilterPerf.basic 1000000 thrpt 5 47,684 ┬▒ 1,866 ops/s
40+
/// StreamableFilterPerf.enumerated 1 thrpt 5 9351050,757 ┬▒ 267315,421 ops/s
41+
/// StreamableFilterPerf.enumerated 10 thrpt 5 3315310,614 ┬▒ 72388,028 ops/s
42+
/// StreamableFilterPerf.enumerated 100 thrpt 5 377616,877 ┬▒ 8096,437 ops/s
43+
/// StreamableFilterPerf.enumerated 1000 thrpt 5 41714,551 ┬▒ 813,698 ops/s
44+
/// StreamableFilterPerf.enumerated 10000 thrpt 5 4084,944 ┬▒ 64,690 ops/s
45+
/// StreamableFilterPerf.enumerated 100000 thrpt 5 367,438 ┬▒ 10,352 ops/s
46+
/// StreamableFilterPerf.enumerated 1000000 thrpt 5 37,380 ┬▒ 0,771 ops/s
47+
/// StreamableFilterPerf.indexed 1 thrpt 5 8582157,437 ┬▒ 215730,754 ops/s
48+
/// StreamableFilterPerf.indexed 10 thrpt 5 3370403,716 ┬▒ 44180,727 ops/s
49+
/// StreamableFilterPerf.indexed 100 thrpt 5 392725,429 ┬▒ 5825,883 ops/s
50+
/// StreamableFilterPerf.indexed 1000 thrpt 5 42956,064 ┬▒ 774,701 ops/s
51+
/// StreamableFilterPerf.indexed 10000 thrpt 5 4274,859 ┬▒ 48,781 ops/s
52+
/// StreamableFilterPerf.indexed 100000 thrpt 5 427,545 ┬▒ 6,204 ops/s
53+
/// StreamableFilterPerf.indexed 1000000 thrpt 5 38,761 ┬▒ 1,574 ops/s
54+
/// ```
55+
///
56+
/// # 1. Avoid whenComplete and tidy up the internals, better wip management
57+
///
58+
/// 1 times regression -22%, 1 million case +63%
59+
///
60+
/// ```
61+
/// Benchmark (times) Mode Cnt Score Error Units
62+
/// StreamableFilterPerf.basic 1 thrpt 5 20450133,688 ┬▒ 292577,808 ops/s
63+
/// StreamableFilterPerf.basic 10 thrpt 5 7001560,283 ┬▒ 95757,407 ops/s
64+
/// StreamableFilterPerf.basic 100 thrpt 5 801954,196 ┬▒ 25583,620 ops/s
65+
/// StreamableFilterPerf.basic 1000 thrpt 5 81090,359 ┬▒ 1422,785 ops/s
66+
/// StreamableFilterPerf.basic 10000 thrpt 5 7861,791 ┬▒ 174,594 ops/s
67+
/// StreamableFilterPerf.basic 100000 thrpt 5 791,253 ┬▒ 10,498 ops/s
68+
/// StreamableFilterPerf.basic 1000000 thrpt 5 78,027 ┬▒ 1,068 ops/s
69+
/// StreamableFilterPerf.enumerated 1 thrpt 5 11758684,924 ┬▒ 308081,461 ops/s
70+
/// StreamableFilterPerf.enumerated 10 thrpt 5 4399154,841 ┬▒ 143732,147 ops/s
71+
/// StreamableFilterPerf.enumerated 100 thrpt 5 581006,390 ┬▒ 15226,107 ops/s
72+
/// StreamableFilterPerf.enumerated 1000 thrpt 5 52474,101 ┬▒ 900,684 ops/s
73+
/// StreamableFilterPerf.enumerated 10000 thrpt 5 5364,237 ┬▒ 55,065 ops/s
74+
/// StreamableFilterPerf.enumerated 100000 thrpt 5 575,620 ┬▒ 11,915 ops/s
75+
/// StreamableFilterPerf.enumerated 1000000 thrpt 5 53,243 ┬▒ 0,953 ops/s
76+
/// StreamableFilterPerf.indexed 1 thrpt 5 10378233,365 ┬▒ 320717,352 ops/s
77+
/// StreamableFilterPerf.indexed 10 thrpt 5 4450787,764 ┬▒ 16570,500 ops/s
78+
/// StreamableFilterPerf.indexed 100 thrpt 5 589007,872 ┬▒ 17229,304 ops/s
79+
/// StreamableFilterPerf.indexed 1000 thrpt 5 61222,108 ┬▒ 1699,621 ops/s
80+
/// StreamableFilterPerf.indexed 10000 thrpt 5 6181,191 ┬▒ 174,083 ops/s
81+
/// StreamableFilterPerf.indexed 100000 thrpt 5 582,777 ┬▒ 14,846 ops/s
82+
/// StreamableFilterPerf.indexed 1000000 thrpt 5 54,477 ┬▒ 1,366 ops/s
83+
/// ```
84+
///
85+
/// # 2. Synchronous bias via CAS-based state management, via Claude Fable suggestions
86+
///
87+
/// times-1 +70% vs baseline, times million +240% vs baseline
88+
/// ```
89+
/// Benchmark (times) Mode Cnt Score Error Units
90+
/// StreamableFilterPerf.basic 1 thrpt 5 44649205,329 ┬▒ 1318761,330 ops/s
91+
/// StreamableFilterPerf.basic 10 thrpt 5 14573791,114 ┬▒ 307959,449 ops/s
92+
/// StreamableFilterPerf.basic 100 thrpt 5 1704743,707 ┬▒ 27297,095 ops/s
93+
/// StreamableFilterPerf.basic 1000 thrpt 5 169935,412 ┬▒ 1894,944 ops/s
94+
/// StreamableFilterPerf.basic 10000 thrpt 5 16747,005 ┬▒ 149,331 ops/s
95+
/// StreamableFilterPerf.basic 100000 thrpt 5 1630,870 ┬▒ 39,286 ops/s
96+
/// StreamableFilterPerf.basic 1000000 thrpt 5 162,566 ┬▒ 4,036 ops/s
97+
/// StreamableFilterPerf.enumerated 1 thrpt 5 13220834,886 ┬▒ 157990,717 ops/s
98+
/// StreamableFilterPerf.enumerated 10 thrpt 5 6266989,179 ┬▒ 168298,570 ops/s
99+
/// StreamableFilterPerf.enumerated 100 thrpt 5 894912,487 ┬▒ 4979,227 ops/s
100+
/// StreamableFilterPerf.enumerated 1000 thrpt 5 89633,958 ┬▒ 1631,029 ops/s
101+
/// StreamableFilterPerf.enumerated 10000 thrpt 5 8973,714 ┬▒ 197,897 ops/s
102+
/// StreamableFilterPerf.enumerated 100000 thrpt 5 887,542 ┬▒ 12,736 ops/s
103+
/// StreamableFilterPerf.enumerated 1000000 thrpt 5 88,537 ┬▒ 0,994 ops/s
104+
/// StreamableFilterPerf.indexed 1 thrpt 5 11534659,877 ┬▒ 281862,191 ops/s
105+
/// StreamableFilterPerf.indexed 10 thrpt 5 6663720,805 ┬▒ 127958,849 ops/s
106+
/// StreamableFilterPerf.indexed 100 thrpt 5 973299,004 ┬▒ 35818,034 ops/s
107+
/// StreamableFilterPerf.indexed 1000 thrpt 5 98248,858 ┬▒ 1445,809 ops/s
108+
/// StreamableFilterPerf.indexed 10000 thrpt 5 9801,805 ┬▒ 183,524 ops/s
109+
/// StreamableFilterPerf.indexed 100000 thrpt 5 983,497 ┬▒ 23,374 ops/s
110+
/// StreamableFilterPerf.indexed 1000000 thrpt 5 97,605 ┬▒ 1,170 ops/s
111+
/// ```
112+
///
113+
/// # 3. Add EnumerableSource and DeferredEnumerableSource paths.
114+
///
115+
/// +193% times 1 vs baseline, +607% times million vs baseline
116+
/// ```
117+
/// Benchmark (times) Mode Cnt Score Error Units
118+
/// StreamableFilterPerf.basic 1 thrpt 5 44129567,500 ┬▒ 3907413,532 ops/s
119+
/// StreamableFilterPerf.basic 10 thrpt 5 15360293,459 ┬▒ 134453,730 ops/s
120+
/// StreamableFilterPerf.basic 100 thrpt 5 1811371,583 ┬▒ 32392,354 ops/s
121+
/// StreamableFilterPerf.basic 1000 thrpt 5 163322,591 ┬▒ 4944,111 ops/s
122+
/// StreamableFilterPerf.basic 10000 thrpt 5 16124,871 ┬▒ 578,538 ops/s
123+
/// StreamableFilterPerf.basic 100000 thrpt 5 1618,494 ┬▒ 68,714 ops/s
124+
/// StreamableFilterPerf.basic 1000000 thrpt 5 157,161 ┬▒ 1,401 ops/s
125+
/// StreamableFilterPerf.enumerated 1 thrpt 5 27449227,121 ┬▒ 639347,534 ops/s
126+
/// StreamableFilterPerf.enumerated 10 thrpt 5 43288619,923 ┬▒ 787987,036 ops/s
127+
/// StreamableFilterPerf.enumerated 100 thrpt 5 3013969,186 ┬▒ 83719,137 ops/s
128+
/// StreamableFilterPerf.enumerated 1000 thrpt 5 272763,930 ┬▒ 4591,739 ops/s
129+
/// StreamableFilterPerf.enumerated 10000 thrpt 5 27699,074 ┬▒ 591,416 ops/s
130+
/// StreamableFilterPerf.enumerated 100000 thrpt 5 2769,250 ┬▒ 75,561 ops/s
131+
/// StreamableFilterPerf.enumerated 1000000 thrpt 5 264,589 ┬▒ 6,425 ops/s
132+
/// StreamableFilterPerf.indexed 1 thrpt 5 11637720,830 ┬▒ 318248,700 ops/s
133+
/// StreamableFilterPerf.indexed 10 thrpt 5 6651438,800 ┬▒ 179861,878 ops/s
134+
/// StreamableFilterPerf.indexed 100 thrpt 5 978543,147 ┬▒ 34745,778 ops/s
135+
/// StreamableFilterPerf.indexed 1000 thrpt 5 97398,438 ┬▒ 1071,570 ops/s
136+
/// StreamableFilterPerf.indexed 10000 thrpt 5 9768,628 ┬▒ 221,269 ops/s
137+
/// StreamableFilterPerf.indexed 100000 thrpt 5 972,717 ┬▒ 31,004 ops/s
138+
/// StreamableFilterPerf.indexed 1000000 thrpt 5 95,972 ┬▒ 2,552 ops/s
139+
/// ```
140+
@BenchmarkMode(Mode.Throughput)
141+
@Warmup(iterations = 5, time = 1, timeUnit = TimeUnit.SECONDS)
142+
@Measurement(iterations = 5, time = 1, timeUnit = TimeUnit.SECONDS)
143+
@OutputTimeUnit(TimeUnit.SECONDS)
144+
@Fork(value = 1)
145+
@State(Scope.Thread)
146+
public class StreamableFilterPerf {
147+
@Param({ "1", "10", "100", "1000", "10000", "100000", "1000000" })
148+
public int times;
149+
150+
Streamable<Integer> result;
151+
Streamable<Optional<Integer>> indexedMax;
152+
Streamable<Optional<Integer>> enumeratedMax;
153+
154+
@Setup
155+
public void setup() {
156+
result = Streamable.range(1, times).filter(v -> (v & 1) != 0);
157+
indexedMax = result.collect(Collectors.maxBy(Comparator.naturalOrder()));
158+
enumeratedMax = Streamable.fromIterable(() -> IntStream.range(1, 1 + times).iterator())
159+
.filter(v -> (v & 1) != 0)
160+
.collect(Collectors.maxBy(Comparator.naturalOrder()));
161+
}
162+
163+
@Benchmark
164+
public Object basic() {
165+
return result.blockingLast();
166+
}
167+
168+
@Benchmark
169+
public Object indexed() {
170+
return indexedMax.blockingLast();
171+
}
172+
173+
@Benchmark
174+
public Object enumerated() {
175+
return enumeratedMax.blockingLast();
176+
}
177+
}

‎src/jmh/java/io/reactivex/rxjava4/streamable/StreamableLastAsSinglePerf.java‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -67,6 +67,21 @@
6767
/// StreamableLastAsSinglePerf.benchmark 100000 thrpt 5 1408,117 ┬▒ 134,934 ops/s
6868
/// StreamableLastAsSinglePerf.benchmark 1000000 thrpt 5 123,326 ┬▒ 29,358 ops/s
6969
/// ```
70+
///
71+
/// # 3. synchronous biased consumption loop, similar to filter, via Claude Fable
72+
///
73+
/// +55% for times 1 vs baseline, +582% for times million vs baseline
74+
///
75+
/// ```
76+
/// Benchmark (times) Mode Cnt Score Error Units
77+
/// StreamableLastAsSinglePerf.benchmark 1 thrpt 5 22006265,959 ┬▒ 1776391,634 ops/s
78+
/// StreamableLastAsSinglePerf.benchmark 10 thrpt 5 28047951,727 ┬▒ 302931,360 ops/s
79+
/// StreamableLastAsSinglePerf.benchmark 100 thrpt 5 6137837,568 ┬▒ 100371,047 ops/s
80+
/// StreamableLastAsSinglePerf.benchmark 1000 thrpt 5 496011,671 ┬▒ 114738,714 ops/s
81+
/// StreamableLastAsSinglePerf.benchmark 10000 thrpt 5 51837,709 ┬▒ 3228,185 ops/s
82+
/// StreamableLastAsSinglePerf.benchmark 100000 thrpt 5 5094,499 ┬▒ 405,946 ops/s
83+
/// StreamableLastAsSinglePerf.benchmark 1000000 thrpt 5 456,867 ┬▒ 15,351 ops/s
84+
/// ```
7085
@BenchmarkMode(Mode.Throughput)
7186
@Warmup(iterations = 5, time = 1, timeUnit = TimeUnit.SECONDS)
7287
@Measurement(iterations = 5, time = 1, timeUnit = TimeUnit.SECONDS)

‎src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableCollector.java‎

Lines changed: 40 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -13,7 +13,9 @@
1313

1414
package io.reactivex.rxjava4.internal.operators.streamable;
1515

16+
import java.io.Serial;
1617
import java.util.concurrent.*;
18+
import java.util.concurrent.Future.State;
1719
import java.util.concurrent.atomic.AtomicInteger;
1820
import java.util.function.*;
1921
import java.util.stream.Collector;
@@ -41,9 +43,11 @@ public record StreamableCollector<T, A, R>(
4143
}
4244

4345
static final class CollectorStreamable<T, A, R>
46+
extends AtomicInteger
4447
implements Streamer<R>, BiConsumer<Object, Throwable> {
4548

46-
final AtomicInteger wip;
49+
@Serial
50+
private static final long serialVersionUID = -2893736592415047972L;
4751

4852
final Streamer<T> upstream;
4953

@@ -70,7 +74,6 @@ static final class CollectorStreamable<T, A, R>
7074
Function<A, R> finisher,
7175
StreamerCancellation cancellation) {
7276
this.upstream = upstream;
73-
this.wip = new AtomicInteger();
7477
this.storage = storage;
7578
this.accumulator = accumulator;
7679
this.finisher = finisher;
@@ -170,32 +173,36 @@ void deferredEnumerate(boolean hasInitialValue, Throwable error) {
170173
}
171174

172175
void drain() {
173-
if (wip.getAndIncrement() != 0) {
174-
return;
175-
}
176-
177-
int wipMax = 1;
178-
int wipIndex = 0;
179-
do {
176+
for (;;) {
180177
if (done) {
181178
StreamableHelper.whenComplete(upstream.finish(), this);
182-
break;
179+
return;
183180
} else {
184-
StreamableHelper.whenComplete(upstream.next(), this);
185-
}
186-
if (++wipIndex == wipMax) {
187-
var newWip = wip.get();
188-
if (newWip != wipMax) {
189-
wipMax = newWip;
190-
} else {
191-
wipMax = wip.addAndGet(-wipMax);
192-
if (wipMax == 0) {
193-
break;
181+
var upstreamNext = upstream.next().toCompletableFuture();
182+
var state = upstreamNext.state();
183+
if (state == State.RUNNING) {
184+
set(1);
185+
upstreamNext.whenComplete(this);
186+
if (compareAndSet(1, 0)) {
187+
return;
194188
}
195-
wipIndex = 0;
189+
state = upstreamNext.state();
190+
}
191+
192+
if (state == State.SUCCESS) {
193+
if (upstreamNext.getNow((false))) {
194+
accumulator.accept(storage, upstream.current());
195+
} else {
196+
current = finisher.apply(storage);
197+
nextReady.complete(true);
198+
return;
199+
}
200+
} else {
201+
nextReady.completeExceptionally(upstreamNext.exceptionNow());
202+
return;
196203
}
197204
}
198-
} while (true);
205+
}
199206
}
200207

201208
@Override
@@ -207,15 +214,17 @@ public void accept(Object t, Throwable u) {
207214
finishReady.complete(null);
208215
}
209216
} else {
210-
if (u != null) {
211-
nextReady.completeExceptionally(u);
212-
} else
213-
if ((Boolean)t) {
214-
accumulator.accept(storage, upstream.current());
215-
drain();
216-
} else {
217-
current = finisher.apply(storage);
218-
nextReady.complete(true);
217+
if (!compareAndSet(1, 2)) {
218+
if (u != null) {
219+
nextReady.completeExceptionally(u);
220+
} else
221+
if ((Boolean)t) {
222+
accumulator.accept(storage, upstream.current());
223+
drain();
224+
} else {
225+
current = finisher.apply(storage);
226+
nextReady.complete(true);
227+
}
219228
}
220229
}
221230
}

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

Lines changed: 19 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -14,11 +14,12 @@
1414
package io.reactivex.rxjava4.internal.operators.streamable;
1515

1616
import java.util.NoSuchElementException;
17-
import java.util.concurrent.*;
17+
import java.util.concurrent.CompletionStage;
1818

1919
import io.reactivex.rxjava4.annotations.NonNull;
2020
import io.reactivex.rxjava4.core.*;
21-
import io.reactivex.rxjava4.disposables.*;
21+
import io.reactivex.rxjava4.disposables.StreamerCancellation;
22+
import io.reactivex.rxjava4.operators.*;
2223

2324
public enum StreamableEmpty implements Streamable<Object> {
2425

@@ -34,7 +35,7 @@ public static <T> Streamer<T> createEmpty() {
3435
return (Streamer<T>)EmptyStreamer.INSTANCE;
3536
}
3637

37-
enum EmptyStreamer implements Streamer<Object> {
38+
enum EmptyStreamer implements Streamer<Object>, IndexableSource<Object>, EnumerableSource<Object> {
3839

3940
INSTANCE;
4041

@@ -52,5 +53,20 @@ enum EmptyStreamer implements Streamer<Object> {
5253
public @NonNull CompletionStage<Void> finish() {
5354
return FINISHED;
5455
}
56+
57+
@Override
58+
public boolean nextSync() throws Throwable {
59+
return false;
60+
}
61+
62+
@Override
63+
public @NonNull Object elementAt(long index) throws Throwable {
64+
throw new NoSuchElementException("This Streamable/Streamer never has elements");
65+
}
66+
67+
@Override
68+
public long limit() {
69+
return 0;
70+
}
5571
}
5672
}

0 commit comments

Comments
 (0)