Fix concurrency bug in ScheduledObserver #269
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
This is a followup to ReactiveX@1fa6ae3 that fixed one issue (concurrency) and created another (broke Rx contract by allowing concurrent execution of onNext). I have reverted back to the previous implementatio and then attempted to fix the concurrency issue again. I think it ended up being a simple fix … just re-ordering the `enqueue` method to remove the race-condition between the logic protected by the AtomicInteger and adding to the queue. It's not an atomic operation (adding then processing) so we need to just add to the queue and treat it as an async data structure and keep the AtomicInteger portion to only protecting the "process or not process" logic. ```java // this must happen before 'counter' is used to provide synchronization between threads queue.offer(notification); ``` This may still have issues but it's now working in all of my concurrency tests (the ones that broken with the original and then my modified version). The tests are not easy to build unit tests out of as they require running for many seconds and non-deterministically causing a race condition so I have not yet spend the time to try and figure out a deterministic unit test hence them not being committed.
|
RxJava-pull-requests #141 SUCCESS |
benjchristensen
added a commit
that referenced
this pull request
May 11, 2013
Fix concurrency bug in ScheduledObserver
neerajrj
pushed a commit
to neerajrj/Hystrix
that referenced
this pull request
Nov 2, 2013
- pick up concurrency fix to ScheduledObserver ReactiveX/RxJava#269
rickbw
pushed a commit
to rickbw/RxJava
that referenced
this pull request
Jan 9, 2014
Fix concurrency bug in ScheduledObserver
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.
This is a followup to 1fa6ae3 that fixed one issue (concurrency) and created another (broke Rx contract by allowing concurrent execution of onNext).
I have reverted back to the previous implementatio and then attempted to fix the concurrency issue again.
I think it ended up being a simple fix … just re-ordering the
enqueuemethod to remove the race-condition between the logic protected by the AtomicInteger and adding to the queue.It's not an atomic operation (adding then processing) so we need to just add to the queue and treat it as an async data structure and keep the AtomicInteger portion to only protecting the "process or not process" logic.
This may still have issues but it's now working in all of my concurrency tests (the ones that broken with the original and then my modified version). The tests are not easy to build unit tests out of as they require running for many seconds and non-deterministically causing a race condition so I have not yet spend the time to try and figure out a deterministic unit test hence them not being committed.