Skip to content

Commit 38a3ffe

Browse files
Copilotakarnokd
Copilot
and
akarnokd
authored
Fix DispatchStreamProcessorTest.raceToStream flakiness (#8275)
* Initial plan * Fix raceToStream flakiness in DispatchStreamProcessorTest Co-authored-by: akarnokd <1269832+akarnokd@users.noreply.github.com> * Polish awaitCondition to use TimeUnit wall-clock deadline Co-authored-by: akarnokd <1269832+akarnokd@users.noreply.github.com> --------- Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com> Co-authored-by: akarnokd <1269832+akarnokd@users.noreply.github.com>
1 parent 6f02d52 commit 38a3ffe

3 files changed

Lines changed: 107 additions & 56 deletions

File tree

‎src/main/java/io/reactivex/rxjava4/processors/DispatchStreamProcessor.java‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -46,8 +46,14 @@ public final class DispatchStreamProcessor<T> implements StreamProcessor<T, T> {
4646
@Override
4747
public @NonNull Streamer<@NonNull T> stream(@NonNull StreamerCancellation cancellation) {
4848
var result = new DispatchStreamer<T>(this);
49-
cancellation.add(result);
5049
if (add(result)) {
50+
// Register with cancellation after a successful add. If cancellation
51+
// already disposed the streamer (or dispose races with add), remove again
52+
// because the first remove may have observed the pre-add array.
53+
cancellation.add(result);
54+
if (result.isDisposed()) {
55+
remove(result);
56+
}
5157
return result;
5258
}
5359
var t = terminalEvent;

‎src/test/java/io/reactivex/rxjava4/internal/operators/streamable/DispatchStreamProcessorTest.java‎

Lines changed: 23 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -40,7 +40,7 @@ public void normal() throws Throwable {
4040

4141
ts.awaitOnSubscribe(1, TimeUnit.SECONDS);
4242

43-
awaitStreamers(dsp, 1000);
43+
assertHasStreamers(dsp, 1000);
4444

4545
for (int i = 1; i < 6; i++) {
4646
dsp.next(i).toCompletableFuture().join();
@@ -50,7 +50,7 @@ public void normal() throws Throwable {
5050
ts.awaitDone(5, TimeUnit.SECONDS)
5151
.assertResult(1, 2, 3, 4, 5);
5252

53-
assertFalse(dsp.hasStreamers(), "dsp has streamers?");
53+
assertNoStreamers(dsp, 1000);
5454
assertTrue(dsp.hasComplete(), "dsp has completed?");
5555
assertFalse(dsp.hasThrowable(), "dsp has throwable?");
5656
assertNull(dsp.getThrowable(), "dsp has a non-null throwable?");
@@ -69,7 +69,7 @@ public void endsInError() throws Throwable {
6969

7070
ts.awaitOnSubscribe(1, TimeUnit.SECONDS);
7171

72-
awaitStreamers(dsp, 1000);
72+
assertHasStreamers(dsp, 1000);
7373

7474
for (int i = 1; i < 6; i++) {
7575
dsp.next(i).toCompletableFuture().join();
@@ -80,7 +80,7 @@ public void endsInError() throws Throwable {
8080
ts.awaitDone(5, TimeUnit.SECONDS)
8181
.assertFailure(TestException.class, 1, 2, 3, 4, 5);
8282

83-
assertFalse(dsp.hasStreamers(), "dsp has streamers?");
83+
assertNoStreamers(dsp, 1000);
8484
assertFalse(dsp.hasComplete(), "dsp has completed?");
8585
assertTrue(dsp.hasThrowable(), "dsp has no throwable?");
8686
assertSame(te, dsp.getThrowable(), "dsp has the wrong throwable?");
@@ -100,9 +100,7 @@ public void normalDebug() throws Throwable {
100100

101101
ts.awaitOnSubscribe(1, TimeUnit.SECONDS);
102102

103-
awaitStreamers(dsp, 1000);
104-
105-
assertTrue(dsp.hasStreamers(), "dsp has no streamers?");
103+
assertHasStreamers(dsp, 1000);
106104

107105
for (int i = 1; i < 6; i++) {
108106
dsp.next(i).toCompletableFuture().join();
@@ -112,7 +110,7 @@ public void normalDebug() throws Throwable {
112110
ts.awaitDone(5, TimeUnit.SECONDS)
113111
.assertResult(1, 2, 3, 4, 5);
114112

115-
assertFalse(dsp.hasStreamers(), "dsp has streamers?");
113+
assertNoStreamers(dsp, 1000);
116114
assertTrue(dsp.hasComplete(), "dsp has completed?");
117115
assertFalse(dsp.hasThrowable(), "dsp has throwable?");
118116
assertNull(dsp.getThrowable(), "dsp has a non-null throwable?");
@@ -178,16 +176,14 @@ public void normalTake3AltDebug() throws Throwable {
178176

179177
ts.awaitOnSubscribe(1, TimeUnit.SECONDS);
180178

181-
awaitStreamers(dsp, 1000);
179+
assertHasStreamers(dsp, 1000);
182180

183181
for (int i = 1; i < 4; i++) {
184182
IO.println(i + " -> next");
185183
dsp.next(i).toCompletableFuture().join();
186184
}
187185

188-
awaitNoStreamers(dsp, 1000);
189-
190-
assertFalse(dsp.hasStreamers(), "dsp has streamers?");
186+
assertNoStreamers(dsp, 1000);
191187

192188
for (int i = 4; i < 6; i++) {
193189
IO.println(i + " -> next");
@@ -201,7 +197,7 @@ public void normalTake3AltDebug() throws Throwable {
201197
ts.awaitDone(5, TimeUnit.SECONDS)
202198
.assertResult(1, 2, 3);
203199

204-
assertFalse(dsp.hasStreamers(), "dsp has streamers?");
200+
assertNoStreamers(dsp, 1000);
205201
assertTrue(dsp.hasComplete(), "dsp has completed?");
206202
assertFalse(dsp.hasThrowable(), "dsp has throwable?");
207203
assertNull(dsp.getThrowable(), "dsp has a non-null throwable?");
@@ -222,7 +218,7 @@ public void normalTake3Debug() throws Throwable {
222218

223219
ts.awaitOnSubscribe(1, TimeUnit.SECONDS);
224220

225-
awaitStreamers(dsp, 1000);
221+
assertHasStreamers(dsp, 1000);
226222

227223
for (int i = 1; i < 6; i++) {
228224
dsp.next(i).toCompletableFuture().join();
@@ -235,7 +231,7 @@ public void normalTake3Debug() throws Throwable {
235231
ts.awaitDone(5, TimeUnit.SECONDS)
236232
.assertResult(1, 2, 3);
237233

238-
assertFalse(dsp.hasStreamers(), "dsp has streamers?");
234+
assertNoStreamers(dsp, 1000);
239235
assertTrue(dsp.hasComplete(), "dsp has completed?");
240236
assertFalse(dsp.hasThrowable(), "dsp has throwable?");
241237
assertNull(dsp.getThrowable(), "dsp has a non-null throwable?");
@@ -257,7 +253,7 @@ public void normalMulti() throws Throwable {
257253
ts.awaitOnSubscribe(1, TimeUnit.SECONDS);
258254
ts2.awaitOnSubscribe(1, TimeUnit.SECONDS);
259255

260-
awaitStreamers(dsp, 1000, 2);
256+
assertHasStreamers(dsp, 1000, 2);
261257

262258
for (int i = 1; i < 6; i++) {
263259
dsp.next(i).toCompletableFuture().join();
@@ -270,7 +266,7 @@ public void normalMulti() throws Throwable {
270266
ts2.awaitDone(5, TimeUnit.SECONDS)
271267
.assertResult(1, 2, 3, 4, 5);
272268

273-
assertFalse(dsp.hasStreamers(), "dsp has streamers?");
269+
assertNoStreamers(dsp, 1000);
274270
assertTrue(dsp.hasComplete(), "dsp has completed?");
275271
assertFalse(dsp.hasThrowable(), "dsp has throwable?");
276272
assertNull(dsp.getThrowable(), "dsp has a non-null throwable?");
@@ -291,7 +287,7 @@ public void normalMultiOtherCancels() throws Throwable {
291287
ts.awaitOnSubscribe(1, TimeUnit.SECONDS);
292288
ts2.awaitOnSubscribe(1, TimeUnit.SECONDS);
293289

294-
awaitStreamers(dsp, 1000, 2);
290+
assertHasStreamers(dsp, 1000, 2);
295291

296292
ts2.cancel();
297293

@@ -305,7 +301,7 @@ public void normalMultiOtherCancels() throws Throwable {
305301

306302
ts2.assertEmpty();
307303

308-
assertFalse(dsp.hasStreamers(), "dsp has streamers?");
304+
assertNoStreamers(dsp, 1000);
309305
assertTrue(dsp.hasComplete(), "dsp has completed?");
310306
assertFalse(dsp.hasThrowable(), "dsp has throwable?");
311307
assertNull(dsp.getThrowable(), "dsp has a non-null throwable?");
@@ -326,7 +322,7 @@ public void normalMultiFirstCancels() throws Throwable {
326322
ts.awaitOnSubscribe(1, TimeUnit.SECONDS);
327323
ts2.awaitOnSubscribe(1, TimeUnit.SECONDS);
328324

329-
awaitStreamers(dsp, 1000, 2);
325+
assertHasStreamers(dsp, 1000, 2);
330326

331327
ts.cancel();
332328

@@ -340,7 +336,7 @@ public void normalMultiFirstCancels() throws Throwable {
340336

341337
ts.assertEmpty();
342338

343-
assertFalse(dsp.hasStreamers(), "dsp has streamers?");
339+
assertNoStreamers(dsp, 1000);
344340
assertTrue(dsp.hasComplete(), "dsp has completed?");
345341
assertFalse(dsp.hasThrowable(), "dsp has throwable?");
346342
assertNull(dsp.getThrowable(), "dsp has a non-null throwable?");
@@ -367,14 +363,14 @@ public void raceToStream() throws Throwable {
367363
ts.awaitOnSubscribe(1, TimeUnit.SECONDS);
368364
ts2.awaitOnSubscribe(1, TimeUnit.SECONDS);
369365

370-
awaitStreamers(dsp, 1000);
366+
// Both subscribers must be fully attached before cancellation,
367+
// otherwise a late stream() can re-add a streamer after the wait.
368+
assertHasStreamers(dsp, 1000, 2);
371369

372370
ts.cancel();
373371
ts2.cancel();
374372

375-
awaitNoStreamers(dsp, 1000);
376-
377-
assertFalse(dsp.hasStreamers(), "dsp has streamers?");
373+
assertNoStreamers(dsp, 1000);
378374
assertFalse(dsp.hasComplete(), "dsp has completed?");
379375
assertFalse(dsp.hasThrowable(), "dsp has throwable?");
380376
assertNull(dsp.getThrowable(), "dsp has a non-null throwable?");
@@ -394,13 +390,11 @@ public void comeAndGo() throws Throwable {
394390

395391
ts.awaitOnSubscribe(1, TimeUnit.SECONDS);
396392

397-
awaitStreamers(dsp, 1000);
393+
assertHasStreamers(dsp, 1000);
398394

399395
ts.cancel();
400396

401-
awaitNoStreamers(dsp, 1000);
402-
403-
assertFalse(dsp.hasStreamers(), "dsp has streamers?");
397+
assertNoStreamers(dsp, 1000);
404398
assertFalse(dsp.hasComplete(), "dsp has completed?");
405399
assertFalse(dsp.hasThrowable(), "dsp has throwable?");
406400
assertNull(dsp.getThrowable(), "dsp has a non-null throwable?");

‎src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableBaseTest.java‎

Lines changed: 77 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,7 @@
1515

1616
import java.lang.ref.Cleaner;
1717
import java.util.*;
18-
import java.util.concurrent.TimeoutException;
18+
import java.util.concurrent.*;
1919
import java.util.function.*;
2020

2121
import org.junit.jupiter.api.*;
@@ -119,13 +119,7 @@ public static <T> StreamableInterceptConfig<T> debugIntercept() {
119119
public static void awaitStreamers(StreamProcessor<?, ?> sp, long timeoutMillis)
120120
throws InterruptedException, TimeoutException
121121
{
122-
long timeout = timeoutMillis * 1_000_000L;
123-
while (!sp.hasStreamers()) {
124-
Thread.sleep(0, 1000);
125-
if (--timeout <= 0L) {
126-
throw new TimeoutException("hasStreamers still false");
127-
}
128-
}
122+
awaitCondition(true, sp::hasStreamers, timeoutMillis, "hasStreamers still false");
129123
}
130124

131125
/**
@@ -140,13 +134,8 @@ public static void awaitStreamers(StreamProcessor<?, ?> sp, long timeoutMillis)
140134
public static void awaitStreamers(StreamProcessor<?, ?> sp, long timeoutMillis, int atLeast)
141135
throws InterruptedException, TimeoutException
142136
{
143-
long timeout = timeoutMillis * 1_000_000L;
144-
while (sp.streamerCount() < atLeast) {
145-
Thread.sleep(0, 1000);
146-
if (--timeout <= 0L) {
147-
throw new TimeoutException("hasStreamers still false");
148-
}
149-
}
137+
awaitCondition(true, () -> sp.streamerCount() >= atLeast, timeoutMillis,
138+
"streamerCount still below " + atLeast + " (was " + sp.streamerCount() + ")");
150139
}
151140

152141
/**
@@ -160,12 +149,55 @@ public static void awaitStreamers(StreamProcessor<?, ?> sp, long timeoutMillis,
160149
public static void awaitNoStreamers(StreamProcessor<?, ?> sp, long timeoutMillis)
161150
throws InterruptedException, TimeoutException
162151
{
163-
long timeout = timeoutMillis * 1_000_000L;
164-
while (sp.hasStreamers()) {
165-
Thread.sleep(0, 1000);
166-
if (--timeout <= 0L) {
167-
throw new TimeoutException("hasStreamers still false");
168-
}
152+
awaitCondition(false, sp::hasStreamers, timeoutMillis, "hasStreamers still true");
153+
}
154+
155+
/**
156+
* Awaits until the processor has no streamers and asserts that state.
157+
* @param sp the processor
158+
* @param timeoutMillis how long to wait for the streamer(s) to leave
159+
* @throws InterruptedException if the sleep is interrupted
160+
* @throws TimeoutException if the wait times out
161+
*/
162+
public static void assertNoStreamers(StreamProcessor<?, ?> sp, long timeoutMillis)
163+
throws InterruptedException, TimeoutException
164+
{
165+
awaitNoStreamers(sp, timeoutMillis);
166+
if (sp.hasStreamers()) {
167+
throw new AssertionError("Processor still has streamers: " + sp.streamerCount());
168+
}
169+
}
170+
171+
/**
172+
* Awaits until the processor has at least one streamer and asserts that state.
173+
* @param sp the processor
174+
* @param timeoutMillis how long to wait for the streamer(s) to arrive
175+
* @throws InterruptedException if the sleep is interrupted
176+
* @throws TimeoutException if the wait times out
177+
*/
178+
public static void assertHasStreamers(StreamProcessor<?, ?> sp, long timeoutMillis)
179+
throws InterruptedException, TimeoutException
180+
{
181+
awaitStreamers(sp, timeoutMillis);
182+
if (!sp.hasStreamers()) {
183+
throw new AssertionError("Processor has no streamers");
184+
}
185+
}
186+
187+
/**
188+
* Awaits until the processor has at least {@code atLeast} streamers and asserts that state.
189+
* @param sp the processor
190+
* @param timeoutMillis how long to wait for the streamer(s) to arrive
191+
* @param atLeast the minimum number of streamers expected
192+
* @throws InterruptedException if the sleep is interrupted
193+
* @throws TimeoutException if the wait times out
194+
*/
195+
public static void assertHasStreamers(StreamProcessor<?, ?> sp, long timeoutMillis, int atLeast)
196+
throws InterruptedException, TimeoutException
197+
{
198+
awaitStreamers(sp, timeoutMillis, atLeast);
199+
if (sp.streamerCount() < atLeast) {
200+
throw new AssertionError("Processor streamerCount below " + atLeast + ": " + sp.streamerCount());
169201
}
170202
}
171203

@@ -181,12 +213,31 @@ public static void awaitNoStreamers(StreamProcessor<?, ?> sp, long timeoutMillis
181213
*/
182214
public static void awaitCondition(boolean value, @NonNull BooleanSupplier condition, long timeoutMillis)
183215
throws InterruptedException, TimeoutException {
184-
long timeout = timeoutMillis * 1_000_000L;
185-
while (condition.getAsBoolean() != value) {
186-
Thread.sleep(0, 1000);
187-
if (--timeout <= 0L) {
188-
throw new TimeoutException("condition still " + (!value));
216+
awaitCondition(value, condition, timeoutMillis, "condition still " + (!value));
217+
}
218+
219+
/**
220+
* Awaits a given {@link BooleanSupplier} to return the expected {@code value}
221+
* within the given wall-clock time period.
222+
* @param value the expected value within the timeout period
223+
* @param condition the condition to repeatedly call to assess the state
224+
* @param timeoutMillis how long to wait for the condition to become as expected
225+
* @param timeoutMessage the message used when the wait times out
226+
* @throws InterruptedException if the sleep is interrupted
227+
* @throws TimeoutException if the wait times out
228+
*/
229+
public static void awaitCondition(boolean value, @NonNull BooleanSupplier condition, long timeoutMillis,
230+
@NonNull String timeoutMessage)
231+
throws InterruptedException, TimeoutException {
232+
long end = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(timeoutMillis);
233+
for (;;) {
234+
if (condition.getAsBoolean() == value) {
235+
return;
236+
}
237+
if (System.nanoTime() >= end) {
238+
throw new TimeoutException(timeoutMessage);
189239
}
240+
Thread.sleep(0, 1000);
190241
}
191242
}
192243
}

0 commit comments

Comments
 (0)