Fix a race in some tests for JavaRX integration #1801
Merged
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
dkhalanskyjb
force-pushed
the
rx-withExceptionHandler-race
branch
from
f84bf69 to
4ba9802
Compare
An extremely rare race could happen in any of the tests in `LeakedExceptionTest` in the following case: * `withExceptionHandler` runs the block passed to it; * In one of the last iterations of `repeat`, `select` in `combine` happens on both flows at the same time, that is, the block that was passed to `rx[Something]` runs in two threads simultaneously; * One of these two threads (thread A) runs anomalously slow; * The other thread successfully throws an exception; * This exception is propagated to `catch`, so `collect` is finished; * `repeat` is exited, the block passed to `withExceptionHandler` is done executing; * `withExceptionHandler` sets back the usual exception handler, which fails when an exception in JavaRX happens (see https://github.com/ReactiveX/RxJava/wiki/What's-different-in-2.0#error-handling); * Thread A wakes up and throws an exception. This time, it is passed not to `handler`, which is made specifically to deal with this, but to the default handler. As a fix, now a special coroutine context passed to `rx[Something]` ensures that the spawned executions are run in a thread pool that blocks until all the tasks are done.
dkhalanskyjb
force-pushed
the
rx-withExceptionHandler-race
branch
from
4ba9802 to
5a32d76
Compare
elizarov
suggested changes
Feb 13, 2020
| repeat(10000) { | ||
| combine(flow, flow) { _, _ -> Unit } | ||
| .catch {} | ||
| .collect { } |
There was a problem hiding this comment.
Style nit: Continuation lines should be formatted with 4 spaces of indentation (not 8).
Also, it as we are touching this code anyway, it would be create to fix { } -> {} (remove space)
| val dispatcher = pool.asCoroutineDispatcher() | ||
| block(dispatcher) | ||
| pool.shutdown() | ||
| while (!pool.awaitTermination(10, TimeUnit.SECONDS)); |
There was a problem hiding this comment.
when you write while with empty body write { /* deliberately empty */ } (with comment!) as its body.
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
See the commit message for the rationale.
I looked at the other uses of
withExceptionHandler, and it seems that all the other executions are deterministic. I have slight doubts abouttestExceptionAfterCancellation, but, to my understanding,blockingSubscribeensures that every execution is finished.